diff --git a/apps/desktop/src/main/__tests__/executor-selection.test.ts b/apps/desktop/src/main/__tests__/executor-selection.test.ts index 17efe588738..1616973d24e 100644 --- a/apps/desktop/src/main/__tests__/executor-selection.test.ts +++ b/apps/desktop/src/main/__tests__/executor-selection.test.ts @@ -28,6 +28,54 @@ import { useExecutorSelection, newTaskConfiguration, ConversationServicesProvide const entry: ExecutorCatalogEntry = { id: 'external', displayName: 'External', readiness: 'ready', models: [{ id: 'selected', name: 'Selected' }], supportsAttachments: false, supportsModelChange: true }; +test('explicit restore confirms the saved model while preserving the current task', async () => { + const { document, window } = parseHTML('
'); + const values = { document, window, HTMLElement: window.HTMLElement, Node: window.Node, IS_REACT_ACT_ENVIRONMENT: true }; + const originals = new Map(Object.keys(values).map(key => [key, Object.getOwnPropertyDescriptor(globalThis, key)])); + for (const [key, value] of Object.entries(values)) Object.defineProperty(globalThis, key, { configurable: true, writable: true, value }); + const root = createRoot(document.getElementById('root')!); + let latest!: ReturnType; + let ready = false; + let finishRestore!: () => void; + const confirmation = new Promise((resolve) => { finishRestore = resolve; }); + const writes: Array<{ sessionId: string; model: string | undefined }> = []; + const services = { + subscribeChanges: () => () => {}, + newTasks: { subscribeChanges: () => () => {}, getExecutors: async () => [entry] }, + sessions: { + getExecutorState: async () => [{ ...entry, readiness: ready ? 'ready' : 'restorable', currentModel: 'selected' }], + setExecutorModelConfiguration: async (sessionId: string, config: { model?: string }) => { + writes.push({ sessionId, model: config.model }); + await confirmation; + ready = true; + return { ok: true, session: { executorConfig: config } }; + }, + }, + } as unknown as ConversationServices; + function Probe() { + latest = useExecutorSelection({ + key: 'saved', cwd: '/fixture', + target: { hostId: 'host', profileId: 'profile', projectId: null }, + session: { id: 'saved', executorId: 'external', executorConfig: { model: 'selected' } } as SessionSummary, + }); + return null; + } + try { + await act(async () => root.render(createElement(ConversationServicesProvider, { services, children: createElement(Probe) }))); + assert.equal(latest.entry?.readiness, 'restorable'); + let restoring!: Promise; + await act(async () => { restoring = latest.restore(); }); + assert.equal(latest.entry?.readiness, 'restoring'); + await act(async () => { finishRestore(); await restoring; }); + assert.deepEqual(writes, [{ sessionId: 'saved', model: 'selected' }]); + assert.equal(latest.entry?.readiness, 'ready'); + assert.equal(latest.selection?.executorId, 'external'); + } finally { + await act(async () => root.unmount()); + for (const [key, descriptor] of originals) { if (descriptor) Object.defineProperty(globalThis, key, descriptor); else Reflect.deleteProperty(globalThis, key); } + } +}); + test('a catalog change during failed discovery retries after the in-flight result settles', async () => { const { document, window } = parseHTML('
'); const values = { document, window, HTMLElement: window.HTMLElement, Node: window.Node, IS_REACT_ACT_ENVIRONMENT: true }; @@ -147,6 +195,8 @@ test('late draft discovery cannot replace the current target; existing tasks ins assert.equal(latest.changing, true); await act(async () => { await assert.rejects(latest.select({ executorId: 'external', configuration: { model: 'other' } }), /pending/); + await assert.rejects(latest.restore(), /pending/); + assert.equal(latest.entry?.readiness, 'ready', 'a rejected restore is not shown as active'); serverModel = 'fast'; confirm({ ok: true, session: { executorConfig: { model: 'fast' } } }); await pending; diff --git a/apps/desktop/src/renderer/features/conversation/controller/use-executor-selection.ts b/apps/desktop/src/renderer/features/conversation/controller/use-executor-selection.ts index 52d71225504..587594f3d88 100644 --- a/apps/desktop/src/renderer/features/conversation/controller/use-executor-selection.ts +++ b/apps/desktop/src/renderer/features/conversation/controller/use-executor-selection.ts @@ -185,11 +185,29 @@ export function useExecutorSelection(input: { } }; const entry = catalog.find((candidate) => candidate.id === selection?.executorId); + const restore = async () => { + if (!sessionId || !executorId) throw new Error('Executor Session is unavailable'); + if (inFlight.current === key) throw new Error('Executor configuration is pending'); + const model = input.session?.executorConfig?.model ?? inspected?.currentModel; + if (!model) throw new Error('Executor model is unavailable'); + setSnapshot((previous) => + previous?.key === key + ? { + ...previous, + catalog: previous.catalog.map((entry) => + entry.id === executorId ? { ...entry, readiness: 'restoring' } : entry, + ), + } + : previous, + ); + await select({ executorId, configuration: { model } }); + }; return { selection, catalog, entry, select, + restore, refresh, changing: changingKey === key, loading: snapshot?.key !== key || snapshot.loading, diff --git a/apps/desktop/src/renderer/features/conversation/model/executor-composer.ts b/apps/desktop/src/renderer/features/conversation/model/executor-composer.ts index 489f17ed024..9be292d6207 100644 --- a/apps/desktop/src/renderer/features/conversation/model/executor-composer.ts +++ b/apps/desktop/src/renderer/features/conversation/model/executor-composer.ts @@ -43,6 +43,7 @@ export function executorComposerProps( loading: executor.loading, error: executor.error, onSelect: (selection) => executor.select(selection), + onRestore: () => executor.restore(), onRetry: () => { void executor.refresh(); }, diff --git a/docs/antigravity-acp-plugin-rebuild.md b/docs/antigravity-acp-plugin-rebuild.md index 1c01a0db1af..edc33aa862e 100644 --- a/docs/antigravity-acp-plugin-rebuild.md +++ b/docs/antigravity-acp-plugin-rebuild.md @@ -69,8 +69,9 @@ policy, initial model configuration, and Antigravity question/failure recognitio - cancellation drain with the actual provider stop reason retained in the durable runtime ledger; - bounded, cached model discovery and Agent-confirmed idle model changes; - generic initial ACP configuration validation/application; -- durable history-only detection so a Host/Plugin restart cannot silently fork an existing external - conversation into a new ACP Session. +- Plugin-private external Session identity and write-ahead prompt state; a clean + completed task can resume the same ACP Session after a Host/Plugin restart, + while uncertain history stays readable without creating a replacement Session. The Antigravity adapter owns: @@ -83,6 +84,27 @@ The Host owns only executor visibility and binding, Maka Session/run identity, c persistence, hosted form admission, and Plugin retirement. No ACP process, credential, or external Session identifier crosses the Plugin boundary. +## Restoring an existing task + +After a restart, open the Antigravity task and choose **Restore** in the model +menu or the notice beside the composer. Maka starts the Agent only when you +choose to restore or continue. The Agent must support ACP `session/resume`; a +successful restore uses the original external Session and keeps the saved task +and model. No earlier prompt is sent again. + +If restoration fails, the task history remains readable. Check the installed +Agent and helper, Google sign-in, and network access in **Settings → External +Agents**, then retry Restore. Maka will not create a new external Session for +that task. An executable or helper change also requires a new task because the +saved Session is bound to the previous installation. + +**History gap** means a prompt may have reached the Agent before Maka finished +saving its terminal event. Maka inspects ACP load replay separately and does +not append possibly duplicate output to the saved history or resend that +prompt. Start a new task if you need to continue working. Tasks created before +this restoration version have no saved external Session ID and remain +history-only. + ## Deliberately not ported The rebuild does not carry forward the PR #5224 provider catalog protocol, external-Agent Session diff --git a/docs/archive/antigravity-acp-pr2-acceptance.md b/docs/archive/antigravity-acp-pr2-acceptance.md index 8e6680a5290..acfd6bdc9e6 100644 --- a/docs/archive/antigravity-acp-pr2-acceptance.md +++ b/docs/archive/antigravity-acp-pr2-acceptance.md @@ -287,7 +287,7 @@ The earlier completed-task capture remains [historical execution evidence](https://github.com/user-attachments/assets/4fc2827b-feed-4639-9a6a-d50caf203055), not a claim about the current picker or a repeated coding run. -Public CI, independent human approval and merge remain repository gates. This document does not -mark issue #5103's PR 2 checkbox complete. Unknown future label shapes remain raw models until +Public CI and independent human approval passed before PR 2 merged as #5224 on 2026-09-23. +Issue #5103's PR 2 checklist is complete. Unknown future label shapes remain raw models until verified. Cross-process restoration belongs to PR 3; modes and expanded catalog lifecycle belong -to PR 4. Neither is claimed by this PR. +to PR 4. Neither is claimed by PR 2. diff --git a/docs/archive/antigravity-acp-pr3-acceptance.md b/docs/archive/antigravity-acp-pr3-acceptance.md new file mode 100644 index 00000000000..20e28fb1471 --- /dev/null +++ b/docs/archive/antigravity-acp-pr3-acceptance.md @@ -0,0 +1,148 @@ + + +# Antigravity ACP PR 3 acceptance + +PR 3 is tracked by [issue #5103](https://github.com/apache/maka/issues/5103). +The results below are from a new probe, separate from the successful PR 2 +execution and earlier cross-process feasibility checks in +[the PR 2 record](antigravity-acp-pr2-acceptance.md). + +## Real Agent prerequisite probe, 2026-09-24 + +- Official macOS arm64 `agy_acp_server.par` **1.1.1** with its matching + `localharness_external`, ACP SDK **1.4.0**. +- The probe used a temporary toy project with a randomly generated synthetic + token. An isolated temporary home copied only the existing OAuth token and + ACP settings into mode-restricted files, but stalled before `session/new` + completed. The method results below used the existing authenticated Agent + home with the toy project. No credentials, Session IDs, token values, or + project contents were recorded. +- `initialize` returned protocol version 1, `loadSession: true`, and + `sessionCapabilities.resume`. `session/new` returned an external Session ID + and confirmed `gemini-3.7-flash-high`. +- A fresh process accepted `session/resume` with the same ID and project + directory and returned the same model. A second independent process also + accepted `session/resume` for that ID. `session/load` accepted it and sent + two `user_message_chunk` notifications and an + `available_commands_update`. No successful Agent output existed to replay + in this run. +- The first prompt and post-resume prompt each ended with `end_turn` but + emitted an Agent execution error: the model request returned HTTP 403, + reporting that the account was not eligible for Gemini Code Assist for + individuals in the current location. The temporary profile with and without + explicit HTTP(S) proxy variables stalled before Session creation. Direct access with + the machine's SOCKS proxy configuration failed because the bundled Python + lacked `python-socks`. Explicit local HTTP(S) proxy variables allowed + Session creation but produced the same model 403. +- All probe-owned Agent process groups and temporary profiles were removed. + +This initial probe **did not pass the PR 3 prerequisite gate**. It confirms +cross-process method acceptance, but cannot establish continued model context, +successful replay shape, duplicate or reordered replay behavior, or a crash +window where the Agent advanced beyond Maka's durable history. The 2026-09-12 +PR 2 feasibility probe observed one successful output chunk and tool replay; +it did not send a post-resume prompt or establish reconciliation semantics. +The later route-change probe below supplied the missing successful turns. + +## Post-login recheck, 2026-09-24 + +After the user completed login in a locally built Maka, the official 1.1.1 +Agent was probed again with the authenticated home and a new temporary toy +project. It negotiated the same resume/load capabilities, created a Session, +and confirmed `gemini-3.7-flash-high`. After a synthetic-token prompt, a fresh +process resumed the same Session ID and accepted a follow-up asking for that +token. Both prompts emitted the same location-eligibility HTTP 403 Agent error, +so the follow-up did not recall the token. A third process loaded the Session; +it replayed two user-message chunks and an available-commands update, with no +successful assistant or tool output to reconcile. The configured local HTTP(S) +proxy's observed egress region was Singapore. All probe processes and the toy +project were removed. Successful login therefore has not cleared the real-model +prerequisite gate. + +## Route-change verification, 2026-09-24 + +After the user changed the proxy exit, a new probe used the same official +macOS arm64 1.1.1 server and matching helper, the user's authenticated Agent +home, and a temporary toy project. Synthetic random values were used only to +test context; no token values, credentials, external Session IDs, or private +project content are recorded here. Intermittent HTTP 403 responses still +occurred between successful requests, so the probe retried affected tool turns. + +- `initialize` again negotiated protocol 1, `loadSession: true`, and + `sessionCapabilities.resume`. `session/new` returned a Session ID and + `gemini-3.7-flash-high`. +- A successful prompt gave the Agent one synthetic token. A new process used + `session/resume` with that same ID and project path, returned the same model, + and a follow-up answer recalled the exact token. The Session was not + replaced and the prompt was not resent. +- A successful tool turn read a second synthetic token from a toy file. The + file was removed, then a new process used `session/load` and replayed user, + Agent, and tool updates. A follow-up recalled the file-only token after the + file was gone, proving that tool context survived the process change. +- Two independent `session/load` calls produced the same update counts and + replay tool IDs for the completed history. A failed tool turn showed that + live tool IDs can differ from replayed IDs. Repeated identical user prompts + appeared as distinct replayed user chunks. Thus neither prompt text nor + live tool ID alone is a safe general deduplication key. +- The probe terminated all child process groups and removed its toy project. + +These observations support `session/resume` for a fully committed Maka turn. +For an interrupted turn, Maka captures `session/load` replay separately and +reports a history gap. It does not append unaligned replay to the canonical +conversation or submit another prompt. This is a conservative reconciliation +decision because the official replay did not provide stable canonical event +identities across every observed turn. The user can read the saved history and +start a new task; the external Session ID is never silently replaced. + +## Built Plugin smoke, 2026-09-24 + +The built production ACP Runtime and Antigravity adapter bundles were loaded +with the official 1.1.1 executable/helper and the authenticated Agent home. +A temporary toy task completed a synthetic-token prompt. The Runtime +acknowledged that turn, disposed its process, and a fresh `AcpExecutor` +instance reported `restorable` from the saved Plugin-private record. The +explicit restore changed readiness to `ready`; a follow-up prompt completed +and recalled the synthetic token. The test removed the toy project and did +not print the token or external Session ID. This exercises the shipped +Plugin code path in addition to the direct protocol probe. It does not stand +in for a full Desktop UI restart test. + +## Implementation checks + +On the branch tested at `bd8661f3a` and rebased without conflicts to +`b62ca805e` before opening the draft PR: + +- `npm run build`, `npm run typecheck`, `npm run lint`, and + `npm run format:check` passed. +- Renderer architecture, locale hygiene, and the protocol compatibility epoch + guard passed; the epoch advanced from 183 to 184 for the new readiness values. +- Every workspace test suite passed in a final serial run using + `node scripts/run-workspace-tests-parallel.mjs --concurrency=1`. Controlled + ACP tests include a forcibly killed, durably acknowledged child process + restored under the same external Session in a fresh process; uncertain load + replay remains outside the canonical turn. +- An earlier serial run had one intermittent Runtime Host Goal handoff timing + assertion. The full Host workspace passed independently, that exact test + passed in isolation, and the final serial run passed. The failing test does + not touch ACP restoration. + +A full Desktop UI restart with a signed-in real Agent was not executed. The +production Plugin bundle smoke, Desktop picker/controller tests, and the real +Agent protocol probes cover the corresponding layers separately. diff --git a/packages/acp-executor-plugin/src/__tests__/acp-executor-plugin.test.ts b/packages/acp-executor-plugin/src/__tests__/acp-executor-plugin.test.ts index d78c76c1800..7b50b0f4d95 100644 --- a/packages/acp-executor-plugin/src/__tests__/acp-executor-plugin.test.ts +++ b/packages/acp-executor-plugin/src/__tests__/acp-executor-plugin.test.ts @@ -29,6 +29,7 @@ import { type AcpAgentAdapter, type AcpConnectionFactory, type AcpConversationStateStore, + type AcpContinuityRecord, } from '../index.js'; const adapter: AcpAgentAdapter<{ executable: string; model?: string }> = { @@ -42,6 +43,451 @@ const adapter: AcpAgentAdapter<{ executable: string; model?: string }> = { }), }; +function durableState() { + const values = new Map(); + let rejectCommit = false; + const state: AcpConversationStateStore = { + has: async (key) => values.has(key), + mark: async (key, cwd) => { + values.set(key, { version: 1, cwd }); + }, + read: async (key) => structuredClone(values.get(key)), + write: async (key, record) => { + if (rejectCommit && record.phase === 'committed') throw new Error('Commit unavailable'); + values.set(key, structuredClone(record)); + }, + }; + return { + state, + values, + rejectCommit: () => { + rejectCommit = true; + }, + record: (key = 'session-a') => values.get(key) as AcpContinuityRecord, + }; +} + +test('durably acknowledged ACP Session resumes the same external ID without replay or session/new', async () => { + const fixture = await executableFixture(); + const protocol = fakeProtocol(); + protocol.supportsRestore = true; + const storage = durableState(); + const make = () => + new AcpExecutor( + adapter, + { executable: fixture.executable }, + { + state: storage.state, + createConnection: protocol.factory, + }, + ); + const first = make(); + try { + assert.equal((await first.execute(request('first'), executorContext([]))).status, 'completed'); + assert.equal(storage.record().phase, 'prompt_pending'); + await first.acknowledgeExecution('session-a', 'turn-first'); + assert.equal(storage.record().phase, 'committed'); + assert.equal(storage.record().committedPrompts, 1); + await first.dispose(); + const restored = make(); + try { + assert.equal( + (await restored.inspectConversation({ conversationKey: 'session-a', cwd: process.cwd() })) + .readiness, + 'restorable', + ); + await restored.configureConversation( + { conversationKey: 'session-a', cwd: process.cwd(), configuration: { model: 'default' } }, + new AbortController().signal, + ); + assert.equal(protocol.sessions, 1); + assert.equal(protocol.resumes, 1); + assert.equal(protocol.loads, 0); + assert.equal( + (await restored.inspectConversation({ conversationKey: 'session-a', cwd: process.cwd() })) + .readiness, + 'ready', + ); + assert.equal( + (await restored.execute(request('second'), executorContext([]))).status, + 'completed', + ); + await restored.acknowledgeExecution('session-a', 'turn-second'); + assert.equal(storage.record().committedPrompts, 2); + assert.equal(protocol.sessions, 1); + } finally { + await restored.dispose(); + } + } finally { + await first.dispose(); + await rm(fixture.root, { recursive: true, force: true }); + } +}); + +test('unacknowledged prompt loads replay as restoration and exposes history gap without a second prompt', async () => { + const fixture = await executableFixture(); + const protocol = fakeProtocol(); + protocol.supportsRestore = true; + const storage = durableState(); + const make = () => + new AcpExecutor( + adapter, + { executable: fixture.executable }, + { + state: storage.state, + createConnection: protocol.factory, + }, + ); + const first = make(); + try { + assert.equal((await first.execute(request('first'), executorContext([]))).status, 'completed'); + await first.dispose(); + const restored = make(); + const events: unknown[] = []; + try { + assert.equal( + (await restored.inspectConversation({ conversationKey: 'session-a', cwd: process.cwd() })) + .readiness, + 'history_gap', + ); + const result = await restored.execute(request('new-input'), executorContext(events)); + assert.equal(result.status, 'failed'); + if (result.status === 'failed') assert.equal(result.code, 'acp_history_gap'); + assert.deepEqual(events, []); + assert.equal(protocol.loads, 1); + assert.equal(protocol.resumes, 0); + assert.equal(protocol.sessions, 1); + assert.equal(protocol.prompts, 1); + assert.equal(storage.record().phase, 'history_gap'); + assert.equal(storage.record().gapEvidence?.replayedUserChunks, 1); + } finally { + await restored.dispose(); + } + const reopened = make(); + try { + const repeated = await reopened.execute(request('still-blocked'), executorContext(events)); + assert.equal(repeated.status, 'failed'); + if (repeated.status === 'failed') assert.equal(repeated.code, 'acp_history_gap'); + assert.equal(protocol.loads, 1); + assert.equal(protocol.prompts, 1); + } finally { + await reopened.dispose(); + } + } finally { + await first.dispose(); + await rm(fixture.root, { recursive: true, force: true }); + } +}); + +test('late restored-session updates stay quarantined while the history gap is persisted', async () => { + const fixture = await executableFixture(); + const protocol = fakeProtocol(); + protocol.supportsRestore = true; + const storage = durableState(); + let releaseGapWrite!: () => void; + const gapWriteBlocked = new Promise((resolve) => { + releaseGapWrite = resolve; + }); + let gapWriteStarted!: () => void; + const gapWriteReached = new Promise((resolve) => { + gapWriteStarted = resolve; + }); + const state: AcpConversationStateStore = { + ...storage.state, + write: async (key, record) => { + if (record.phase === 'history_gap') { + gapWriteStarted(); + await gapWriteBlocked; + } + await storage.state.write!(key, record); + }, + }; + const make = () => + new AcpExecutor( + adapter, + { executable: fixture.executable }, + { state, createConnection: protocol.factory }, + ); + const first = make(); + const blockedWrite = join(process.cwd(), 'node_modules/.maka-acp-restore-write'); + try { + assert.equal((await first.execute(request('first'), executorContext([]))).status, 'completed'); + await first.dispose(); + const restored = make(); + const events: unknown[] = []; + try { + const execution = restored.execute(request('unsent-new-turn'), executorContext(events)); + await gapWriteReached; + protocol.notifyUpdate({ + sessionUpdate: 'agent_message_chunk', + content: { type: 'text', text: 'LATE OLD EXTERNAL OUTPUT' }, + }); + protocol.notifyUpdate({ + sessionUpdate: 'tool_call', + toolCallId: 'late-old-tool', + title: 'Historical tool', + }); + assert.deepEqual(await protocol.requestPermission(), { outcome: { outcome: 'cancelled' } }); + await Promise.all([ + assert.rejects(protocol.readTextFile(join(process.cwd(), 'package.json')), { + code: 'acp_history_gap', + }), + assert.rejects(protocol.writeTextFile(blockedWrite, 'must not be written'), { + code: 'acp_history_gap', + }), + ]); + releaseGapWrite(); + const result = await execution; + assert.equal(result.status, 'failed'); + if (result.status === 'failed') assert.equal(result.code, 'acp_history_gap'); + assert.deepEqual(events, []); + assert.equal(protocol.prompts, 1); + assert.equal(storage.record().phase, 'history_gap'); + } finally { + releaseGapWrite(); + await restored.dispose(); + } + } finally { + await first.dispose(); + await rm(blockedWrite, { force: true }); + await rm(fixture.root, { recursive: true, force: true }); + } +}); + +test('failed restore permits explicit retry and never creates a replacement Session', async () => { + const fixture = await executableFixture(); + const protocol = fakeProtocol(); + protocol.supportsRestore = true; + const storage = durableState(); + const make = () => + new AcpExecutor( + adapter, + { executable: fixture.executable }, + { + state: storage.state, + createConnection: protocol.factory, + }, + ); + const first = make(); + try { + await first.execute(request('first'), executorContext([])); + await first.acknowledgeExecution('session-a', 'turn-first'); + await first.dispose(); + const restored = make(); + const input = { + conversationKey: 'session-a', + cwd: process.cwd(), + configuration: { model: 'default' }, + }; + try { + protocol.restoreFailure = true; + await assert.rejects(restored.configureConversation(input, new AbortController().signal)); + assert.equal((await restored.inspectConversation(input)).readiness, 'restore_failed'); + protocol.restoreFailure = false; + await restored.configureConversation(input, new AbortController().signal); + assert.equal((await restored.inspectConversation(input)).readiness, 'ready'); + assert.equal(protocol.sessions, 1); + assert.equal(protocol.resumes, 1); + } finally { + await restored.dispose(); + } + } finally { + await first.dispose(); + await rm(fixture.root, { recursive: true, force: true }); + } +}); + +test('missing restore capability, corrupted record and changed executable fail closed', async () => { + const fixture = await executableFixture(); + const protocol = fakeProtocol(); + protocol.supportsRestore = true; + const storage = durableState(); + const make = () => + new AcpExecutor( + adapter, + { executable: fixture.executable }, + { + state: storage.state, + createConnection: protocol.factory, + }, + ); + const first = make(); + try { + await first.execute(request('first'), executorContext([])); + await first.acknowledgeExecution('session-a', 'turn-first'); + await first.dispose(); + const input = { + conversationKey: 'session-a', + cwd: process.cwd(), + configuration: { model: 'default' }, + }; + protocol.supportsRestore = false; + const unsupported = make(); + await assert.rejects(unsupported.configureConversation(input, new AbortController().signal)); + await unsupported.dispose(); + protocol.supportsRestore = true; + await writeFile(fixture.executable, 'changed fixture'); + const changed = make(); + await assert.rejects(changed.configureConversation(input, new AbortController().signal)); + assert.equal((await changed.inspectConversation(input)).readiness, 'restore_failed'); + await writeFile(fixture.executable, 'fixture'); + await changed.configureConversation(input, new AbortController().signal); + assert.equal((await changed.inspectConversation(input)).readiness, 'ready'); + await changed.dispose(); + storage.values.set('session-a', { version: 2, cwd: process.cwd(), sessionId: 'acp-session' }); + const corrupt = make(); + assert.equal((await corrupt.inspectConversation(input)).readiness, 'history_only'); + assert.equal((await corrupt.execute(request('second'), executorContext([]))).status, 'failed'); + await corrupt.dispose(); + assert.equal(protocol.sessions, 1); + assert.equal(protocol.prompts, 1); + } finally { + await first.dispose(); + await rm(fixture.root, { recursive: true, force: true }); + } +}); + +test('failed Plugin acknowledgement leaves a visible gap', async () => { + const fixture = await executableFixture(); + const protocol = fakeProtocol(); + protocol.supportsRestore = true; + const storage = durableState(); + const executor = new AcpExecutor( + adapter, + { executable: fixture.executable }, + { + state: storage.state, + createConnection: protocol.factory, + }, + ); + try { + await executor.execute(request('first'), executorContext([])); + storage.rejectCommit(); + await assert.rejects(executor.acknowledgeExecution('session-a', 'turn-first')); + assert.equal( + (await executor.inspectConversation({ conversationKey: 'session-a', cwd: process.cwd() })) + .readiness, + 'history_gap', + ); + assert.equal(storage.record().phase, 'prompt_pending'); + } finally { + await executor.dispose(); + await rm(fixture.root, { recursive: true, force: true }); + } +}); + +test('a terminal result abandoned before durable consumption becomes a history gap', async () => { + const fixture = await executableFixture(); + const protocol = fakeProtocol(); + const storage = durableState(); + const executor = new AcpExecutor( + adapter, + { executable: fixture.executable }, + { state: storage.state, createConnection: protocol.factory }, + ); + try { + assert.equal( + (await executor.execute(request('first'), executorContext([]))).status, + 'completed', + ); + await executor.abandonExecution('session-a', 'turn-first'); + assert.equal(storage.record().phase, 'history_gap'); + assert.equal( + (await executor.inspectConversation({ conversationKey: 'session-a', cwd: process.cwd() })) + .readiness, + 'history_gap', + ); + const blocked = await executor.execute(request('second'), executorContext([])); + assert.equal(blocked.status, 'failed'); + if (blocked.status === 'failed') assert.equal(blocked.code, 'acp_history_gap'); + } finally { + await executor.dispose(); + await rm(fixture.root, { recursive: true, force: true }); + } +}); + +test('fresh configure initialization failure remains retryable', async () => { + const fixture = await executableFixture(); + const protocol = fakeProtocol(); + const storage = durableState(); + let rejectRead = true; + const executor = new AcpExecutor( + adapter, + { executable: fixture.executable }, + { + createConnection: protocol.factory, + state: { + ...storage.state, + read: async (key) => { + if (rejectRead) throw new Error('Storage unavailable'); + return await storage.state.read!(key); + }, + }, + }, + ); + const input = { + conversationKey: 'session-a', + cwd: process.cwd(), + configuration: { model: 'default' }, + }; + try { + await assert.rejects( + executor.configureConversation(input, new AbortController().signal), + /Storage unavailable/u, + ); + rejectRead = false; + assert.equal((await executor.inspectConversation(input)).readiness, 'ready'); + await executor.configureConversation(input, new AbortController().signal); + assert.equal((await executor.inspectConversation(input)).readiness, 'ready'); + assert.equal(protocol.sessions, 1); + } finally { + await executor.dispose(); + await rm(fixture.root, { recursive: true, force: true }); + } +}); + +test('concurrent restore retries share one replacement Session', async () => { + const fixture = await executableFixture(); + const protocol = fakeProtocol(); + protocol.supportsRestore = true; + const storage = durableState(); + const make = () => + new AcpExecutor( + adapter, + { executable: fixture.executable }, + { state: storage.state, createConnection: protocol.factory }, + ); + const first = make(); + try { + await first.execute(request('first'), executorContext([])); + await first.acknowledgeExecution('session-a', 'turn-first'); + await first.dispose(); + const restored = make(); + const input = { + conversationKey: 'session-a', + cwd: process.cwd(), + configuration: { model: 'default' }, + }; + try { + protocol.restoreFailure = true; + await assert.rejects(restored.configureConversation(input, new AbortController().signal)); + protocol.restoreFailure = false; + const retries = await Promise.allSettled([ + restored.configureConversation(input, new AbortController().signal), + restored.configureConversation(input, new AbortController().signal), + ]); + assert.equal(retries.filter((result) => result.status === 'fulfilled').length, 1); + assert.equal(protocol.resumes, 1); + assert.equal((await restored.inspectConversation(input)).readiness, 'ready'); + } finally { + await restored.dispose(); + } + } finally { + await first.dispose(); + await rm(fixture.root, { recursive: true, force: true }); + } +}); + test('runtime retains one ACP process and Session across prompts', async () => { const fixture = await executableFixture(); const protocol = fakeProtocol(); @@ -385,6 +831,7 @@ for (const stopReason of [ ]) { test(`runtime drains cancellation and preserves ${stopReason}`, async () => { const fixture = await executableFixture(); + const storage = durableState(); let started!: () => void; const ready = new Promise((resolve) => { started = resolve; @@ -429,7 +876,7 @@ for (const stopReason of [ const executor = new AcpExecutor( adapter, { executable: fixture.executable }, - { createConnection: factory }, + { createConnection: factory, state: storage.state }, ); const abort = new AbortController(); const execution = executor.execute(request('cancel'), executorContext([], abort.signal)); @@ -443,6 +890,16 @@ for (const stopReason of [ : { providerStopReason: stopReason }), }); assert.equal(cancellations, 1); + assert.equal(storage.record().phase, 'prompt_pending'); + await executor.acknowledgeExecution('session-a', 'turn-cancel'); + const uncertain = ['request_error', 'process_crash'].includes(stopReason); + assert.equal(storage.record().phase, uncertain ? 'prompt_pending' : 'committed'); + await executor.disposeConversation('session-a'); + assert.equal( + (await executor.inspectConversation({ conversationKey: 'session-a', cwd: process.cwd() })) + .readiness, + uncertain ? 'restore_failed' : 'restorable', + ); } finally { await executor.dispose(); await rm(fixture.root, { recursive: true, force: true }); @@ -532,26 +989,26 @@ test('discovery shares a disposable probe, does not mark a task, and first promp const before = protocol.connections; const status = await executor.inspectConversation({ conversationKey: 'session-a', - cwd: fixture.root, + cwd: process.cwd(), }); assert.equal(status.currentModel, 'fast'); assert.equal(protocol.connections, before); await executor.configureConversation( - { conversationKey: 'session-a', cwd: fixture.root, configuration: { model: 'default' } }, + { conversationKey: 'session-a', cwd: process.cwd(), configuration: { model: 'default' } }, new AbortController().signal, ); assert.equal(protocol.selectedModel, 'default'); await assert.rejects( () => executor.configureConversation( - { conversationKey: 'session-a', cwd: fixture.root, configuration: { model: 'removed' } }, + { conversationKey: 'session-a', cwd: process.cwd(), configuration: { model: 'removed' } }, new AbortController().signal, ), /unavailable/u, ); assert.equal(protocol.selectedModel, 'default'); assert.equal( - (await executor.inspectConversation({ conversationKey: 'session-a', cwd: fixture.root })) + (await executor.inspectConversation({ conversationKey: 'session-a', cwd: process.cwd() })) .readiness, 'ready', ); @@ -586,14 +1043,14 @@ for (const failure of ['response_lost', 'timeout'] as const) { timeout = setTimeout(() => abort.abort(new DOMException('Timed out', 'TimeoutError')), 50); await assert.rejects( executor.configureConversation( - { conversationKey: 'session-a', cwd: fixture.root, configuration: { model: 'fast' } }, + { conversationKey: 'session-a', cwd: process.cwd(), configuration: { model: 'fast' } }, abort.signal, ), ); // The external mutation happened even though no matching confirmation arrived. assert.equal(protocol.selectedModel, failure === 'timeout' ? 'fast' : 'default'); assert.equal( - (await executor.inspectConversation({ conversationKey: 'session-a', cwd: fixture.root })) + (await executor.inspectConversation({ conversationKey: 'session-a', cwd: process.cwd() })) .readiness, 'history_only', ); @@ -706,6 +1163,14 @@ function fakeProtocol(): { configurationFailure?: 'response_lost' | 'unconfirmed' | 'timeout' | 'once'; sessionCreationFailure?: 'once'; promptModels: string[]; + supportsRestore: boolean; + restoreFailure?: boolean; + resumes: number; + loads: number; + notifyUpdate(update: unknown): void; + requestPermission(): Promise; + readTextFile(path: string): Promise; + writeTextFile(path: string, content: string): Promise; notifyConfiguration(model: string): void; } { const fixture = { @@ -722,6 +1187,14 @@ function fakeProtocol(): { | undefined, sessionCreationFailure: undefined as 'once' | undefined, promptModels: [] as string[], + supportsRestore: false, + restoreFailure: false, + resumes: 0, + loads: 0, + notifyUpdate: (_update: unknown): void => {}, + requestPermission: async (): Promise => undefined, + readTextFile: async (_path: string): Promise => undefined, + writeTextFile: async (_path: string, _content: string): Promise => undefined, notifyConfiguration: (_model: string): void => {}, factory: undefined as unknown as AcpConnectionFactory, }; @@ -740,6 +1213,27 @@ function fakeProtocol(): { }, } as unknown as ClientApp; input.configureClient(app); + fixture.notifyUpdate = (update) => { + notifications.get(methods.client.session.update)?.({ + params: { sessionId: 'acp-session', update } as never, + }); + }; + fixture.requestPermission = async () => + await requests.get(methods.client.session.requestPermission)?.({ + params: { + sessionId: 'acp-session', + toolCall: { toolCallId: 'late-old-tool', title: 'Allow edit?' }, + options: [{ optionId: 'allow_once', name: 'Allow once', kind: 'allow_once' }], + } as never, + }); + fixture.readTextFile = async (path) => + await requests.get(methods.client.fs.readTextFile)?.({ + params: { sessionId: 'acp-session', path } as never, + }); + fixture.writeTextFile = async (path, content) => + await requests.get(methods.client.fs.writeTextFile)?.({ + params: { sessionId: 'acp-session', path, content } as never, + }); fixture.notifyConfiguration = (model) => { notifications.get(methods.client.session.update)?.({ params: { @@ -769,7 +1263,44 @@ function fakeProtocol(): { params: Record, options?: { cancellationSignal?: AbortSignal }, ) => { - if (method === methods.agent.initialize) return { protocolVersion: 1 }; + if (method === methods.agent.initialize) + return { + protocolVersion: 1, + agentCapabilities: fixture.supportsRestore + ? { loadSession: true, sessionCapabilities: { resume: {} } } + : {}, + }; + if (method === methods.agent.session.resume || method === methods.agent.session.load) { + assert.equal(params.sessionId, 'acp-session'); + assert.equal(params.cwd, process.cwd()); + if (fixture.restoreFailure) throw new Error('Session unavailable'); + if (method === methods.agent.session.resume) fixture.resumes += 1; + else { + fixture.loads += 1; + for (const update of [ + { sessionUpdate: 'user_message_chunk', content: { type: 'text', text: 'first' } }, + { sessionUpdate: 'agent_message_chunk', content: { type: 'text', text: 'replay' } }, + { sessionUpdate: 'tool_call', toolCallId: 'replay-tool', title: 'Historical tool' }, + ]) + notifications.get(methods.client.session.update)?.({ + params: { sessionId: 'acp-session', update } as never, + }); + } + return { + configOptions: [ + { + type: 'select', + id: 'model', + name: 'Model', + currentValue: fixture.selectedModel ?? 'default', + options: [ + { value: 'default', name: 'Default' }, + { value: 'fast', name: 'Fast' }, + ], + }, + ], + }; + } if (method === methods.agent.session.new) { fixture.sessions += 1; if (fixture.sessionCreationFailure === 'once') { @@ -931,7 +1462,7 @@ for (const failure of ['once', 'unconfirmed'] as const) ); const input = { conversationKey: 'session-a', - cwd: fixture.root, + cwd: process.cwd(), configuration: { model: 'fast' }, }; try { @@ -970,7 +1501,7 @@ test('idle Agent configuration notifications update the inspected model without ); protocol.notifyConfiguration('fast'); assert.equal( - (await executor.inspectConversation({ conversationKey: 'session-a', cwd: fixture.root })) + (await executor.inspectConversation({ conversationKey: 'session-a', cwd: process.cwd() })) .currentModel, 'fast', ); diff --git a/packages/acp-executor-plugin/src/__tests__/acp-process.test.ts b/packages/acp-executor-plugin/src/__tests__/acp-process.test.ts index 0dad5beb094..aa3e2902ea1 100644 --- a/packages/acp-executor-plugin/src/__tests__/acp-process.test.ts +++ b/packages/acp-executor-plugin/src/__tests__/acp-process.test.ts @@ -22,7 +22,7 @@ import { test } from 'node:test'; import { mkdtemp, writeFile, readFile, rm, symlink } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import { join } from 'node:path'; -import { AcpExecutor, type AcpAgentAdapter } from '../index.js'; +import { AcpExecutor, type AcpAgentAdapter, type AcpContinuityRecord } from '../index.js'; import { readWorkspaceTextFile } from '../acp-filesystem.js'; import { Context } from '@maka/runtime/plugin-kernel'; import { PluginExecutorService } from '@maka/runtime/plugin-executor-service'; @@ -36,21 +36,40 @@ import type { const program = String.raw` const {createInterface}=require('node:readline'); const {spawn}=require('node:child_process'); -const pending=new Map(); let seq=0, turn=0, cwd, promptId; +const fs=require('node:fs'); +const pending=new Map(); let seq=0, turn=0, cwd, promptId, history=[]; +const historyPath=()=>cwd+'/.acp-fixture-history.json'; +const persist=()=>fs.writeFileSync(historyPath(),JSON.stringify(history)); const send=(value)=>process.stdout.write(JSON.stringify({jsonrpc:'2.0',...value})+'\n'); const respond=(id,result)=>send({id,result}); -const update=(update)=>send({method:'session/update',params:{sessionId:'fixture',update}}); +const update=(update)=>{history.push(update);persist();send({method:'session/update',params:{sessionId:'fixture',update}});}; const text=(text)=>update({sessionUpdate:'agent_message_chunk',content:{type:'text',text}}); const call=(method,params)=>new Promise(resolve=>{const id='client-'+(++seq);pending.set(id,resolve);send({id,method,params});}); createInterface({input:process.stdin}).on('line',async line=>{ const m=JSON.parse(line); if(pending.has(m.id)){pending.get(m.id)(m);pending.delete(m.id);return;} - if(m.method==='initialize')return respond(m.id,{protocolVersion:1}); - if(m.method==='session/new'){cwd=m.params.cwd;return respond(m.id,{sessionId:'fixture',configOptions:[]});} + if(m.method==='initialize')return respond(m.id,{protocolVersion:1,agentCapabilities:{loadSession:true,sessionCapabilities:{resume:{}}}}); + if(m.method==='session/new'){ + cwd=m.params.cwd; + if(fs.existsSync(historyPath()))return send({id:m.id,error:{code:-32000,message:'Session already exists'}}); + history=[];persist();return respond(m.id,{sessionId:'fixture',configOptions:[]}); + } + if(m.method==='session/resume'||m.method==='session/load'){ + cwd=m.params.cwd; + if(m.params.sessionId!=='fixture'||!fs.existsSync(historyPath())) + return send({id:m.id,error:{code:-32000,message:'Unknown Session'}}); + history=JSON.parse(fs.readFileSync(historyPath(),'utf8')); + turn=history.filter(item=>item.sessionUpdate==='user_message_chunk').length; + if(m.method==='session/load')for(const item of history) + send({method:'session/update',params:{sessionId:'fixture',update:item}}); + return respond(m.id,{configOptions:[]}); + } if(m.method==='session/cancel'){if(promptId!==undefined){respond(promptId,{stopReason:'cancelled'});promptId=undefined;}return;} if(m.method!=='session/prompt')return; const value=m.params.prompt[0].text; + update({sessionUpdate:'user_message_chunk',content:{type:'text',text:value}}); if(value==='crash'){process.exit(3);return;} + if(value==='refusal'){respond(m.id,{stopReason:'refusal'});return;} if(value==='ignore-cancel'){text('waiting');return;} if(value==='wait'){promptId=m.id;text('waiting');return;} if(value==='helper'){const child=spawn(process.execPath,['-e','setInterval(()=>{},1000)'],{stdio:'ignore'});text(String(child.pid));respond(m.id,{stopReason:'end_turn'});return;} @@ -123,6 +142,7 @@ async function fixture() { }); return { root, + script, executor, request, context, @@ -167,6 +187,204 @@ test('real stdio retains a conversation and returns original question option ids } }); +test('real stdio restores the same Session in a new process after durable acknowledgement', async () => { + const f = await fixture(); + const values = new Map(); + const state = { + has: async (key: string) => values.has(key), + mark: async (key: string, cwd: string) => { + values.set(key, { version: 1, cwd }); + }, + read: async (key: string) => values.get(key), + write: async (key: string, record: unknown) => { + values.set(key, record); + }, + }; + const adapter: AcpAgentAdapter = { + id: 'stdio-restore-fixture', + displayName: 'Fixture', + configure: () => ({ launch: { executable: process.execPath, args: [f.script] } }), + permissionKind: () => 'question', + }; + const make = () => new AcpExecutor(adapter, {}, { state }); + const first = make(); + try { + const initial = await first.execute(f.request('one'), f.context()); + assert.equal(initial.status, 'completed'); + if (initial.status !== 'completed') throw new Error('Expected first completion'); + await first.acknowledgeExecution('task', 'one'); + const firstPid = JSON.parse(initial.text).pid as number; + assert.notEqual(firstPid, process.pid); + process.kill(firstPid, 'SIGKILL'); + let readiness: string | undefined; + for (let attempt = 0; attempt < 60; attempt++) { + readiness = (await first.inspectConversation({ conversationKey: 'task', cwd: f.root })) + .readiness; + if (readiness === 'restorable') break; + await new Promise((resolve) => setTimeout(resolve, 50)); + } + assert.equal(readiness, 'restorable'); + await first.dispose(); + const restored = make(); + try { + assert.equal( + (await restored.inspectConversation({ conversationKey: 'task', cwd: f.root })).readiness, + 'restorable', + ); + const later = await restored.execute(f.request('two'), f.context()); + assert.equal(later.status, 'completed'); + if (later.status !== 'completed') throw new Error('Expected restored completion'); + assert.notEqual(JSON.parse(initial.text).pid, JSON.parse(later.text).pid); + assert.equal(JSON.parse(later.text).turn, 2); + await restored.acknowledgeExecution('task', 'two'); + assert.equal( + (await restored.inspectConversation({ conversationKey: 'task', cwd: f.root })).readiness, + 'ready', + ); + } finally { + await restored.dispose(); + } + } finally { + await first.dispose(); + await f.dispose(); + } +}); + +for (const prompt of ['wait', 'refusal', 'ignore-cancel', 'crash']) { + test(`real stdio checkpoint after ${prompt} preserves safe restart behavior`, async () => { + const f = await fixture(); + const values = new Map(); + const state = { + has: async (key: string) => values.has(key), + mark: async () => { + throw new Error('Expected versioned storage'); + }, + read: async (key: string) => values.get(key), + write: async (key: string, record: AcpContinuityRecord) => { + values.set(key, record); + }, + }; + const adapter: AcpAgentAdapter = { + id: 'stdio-checkpoint-fixture', + displayName: 'Fixture', + configure: () => ({ launch: { executable: process.execPath, args: [f.script] } }), + permissionKind: () => 'question', + }; + const make = () => new AcpExecutor(adapter, {}, { state }); + const first = make(); + const root = new Context(); + const service = new PluginExecutorService(root); + root + .extend({ + maka: { rootId: 'profile', packageId: 'fixture', entryId: 'stdio', generation: 1 }, + }) + .executors.register(first); + const backend = new PluginExecutorBackend({ + sessionId: 'task', + cwd: f.root, + binding: service.bind('task', first.id), + }); + const restored = make(); + try { + let firstPid: number | undefined; + for await (const event of backend.send({ turnId: 'one', text: 'one' })) { + if (event.type === 'text_complete') firstPid = JSON.parse(event.text).pid; + } + assert.equal(values.get('task')?.phase, 'committed'); + const terminalEvents: string[] = []; + for await (const event of backend.send({ turnId: 'stopped', text: prompt })) { + terminalEvents.push(event.type); + if (event.type === 'text_delta') await backend.stop('user_stop'); + if (event.type === 'complete') + assert.equal(values.get('task')?.phase, 'prompt_pending', 'delivery alone cannot commit'); + } + assert.equal(terminalEvents.at(-1), 'complete'); + assert.ok(terminalEvents.includes(['crash', 'refusal'].includes(prompt) ? 'error' : 'abort')); + const settled = prompt === 'wait' || prompt === 'refusal'; + assert.equal(values.get('task')?.phase, settled ? 'committed' : 'prompt_pending'); + assert.equal(values.get('task')?.committedPrompts, settled ? 2 : 1); + await first.dispose(); + assert.equal( + (await restored.inspectConversation({ conversationKey: 'task', cwd: f.root })).readiness, + settled ? 'restorable' : 'history_gap', + ); + const events: PluginExecutorOutputEvent[] = []; + const result = await restored.execute( + f.request('after'), + f.context(undefined, (event) => events.push(event)), + ); + if (settled) { + assert.equal(result.status, 'completed'); + if (result.status !== 'completed') throw new Error('Expected restored completion'); + assert.equal(JSON.parse(result.text).turn, 3); + assert.notEqual(JSON.parse(result.text).pid, firstPid); + await restored.acknowledgeExecution('task', 'after'); + assert.equal(values.get('task')?.committedPrompts, 3); + } else { + assert.equal(result.status, 'failed'); + if (result.status !== 'failed') throw new Error('Expected history gap'); + assert.equal(result.code, 'acp_history_gap'); + assert.deepEqual(events, [], 'uncertain replay must not reach canonical history'); + } + } finally { + await backend.dispose(); + await root.fiber.dispose(); + await first.dispose(); + await restored.dispose(); + await f.dispose(); + } + }); +} + +test('real stdio load replay stays separate when the Agent may be ahead of durable history', async () => { + const f = await fixture(); + const values = new Map(); + const state = { + has: async (key: string) => values.has(key), + mark: async (key: string, cwd: string) => { + values.set(key, { version: 1, cwd }); + }, + read: async (key: string) => values.get(key), + write: async (key: string, record: unknown) => { + values.set(key, record); + }, + }; + const adapter: AcpAgentAdapter = { + id: 'stdio-restore-fixture', + displayName: 'Fixture', + configure: () => ({ launch: { executable: process.execPath, args: [f.script] } }), + permissionKind: () => 'question', + }; + const make = () => new AcpExecutor(adapter, {}, { state }); + const first = make(); + try { + assert.equal((await first.execute(f.request('one'), f.context())).status, 'completed'); + // Simulate process/Host loss after the Agent finished but before Maka's + // terminal event was acknowledged to the Plugin. + await first.dispose(); + const restored = make(); + const projected: PluginExecutorOutputEvent[] = []; + try { + const result = await restored.execute( + f.request('new-user-input'), + f.context(undefined, (event) => projected.push(event)), + ); + assert.equal(result.status, 'failed'); + if (result.status === 'failed') assert.equal(result.code, 'acp_history_gap'); + assert.deepEqual(projected, []); + assert.equal( + (await restored.inspectConversation({ conversationKey: 'task', cwd: f.root })).readiness, + 'history_gap', + ); + } finally { + await restored.dispose(); + } + } finally { + await first.dispose(); + await f.dispose(); + } +}); + test('real filesystem callbacks reject a symlink escape and permit workspace files', async () => { const f = await fixture(); const outside = await mkdtemp(join(tmpdir(), 'maka-acp-outside-')); diff --git a/packages/acp-executor-plugin/src/index.ts b/packages/acp-executor-plugin/src/index.ts index e323d6f5ee1..af24887252a 100644 --- a/packages/acp-executor-plugin/src/index.ts +++ b/packages/acp-executor-plugin/src/index.ts @@ -18,7 +18,7 @@ */ import { createHash } from 'node:crypto'; -import { constants } from 'node:fs'; +import { constants, createReadStream } from 'node:fs'; import { access, mkdtemp, rm, realpath, stat } from 'node:fs/promises'; import { tmpdir } from 'node:os'; import type { ExecutorCatalogEntry, ExecutorConfiguration } from '@maka/core/executor-catalog'; @@ -105,6 +105,25 @@ export type AcpConnectionFactory = (input: AcpConnectionFactoryInput) => AcpConn export interface AcpConversationStateStore { has(conversationKey: string, cwd: string): Promise; mark(conversationKey: string, cwd: string): Promise; + /** Versioned Plugin-private state. Older test providers may expose only v1 markers. */ + read?(conversationKey: string): Promise; + write?(conversationKey: string, state: AcpContinuityRecord): Promise; +} + +export interface AcpContinuityRecord { + readonly version: 2; + readonly cwd: string; + /** Digest of the adapter and exact executable/helper bytes; never sent to the Host. */ + readonly binding: string; + readonly phase: 'reserved' | 'established' | 'prompt_pending' | 'committed' | 'history_gap'; + readonly committedPrompts: number; + readonly sessionId?: string; + readonly pendingTurnId?: string; + readonly confirmedModel?: string; + readonly gapEvidence?: { + readonly replayedUpdates: number; + readonly replayedUserChunks: number; + }; } interface ActivePrompt { @@ -135,6 +154,12 @@ interface RetainedSession { owner?: AcpConnectionOwner; connection?: ClientConnection; acpSessionId?: string; + record?: AcpContinuityRecord; + restoring?: boolean; + replay?: { updates: number; userChunks: number }; + restoreFailed?: boolean; + historyGap?: boolean; + awaitingAck?: string; continuityReserved?: boolean; configOptions: readonly SessionConfigOption[]; initialization?: Promise; @@ -215,27 +240,62 @@ export class AcpExecutor implements PluginExecutorProvider { }): Promise { const session = this.#sessions.get(input.conversationKey); if (this.#disposed) return this.#catalogEntry('unavailable'); - if (session?.lost || (!session && (await this.#state?.has(input.conversationKey, input.cwd)))) - return this.#catalogEntry('history_only'); - return this.#catalogEntry('ready', session?.configOptions ?? [], input.configuration?.model); + if (session && session.cwd !== resolve(input.cwd)) + return this.#catalogEntry('history_only', [], input.configuration?.model); + if (session?.restoring) + return this.#catalogEntry('restoring', session.configOptions, input.configuration?.model); + if (session?.restoreFailed) + return this.#catalogEntry( + 'restore_failed', + session.configOptions, + input.configuration?.model, + ); + if (session?.historyGap) + return this.#catalogEntry('history_gap', session.configOptions, input.configuration?.model); + if (session && !session.lost) + return this.#catalogEntry('ready', session.configOptions, input.configuration?.model); + if (session?.lost && !session.record) + return this.#catalogEntry('history_only', session.configOptions, input.configuration?.model); + const stored = this.#state?.read + ? decodeContinuity(await this.#state.read(input.conversationKey)) + : (await this.#state?.has(input.conversationKey, input.cwd)) + ? 'legacy' + : undefined; + if (!stored) return this.#catalogEntry('ready', [], input.configuration?.model); + if (stored === 'invalid' || stored === 'legacy' || stored.cwd !== resolve(input.cwd)) + return this.#catalogEntry('history_only', [], input.configuration?.model); + if (stored.phase === 'reserved') + return this.#catalogEntry('history_only', [], input.configuration?.model); + return this.#catalogEntry( + stored.phase === 'prompt_pending' || stored.phase === 'history_gap' + ? 'history_gap' + : 'restorable', + [], + input.configuration?.model ?? stored.confirmedModel, + ); } async configureConversation( input: { conversationKey: string; cwd: string; configuration?: ExecutorConfiguration }, signal: AbortSignal, ): Promise { - const session = this.#sessions.get(input.conversationKey); - if ( - !session?.connection || - session.lost || - session.active || - session.configuring || - !input.configuration?.model - ) - throw new Error('ACP model change is unavailable'); + if (!input.configuration?.model) throw new Error('ACP model change is unavailable'); + const session = await this.#session(input); + if (session.active || session.configuring) throw new Error('ACP model change is unavailable'); session.configuring = true; + const wasConnected = !!session.connection; try { + await this.#ensureInitialized(session, signal); await this.#applyInitialConfig(session, { model: input.configuration.model }, signal, true); + } catch (error) { + if (!wasConnected || session.lost) { + await this.#lose(session); + if (session.record?.sessionId && !session.historyGap) session.restoreFailed = true; + const retryable = !session.acpSessionId && !session.continuityReserved && !session.record; + if (retryable && this.#sessions.get(session.conversationKey) === session) + this.#sessions.delete(session.conversationKey); + } + throw error; } finally { session.configuring = false; } @@ -255,12 +315,17 @@ export class AcpExecutor implements PluginExecutorProvider { ? model.options .flatMap((entry) => ('options' in entry ? entry.options : [entry])) .map((entry) => ({ id: entry.value, name: entry.name })) - : []; + : (this.#catalog?.models ?? []); return { id: this.id, displayName: this.displayName, readiness, - ...(this.#adapter.describeModels?.(models) ?? { models }), + ...(model?.type === 'select' + ? (this.#adapter.describeModels?.(models) ?? { models }) + : { + models, + ...(this.#catalog?.modelGroups ? { modelGroups: this.#catalog.modelGroups } : {}), + }), ...(model?.type === 'select' ? { currentModel: model.currentValue } : selected @@ -294,11 +359,11 @@ export class AcpExecutor implements PluginExecutorProvider { if (model) request = { ...request, configuration: { model } }; let session: RetainedSession; try { - session = this.#session(request); + session = await this.#session(request); } catch (error) { return failure(safeErrorMessage(error), errorCode(error)); } - if (session.active || session.configuring) + if (session.active || session.configuring || session.awaitingAck) return failure(`${this.displayName} ACP Session is busy`, 'acp_busy', true); const active: ActivePrompt = { context, tools: new Map(), text: '' }; session.active = active; @@ -311,12 +376,17 @@ export class AcpExecutor implements PluginExecutorProvider { { model: request.configuration.model }, context.signal, ); + await this.#beginPrompt(session, request.turnId); const prompt = session.connection!.agent.request(methods.agent.session.prompt, { sessionId: session.acpSessionId!, prompt: [{ type: 'text', text: promptText(request) }], }); const response = await this.#awaitPrompt(session, prompt, context.signal); if (!response) return { status: 'cancelled', reason: 'timeout' }; + // A settled prompt can be checkpointed even when it was cancelled or failed. + // Timeouts and transport failures never reach this point, and the Runtime + // must still durably consume the terminal event before acknowledging it. + if (session.record) session.awaitingAck = request.turnId; if (context.signal.aborted || response.stopReason === 'cancelled') { return { status: 'cancelled', providerStopReason: response.stopReason }; } @@ -331,13 +401,16 @@ export class AcpExecutor implements PluginExecutorProvider { } return { status: 'completed', text: active.text }; } catch (error) { + if (errorCode(error) === 'acp_history_gap') session.historyGap = true; await this.#lose(session); + if (session.record?.sessionId && !session.historyGap) session.restoreFailed = true; // Retry only when no session/new was attempted. Once continuity is // reserved, a lost response may hide an established Agent Session. const retryable = !session.acpSessionId && !session.continuityReserved && - errorCode(error) !== 'acp_history_only'; + errorCode(error) !== 'acp_history_only' && + errorCode(error) !== 'acp_history_gap'; if (retryable && this.#sessions.get(session.conversationKey) === session) this.#sessions.delete(session.conversationKey); if (context.signal.aborted) { @@ -373,18 +446,27 @@ export class AcpExecutor implements PluginExecutorProvider { if (failures.length) throw new AggregateError(failures, 'ACP process cleanup failed'); } - #session(request: Readonly): RetainedSession { + async #session(request: { + readonly conversationKey: string; + readonly cwd: string; + readonly configuration?: ExecutorConfiguration; + }): Promise { const cwd = resolve(request.cwd); const existing = this.#sessions.get(request.conversationKey); if (existing) { if (existing.cwd !== cwd) throw new AcpRuntimeError('ACP conversation cannot change workspace', 'acp_cwd_changed'); - if (existing.lost) + if (!existing.lost) return existing; + if (existing.active || existing.configuring) + throw new AcpRuntimeError('ACP Session is busy', 'acp_busy'); + if (!existing.record) throw new AcpRuntimeError( 'ACP conversation is history-only because its external process was lost', 'acp_history_only', ); - return existing; + await existing.loss; + if (this.#sessions.get(request.conversationKey) !== existing) return this.#session(request); + this.#sessions.delete(request.conversationKey); } const created: RetainedSession = { conversationKey: request.conversationKey, @@ -405,21 +487,54 @@ export class AcpExecutor implements PluginExecutorProvider { await initialization; } catch (error) { session.initialization = undefined; + session.restoring = false; + session.replay = undefined; throw error; } } async #initialize(session: RetainedSession, signal: AbortSignal, probe = false): Promise { - if (!probe && (await this.#state?.has(session.conversationKey, session.cwd))) { + const timeout = AbortSignal.timeout(INITIALIZE_TIMEOUT_MS); + const startupSignal = AbortSignal.any([signal, timeout]); + startupSignal.throwIfAborted(); + const launch = this.#configured.launch; + const stored = probe + ? undefined + : this.#state?.read + ? decodeContinuity(await this.#state.read(session.conversationKey)) + : (await this.#state?.has(session.conversationKey, session.cwd)) + ? 'legacy' + : undefined; + if (stored === 'legacy') throw new AcpRuntimeError( 'ACP conversation is history-only after the Plugin or Host was restarted', 'acp_history_only', ); + if ( + stored === 'invalid' || + (stored && stored.cwd !== session.cwd) || + stored?.phase === 'reserved' + ) + throw new AcpRuntimeError( + 'ACP conversation has no recoverable external Session identity', + 'acp_history_only', + ); + if (stored?.phase === 'history_gap') + throw new AcpRuntimeError( + 'External Session history cannot be aligned with saved conversation events', + 'acp_history_gap', + ); + if (stored) { + session.record = stored; + session.acpSessionId = stored.sessionId; + session.restoring = true; } - const timeout = AbortSignal.timeout(INITIALIZE_TIMEOUT_MS); - const startupSignal = AbortSignal.any([signal, timeout]); - startupSignal.throwIfAborted(); - const launch = this.#configured.launch; + const binding = + !probe && this.#state?.write + ? await launchBinding(this.id, launch, startupSignal) + : undefined; + if (stored && stored.binding !== binding) + throw new AcpRuntimeError('ACP Agent installation changed', 'acp_restore_identity_changed'); const executable = await checkedExecutable(launch.executable); await Promise.all((launch.requiredExecutables ?? []).map(checkedExecutable)); const owner = this.#createConnection({ @@ -449,11 +564,70 @@ export class AcpExecutor implements PluginExecutorProvider { owner.failed, ]); if (initialized.protocolVersion !== 1) throw new Error('Unsupported ACP protocol version'); + if (stored) { + session.replay = { updates: 0, userChunks: 0 }; + const pending = stored.phase === 'prompt_pending'; + const supported = pending + ? initialized.agentCapabilities?.loadSession + : initialized.agentCapabilities?.sessionCapabilities?.resume; + if (!supported) + throw new AcpRuntimeError( + 'ACP Agent cannot restore this Session', + 'acp_restore_unsupported', + ); + const restored = await Promise.race([ + owner.connection.agent.request( + pending ? methods.agent.session.load : methods.agent.session.resume, + { sessionId: stored.sessionId!, cwd: session.cwd, mcpServers: [] }, + { cancellationSignal: startupSignal }, + ), + owner.failed, + ]); + session.configOptions = restored.configOptions ?? []; + const restoredModel = currentAcpModel(session.configOptions); + if (restoredModel) await this.#persistConfirmedModel(session, restoredModel); + session.restoring = false; + if (pending) { + // The Agent may have progressed beyond Maka's last durable event. Replay + // identities are not sufficient to fabricate canonical history: failed + // turns can change tool IDs between live output and load. Keep every + // replay notification separate and expose an explicit history gap. + session.historyGap = true; + session.record = { + ...session.record!, + phase: 'history_gap', + pendingTurnId: undefined, + gapEvidence: { + replayedUpdates: session.replay?.updates ?? 0, + replayedUserChunks: session.replay?.userChunks ?? 0, + }, + }; + await this.#state?.write?.(session.conversationKey, session.record); + session.replay = undefined; + throw new AcpRuntimeError( + 'External Session was restored, but conversation history may have a gap', + 'acp_history_gap', + ); + } + session.replay = undefined; + return; + } if (!probe) { startupSignal.throwIfAborted(); // Reserve continuity before asking the Agent to create a Session. A failed // or interrupted session/new may still have created one remotely. - await this.#state?.mark(session.conversationKey, session.cwd); + if (this.#state?.write && binding) { + session.record = { + version: 2, + cwd: session.cwd, + binding, + phase: 'reserved', + committedPrompts: 0, + }; + await this.#state.write(session.conversationKey, session.record); + } else { + await this.#state?.mark(session.conversationKey, session.cwd); + } session.continuityReserved = true; } const created = await Promise.race([ @@ -466,6 +640,17 @@ export class AcpExecutor implements PluginExecutorProvider { ]); session.acpSessionId = created.sessionId; session.configOptions = created.configOptions ?? []; + if (session.record) { + const established: AcpContinuityRecord = { + ...session.record, + phase: 'established', + sessionId: created.sessionId, + }; + await this.#state?.write?.(session.conversationKey, established); + session.record = established; + const createdModel = currentAcpModel(session.configOptions); + if (createdModel) await this.#persistConfirmedModel(session, createdModel); + } if (!probe) { await this.#applyInitialConfig( session, @@ -477,6 +662,68 @@ export class AcpExecutor implements PluginExecutorProvider { } } + async #beginPrompt(session: RetainedSession, turnId: string): Promise { + if (!session.record) return; + const pending: AcpContinuityRecord = { + ...session.record, + phase: 'prompt_pending', + pendingTurnId: turnId, + }; + await this.#state?.write?.(session.conversationKey, pending); + session.record = pending; + } + + /** Called only after the Runtime Kernel has durably accepted the terminal event. */ + async acknowledgeExecution(conversationKey: string, turnId: string): Promise { + const session = this.#sessions.get(conversationKey); + if ( + !session?.record || + session.awaitingAck !== turnId || + session.record.phase !== 'prompt_pending' || + session.record.pendingTurnId !== turnId + ) + return; + const committed: AcpContinuityRecord = { + ...session.record, + phase: 'committed', + committedPrompts: session.record.committedPrompts + 1, + pendingTurnId: undefined, + }; + try { + await this.#state?.write?.(conversationKey, committed); + } catch (error) { + session.historyGap = true; + await this.#lose(session); + throw error; + } + session.record = committed; + session.awaitingAck = undefined; + } + + /** The external turn settled, but Maka did not durably consume its terminal event. */ + async abandonExecution(conversationKey: string, turnId: string): Promise { + const session = this.#sessions.get(conversationKey); + if ( + !session?.record || + session.awaitingAck !== turnId || + session.record.phase !== 'prompt_pending' || + session.record.pendingTurnId !== turnId + ) + return; + session.historyGap = true; + session.awaitingAck = undefined; + session.record = { + ...session.record, + phase: 'history_gap', + pendingTurnId: undefined, + }; + try { + await this.#state?.write?.(conversationKey, session.record); + } finally { + await this.#lose(session); + } + } + async #applyInitialConfig( session: RetainedSession, values: Readonly>, @@ -501,7 +748,10 @@ export class AcpExecutor implements PluginExecutorProvider { `ACP configuration value is unavailable: ${key}`, 'acp_config_invalid', ); - if (option.currentValue === value) continue; + if (option.currentValue === value) { + if (key === 'model') await this.#persistConfirmedModel(session, value); + continue; + } signal.throwIfAborted(); try { const updated = await session.connection!.agent.request( @@ -516,6 +766,7 @@ export class AcpExecutor implements PluginExecutorProvider { 'acp_config_unconfirmed', ); session.configOptions = updated.configOptions; + if (key === 'model') await this.#persistConfirmedModel(session, value); } catch (error) { // A rejected/unconfirmed mutation can have reached the Agent. Restore the // previous real ID and require an acknowledgement before permitting retry. @@ -536,6 +787,7 @@ export class AcpExecutor implements PluginExecutorProvider { !session.lost ) { session.configOptions = rollback.configOptions; + if (key === 'model') await this.#persistConfirmedModel(session, option.currentValue); restored = true; } } catch { @@ -548,19 +800,26 @@ export class AcpExecutor implements PluginExecutorProvider { } } + async #persistConfirmedModel(session: RetainedSession, model: string): Promise { + if (!session.record || session.record.confirmedModel === model) return; + const record = { ...session.record, confirmedModel: model }; + await this.#state?.write?.(session.conversationKey, record); + session.record = record; + } + #configureClient(session: RetainedSession, app: ClientApp): void { app .onNotification(methods.client.session.update, ({ params }) => { if (params.sessionId === session.acpSessionId) this.#acceptUpdate(session, params.update); }) .onRequest(methods.client.fs.readTextFile, async ({ params }) => { - this.#assertSession(session, params.sessionId); + this.#assertWorkspaceAccess(session, params.sessionId); return { content: await readWorkspaceTextFile(session.cwd, params.path, params.line, params.limit), }; }) .onRequest(methods.client.fs.writeTextFile, async ({ params }) => { - this.#assertSession(session, params.sessionId); + this.#assertWorkspaceAccess(session, params.sessionId); await writeWorkspaceTextFile(session.cwd, params.path, params.content); return {}; }) @@ -571,6 +830,8 @@ export class AcpExecutor implements PluginExecutorProvider { async #requestPermission(session: RetainedSession, request: RequestPermissionRequest) { this.#assertSession(session, request.sessionId); + if (session.restoring || session.historyGap || session.lost) + return { outcome: { outcome: 'cancelled' as const } }; const active = session.active; if (!active) return { outcome: { outcome: 'cancelled' as const } }; const outcome = await active.context.requestPermission({ @@ -583,6 +844,14 @@ export class AcpExecutor implements PluginExecutorProvider { } #acceptUpdate(session: RetainedSession, update: SessionUpdate): void { + if (session.restoring) { + if (session.replay) { + session.replay.updates++; + if (update.sessionUpdate === 'user_message_chunk') session.replay.userChunks++; + } + return; + } + if (session.historyGap || session.lost) return; if (update.sessionUpdate === 'config_option_update') { // During our mutation, its response (or rollback response) is authoritative. if (!session.configuring && !session.lost) session.configOptions = update.configOptions; @@ -603,6 +872,16 @@ export class AcpExecutor implements PluginExecutorProvider { if (update.sessionUpdate === 'tool_call_update') this.#acceptTool(active, update); } + #assertWorkspaceAccess(session: RetainedSession, sessionId: string): void { + this.#assertSession(session, sessionId); + if (session.restoring || session.historyGap || session.lost) { + throw new AcpRuntimeError( + 'ACP filesystem access is unavailable while Session history is untrusted', + 'acp_history_gap', + ); + } + } + #acceptTool(active: ActivePrompt, update: ToolCall | ToolCallUpdate): void { const snapshot = active.tools.get(update.toolCallId) ?? { id: update.toolCallId, @@ -715,6 +994,14 @@ export class AcpExecutor implements PluginExecutorProvider { } } +function currentAcpModel(options: readonly SessionConfigOption[]): string | undefined { + const option = options.find( + (candidate) => + candidate.type === 'select' && (candidate.id === 'model' || candidate.category === 'model'), + ); + return option?.type === 'select' ? option.currentValue : undefined; +} + function toolTextContent(content: readonly ToolCallContent[]): string { return content .flatMap((item) => @@ -760,16 +1047,94 @@ function pluginStateStore( `acp/${executorId}/${createHash('sha256').update(conversationKey).digest('hex')}`; return { async has(conversationKey) { - const value = (await storage.get<{ version?: unknown; cwd?: unknown }>(key(conversationKey))) - .value; - return value?.version === 1; + return (await storage.get(key(conversationKey))).value !== undefined; }, async mark(conversationKey, cwd) { await storage.set(key(conversationKey), { version: 1, cwd }); }, + async read(conversationKey) { + return (await storage.get(key(conversationKey))).value; + }, + async write(conversationKey, state) { + await storage.set(key(conversationKey), state); + }, }; } +type StoredContinuity = AcpContinuityRecord | 'legacy' | 'invalid' | undefined; + +function decodeContinuity(value: unknown): StoredContinuity { + if (value === undefined) return undefined; + if (!value || typeof value !== 'object' || Array.isArray(value)) return 'invalid'; + const record = value as Record; + if (record.version === 1) return 'legacy'; + if ( + record.version !== 2 || + typeof record.cwd !== 'string' || + !isAbsolute(record.cwd) || + typeof record.binding !== 'string' || + !/^sha256:[a-f0-9]{64}$/u.test(record.binding) || + !Number.isSafeInteger(record.committedPrompts) || + (record.committedPrompts as number) < 0 || + !['reserved', 'established', 'prompt_pending', 'committed', 'history_gap'].includes( + String(record.phase), + ) || + (record.sessionId !== undefined && + (typeof record.sessionId !== 'string' || + !record.sessionId || + record.sessionId.length > 4096)) || + (record.pendingTurnId !== undefined && + (typeof record.pendingTurnId !== 'string' || !record.pendingTurnId)) || + (record.confirmedModel !== undefined && + (typeof record.confirmedModel !== 'string' || !record.confirmedModel)) || + (record.phase === 'reserved' && record.sessionId !== undefined) || + (record.phase !== 'reserved' && !record.sessionId) || + (record.phase === 'prompt_pending' && !record.pendingTurnId) || + (record.phase !== 'prompt_pending' && record.pendingTurnId !== undefined) || + (record.gapEvidence !== undefined && + (record.phase !== 'history_gap' || + !record.gapEvidence || + typeof record.gapEvidence !== 'object' || + !Number.isSafeInteger((record.gapEvidence as Record).replayedUpdates) || + !Number.isSafeInteger((record.gapEvidence as Record).replayedUserChunks) || + ((record.gapEvidence as Record).replayedUpdates as number) < 0 || + ((record.gapEvidence as Record).replayedUserChunks as number) < 0)) + ) + return 'invalid'; + return record as unknown as AcpContinuityRecord; +} + +async function launchBinding( + adapterId: string, + launch: AcpLaunchSpec, + signal: AbortSignal, +): Promise { + const paths = [launch.executable, ...(launch.requiredExecutables ?? [])]; + const files = []; + for (const path of paths) { + signal.throwIfAborted(); + const resolved = await checkedExecutable(path); + const hash = createHash('sha256'); + for await (const chunk of createReadStream(resolved)) { + signal.throwIfAborted(); + hash.update(chunk); + } + files.push({ path: resolved, digest: hash.digest('hex') }); + } + return `sha256:${createHash('sha256') + .update( + JSON.stringify({ + adapterId, + files, + args: launch.args ?? [], + cwd: launch.cwd ?? dirname(files[0]!.path), + platform: process.platform, + arch: process.arch, + }), + ) + .digest('hex')}`; +} + function validateAdapter(adapter: AcpAgentAdapter): AcpAgentAdapter { if ( !adapter || diff --git a/packages/core/src/__tests__/executor-catalog.test.ts b/packages/core/src/__tests__/executor-catalog.test.ts index 8695410bd1e..374de122851 100644 --- a/packages/core/src/__tests__/executor-catalog.test.ts +++ b/packages/core/src/__tests__/executor-catalog.test.ts @@ -48,6 +48,16 @@ test('structured capabilities survive wire normalization with immutable exact re assert.ok(Object.isFrozen(output.modelGroups![0]!.variants[0])); assert.notEqual(output.modelGroups, catalog.modelGroups); }); +for (const readiness of ['restorable', 'restoring', 'restore_failed', 'history_gap'] as const) + test(`restoration readiness survives executor catalog normalization: ${readiness}`, () => { + assert.equal( + normalizeCatalogEntry( + { ...catalog, readiness, models: [], modelGroups: undefined }, + 'external', + ).readiness, + readiness, + ); + }); for (const variants of [ [ { modelId: 'invented', level: 'high' }, diff --git a/packages/core/src/executor-catalog.ts b/packages/core/src/executor-catalog.ts index 538ac5053b2..5d821d7083e 100644 --- a/packages/core/src/executor-catalog.ts +++ b/packages/core/src/executor-catalog.ts @@ -35,7 +35,11 @@ export type ExecutorReadiness = | 'ready' | 'unavailable' | 'authentication_required' - | 'history_only'; + | 'history_only' + | 'restorable' + | 'restoring' + | 'restore_failed' + | 'history_gap'; export interface ExecutorModelChoice { readonly id: string; @@ -83,9 +87,16 @@ export function normalizeCatalogEntry( !isExecutorId(id) || value.id !== id || !isCatalogText(value.displayName) || - !['ready', 'unavailable', 'authentication_required', 'history_only'].includes( - value.readiness, - ) || + ![ + 'ready', + 'unavailable', + 'authentication_required', + 'history_only', + 'restorable', + 'restoring', + 'restore_failed', + 'history_gap', + ].includes(value.readiness) || !Array.isArray(value.models) || value.models.length > 256 || !Array.from(value.models).every( diff --git a/packages/runtime-host/src/protocol/index.ts b/packages/runtime-host/src/protocol/index.ts index 57c809310c1..2f41d5595c1 100644 --- a/packages/runtime-host/src/protocol/index.ts +++ b/packages/runtime-host/src/protocol/index.ts @@ -104,7 +104,8 @@ export const RUNTIME_HOST_REGISTRATION_SCHEMA_VERSION = 1 as const; export const RUNTIME_HOST_PROTOCOL_VERSION = 0 as const; // Increment when the same protocol version no longer guarantees safe Client-Host // interoperability. Mismatches are rejected before domain commands are admitted. -export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 197 as const; +export const RUNTIME_HOST_COMPATIBILITY_EPOCH = 198 as const; +// 198: Executor readiness exposes explicit restore, restore-failed and history-gap states. // 197: Queue reorder requests carry the expected queue revision. Epoch-196 // peers reject the required field or send an unfenced reorder request. // 196: `turn.start` carries an optional durable external-message origin; older diff --git a/packages/runtime/src/__tests__/plugin-executor-backend.test.ts b/packages/runtime/src/__tests__/plugin-executor-backend.test.ts index 048cb0e2265..f521703effa 100644 --- a/packages/runtime/src/__tests__/plugin-executor-backend.test.ts +++ b/packages/runtime/src/__tests__/plugin-executor-backend.test.ts @@ -23,7 +23,100 @@ import type { HostedFormSettlement } from '@maka/core/backend-types'; import type { SessionEvent } from '@maka/core/events'; import { PluginExecutorBackend } from '../plugin-executor-backend.js'; import { Context } from '../plugin-kernel.js'; -import { PluginExecutorService } from '../plugin-executor-service.js'; +import { PluginExecutorService, type PluginExecutorResult } from '../plugin-executor-service.js'; + +for (const result of [ + { status: 'completed', text: 'done' }, + { status: 'cancelled', providerStopReason: 'cancelled' }, + { status: 'failed', message: 'Agent refused the request' }, +] satisfies PluginExecutorResult[]) { + for (const detached of [false, true]) { + test(`Plugin acknowledgement for ${result.status} requires terminal consumption (detached: ${detached})`, async () => { + const root = new Context(); + const service = new PluginExecutorService(root); + const acknowledgements: string[] = []; + const abandonments: string[] = []; + root + .extend({ + maka: { rootId: 'profile', packageId: 'fixture', entryId: 'provider', generation: 1 }, + }) + .executors.register({ + id: 'remote', + execute: async () => result, + acknowledgeExecution: async (conversationKey, turnId) => { + acknowledgements.push(`${conversationKey}/${turnId}`); + }, + abandonExecution: async (conversationKey, turnId) => { + abandonments.push(`${conversationKey}/${turnId}`); + }, + }); + const backend = new PluginExecutorBackend({ + sessionId: 'session-a', + cwd: '/workspace', + binding: service.bind('session-a', 'remote'), + }); + try { + const iterator = backend.send({ turnId: 'turn-a', text: 'task' })[Symbol.asyncIterator](); + assert.equal( + (await iterator.next()).value?.type, + result.status === 'completed' + ? 'text_complete' + : result.status === 'cancelled' + ? 'abort' + : 'error', + ); + assert.deepEqual(acknowledgements, []); + assert.equal((await iterator.next()).value?.type, 'complete'); + assert.deepEqual( + acknowledgements, + [], + 'terminal delivery is not itself an acknowledgement', + ); + if (detached) { + await iterator.return?.(); + assert.deepEqual( + acknowledgements, + [], + 'an unconsumed terminal event cannot be acknowledged', + ); + assert.deepEqual(abandonments, ['session-a/turn-a']); + } else { + assert.equal((await iterator.next()).done, true); + assert.deepEqual(acknowledgements, ['session-a/turn-a']); + assert.deepEqual(abandonments, []); + } + } finally { + await backend.dispose(); + await root.fiber.dispose(); + } + }); + } +} + +test('failed Plugin acknowledgement abandons the settled external execution', async () => { + const { root, binding } = fixture(async () => ({ status: 'completed', text: 'done' })); + const abandoned: string[] = []; + const backend = new PluginExecutorBackend({ + sessionId: 'session-a', + cwd: '/workspace', + binding: { + ...binding, + acknowledgeExecution: async () => { + throw new Error('checkpoint unavailable'); + }, + abandonExecution: async (conversationKey, turnId) => { + abandoned.push(`${conversationKey}/${turnId}`); + }, + }, + }); + try { + await collect(backend.send({ turnId: 'turn-a', text: 'task' })); + assert.deepEqual(abandoned, ['session-a/turn-a']); + } finally { + await backend.dispose(); + await root.fiber.dispose(); + } +}); test('executor backend converts plugin output and result to ordinary Session events', async () => { const { root, binding } = fixture(async (request, context) => { diff --git a/packages/runtime/src/__tests__/plugin-executor-service.test.ts b/packages/runtime/src/__tests__/plugin-executor-service.test.ts index 1bee62870a4..aa37df2d481 100644 --- a/packages/runtime/src/__tests__/plugin-executor-service.test.ts +++ b/packages/runtime/src/__tests__/plugin-executor-service.test.ts @@ -155,6 +155,37 @@ test('executor bindings pin one provider generation', async () => { await root.fiber.dispose(); }); +test('executor acknowledgement is bound to one Session and provider generation', async () => { + const root = new Context(); + const service = new PluginExecutorService(root); + const owner = plugin(root, 'profile', 'provider', 1); + const acknowledged: string[] = []; + const abandoned: string[] = []; + const dispose = owner.executors.register({ + id: 'remote', + execute: async () => ({ status: 'completed', text: 'ok' }), + acknowledgeExecution: async (conversationKey, turnId) => { + acknowledged.push(`${conversationKey}:${turnId}`); + }, + abandonExecution: async (conversationKey, turnId) => { + abandoned.push(`${conversationKey}:${turnId}`); + }, + }); + const binding = service.bind('session-a', 'remote'); + await binding.acknowledgeExecution!('session-a', 'turn-a'); + await assert.rejects(binding.acknowledgeExecution!('session-b', 'turn-b')); + await binding.abandonExecution!('session-a', 'turn-b'); + await assert.rejects(binding.abandonExecution!('session-b', 'turn-c')); + assert.deepEqual(acknowledged, ['session-a:turn-a']); + assert.deepEqual(abandoned, ['session-a:turn-b']); + await dispose(); + await assert.rejects(binding.acknowledgeExecution!('session-a', 'turn-c')); + await assert.rejects(binding.abandonExecution!('session-a', 'turn-c')); + assert.deepEqual(acknowledged, ['session-a:turn-a']); + assert.deepEqual(abandoned, ['session-a:turn-b']); + await root.fiber.dispose(); +}); + test('executor completion after caller cancellation is normalized to cancelled', async () => { const root = new Context(); const service = new PluginExecutorService(root); diff --git a/packages/runtime/src/plugin-executor-backend.ts b/packages/runtime/src/plugin-executor-backend.ts index e4032175fce..81bbed40212 100644 --- a/packages/runtime/src/plugin-executor-backend.ts +++ b/packages/runtime/src/plugin-executor-backend.ts @@ -91,18 +91,39 @@ export class PluginExecutorBackend implements AgentBackend { const producer = this.#produce(input, messageId, abort.signal, queue).finally(() => queue.close(), ); - const active: ActiveExecution = { abort, settled: producer }; + const active: ActiveExecution = { abort, settled: producer.then(() => undefined) }; this.#active.add(active); + let acknowledged = false; try { for await (const event of queue) { yield event; queue.ackConsumed(); } - await producer; + const returnedResult = await producer; + if (returnedResult) { + // The Runtime Kernel requests the next item only after onSessionEvent + // resolves. Reaching this point means its terminal event was accepted. + // The Plugin decides whether its external execution actually settled; + // uncertain execution or a failed checkpoint keeps the pending marker. + if (this.#binding.acknowledgeExecution) + try { + await this.#binding.acknowledgeExecution(this.sessionId, input.turnId); + acknowledged = true; + } catch { + /* The provider abandonment below makes the uncertainty explicit. */ + } + } } finally { queue.noteConsumerDetached(); abort.abort(new Error('Plugin executor event consumer detached')); - await producer.catch(() => undefined); + const returnedResult = await producer.catch(() => false); + if ( + returnedResult && + this.#binding.acknowledgeExecution && + !acknowledged && + this.#binding.abandonExecution + ) + await this.#binding.abandonExecution(this.sessionId, input.turnId).catch(() => undefined); this.#active.delete(active); } } @@ -132,7 +153,7 @@ export class PluginExecutorBackend implements AgentBackend { messageId: string, signal: AbortSignal, queue: AsyncEventQueue, - ): Promise { + ): Promise { const turnId = input.turnId; let thinkingText = ''; const toolUseIds = new Map(); @@ -191,7 +212,7 @@ export class PluginExecutorBackend implements AgentBackend { false, queue, ); - return; + return false; } if (result === undefined) { this.#publishFailure( @@ -201,9 +222,10 @@ export class PluginExecutorBackend implements AgentBackend { false, queue, ); - return; + return false; } this.#publishResult(turnId, messageId, result, queue); + return true; } async #requestPermission( diff --git a/packages/runtime/src/plugin-executor-service.ts b/packages/runtime/src/plugin-executor-service.ts index 3a54668d22d..a8636d3658b 100644 --- a/packages/runtime/src/plugin-executor-service.ts +++ b/packages/runtime/src/plugin-executor-service.ts @@ -175,6 +175,10 @@ export interface PluginExecutorProvider { signal: AbortSignal, ): Promise; inspectConversation?(input: PluginExecutorConversationInput): Promise; + /** Confirm terminal consumption for any result; the provider decides whether it can checkpoint. */ + acknowledgeExecution?(conversationKey: string, turnId: string): Promise; + /** Mark a settled result uncertain when its terminal event was not durably consumed. */ + abandonExecution?(conversationKey: string, turnId: string): Promise; execute( request: Readonly, context: PluginExecutorContext, @@ -203,6 +207,8 @@ export interface PluginExecutorBinding { request: PluginExecutorRequest, options?: PluginExecutorExecutionOptions, ): Promise; + acknowledgeExecution?(conversationKey: string, turnId: string): Promise; + abandonExecution?(conversationKey: string, turnId: string): Promise; } export interface PluginExecutorServiceOptions { @@ -273,6 +279,12 @@ export class PluginExecutorService extends Service { ...(provider.inspectConversation ? { inspectConversation: provider.inspectConversation.bind(provider) } : {}), + ...(provider.acknowledgeExecution + ? { acknowledgeExecution: provider.acknowledgeExecution.bind(provider) } + : {}), + ...(provider.abandonExecution + ? { abandonExecution: provider.abandonExecution.bind(provider) } + : {}), }); const entry: RegisteredExecutor = { ...identity, @@ -443,6 +455,29 @@ export class PluginExecutorService extends Service { } return this.executeEntry(entry, request, options); }, + ...(entry.provider.acknowledgeExecution + ? { + acknowledgeExecution: (conversationKey: string, turnId: string) => + conversationKey !== sessionId + ? Promise.reject(new Error('Executor binding cannot cross Session scope')) + : this.withActiveOperation(entry, { conversationKey }, async (signal) => { + signal.throwIfAborted(); + if (entry.retired) throw new ExecutorRetiredAbort(entry.provider.id); + await entry.provider.acknowledgeExecution!(conversationKey, turnId); + }), + } + : {}), + ...(entry.provider.abandonExecution + ? { + abandonExecution: (conversationKey: string, turnId: string) => + conversationKey !== sessionId + ? Promise.reject(new Error('Executor binding cannot cross Session scope')) + : this.withActiveOperation(entry, { conversationKey }, async () => { + if (entry.retired) throw new ExecutorRetiredAbort(entry.provider.id); + await entry.provider.abandonExecution!(conversationKey, turnId); + }), + } + : {}), }); } @@ -564,6 +599,15 @@ function validateProvider(provider: PluginExecutorProvider): void { if (typeof provider.execute !== 'function') { throw new TypeError(`Executor implementation is invalid: ${provider.id}`); } + if ( + provider.acknowledgeExecution !== undefined && + typeof provider.acknowledgeExecution !== 'function' + ) { + throw new TypeError(`Executor acknowledgement is invalid: ${provider.id}`); + } + if (provider.abandonExecution !== undefined && typeof provider.abandonExecution !== 'function') { + throw new TypeError(`Executor abandonment is invalid: ${provider.id}`); + } if ( provider.displayName !== undefined && (typeof provider.displayName !== 'string' || !provider.displayName.trim()) diff --git a/packages/ui/src/__tests__/executor-model-picker.test.tsx b/packages/ui/src/__tests__/executor-model-picker.test.tsx index 1ee7fe7fad0..c143f8cd23f 100644 --- a/packages/ui/src/__tests__/executor-model-picker.test.tsx +++ b/packages/ui/src/__tests__/executor-model-picker.test.tsx @@ -325,14 +325,17 @@ test('executor choice shares the native searchable list and exposes every extern } }); -test('history-only state keeps the executor fixed and offers a new task', async () => { +for (const [readiness, notice] of [ + ['history_only', '歷史仍可閱讀'], + ['history_gap', '進度可能超出已儲存歷史'], +] as const) test(`${readiness} keeps the executor fixed and offers a new task`, async () => { const dom = installTranscriptDom(); let newTasks = 0; try { await dom.render( assert.fail('History cannot change executor')} @@ -344,7 +347,7 @@ test('history-only state keeps the executor fixed and offers a new task', async /> , ); - assert.ok(dom.document.body.textContent?.includes('歷史仍可閱讀')); + assert.ok(dom.document.body.textContent?.includes(notice)); const executorTrigger = dom.document.querySelector('.maka-executor-selector'); assert.notEqual(executorTrigger?.getAttribute('aria-disabled'), 'true'); const button = [...dom.document.querySelectorAll('button')].find( @@ -358,6 +361,56 @@ test('history-only state keeps the executor fixed and offers a new task', async } }); +test('restorable task offers an explicit restore action without selecting a model', async () => { + const dom = installTranscriptDom(); + let restores = 0; + try { + await dom.render( + + assert.fail('Restoration must not commit a model choice')} + onRestore={async () => { restores++; }} + onSetup={() => {}} + onRetry={() => {}} + onNewTask={() => assert.fail('The Session is restorable')} + /> + , + ); + assert.match(dom.document.body.textContent ?? '', /Restore it before continuing/u); + const button = [...dom.document.querySelectorAll('button')].find( + candidate => candidate.textContent === 'Restore Session', + ); + assert.ok(button); + await act(async () => button.dispatchEvent(new dom.window.Event('click', { bubbles: true }))); + assert.equal(restores, 1); + } finally { await dom.cleanup(); } +}); + +test('restoring task shows progress without offering another action', async () => { + const dom = installTranscriptDom(); + try { + await dom.render( + + assert.fail('Restoration is pending')} + onSetup={() => assert.fail('Setup is unavailable while restoring')} + onRetry={() => assert.fail('Retry is unavailable while restoring')} + onNewTask={() => assert.fail('New Task is unavailable while restoring')} + /> + , + ); + assert.match(dom.document.body.textContent ?? '', /Restoring the external Session/u); + assert.equal([...dom.document.querySelectorAll('button')].some(button => + ['Manage agents', 'Restore Session', 'New Task'].includes(button.textContent ?? '')), false); + } finally { await dom.cleanup(); } +}); + test('unavailable executors can be inspected but never committed', async () => { const dom = installTranscriptDom(); const selections: unknown[] = []; diff --git a/packages/ui/src/executor-model-picker.tsx b/packages/ui/src/executor-model-picker.tsx index 66592815369..7b66869718e 100644 --- a/packages/ui/src/executor-model-picker.tsx +++ b/packages/ui/src/executor-model-picker.tsx @@ -53,6 +53,7 @@ export interface ExecutorModelPickerProps { onSetup(): void; onRetry(): void; onNewTask(): void; + onRestore?(): void | Promise; } interface ExecutorCopy { @@ -65,6 +66,11 @@ interface ExecutorCopy { unavailable: string; authentication_required: string; history_only: string; + restorable: string; + restoring: string; + restore_failed: string; + history_gap: string; + restore: string; fixed: string; attachments: string; retry: string; @@ -84,6 +90,11 @@ const EXECUTOR_COPY = { authentication_required: 'Sign in from External Agents settings.', history_only: 'The external process was lost. History is readable; start a new task to continue.', + restorable: 'The external Session can be restored. Restore it before continuing.', + restoring: 'Restoring the external Session…', + restore_failed: 'Restoration failed. History is readable; retry or start a new task.', + history_gap: 'The Agent may be ahead of saved history. Start a new task to avoid an incomplete transcript.', + restore: 'Restore Session', fixed: 'Start a new task to switch executors.', attachments: 'This executor does not support these attachments. Remove them or select Maka. Your draft is preserved.', @@ -101,6 +112,11 @@ const EXECUTOR_COPY = { unavailable: '当前不可用，请检查设置后重试。', authentication_required: '需要登录，请前往外部 Agent 设置。', history_only: '外部进程已丢失。历史仍可阅读，请新建任务继续。', + restorable: '外部会话可以恢复。请先恢复，再继续对话。', + restoring: '正在恢复外部会话…', + restore_failed: '恢复失败，历史仍可阅读。请重试或新建任务。', + history_gap: 'Agent 的进度可能超出已保存历史。请新建任务，避免记录不完整。', + restore: '恢复会话', fixed: '切换执行者需要新建任务。', attachments: '此执行者不支持这些附件。请移除附件或选择 Maka，草稿会保留。', retry: '重试', @@ -117,6 +133,11 @@ const EXECUTOR_COPY = { unavailable: '目前無法使用，請檢查設定後重試。', authentication_required: '需要登入，請前往外部 Agent 設定。', history_only: '外部程序已遺失。歷史仍可閱讀，請建立新任務繼續。', + restorable: '外部會話可以恢復。請先恢復，再繼續對話。', + restoring: '正在恢復外部會話…', + restore_failed: '恢復失敗，歷史仍可閱讀。請重試或建立新任務。', + history_gap: 'Agent 的進度可能超出已儲存歷史。請建立新任務，避免記錄不完整。', + restore: '恢復會話', fixed: '切換執行者需要建立新任務。', attachments: '此執行者不支援這些附件。請移除附件或選擇 Maka，草稿會保留。', retry: '重試', @@ -189,6 +210,12 @@ export function ExecutorModelPicker(props: ExecutorModelPickerProps) { setOpen(false); props.onNewTask(); }; + const restore = () => { + if (props.onRestore && !selecting) { + setSelecting(true); + void Promise.resolve(props.onRestore()).catch(() => setSelectionFailed(true)).finally(() => setSelecting(false)); + } + }; const lockedTo = props.fixed ? props.selection?.executorId ?? NATIVE : undefined; return ( <> @@ -284,9 +311,11 @@ export function ExecutorModelPicker(props: ExecutorModelPickerProps) { ) : (

{browsed ? copy[browsed.readiness] : copy.unavailable}

- {browsed?.readiness === 'history_only' ? ( + {browsed?.readiness === 'restorable' || browsed?.readiness === 'restore_failed' ? ( +
@@ -310,9 +339,11 @@ export function ExecutorModelPicker(props: ExecutorModelPickerProps) { {(selectedUnavailable || props.error) && ( {selected && selected.readiness !== 'ready' ? copy[selected.readiness] : copy.unavailable} - {selected?.readiness === 'history_only' ? ( + {selected?.readiness === 'restorable' || selected?.readiness === 'restore_failed' ? ( +