diff --git a/.project-docs/30-worklog/tasks/20260823-pi-extension-host-7e4c91a2.md b/.project-docs/30-worklog/tasks/20260823-pi-extension-host-7e4c91a2.md index cf89c36..d558588 100644 --- a/.project-docs/30-worklog/tasks/20260823-pi-extension-host-7e4c91a2.md +++ b/.project-docs/30-worklog/tasks/20260823-pi-extension-host-7e4c91a2.md @@ -8,7 +8,7 @@ - Worktree: D:\Datas\OthersProjects\makelore-pi-extension-host-7e4c91a2 - Base commit: 47159b3cbdf06ea66db9a4425e0a9c13bcd73f7f - Owner: codex -- Status: Completed — pending planner review +- Status: Completed — correction candidate pending planner re-review ## Scope @@ -99,6 +99,18 @@ - The Pi 0.84.2 real-child smoke now loads the materialized managed extension through Electron Node and confirms RPC readiness without extension-load or stdout-protocol failure. No real external Provider was contacted. +- Planner review of initial candidate `3861c32` returned **Needs Fix / Not + Done** with four deterministic supported-path findings. The correction + closes each finding without changing the PI DAG: recover clears the old run + before worker replacement and replacement only continues a run still active + in the pool; interaction response ownership is atomic across double-submit + and generation invalidation; Host close stops new connections, waits tracked + request handlers, and terminates stale lease waiters; UI diagnostics retain + only the newest 256 bounded entries. +- Added exact regressions for active-run recover, delayed interaction send plus + generation cancellation, holder+waiter Host shutdown, and 1,000 unknown UI + events. All four planner reproductions now pass locally; PI-070 remains the + sole frontier until planner re-review marks the correction Done. - Real Provider validation remains **Explicitly Waived / Accepted Risk** with `realTurnVerified=false`; provider concurrency, credential isolation, and protocol compatibility are not Pass. macOS x64/arm64 remains deferred to @@ -123,6 +135,22 @@ - `corepack pnpm test` — 199 files / 2192 tests passed before the final bundle integration test was added; that new test then passed independently, making all 200 current test files green across the two recorded runs. +- Planner correction focused suite — 4 files / 9 tests passed, including all + four deterministic review regressions. +- Post-correction all-Pi run — 19 files / 91 tests passed; the existing + `pi-conversation-runtime` Windows temporary JSON `rename EPERM` occurred in + the aggregate run, then the same test passed 1/1 in isolation. No Pi behavior + assertion failed. +- Post-correction full suite — 199 files / 2194 tests passed; only that same + pre-existing Windows temporary-file `rename EPERM` prevented a single green + aggregate result. The affected runtime test passed immediately in isolation, + so all 200 files / 2195 tests are green across the recorded aggregate plus + isolated rerun. The failure is not in a PI-070-owned path and no retry or + filesystem workaround was added. +- Post-correction `corepack pnpm run typecheck` and `lint:check` — passed; lint + remains 0 errors with the same 6 unrelated warnings. +- Post-correction `corepack pnpm run build:vite` — passed for all four build + targets with unchanged existing warnings. - `corepack pnpm run build:vite` — passed for Renderer, Electron Main, Preload, and release utility output. Existing dynamic-import and chunk-size warnings remain unchanged. diff --git a/electron/coding-runtime/pi/extension-host.ts b/electron/coding-runtime/pi/extension-host.ts index ee2617b..83a713f 100644 --- a/electron/coding-runtime/pi/extension-host.ts +++ b/electron/coding-runtime/pi/extension-host.ts @@ -63,9 +63,11 @@ export class PiManagedExtensionHost { private readonly leases: PiProjectWriteLeaseCoordinator; private readonly registrations = new Map(); private readonly runBindings = new Map(); + private readonly requestFlights = new Set>(); private server: Server | null = null; private bridgeUrl: string | null = null; private startFlight: Promise | null = null; + private closing = false; constructor(leases = new PiProjectWriteLeaseCoordinator()) { this.leases = leases; @@ -131,16 +133,20 @@ export class PiManagedExtensionHost { } async close(): Promise { - for (const record of [...this.registrations.values()]) this.disposeRecord(record); - this.runBindings.clear(); const server = this.server; this.server = null; this.bridgeUrl = null; this.startFlight = null; - if (!server) return; - await new Promise((resolve, reject) => { + this.closing = true; + const closed = server ? new Promise((resolve, reject) => { server.close((error) => error ? reject(error) : resolve()); - }); + }) : Promise.resolve(); + for (const record of [...this.registrations.values()]) this.disposeRecord(record); + this.runBindings.clear(); + await Promise.allSettled([...this.requestFlights]); + server?.closeIdleConnections(); + await closed; + this.closing = false; } private start(): Promise { @@ -148,7 +154,10 @@ export class PiManagedExtensionHost { if (this.startFlight) return this.startFlight; this.startFlight = new Promise((resolve, reject) => { const server = createServer((request, response) => { - void this.handle(request, response); + const flight = this.handle(request, response).finally(() => { + this.requestFlights.delete(flight); + }); + this.requestFlights.add(flight); }); server.once('error', reject); server.listen(0, '127.0.0.1', () => { @@ -159,6 +168,7 @@ export class PiManagedExtensionHost { return; } this.server = server; + this.closing = false; this.bridgeUrl = `http://127.0.0.1:${address.port}/v1/worker`; resolve(this.bridgeUrl); }); @@ -226,6 +236,7 @@ export class PiManagedExtensionHost { ); if (record.runId !== value.runId || this.registrations.get(token) !== record) { lease.release(); + this.respond(response, 409, { error: 'Worker run identity is stale' }); return; } record.leases.set(value.resourceId, lease); @@ -266,7 +277,10 @@ export class PiManagedExtensionHost { private respond(response: ServerResponse, status: number, body: Record): void { if (response.writableEnded) return; - response.writeHead(status, { 'content-type': 'application/json; charset=utf-8' }); + response.writeHead(status, { + 'content-type': 'application/json; charset=utf-8', + ...(this.closing ? { connection: 'close' } : {}), + }); response.end(JSON.stringify(body)); } diff --git a/electron/coding-runtime/pi/extension-ui-projector.ts b/electron/coding-runtime/pi/extension-ui-projector.ts index 359fe9c..418c3a9 100644 --- a/electron/coding-runtime/pi/extension-ui-projector.ts +++ b/electron/coding-runtime/pi/extension-ui-projector.ts @@ -5,6 +5,7 @@ const MAX_TITLE_LENGTH = 256; const MAX_EDITOR_TEXT_LENGTH = 64 * 1024; const MAX_WIDGET_LINES = 32; const MAX_WIDGET_LINE_LENGTH = 512; +const MAX_DIAGNOSTICS = 256; export type PiExtensionUiProjection = | { kind: 'notify'; conversationId: string; message: string; level: 'info' | 'warning' | 'error' } @@ -136,7 +137,7 @@ export class PiExtensionUiProjector { } if (event.method === 'set_editor_text' && typeof event.text === 'string') { if (this.getDraftRevision(conversationId) !== run.draftRevision) { - this.diagnostics.push({ method: event.method, reason: 'stale-draft-revision' }); + this.recordDiagnostic({ method: event.method, reason: 'stale-draft-revision' }); return null; } return { @@ -151,12 +152,17 @@ export class PiExtensionUiProjector { } private unsupported(method: string): null { - this.diagnostics.push({ method: bounded(method, 128), reason: 'unsupported-ui-method' }); + this.recordDiagnostic({ method: bounded(method, 128), reason: 'unsupported-ui-method' }); return null; } private invalid(method: string): null { - this.diagnostics.push({ method: bounded(method, 128), reason: 'invalid-ui-payload' }); + this.recordDiagnostic({ method: bounded(method, 128), reason: 'invalid-ui-payload' }); return null; } + + private recordDiagnostic(diagnostic: PiExtensionUiDiagnostic): void { + this.diagnostics.push(diagnostic); + if (this.diagnostics.length > MAX_DIAGNOSTICS) this.diagnostics.shift(); + } } diff --git a/electron/coding-runtime/pi/interaction.ts b/electron/coding-runtime/pi/interaction.ts index 9ec7c68..b651dd6 100644 --- a/electron/coding-runtime/pi/interaction.ts +++ b/electron/coding-runtime/pi/interaction.ts @@ -13,6 +13,7 @@ interface StoredInteraction { interaction: ConversationInteraction; generation: number; labels: Map; + phase: 'pending' | 'responding'; untrack(): void; } @@ -85,6 +86,7 @@ export class PiInteractionStore { interaction, generation, labels, + phase: 'pending', untrack: () => undefined, }; stored.untrack = this.transport.trackGenerationResource({ @@ -102,6 +104,7 @@ export class PiInteractionStore { async respond(conversationId: string, response: PiInteractionResponse): Promise { const stored = this.pending.get(this.key(conversationId, response.interactionId)); if (!stored) throw new Error('Pi interaction is not pending'); + if (stored.phase === 'responding') throw new Error('Pi interaction response is already in progress'); const state = this.transport.getState(conversationId); const active = this.transport.getActiveRun(conversationId); if (state?.generation !== stored.generation @@ -126,7 +129,18 @@ export class PiInteractionStore { } else { throw new Error('Pi interaction response does not match its kind'); } - await this.transport.send(conversationId, command); + stored.phase = 'responding'; + try { + await this.transport.send(conversationId, command); + } catch (error) { + if (this.pending.get(this.key(conversationId, stored.interaction.id)) === stored) { + stored.phase = 'pending'; + } + throw error; + } + if (this.pending.get(this.key(conversationId, stored.interaction.id)) !== stored) { + throw new Error('Pi interaction belongs to a stale worker run'); + } return this.finish(stored, 'cancelled' in response ? 'cancelled' : stored.interaction.kind === 'confirm' && 'confirmed' in response && !response.confirmed diff --git a/electron/coding-runtime/pi/runtime.ts b/electron/coding-runtime/pi/runtime.ts index 6579592..8573870 100644 --- a/electron/coding-runtime/pi/runtime.ts +++ b/electron/coding-runtime/pi/runtime.ts @@ -798,6 +798,13 @@ export class PiConversationRuntime implements CodingConversationRuntime { async recover(conversationId: string): Promise { await this.waitForProjection(conversationId); + const before = this.snapshot(conversationId); + const generation = this.pool.getState(conversationId)?.generation; + if (before.run.runId && generation) { + await this.interactions.cancelRun(conversationId, before.run.runId, true); + this.extensionUi.endRun(conversationId, before.run.runId); + await this.extensionHost?.clearRun(conversationId, generation, before.run.runId); + } const state = await this.pool.recover(conversationId); await this.requestHydration(conversationId, state, false); this.settleRecoveredRun(conversationId); @@ -1003,8 +1010,12 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.replaceWorkerGeneration(event.conversationId, event.state, true); this.resetProjector(event.conversationId); const runId = this.states.get(event.conversationId)?.snapshot.run.runId; - if (runId) this.extensionUi.replaceGeneration(event.conversationId, event.generation, runId); - if (this.extensionHost && runId) { + const activeRun = this.pool.getActiveRun(event.conversationId); + const continuesActiveRun = Boolean(runId && activeRun?.runId === runId); + if (runId && continuesActiveRun) { + this.extensionUi.replaceGeneration(event.conversationId, event.generation, runId); + } + if (this.extensionHost && runId && continuesActiveRun) { void this.extensionHost.bindRun(event.conversationId, event.generation, runId).catch((error) => { this.recordProjectionFailure(event.conversationId, event.generation, error); }); diff --git a/tests/unit/pi-conversation-runtime.test.ts b/tests/unit/pi-conversation-runtime.test.ts index 73ff7eb..38ffed1 100644 --- a/tests/unit/pi-conversation-runtime.test.ts +++ b/tests/unit/pi-conversation-runtime.test.ts @@ -14,6 +14,7 @@ import { import { PiConversationRuntime } from '../../electron/coding-runtime/pi/runtime'; import { PiSessionProjectionError } from '../../electron/coding-runtime/pi/session-projector'; import { PiSessionRegistry } from '../../electron/coding-runtime/pi/session-registry'; +import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host'; import { PiWorkerPool, type PiConversationWorker, @@ -114,6 +115,18 @@ class RuntimeFakeWorker implements PiConversationWorker { } } +class TrackingExtensionHost extends PiManagedExtensionHost { + readonly runs = new Map(); + + override async bindRun(conversationId: string, generation: number, runId: string): Promise { + this.runs.set(conversationId, { generation, runId }); + } + + override async clearRun(conversationId: string, _generation: number, runId?: string): Promise { + if (!runId || this.runs.get(conversationId)?.runId === runId) this.runs.delete(conversationId); + } +} + afterEach(async () => { await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); }); @@ -189,10 +202,12 @@ describe('Pi Conversation runtime', () => { }; }, }); + const trackingHost = new TrackingExtensionHost(); const runtime = new PiConversationRuntime({ pool, registry: new PiSessionRegistry({ projectStore }), createId: (kind) => `${kind}-fixed`, + extensionHost: trackingHost, resolveModel: async (candidate) => { if (candidate.accountId !== 'account-b' || candidate.modelId !== 'model-b') { throw new Error('model unavailable'); @@ -483,6 +498,17 @@ describe('Pi Conversation runtime', () => { await expect(runtime.getSnapshot(forkTarget.id)).rejects.toMatchObject({ publicError: { code: 'CODING_CONVERSATION_NOT_FOUND' }, }); + await runtime.prompt({ + clientRequestId: 'request-active-recover', + conversationId: left.id, + mode: 'prompt', + text: 'Recover this active run', + attachments: [], + }); + expect(trackingHost.runs.get(left.id)?.runId).toBe('run-fixed'); + await runtime.recover(left.id); + expect(trackingHost.runs.has(left.id)).toBe(false); + expect((await runtime.getSnapshot(left.id)).run).toEqual({ status: 'idle' }); unsubscribe(); }); }); diff --git a/tests/unit/pi-extension-host.test.ts b/tests/unit/pi-extension-host.test.ts index 2a6f619..e27fc93 100644 --- a/tests/unit/pi-extension-host.test.ts +++ b/tests/unit/pi-extension-host.test.ts @@ -5,6 +5,7 @@ import { tmpdir } from 'node:os'; import path from 'node:path'; import { afterEach, describe, expect, it } from 'vitest'; import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host'; +import { PiProjectWriteLeaseCoordinator } from '../../electron/coding-runtime/pi/write-lease'; const roots: string[] = []; const hosts: PiManagedExtensionHost[] = []; @@ -29,6 +30,40 @@ async function post( } describe('managed Pi extension bridge', () => { + it('finishes a queued HTTP request while closing a holder and waiter', async () => { + const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-close-')); + roots.push(root); + const leases = new PiProjectWriteLeaseCoordinator(); + const host = new PiManagedExtensionHost(leases); + hosts.push(host); + const holder = await host.registerWorker({ + conversationId: 'conversation-holder', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + const waiter = await host.registerWorker({ + conversationId: 'conversation-waiter', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + await Promise.all([ + host.bindRun('conversation-holder', 1, 'run-holder'), + host.bindRun('conversation-waiter', 1, 'run-waiter'), + ]); + const held = await post(holder, { + action: 'lease.acquire', conversationId: 'conversation-holder', workerGeneration: 1, + runId: 'run-holder', resourceId: 'held-tool', + }); + expect(held.status).toBe(200); + const waiting = post(waiter, { + action: 'lease.acquire', conversationId: 'conversation-waiter', workerGeneration: 1, + runId: 'run-waiter', resourceId: 'waiting-tool', + }); + await expect.poll(() => leases.waitingCount('project-a')).toBe(1); + + await expect(Promise.race([ + host.close().then(() => 'closed'), + new Promise((resolve) => setTimeout(() => resolve('timeout'), 500)), + ])).resolves.toBe('closed'); + await expect(waiting).resolves.toMatchObject({ status: 409 }); + }); + it('materializes the active run into a replacement generation before spawn', async () => { const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-rebuild-')); roots.push(root); diff --git a/tests/unit/pi-extension-ui-projector.test.ts b/tests/unit/pi-extension-ui-projector.test.ts index e45a027..bca79cd 100644 --- a/tests/unit/pi-extension-ui-projector.test.ts +++ b/tests/unit/pi-extension-ui-projector.test.ts @@ -40,5 +40,15 @@ describe('Pi extension UI projector', () => { expect(projector.getDiagnostics()).toEqual([{ method: 'setWidget', reason: 'unsupported-ui-method', }]); + for (let index = 0; index < 1_000; index += 1) { + projector.project('conversation-a', 1, 'run-1', { + type: 'extension_ui_request', id: `unknown-${index}`, method: `unknown-${index}`, + }); + } + const diagnostics = projector.getDiagnostics(); + expect(diagnostics).toHaveLength(256); + expect(diagnostics.at(-1)).toEqual({ + method: 'unknown-999', reason: 'invalid-ui-payload', + }); }); }); diff --git a/tests/unit/pi-interaction.test.ts b/tests/unit/pi-interaction.test.ts index b9dba50..6fe3c09 100644 --- a/tests/unit/pi-interaction.test.ts +++ b/tests/unit/pi-interaction.test.ts @@ -75,4 +75,43 @@ describe('Pi interaction store', () => { expect(changes.map(({ status }) => status)).toEqual(['cancelled', 'cancelled']); expect(store.list()).toEqual([]); }); + + it('makes response ownership atomic across double submit and generation invalidation', async () => { + const sent: PiRpcCommand[] = []; + const changes: ConversationInteraction[] = []; + let cancelResource = () => undefined; + let releaseSend = () => undefined; + const sendGate = new Promise((resolve) => { releaseSend = resolve; }); + const store = new PiInteractionStore({ + getState: () => ({ + conversationId: 'conversation-a', workerId: 'worker-a', state: 'running', generation: 1, + session: { piSessionId: 'session-a', sessionKey: 'key-a' }, + }), + getActiveRun: () => ({ generation: 1, runId: 'run-1' }), + send: async (_conversationId, command) => { + sent.push(command); + await sendGate; + }, + trackGenerationResource: (input) => { + cancelResource = input.cancel; + return () => undefined; + }, + }, (interaction) => changes.push(interaction)); + store.open('conversation-a', 1, 'run-1', { + type: 'extension_ui_request', id: 'question-race', method: 'confirm', title: 'Continue?', + }); + + const first = store.respond('conversation-a', { + interactionId: 'question-race', confirmed: true, + }); + await expect(store.respond('conversation-a', { + interactionId: 'question-race', confirmed: false, + })).rejects.toThrow('already in progress'); + cancelResource(); + releaseSend(); + + await expect(first).rejects.toThrow('stale'); + expect(sent).toHaveLength(1); + expect(changes.map(({ status }) => status)).toEqual(['cancelled']); + }); });