From 3861c3286a554c9c0e1fc8209f6cd2a3ea337419 Mon Sep 17 00:00:00 2001 From: brother7 <7brother7@gmail.com> Date: Sun, 23 Aug 2026 09:39:12 +0800 Subject: [PATCH] feat(coding-runtime): add managed Pi extension host --- .../20260823-pi-extension-host-7e4c91a2.md | 190 +++++++++++ electron/coding-runtime/pi/extension-host.ts | 299 ++++++++++++++++++ .../pi/extension-ui-projector.ts | 162 ++++++++++ .../pi/extensions/makelore-runtime.ts | 100 ++++++ electron/coding-runtime/pi/interaction.ts | 177 +++++++++++ electron/coding-runtime/pi/rpc-client.ts | 13 + electron/coding-runtime/pi/runtime.ts | 176 ++++++++++- electron/coding-runtime/pi/worker-pool.ts | 13 + electron/coding-runtime/pi/worker-process.ts | 13 +- electron/coding-runtime/pi/write-lease.ts | 103 ++++++ tests/unit/pi-conversation-runtime.test.ts | 23 ++ tests/unit/pi-extension-bundle.test.ts | 95 ++++++ tests/unit/pi-extension-host.test.ts | 120 +++++++ tests/unit/pi-extension-ui-projector.test.ts | 44 +++ tests/unit/pi-interaction.test.ts | 78 +++++ tests/unit/pi-managed-worker-opener.test.ts | 13 + tests/unit/pi-rpc-foundation.test.ts | 3 +- tests/unit/pi-runtime-auth-recovery.test.ts | 2 + ...pi-worker-pool-process-integration.test.ts | 4 + tests/unit/pi-worker-pool.test.ts | 4 + tests/unit/pi-worker-process-real.test.ts | 15 + tests/unit/pi-write-lease.test.ts | 42 +++ 22 files changed, 1673 insertions(+), 16 deletions(-) create mode 100644 .project-docs/30-worklog/tasks/20260823-pi-extension-host-7e4c91a2.md create mode 100644 electron/coding-runtime/pi/extension-host.ts create mode 100644 electron/coding-runtime/pi/extension-ui-projector.ts create mode 100644 electron/coding-runtime/pi/extensions/makelore-runtime.ts create mode 100644 electron/coding-runtime/pi/interaction.ts create mode 100644 electron/coding-runtime/pi/write-lease.ts create mode 100644 tests/unit/pi-extension-bundle.test.ts create mode 100644 tests/unit/pi-extension-host.test.ts create mode 100644 tests/unit/pi-extension-ui-projector.test.ts create mode 100644 tests/unit/pi-interaction.test.ts create mode 100644 tests/unit/pi-write-lease.test.ts 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 new file mode 100644 index 0000000..cf89c36 --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260823-pi-extension-host-7e4c91a2.md @@ -0,0 +1,190 @@ +# Task: Implement PI-070 managed extension host interactions and write lease + +## Identity + +- Task ID: 20260823-pi-extension-host-7e4c91a2 +- Mode: Feature +- Branch: codex/20260823-pi-extension-host-7e4c91a2-pi-extension-host +- Worktree: D:\Datas\OthersProjects\makelore-pi-extension-host-7e4c91a2 +- Base commit: 47159b3cbdf06ea66db9a4425e0a9c13bcd73f7f +- Owner: codex +- Status: Completed — pending planner review + +## Scope + +- Implement `PI-070 — Managed extension host, interactions, and write lease` + from the `【架构】opencode2pi` Spec/ticket graph on cumulative baseline + `47159b3cbdf06ea66db9a4425e0a9c13bcd73f7f`. +- Own the Main-private versioned Makelore Pi extension entry, authenticated + worker bridge/registry, interaction lifecycle, project mutation write lease, + and the smallest PI-060 projector/runtime integration required to expose + safe product state. +- Add focused real-bridge, stale-token/generation, interaction lifecycle, and + lease concurrency/cancellation tests. Do not wire Renderer Host routes, + subagents, product browser/tools, or final packaging owned by later tickets. + +## Intent And Constraints + +- Follow planner task `20260822-pi-runtime-spec-b6e2c9a4`, ticket `PI-070`, and + Spec `LEASE-001` through `LEASE-006`, `EXT-001` through `EXT-006`, and section + 9.3 interaction semantics. +- Load exactly one explicit versioned Makelore extension bundle. Project/user + Pi extension discovery remains disabled; no generic plugin platform, + permission engine, compatibility layer, signature/hash framework, or remote + authorization scaffold is in scope. +- Electron Main owns bridge tokens, worker identity, project identity, + interactions, leases, and diagnostics. Every bridge request must match the + registered conversation, worker generation, run, and short-lived token; + disposed/recovered workers and old generations fail closed without affecting + the current worker. +- `ask_user` supports select/confirm/input/editor and correlates responses only + by the pending interaction ID. Worker abort/crash/dispose cancels pending + interactions. Unknown UI/widget/details are bounded diagnostics, never raw + product state or logs. +- Mutation write/edit/bash work uses one cancellable FIFO project lease; reads + and pure reasoning bypass it, different projects may mutate concurrently, + and complete/error/abort/timeout/crash/dispose always release ownership. +- Draft/editor suggestions are revision-conditional and must not overwrite a + newer user draft. Host routes and Renderer consumption remain for PI-100/ + PI-130; this ticket records the Main-side contract only. +- Real external Provider validation remains **Explicitly Waived / Accepted + Risk** with `realTurnVerified=false`; macOS x64/arm64 remains deferred to + mandatory PI-150 validation. Neither is a Pass. +- Do not create subagents. Keep changes surgical and rooted in this isolated + worktree. + +## Plan + +1. Verify the locked Pi 0.84.2 extension/tool interface and the existing + PI-040/050/060 seams; define the smallest product-owned bridge interface. +2. Implement the versioned managed extension bundle plus authenticated worker + registry/bridge and prove real bridge round trips and stale identity rejection. +3. Implement interaction storage/projection and the cancellable project write + lease, then wire generation/run cleanup into the runtime without widening + Host/Renderer interfaces. +4. Run focused tests, all Pi tests, typecheck, lint, build, and relevant + Electron/package checks; update this task record, pass documentation drift, + commit, and hand the exact implementation HEAD to the planner for review. + +## Outcome + +- Implemented a Main-private, versioned `makelore-runtime-v1.mjs` bundle and + changed Pi startup from `--no-tools` to an explicit built-in plus + `ask_user` allowlist. Discovery remains disabled with `--no-extensions`, + while the managed opener appends exactly one product-owned `--extension`. +- Added `PiManagedExtensionHost`, a loopback-only HTTP bridge with one random + bearer token per worker generation. Every request must match token, + Conversation, generation, and active run; worker invalidation/dispose makes + the token unusable and releases all owned or waiting resources. Active run + binding is inherited synchronously by a replacement generation before its + child process starts, closing the stale-rebuild race. +- Added the self-contained extension `ask_user` tool for select, confirm, + input, and editor requests. Pi's fire-and-forget + `extension_ui_response` records now use a dedicated one-way transport seam + rather than pretending to be request/response commands. +- Added a Host-side `PiInteractionStore` with exact interaction/option + correlation, current generation/run validation, terminal status projection, + and generation-scoped abort/crash/dispose cancellation. Runtime dialog + events pass through the store before entering product patches. +- Added a FIFO `PiProjectWriteLeaseCoordinator` and extension hooks that make + `write`, `edit`, and `bash` share one mutation lease per project. Reads and + reasoning bypass it; different projects proceed in parallel. Tool result, + agent end, session shutdown, abort, crash, recovery, dispose, and Host + shutdown all reach an idempotent release path; waiting bridge requests are + cancellable. +- Added safe projection for notify/status/registered widget/title/editor-text + events. Unknown widget/UI payloads become bounded method-only diagnostics; + editor text is emitted only when the current draft revision still equals the + run-start revision. +- 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. +- 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 + PI-150 and is not Pass. + +## Verification + +- `corepack pnpm run typecheck` — passed. +- `corepack pnpm run lint:check` — passed with 0 errors; the repository's 6 + pre-existing React warnings remain outside PI-070. +- Focused PI-070 plus predecessor regression suite — 9 files / 41 tests + passed, covering worker transport/pool/runtime, exact interaction response, + stale identity, loopback bridge, lease FIFO/cancellation, draft revision, + and managed opener arguments. +- `corepack pnpm vitest run tests/unit/pi-extension-bundle.test.ts` — 1/1 + passed by loading the actual materialized bundle, executing its mutation + hooks, and observing real bridge lease blocking/release; a read hook bypassed + the lease. +- `corepack pnpm vitest run tests/unit/pi-worker-process-real.test.ts` — 1/1 + passed against locked `@earendil-works/pi-coding-agent@0.84.2` through the + Electron Node child process with the managed extension explicitly loaded. +- `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. +- `corepack pnpm run build:vite` — passed for Renderer, Electron Main, + Preload, and release utility output. Existing dynamic-import and chunk-size + warnings remain unchanged. +- `git diff --check` — passed. +- Electron E2E was not applicable: PI-070 intentionally adds no Renderer or + Host API route; PI-100/PI-130 own that user-visible wiring. macOS and real + Provider checks were intentionally not run under the explicit decisions + above. + +## Follow-ups + +- PI-100 must construct one shared `PiManagedExtensionHost` for the managed + opener and `PiConversationRuntime`, expose sanitized interaction methods + through Host API, and keep bridge tokens/leases Main-private. +- PI-130 must consume safe UI projections, supply the actual per-Conversation + draft revision, register only product-owned widget keys, and render terminal + interaction states without exposing Pi wire payloads. +- PI-080/PI-120 child workers must use this same project lease and worker-run + identity model; they must not introduce a second per-child mutation lock. +- PI-150 owns staging the managed extension in final packaged closures and the + deferred macOS x64/arm64 matrix. +- Real Provider concurrency, credential isolation, authentication, abort, + image/model/session behavior, and protocol variation remain accepted + unverified risks with `realTurnVerified=false`. + +## Promotion Candidates + +- Target: Pi runtime extension and process architecture. + Proposal: retain discovery-disabled startup and load exactly one explicitly + materialized, versioned Makelore extension; keep its loopback endpoint, + bearer token registry, run context, interactions, and diagnostics inside + Electron Main. + Evidence: managed opener assertions, the locked Pi 0.84.2 real-child smoke, + stale-token/generation bridge tests, and the actual-bundle integration test + all pass. + Future impact: PI-100 and PI-150 can wire Host routes and packaging without + creating a project/user plugin surface or moving Pi protocol objects into + Renderer. + Semantic conflicts: canonical product-state documents still describe the + active OpenCode product and must not claim Pi is live before the later + cutover ticket is integrated. + Human confirmation required: no for the internal architecture; yes before + canonical product-state text claims the Pi cutover is released. +- Target: Pi mutation concurrency architecture. + Proposal: all parent and future child `write`/`edit`/`bash` executions for a + project share one cancellable FIFO Main-owned lease; read-only tools and + other projects bypass that queue. + Evidence: coordinator and real HTTP/bundle tests prove same-project + serialization, cross-project parallelism, cancellation, terminal release, + and current-worker survival after stale identity rejection. + Future impact: PI-080/PI-120 can add subagents without inventing competing + locks or weakening project isolation. + Semantic conflicts: none found. + Human confirmation required: no. +- Target: PI-130 draft and extension UI projection. + Proposal: capture draft revision at run start and apply `set_editor_text` + only if the revision is unchanged; accept status/widget projections only for + registered product keys and retain bounded method-only diagnostics for + unknown payloads. + Evidence: focused projector tests prove accepted and stale draft paths plus + unknown-widget suppression without raw details. + Future impact: prevents asynchronous agent UI events from overwriting newer + user edits and keeps extension detail schemas product-owned. + Semantic conflicts: none found. + Human confirmation required: no. diff --git a/electron/coding-runtime/pi/extension-host.ts b/electron/coding-runtime/pi/extension-host.ts new file mode 100644 index 0000000..ee2617b --- /dev/null +++ b/electron/coding-runtime/pi/extension-host.ts @@ -0,0 +1,299 @@ +import { randomBytes, randomUUID } from 'node:crypto'; +import { createServer, type IncomingMessage, type Server, type ServerResponse } from 'node:http'; +import { mkdir, rm } from 'node:fs/promises'; +import path from 'node:path'; +import { atomicWriteJson } from '../../coding-projects/atomic-json'; +import { materializeMakelorePiExtension } from './extensions/makelore-runtime'; +import { + PiProjectWriteLeaseCoordinator, + type PiProjectWriteLease, +} from './write-lease'; + +const MAX_REQUEST_BYTES = 64 * 1024; + +interface WorkerRegistrationRecord { + token: string; + conversationId: string; + generation: number; + projectId: string; + contextFile: string; + runId: string | null; + leases: Map; + waiters: Map; +} + +export interface PiExtensionWorkerRegistration { + extensionPath: string; + env: NodeJS.ProcessEnv; + sensitiveValues: string[]; + dispose(): Promise; +} + +export interface RegisterPiExtensionWorkerInput { + conversationId: string; + generation: number; + projectId: string; + extensionsDir: string; +} + +interface BridgeRequest { + action: 'lease.acquire' | 'lease.release'; + conversationId: string; + workerGeneration: number; + runId: string; + resourceId: string; + leaseId?: string; +} + +function recordValue(value: unknown): value is Record { + return value !== null && typeof value === 'object' && !Array.isArray(value); +} + +function bridgeRequest(value: unknown): value is BridgeRequest { + if (!recordValue(value)) return false; + return (value.action === 'lease.acquire' || value.action === 'lease.release') + && typeof value.conversationId === 'string' + && Number.isSafeInteger(value.workerGeneration) + && typeof value.runId === 'string' + && typeof value.resourceId === 'string' + && (value.leaseId === undefined || typeof value.leaseId === 'string'); +} + +export class PiManagedExtensionHost { + private readonly leases: PiProjectWriteLeaseCoordinator; + private readonly registrations = new Map(); + private readonly runBindings = new Map(); + private server: Server | null = null; + private bridgeUrl: string | null = null; + private startFlight: Promise | null = null; + + constructor(leases = new PiProjectWriteLeaseCoordinator()) { + this.leases = leases; + } + + async registerWorker(input: RegisterPiExtensionWorkerInput): Promise { + if (!Number.isSafeInteger(input.generation) || input.generation <= 0) { + throw new Error('Worker generation must be a positive safe integer'); + } + const bridgeUrl = await this.start(); + await mkdir(input.extensionsDir, { recursive: true }); + const extensionPath = await materializeMakelorePiExtension(input.extensionsDir); + const token = randomBytes(32).toString('base64url'); + const contextFile = path.join(input.extensionsDir, `worker-${randomUUID()}.json`); + const record: WorkerRegistrationRecord = { + token, + conversationId: input.conversationId, + generation: input.generation, + projectId: input.projectId, + contextFile, + runId: this.runBindings.get(input.conversationId) ?? null, + leases: new Map(), + waiters: new Map(), + }; + this.registrations.set(token, record); + await this.writeContext(record); + let disposed = false; + return { + extensionPath, + env: { + MAKELORE_PI_BRIDGE_URL: bridgeUrl, + MAKELORE_PI_WORKER_TOKEN: token, + MAKELORE_PI_CONTEXT_FILE: contextFile, + }, + sensitiveValues: [token], + dispose: async () => { + if (disposed) return; + disposed = true; + this.disposeRecord(record); + await rm(contextFile, { force: true }); + }, + }; + } + + async bindRun(conversationId: string, generation: number, runId: string): Promise { + const record = this.findWorker(conversationId, generation); + if (!record) throw new Error('Pi extension worker registration is unavailable'); + this.releaseWorkerResources(record); + this.runBindings.set(conversationId, runId); + record.runId = runId; + await this.writeContext(record); + } + + async clearRun(conversationId: string, generation: number, runId?: string): Promise { + const record = this.findWorker(conversationId, generation); + if (!runId || this.runBindings.get(conversationId) === runId) { + this.runBindings.delete(conversationId); + } + if (!record || (runId && record.runId !== runId)) return; + this.releaseWorkerResources(record); + record.runId = null; + await this.writeContext(record); + } + + 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) => { + server.close((error) => error ? reject(error) : resolve()); + }); + } + + private start(): Promise { + if (this.bridgeUrl) return Promise.resolve(this.bridgeUrl); + if (this.startFlight) return this.startFlight; + this.startFlight = new Promise((resolve, reject) => { + const server = createServer((request, response) => { + void this.handle(request, response); + }); + server.once('error', reject); + server.listen(0, '127.0.0.1', () => { + server.removeListener('error', reject); + const address = server.address(); + if (!address || typeof address === 'string') { + reject(new Error('Pi extension bridge did not bind a loopback port')); + return; + } + this.server = server; + this.bridgeUrl = `http://127.0.0.1:${address.port}/v1/worker`; + resolve(this.bridgeUrl); + }); + }).finally(() => { + if (!this.bridgeUrl) this.startFlight = null; + }); + return this.startFlight; + } + + private async handle(request: IncomingMessage, response: ServerResponse): Promise { + if (request.method !== 'POST' || request.url !== '/v1/worker') { + this.respond(response, 404, { error: 'Not found' }); + return; + } + const authorization = request.headers.authorization; + const token = authorization?.startsWith('Bearer ') ? authorization.slice(7) : ''; + const record = this.registrations.get(token); + if (!record) { + this.respond(response, 401, { error: 'Worker token is invalid' }); + return; + } + try { + const value = await this.readBody(request); + if (!bridgeRequest(value)) { + this.respond(response, 400, { error: 'Bridge request is invalid' }); + return; + } + if (value.conversationId !== record.conversationId + || value.workerGeneration !== record.generation + || value.runId !== record.runId) { + this.respond(response, 409, { error: 'Worker run identity is stale' }); + return; + } + if (!value.resourceId.trim()) { + this.respond(response, 400, { error: 'Bridge resource id is required' }); + return; + } + if (value.action === 'lease.release') { + const lease = record.leases.get(value.resourceId); + if (!lease || !value.leaseId || lease.id !== value.leaseId) { + this.respond(response, 409, { error: 'Project write lease is stale' }); + return; + } + record.leases.delete(value.resourceId); + lease.release(); + this.respond(response, 200, { released: true }); + return; + } + if (record.leases.has(value.resourceId) || record.waiters.has(value.resourceId)) { + this.respond(response, 409, { error: 'Project write lease resource already exists' }); + return; + } + const controller = new AbortController(); + record.waiters.set(value.resourceId, controller); + const cancel = () => { + if (!response.writableEnded) controller.abort(); + }; + request.once('aborted', cancel); + response.once('close', cancel); + try { + const lease = await this.leases.acquire( + record.projectId, + `${record.conversationId}:${record.generation}:${value.resourceId}`, + controller.signal, + ); + if (record.runId !== value.runId || this.registrations.get(token) !== record) { + lease.release(); + return; + } + record.leases.set(value.resourceId, lease); + this.respond(response, 200, { leaseId: lease.id }); + } finally { + request.removeListener('aborted', cancel); + response.removeListener('close', cancel); + record.waiters.delete(value.resourceId); + } + } catch { + if (!response.writableEnded) this.respond(response, 400, { error: 'Bridge request failed' }); + } + } + + private readBody(request: IncomingMessage): Promise { + return new Promise((resolve, reject) => { + const chunks: Buffer[] = []; + let bytes = 0; + request.on('data', (chunk: Buffer) => { + bytes += chunk.length; + if (bytes > MAX_REQUEST_BYTES) { + reject(new Error('Bridge request is too large')); + request.destroy(); + return; + } + chunks.push(chunk); + }); + request.once('end', () => { + try { + resolve(JSON.parse(Buffer.concat(chunks).toString('utf8'))); + } catch (error) { + reject(error); + } + }); + request.once('error', reject); + }); + } + + private respond(response: ServerResponse, status: number, body: Record): void { + if (response.writableEnded) return; + response.writeHead(status, { 'content-type': 'application/json; charset=utf-8' }); + response.end(JSON.stringify(body)); + } + + private findWorker(conversationId: string, generation: number): WorkerRegistrationRecord | undefined { + return [...this.registrations.values()].find((record) => ( + record.conversationId === conversationId && record.generation === generation + )); + } + + private async writeContext(record: WorkerRegistrationRecord): Promise { + await atomicWriteJson(record.contextFile, { + conversationId: record.conversationId, + workerGeneration: record.generation, + ...(record.runId ? { runId: record.runId } : {}), + }); + } + + private disposeRecord(record: WorkerRegistrationRecord): void { + if (this.registrations.get(record.token) !== record) return; + this.registrations.delete(record.token); + this.releaseWorkerResources(record); + } + + private releaseWorkerResources(record: WorkerRegistrationRecord): void { + for (const controller of record.waiters.values()) controller.abort(); + record.waiters.clear(); + for (const lease of record.leases.values()) lease.release(); + record.leases.clear(); + } +} diff --git a/electron/coding-runtime/pi/extension-ui-projector.ts b/electron/coding-runtime/pi/extension-ui-projector.ts new file mode 100644 index 0000000..359fe9c --- /dev/null +++ b/electron/coding-runtime/pi/extension-ui-projector.ts @@ -0,0 +1,162 @@ +import type { PiRpcEvent } from './rpc-client'; + +const MAX_MESSAGE_LENGTH = 2_000; +const MAX_TITLE_LENGTH = 256; +const MAX_EDITOR_TEXT_LENGTH = 64 * 1024; +const MAX_WIDGET_LINES = 32; +const MAX_WIDGET_LINE_LENGTH = 512; + +export type PiExtensionUiProjection = + | { kind: 'notify'; conversationId: string; message: string; level: 'info' | 'warning' | 'error' } + | { kind: 'status'; conversationId: string; key: string; text?: string } + | { + kind: 'widget'; + conversationId: string; + key: string; + lines?: string[]; + placement?: 'aboveEditor' | 'belowEditor'; + } + | { kind: 'title'; conversationId: string; title: string } + | { kind: 'editor-text'; conversationId: string; text: string; draftRevision: number }; + +export interface PiExtensionUiDiagnostic { + method: string; + reason: 'unsupported-ui-method' | 'invalid-ui-payload' | 'stale-draft-revision'; +} + +export interface PiExtensionUiProjectorOptions { + getDraftRevision(conversationId: string): number; + knownStatusKeys?: readonly string[]; + knownWidgetKeys?: readonly string[]; +} + +interface RunProjectionState { + runId: string; + generation: number; + draftRevision: number; +} + +function bounded(value: string, length: number): string { + return value.length <= length ? value : `${value.slice(0, length - 1)}…`; +} + +export class PiExtensionUiProjector { + private readonly getDraftRevision: (conversationId: string) => number; + private readonly knownStatusKeys: Set; + private readonly knownWidgetKeys: Set; + private readonly runs = new Map(); + private readonly diagnostics: PiExtensionUiDiagnostic[] = []; + + constructor(options: PiExtensionUiProjectorOptions) { + this.getDraftRevision = options.getDraftRevision; + this.knownStatusKeys = new Set(options.knownStatusKeys ?? ['makelore.write-lease']); + this.knownWidgetKeys = new Set(options.knownWidgetKeys ?? []); + } + + beginRun(conversationId: string, generation: number, runId: string): void { + this.runs.set(conversationId, { + generation, + runId, + draftRevision: this.getDraftRevision(conversationId), + }); + } + + replaceGeneration(conversationId: string, generation: number, runId: string): void { + const current = this.runs.get(conversationId); + if (current?.runId === runId) current.generation = generation; + } + + endRun(conversationId: string, runId: string): void { + if (this.runs.get(conversationId)?.runId === runId) this.runs.delete(conversationId); + } + + getDiagnostics(): PiExtensionUiDiagnostic[] { + return structuredClone(this.diagnostics); + } + + project( + conversationId: string, + generation: number, + runId: string, + event: PiRpcEvent, + ): PiExtensionUiProjection | null { + if (event.type !== 'extension_ui_request' || typeof event.method !== 'string') return null; + const run = this.runs.get(conversationId); + if (!run || run.runId !== runId || run.generation !== generation) return null; + if (event.method === 'notify' + && typeof event.message === 'string' + && ['info', 'warning', 'error', undefined].includes(event.notifyType as string | undefined)) { + return { + kind: 'notify', + conversationId, + message: bounded(event.message, MAX_MESSAGE_LENGTH), + level: event.notifyType === 'warning' || event.notifyType === 'error' + ? event.notifyType + : 'info', + }; + } + if (event.method === 'setStatus' + && typeof event.statusKey === 'string' + && (event.statusText === undefined || typeof event.statusText === 'string')) { + if (!this.knownStatusKeys.has(event.statusKey)) return this.unsupported(event.method); + return { + kind: 'status', + conversationId, + key: event.statusKey, + ...(typeof event.statusText === 'string' + ? { text: bounded(event.statusText, MAX_MESSAGE_LENGTH) } + : {}), + }; + } + if (event.method === 'setWidget' + && typeof event.widgetKey === 'string' + && (event.widgetLines === undefined || Array.isArray(event.widgetLines))) { + if (!this.knownWidgetKeys.has(event.widgetKey)) return this.unsupported(event.method); + if (Array.isArray(event.widgetLines) + && !event.widgetLines.every((line) => typeof line === 'string')) { + return this.invalid(event.method); + } + return { + kind: 'widget', + conversationId, + key: event.widgetKey, + ...(Array.isArray(event.widgetLines) + ? { + lines: event.widgetLines.slice(0, MAX_WIDGET_LINES) + .map((line) => bounded(line as string, MAX_WIDGET_LINE_LENGTH)), + } + : {}), + ...(event.widgetPlacement === 'aboveEditor' || event.widgetPlacement === 'belowEditor' + ? { placement: event.widgetPlacement } + : {}), + }; + } + if (event.method === 'setTitle' && typeof event.title === 'string') { + return { kind: 'title', conversationId, title: bounded(event.title, MAX_TITLE_LENGTH) }; + } + 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' }); + return null; + } + return { + kind: 'editor-text', + conversationId, + text: bounded(event.text, MAX_EDITOR_TEXT_LENGTH), + draftRevision: run.draftRevision, + }; + } + if (['select', 'confirm', 'input', 'editor'].includes(event.method)) return null; + return this.invalid(event.method); + } + + private unsupported(method: string): null { + this.diagnostics.push({ 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' }); + return null; + } +} diff --git a/electron/coding-runtime/pi/extensions/makelore-runtime.ts b/electron/coding-runtime/pi/extensions/makelore-runtime.ts new file mode 100644 index 0000000..488aa0e --- /dev/null +++ b/electron/coding-runtime/pi/extensions/makelore-runtime.ts @@ -0,0 +1,100 @@ +import path from 'node:path'; +import { atomicWriteText } from '../../../coding-projects/atomic-json'; + +export const MAKELORE_PI_EXTENSION_VERSION = 1; +export const MAKELORE_PI_EXTENSION_FILENAME = `makelore-runtime-v${MAKELORE_PI_EXTENSION_VERSION}.mjs`; + +const BUNDLE_SOURCE = String.raw` +import { readFile } from 'node:fs/promises'; + +const MUTATION_TOOLS = new Set(['bash', 'edit', 'write']); +const leases = new Map(); + +async function runtimeContext() { + const value = JSON.parse(await readFile(process.env.MAKELORE_PI_CONTEXT_FILE, 'utf8')); + if (!value.runId) throw new Error('Makelore run context is unavailable'); + return value; +} + +async function bridge(action, body, signal) { + const context = await runtimeContext(); + const response = await fetch(process.env.MAKELORE_PI_BRIDGE_URL, { + method: 'POST', + headers: { + authorization: 'Bearer ' + process.env.MAKELORE_PI_WORKER_TOKEN, + 'content-type': 'application/json', + }, + body: JSON.stringify({ ...context, action, ...body }), + signal, + }); + const result = await response.json().catch(() => ({})); + if (!response.ok) throw new Error(result.error || 'Makelore runtime bridge rejected the request'); + return result; +} + +async function releaseLease(toolCallId) { + const leaseId = leases.get(toolCallId); + if (!leaseId) return; + leases.delete(toolCallId); + await bridge('lease.release', { leaseId, resourceId: toolCallId }).catch(() => undefined); +} + +async function releaseAll() { + await Promise.all([...leases.keys()].map(releaseLease)); +} + +export default function makeloreRuntime(pi) { + pi.registerTool({ + name: 'ask_user', + label: 'Ask user', + description: 'Ask the user for a selection, confirmation, short input, or editor text.', + parameters: { + type: 'object', + additionalProperties: false, + required: ['kind', 'title'], + properties: { + kind: { type: 'string', enum: ['select', 'confirm', 'input', 'editor'] }, + title: { type: 'string' }, + message: { type: 'string' }, + options: { type: 'array', items: { type: 'string' } }, + }, + }, + async execute(_toolCallId, params, signal, _onUpdate, ctx) { + let value; + if (params.kind === 'select') { + value = await ctx.ui.select(params.title, Array.isArray(params.options) ? params.options : [], { signal }); + } else if (params.kind === 'confirm') { + value = await ctx.ui.confirm(params.title, params.message || '', { signal }); + } else if (params.kind === 'editor') { + value = await ctx.ui.editor(params.title, params.message || '', { signal }); + } else { + value = await ctx.ui.input(params.title, params.message || '', { signal }); + } + return { + content: [{ type: 'text', text: value === undefined ? 'User cancelled' : String(value) }], + details: { kind: params.kind, cancelled: value === undefined }, + }; + }, + }); + + pi.on('tool_call', async (event, ctx) => { + if (!MUTATION_TOOLS.has(event.toolName)) return; + ctx.ui.setStatus('makelore.write-lease', '等待项目写入'); + try { + const result = await bridge('lease.acquire', { resourceId: event.toolCallId }, ctx.signal); + leases.set(event.toolCallId, result.leaseId); + } finally { + ctx.ui.setStatus('makelore.write-lease', undefined); + } + }); + pi.on('tool_result', async (event) => releaseLease(event.toolCallId)); + pi.on('agent_end', releaseAll); + pi.on('session_shutdown', releaseAll); +} +`; + +export async function materializeMakelorePiExtension(extensionsDir: string): Promise { + const extensionPath = path.join(extensionsDir, MAKELORE_PI_EXTENSION_FILENAME); + await atomicWriteText(extensionPath, BUNDLE_SOURCE.trimStart()); + return extensionPath; +} diff --git a/electron/coding-runtime/pi/interaction.ts b/electron/coding-runtime/pi/interaction.ts new file mode 100644 index 0000000..9ec7c68 --- /dev/null +++ b/electron/coding-runtime/pi/interaction.ts @@ -0,0 +1,177 @@ +import type { ConversationInteraction } from '../contracts'; +import type { PiRpcCommand, PiRpcEvent } from './rpc-client'; +import type { PiGenerationResourceInput, PiWorkerPoolState } from './worker-pool'; + +interface PiInteractionTransport { + getState(conversationId: string): PiWorkerPoolState | null; + getActiveRun(conversationId: string): { runId: string; generation: number } | null; + send(conversationId: string, command: PiRpcCommand): Promise; + trackGenerationResource(input: PiGenerationResourceInput): () => void; +} + +interface StoredInteraction { + interaction: ConversationInteraction; + generation: number; + labels: Map; + untrack(): void; +} + +export type PiInteractionResponse = + | { interactionId: string; cancelled: true } + | { interactionId: string; optionId: string } + | { interactionId: string; confirmed: boolean } + | { interactionId: string; value: string }; + +function dialogEvent(event: PiRpcEvent): event is PiRpcEvent & { + id: string; + method: 'select' | 'confirm' | 'input' | 'editor'; + title: string; +} { + return event.type === 'extension_ui_request' + && typeof event.id === 'string' + && ['select', 'confirm', 'input', 'editor'].includes(String(event.method)) + && typeof event.title === 'string'; +} + +export class PiInteractionStore { + private readonly pending = new Map(); + + constructor( + private readonly transport: PiInteractionTransport, + private readonly onChange: (interaction: ConversationInteraction) => void, + ) {} + + list(conversationId?: string): ConversationInteraction[] { + return [...this.pending.values()] + .map(({ interaction }) => interaction) + .filter((interaction) => !conversationId || interaction.conversationId === conversationId) + .map((interaction) => structuredClone(interaction)); + } + + open( + conversationId: string, + generation: number, + runId: string, + event: PiRpcEvent, + ): ConversationInteraction | null { + if (!dialogEvent(event)) return null; + const state = this.transport.getState(conversationId); + const active = this.transport.getActiveRun(conversationId); + if (state?.generation !== generation + || active?.generation !== generation + || active.runId !== runId) return null; + const key = this.key(conversationId, event.id); + if (this.pending.has(key)) throw new Error(`Duplicate Pi interaction id: ${event.id}`); + const labels = new Map(); + const options = event.method === 'select' && Array.isArray(event.options) + ? event.options.flatMap((label, index) => { + if (typeof label !== 'string') return []; + const id = `${event.id}:option:${index}`; + labels.set(id, label); + return [{ id, label }]; + }) + : undefined; + const interaction: ConversationInteraction = { + id: event.id, + conversationId, + runId, + kind: event.method, + title: event.title, + ...(typeof event.message === 'string' ? { message: event.message } : {}), + ...(options ? { options } : {}), + status: 'pending', + }; + const stored: StoredInteraction = { + interaction, + generation, + labels, + untrack: () => undefined, + }; + stored.untrack = this.transport.trackGenerationResource({ + conversationId, + kind: 'interaction', + id: event.id, + cancel: () => { + void this.cancelStored(stored, false); + }, + }); + this.pending.set(key, stored); + return structuredClone(interaction); + } + + 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'); + const state = this.transport.getState(conversationId); + const active = this.transport.getActiveRun(conversationId); + if (state?.generation !== stored.generation + || active?.generation !== stored.generation + || active.runId !== stored.interaction.runId) { + await this.cancelStored(stored, false); + throw new Error('Pi interaction belongs to a stale worker run'); + } + + let command: PiRpcCommand; + if ('cancelled' in response) { + command = { type: 'extension_ui_response', id: stored.interaction.id, cancelled: true }; + } else if (stored.interaction.kind === 'select' && 'optionId' in response) { + const value = stored.labels.get(response.optionId); + if (value === undefined) throw new Error('Pi interaction option is invalid'); + command = { type: 'extension_ui_response', id: stored.interaction.id, value }; + } else if (stored.interaction.kind === 'confirm' && 'confirmed' in response) { + command = { type: 'extension_ui_response', id: stored.interaction.id, confirmed: response.confirmed }; + } else if ((stored.interaction.kind === 'input' || stored.interaction.kind === 'editor') + && 'value' in response) { + command = { type: 'extension_ui_response', id: stored.interaction.id, value: response.value }; + } else { + throw new Error('Pi interaction response does not match its kind'); + } + await this.transport.send(conversationId, command); + return this.finish(stored, 'cancelled' in response + ? 'cancelled' + : stored.interaction.kind === 'confirm' && 'confirmed' in response && !response.confirmed + ? 'rejected' + : 'answered'); + } + + async cancelRun(conversationId: string, runId: string, notifyWorker: boolean): Promise { + const targets = [...this.pending.values()].filter(({ interaction }) => ( + interaction.conversationId === conversationId && interaction.runId === runId + )); + await Promise.all(targets.map((stored) => this.cancelStored(stored, notifyWorker))); + } + + async cancelGeneration(conversationId: string, generation: number): Promise { + const targets = [...this.pending.values()].filter((stored) => ( + stored.interaction.conversationId === conversationId && stored.generation === generation + )); + await Promise.all(targets.map((stored) => this.cancelStored(stored, false))); + } + + private async cancelStored(stored: StoredInteraction, notifyWorker: boolean): Promise { + if (!this.pending.has(this.key(stored.interaction.conversationId, stored.interaction.id))) return; + if (notifyWorker) { + await this.transport.send(stored.interaction.conversationId, { + type: 'extension_ui_response', + id: stored.interaction.id, + cancelled: true, + }).catch(() => undefined); + } + this.finish(stored, 'cancelled'); + } + + private finish( + stored: StoredInteraction, + status: Exclude, + ): ConversationInteraction { + this.pending.delete(this.key(stored.interaction.conversationId, stored.interaction.id)); + stored.untrack(); + const interaction = { ...stored.interaction, status }; + this.onChange(structuredClone(interaction)); + return interaction; + } + + private key(conversationId: string, interactionId: string): string { + return `${conversationId}\u0000${interactionId}`; + } +} diff --git a/electron/coding-runtime/pi/rpc-client.ts b/electron/coding-runtime/pi/rpc-client.ts index 9f2a437..2563943 100644 --- a/electron/coding-runtime/pi/rpc-client.ts +++ b/electron/coding-runtime/pi/rpc-client.ts @@ -127,6 +127,19 @@ export class PiRpcClient { throw lastError; } + async send(command: PiRpcCommand): Promise { + if (!command.type) throw new Error('Pi RPC command type is required'); + if (this.invalidated) throw this.invalidated; + try { + await this.write(`${JSON.stringify(command)}\n`); + } catch (error) { + throw new PiProcessError('PI_RPC_WRITE_FAILED', `Could not write Pi RPC ${command.type}`, { + cause: error, + generation: this.generation, + }); + } + } + accept(value: unknown): void { if (!recordValue(value) || typeof value.type !== 'string') { throw new PiProcessError('PI_RPC_PROTOCOL_ERROR', 'Pi RPC record must be an object with a type'); diff --git a/electron/coding-runtime/pi/runtime.ts b/electron/coding-runtime/pi/runtime.ts index a4e0374..6579592 100644 --- a/electron/coding-runtime/pi/runtime.ts +++ b/electron/coding-runtime/pi/runtime.ts @@ -74,6 +74,15 @@ import { PiSessionProjectionError, projectPiSessionSnapshot, } from './session-projector'; +import { PiManagedExtensionHost } from './extension-host'; +import { + PiInteractionStore, + type PiInteractionResponse, +} from './interaction'; +import { + PiExtensionUiProjector, + type PiExtensionUiProjection, +} from './extension-ui-projector'; type RuntimeIdKind = 'run' | 'queue'; @@ -88,6 +97,10 @@ export interface PiConversationRuntimeOptions { providerRefreshCoordinator?: PiProviderRefreshCoordinator; isAuthenticationError?(error: unknown): boolean; refreshCredential?(accountId: string): Promise; + extensionHost?: PiManagedExtensionHost; + getDraftRevision?(conversationId: string): number; + knownExtensionWidgetKeys?: readonly string[]; + onExtensionUiProjection?(projection: PiExtensionUiProjection): void; } export interface PiWorkerProcessAdapter { @@ -97,6 +110,7 @@ export interface PiWorkerProcessAdapter { command: PiRpcCommand, options?: PiRpcRequestOptions, ): Promise>; + send(command: PiRpcCommand): Promise; subscribe(listener: (event: PiRpcEvent) => void): () => void; subscribeInvalidation(listener: (error: PiProcessError) => void): () => void; stop(): Promise; @@ -120,6 +134,7 @@ export interface PiManagedWorkerOpenerOptions { createProcess?: (options: PiWorkerProcessOptions) => PiWorkerProcessAdapter; now?: () => number; onTelemetry?: (event: PiRuntimeTelemetryEvent) => void; + extensionHost: PiManagedExtensionHost; } interface PiRpcSessionStateProjection { @@ -132,12 +147,18 @@ class ManagedPiConversationWorker implements PiConversationWorker { readonly id: string, readonly generation: number, private readonly process: PiWorkerProcessAdapter, + private readonly disposeExtension: () => Promise, + private readonly unsubscribeExtensionInvalidation: () => void, ) {} request(command: PiRpcCommand, options?: PiRpcRequestOptions): Promise> { return this.process.request(command, options); } + send(command: PiRpcCommand): Promise { + return this.process.send(command); + } + subscribe(listener: (event: PiRpcEvent) => void): () => void { return this.process.subscribe(listener); } @@ -146,8 +167,13 @@ class ManagedPiConversationWorker implements PiConversationWorker { return this.process.subscribeInvalidation(listener); } - stop(): Promise { - return this.process.stop(); + async stop(): Promise { + this.unsubscribeExtensionInvalidation(); + try { + return await this.process.stop(); + } finally { + await this.disposeExtension(); + } } } @@ -193,6 +219,12 @@ export function createPiManagedWorkerOpener( ? { localProxyCredential: await options.getLocalProxyCredential() } : {}), }); + const extension = await options.extensionHost.registerWorker({ + conversationId: input.conversation.conversationId, + generation: input.generation, + projectId: input.conversation.projectId, + extensionsDir: managedPaths.extensionsDir, + }); recordManagedMilestone( options.onTelemetry, input, @@ -214,11 +246,15 @@ export function createPiManagedWorkerOpener( sessionDir: resources.projectSessionsDir, additionalArgs: [ ...buildPiManagedInputArgs(selection, resources), + '--extension', extension.extensionPath, ...(input.fork ? ['--fork', input.fork.sourceSession.piSessionId] : []), '--session-id', sessionKey, ], - env: credential.env, - sensitiveValues: credential.sensitiveValues, + env: { ...credential.env, ...extension.env }, + sensitiveValues: [...credential.sensitiveValues, ...extension.sensitiveValues], + }); + let unsubscribeExtensionInvalidation = process.subscribeInvalidation(() => { + void extension.dispose(); }); try { const spawnStartedAt = now(); @@ -290,11 +326,18 @@ export function createPiManagedWorkerOpener( `${input.conversation.conversationId}:${input.generation}`, input.generation, process, + extension.dispose, + () => { + unsubscribeExtensionInvalidation(); + unsubscribeExtensionInvalidation = () => undefined; + }, ), session: clone(bound.session), }; } catch (error) { + unsubscribeExtensionInvalidation(); await process.stop().catch(() => undefined); + await extension.dispose(); throw error; } }; @@ -401,6 +444,10 @@ export class PiConversationRuntime implements CodingConversationRuntime { private readonly providerRefresh: PiProviderRefreshCoordinator; private readonly isAuthenticationError: ((error: unknown) => boolean) | undefined; private readonly refreshCredential: ((accountId: string) => Promise) | undefined; + private readonly extensionHost: PiManagedExtensionHost | undefined; + private readonly interactions: PiInteractionStore; + private readonly extensionUi: PiExtensionUiProjector; + private readonly onExtensionUiProjection: ((projection: PiExtensionUiProjection) => void) | undefined; private readonly states = new Map(); private readonly inputs = new Map(); private readonly listeners = new Set<(patch: ConversationPatchEnvelope) => void>(); @@ -432,6 +479,17 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.providerRefresh = options.providerRefreshCoordinator ?? new PiProviderRefreshCoordinator(); this.isAuthenticationError = options.isAuthenticationError; this.refreshCredential = options.refreshCredential; + this.extensionHost = options.extensionHost; + this.interactions = new PiInteractionStore(this.pool, (interaction) => { + this.emit(interaction.conversationId, { op: 'interaction.upsert', interaction }, interaction.runId); + }); + this.extensionUi = new PiExtensionUiProjector({ + getDraftRevision: options.getDraftRevision ?? (() => 0), + ...(options.knownExtensionWidgetKeys + ? { knownWidgetKeys: options.knownExtensionWidgetKeys } + : {}), + }); + this.onExtensionUiProjection = options.onExtensionUiProjection; if (Boolean(this.isAuthenticationError) !== Boolean(this.refreshCredential)) { throw new Error('Provider authentication detection and refresh must be configured together'); } @@ -525,11 +583,25 @@ export class PiConversationRuntime implements CodingConversationRuntime { message: input.text, ...(images.length > 0 ? { images } : {}), }; - const ticket = this.pool.startTopLevel({ - conversationId: input.conversationId, - runId, - command, - }); + const generation = this.pool.getState(input.conversationId)?.generation; + if (generation) this.extensionUi.beginRun(input.conversationId, generation, runId); + if (this.extensionHost && generation) { + await this.extensionHost.bindRun(input.conversationId, generation, runId); + } + let ticket; + try { + ticket = this.pool.startTopLevel({ + conversationId: input.conversationId, + runId, + command, + }); + } catch (error) { + this.extensionUi.endRun(input.conversationId, runId); + if (this.extensionHost && generation) { + await this.extensionHost.clearRun(input.conversationId, generation, runId); + } + throw error; + } this.emit(input.conversationId, { op: 'run.state', run: { @@ -567,8 +639,14 @@ export class PiConversationRuntime implements CodingConversationRuntime { op: 'run.state', run: { ...current, status: 'aborting' }, }, current.runId); + if (current.runId) await this.interactions.cancelRun(conversationId, current.runId, true); try { await this.pool.request(conversationId, { type: 'abort' }); + const generation = this.pool.getState(conversationId)?.generation; + if (current.runId && generation) { + await this.extensionHost?.clearRun(conversationId, generation, current.runId); + this.extensionUi.endRun(conversationId, current.runId); + } } catch (error) { const latest = this.snapshot(conversationId).run; if (latest.runId === current.runId && latest.status === 'aborting') { @@ -651,11 +729,25 @@ export class PiConversationRuntime implements CodingConversationRuntime { async compact(conversationId: string): Promise { await this.waitForProjection(conversationId); const runId = this.id('run'); - const ticket = this.pool.startTopLevel({ - conversationId, - runId, - command: { type: 'compact' }, - }); + const generation = this.pool.getState(conversationId)?.generation; + if (generation) this.extensionUi.beginRun(conversationId, generation, runId); + if (this.extensionHost && generation) { + await this.extensionHost.bindRun(conversationId, generation, runId); + } + let ticket; + try { + ticket = this.pool.startTopLevel({ + conversationId, + runId, + command: { type: 'compact' }, + }); + } catch (error) { + this.extensionUi.endRun(conversationId, runId); + if (this.extensionHost && generation) { + await this.extensionHost.clearRun(conversationId, generation, runId); + } + throw error; + } this.emit(conversationId, { op: 'run.state', run: { status: 'compacting', runId, startedAt: this.now() }, @@ -713,6 +805,8 @@ export class PiConversationRuntime implements CodingConversationRuntime { } async dispose(conversationId: string): Promise { + const state = this.pool.getState(conversationId); + if (state) await this.interactions.cancelGeneration(conversationId, state.generation); await this.pool.dispose(conversationId); this.registry.forget(conversationId); this.inputs.delete(conversationId); @@ -727,9 +821,17 @@ export class PiConversationRuntime implements CodingConversationRuntime { return () => this.listeners.delete(listener); } + async respondInteraction( + conversationId: string, + response: PiInteractionResponse, + ): Promise { + await this.interactions.respond(conversationId, response); + } + async shutdown(): Promise { this.unsubscribePool(); await this.pool.shutdown(); + await this.extensionHost?.close(); this.projectors.clear(); this.projectionChains.clear(); this.hydrationFlights.clear(); @@ -900,12 +1002,25 @@ export class PiConversationRuntime implements CodingConversationRuntime { if (!this.states.has(event.conversationId)) return; 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) { + void this.extensionHost.bindRun(event.conversationId, event.generation, runId).catch((error) => { + this.recordProjectionFailure(event.conversationId, event.generation, error); + }); + } void this.requestHydration(event.conversationId, event.state, true).catch((error) => { this.recordProjectionFailure(event.conversationId, event.generation, error); }); return; } if (event.type === 'worker.crashed') { + void this.interactions.cancelGeneration(event.conversationId, event.generation); + const runId = this.states.get(event.conversationId)?.snapshot.run.runId; + if (runId) { + this.extensionUi.endRun(event.conversationId, runId); + void this.extensionHost?.clearRun(event.conversationId, event.generation, runId); + } const state = this.pool.getState(event.conversationId); if (state) this.emit(event.conversationId, { op: 'worker.state', state: publicWorkerState(state) }); return; @@ -913,6 +1028,26 @@ export class PiConversationRuntime implements CodingConversationRuntime { void this.enqueueProjection(event.conversationId, async () => { const snapshot = this.states.get(event.conversationId)?.snapshot; if (!snapshot || snapshot.cursor.workerGeneration !== event.generation) return; + if (event.event.type === 'extension_ui_request' && snapshot.run.runId) { + const interaction = this.interactions.open( + event.conversationId, + event.generation, + snapshot.run.runId, + event.event, + ); + if (interaction) { + this.emit(event.conversationId, { op: 'interaction.upsert', interaction }, interaction.runId); + return; + } + const projection = this.extensionUi.project( + event.conversationId, + event.generation, + snapshot.run.runId, + event.event, + ); + if (projection) this.onExtensionUiProjection?.(projection); + return; + } const projector = this.projector(event.conversationId); const patches = await projector.project(snapshot, event.event); for (const patch of patches) { @@ -928,6 +1063,15 @@ export class PiConversationRuntime implements CodingConversationRuntime { ); } } + if (event.event.type === 'agent_settled' && snapshot.run.runId) { + await this.interactions.cancelRun(event.conversationId, snapshot.run.runId, true); + await this.extensionHost?.clearRun( + event.conversationId, + event.generation, + snapshot.run.runId, + ); + this.extensionUi.endRun(event.conversationId, snapshot.run.runId); + } }).catch((error) => { this.recordProjectionFailure(event.conversationId, event.generation, error); }); @@ -946,6 +1090,10 @@ export class PiConversationRuntime implements CodingConversationRuntime { error: runtimeFailure(error), }, }, runId); + void this.interactions.cancelRun(conversationId, runId, true); + this.extensionUi.endRun(conversationId, runId); + const generation = this.pool.getState(conversationId)?.generation; + if (generation) void this.extensionHost?.clearRun(conversationId, generation, runId); } private requestHydration( diff --git a/electron/coding-runtime/pi/worker-pool.ts b/electron/coding-runtime/pi/worker-pool.ts index 5ca2c98..02c2c43 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -24,6 +24,7 @@ export interface PiConversationWorker { command: PiRpcCommand, options?: PiRpcRequestOptions, ): Promise>; + send(command: PiRpcCommand): Promise; subscribe(listener: (event: PiRpcEvent) => void): () => void; subscribeInvalidation(listener: (error: PiProcessError) => void): () => void; stop(): Promise; @@ -365,6 +366,18 @@ export class PiWorkerPool { } } + async send(conversationId: string, command: PiRpcCommand): Promise { + let record = this.workers.get(conversationId); + if (!record || record.state === 'crashed') throw new Error('Conversation worker is not available'); + if (record.rebuildFlight) record = await record.rebuildFlight; + await record.worker.send(command); + } + + getActiveRun(conversationId: string): { runId: string; generation: number } | null { + const active = this.activeRuns.get(conversationId); + return active ? { ...active } : null; + } + failTopLevel(conversationId: string, runId: string, error: Error): void { const active = this.activeRuns.get(conversationId); if (active?.runId === runId) { diff --git a/electron/coding-runtime/pi/worker-process.ts b/electron/coding-runtime/pi/worker-process.ts index f7b96d2..818e039 100644 --- a/electron/coding-runtime/pi/worker-process.ts +++ b/electron/coding-runtime/pi/worker-process.ts @@ -75,7 +75,7 @@ export function buildPiRpcArgs( '--no-themes', '--no-context-files', '--no-approve', - '--no-tools', + '--tools', 'read,bash,edit,write,grep,find,ls,ask_user', ...additionalArgs, ]; } @@ -310,6 +310,17 @@ export class PiWorkerProcess { return this.rpc.request(command, options); } + send(command: PiRpcCommand): Promise { + if (!this.rpc) { + return Promise.reject(new PiProcessError( + 'PI_WORKER_START_FAILED', + 'Pi worker has not started', + { generation: this.generationValue }, + )); + } + return this.rpc.send(command); + } + subscribe(listener: (event: PiRpcEvent) => void): () => void { if (!this.rpc) throw new Error('Pi worker has not started'); return this.rpc.subscribe(listener); diff --git a/electron/coding-runtime/pi/write-lease.ts b/electron/coding-runtime/pi/write-lease.ts new file mode 100644 index 0000000..031d4d3 --- /dev/null +++ b/electron/coding-runtime/pi/write-lease.ts @@ -0,0 +1,103 @@ +import { randomUUID } from 'node:crypto'; + +export interface PiProjectWriteLease { + id: string; + projectId: string; + holderId: string; + release(): void; +} + +interface WaitingLease { + projectId: string; + holderId: string; + signal?: AbortSignal; + resolve(lease: PiProjectWriteLease): void; + reject(error: Error): void; + onAbort?: () => void; +} + +export class PiProjectWriteLeaseCoordinator { + private readonly active = new Map(); + private readonly waiting = new Map(); + + get activeCount(): number { return this.active.size; } + + waitingCount(projectId?: string): number { + if (projectId) return this.waiting.get(projectId)?.length ?? 0; + return [...this.waiting.values()].reduce((total, queue) => total + queue.length, 0); + } + + acquire(projectId: string, holderId: string, signal?: AbortSignal): Promise { + if (!projectId.trim() || !holderId.trim()) throw new Error('Project and lease holder are required'); + if (signal?.aborted) return Promise.reject(new Error('Project write lease wait cancelled')); + if (!this.active.has(projectId)) return Promise.resolve(this.issue(projectId, holderId)); + + return new Promise((resolve, reject) => { + const waiter: WaitingLease = { projectId, holderId, resolve, reject, ...(signal ? { signal } : {}) }; + if (signal) { + waiter.onAbort = () => { + this.removeWaiter(waiter); + reject(new Error('Project write lease wait cancelled')); + }; + signal.addEventListener('abort', waiter.onAbort, { once: true }); + } + const queue = this.waiting.get(projectId) ?? []; + queue.push(waiter); + this.waiting.set(projectId, queue); + }); + } + + cancelProject(projectId: string): void { + for (const waiter of this.waiting.get(projectId) ?? []) { + this.detachAbort(waiter); + waiter.reject(new Error('Project write lease wait cancelled')); + } + this.waiting.delete(projectId); + this.active.get(projectId)?.release(); + } + + private issue(projectId: string, holderId: string): PiProjectWriteLease { + let released = false; + const lease: PiProjectWriteLease = { + id: randomUUID(), + projectId, + holderId, + release: () => { + if (released) return; + released = true; + if (this.active.get(projectId) === lease) this.active.delete(projectId); + this.advance(projectId); + }, + }; + this.active.set(projectId, lease); + return lease; + } + + private advance(projectId: string): void { + const queue = this.waiting.get(projectId); + while (!this.active.has(projectId) && queue?.length) { + const waiter = queue.shift() as WaitingLease; + this.detachAbort(waiter); + if (waiter.signal?.aborted) { + waiter.reject(new Error('Project write lease wait cancelled')); + continue; + } + waiter.resolve(this.issue(projectId, waiter.holderId)); + } + if (!queue?.length) this.waiting.delete(projectId); + } + + private removeWaiter(waiter: WaitingLease): void { + const queue = this.waiting.get(waiter.projectId); + const index = queue?.indexOf(waiter) ?? -1; + if (index >= 0) queue?.splice(index, 1); + if (!queue?.length) this.waiting.delete(waiter.projectId); + this.detachAbort(waiter); + } + + private detachAbort(waiter: WaitingLease): void { + if (waiter.signal && waiter.onAbort) { + waiter.signal.removeEventListener('abort', waiter.onAbort); + } + } +} diff --git a/tests/unit/pi-conversation-runtime.test.ts b/tests/unit/pi-conversation-runtime.test.ts index d13d22d..73ff7eb 100644 --- a/tests/unit/pi-conversation-runtime.test.ts +++ b/tests/unit/pi-conversation-runtime.test.ts @@ -77,6 +77,10 @@ class RuntimeFakeWorker implements PiConversationWorker { }; } + async send(command: PiRpcCommand): Promise { + this.requests.push(structuredClone(command)); + } + setSessionData(input: { state?: unknown; entries?: unknown; stats?: unknown }): void { if (input.state !== undefined) this.stateData = structuredClone(input.state); if (input.entries !== undefined) this.entriesData = structuredClone(input.entries); @@ -240,6 +244,25 @@ describe('Pi Conversation runtime', () => { status: 'complete', })); expect(JSON.stringify(streamed.nodes)).toContain('Implemented'); + workers.get(left.id)!.emit({ + type: 'extension_ui_request', + id: 'question-runtime', + method: 'select', + title: 'Choose implementation', + options: ['Small seam', 'Large seam'], + }); + await expect.poll(async () => (await runtime.getSnapshot(left.id)).pendingInteractions) + .toContainEqual(expect.objectContaining({ id: 'question-runtime', status: 'pending' })); + await runtime.respondInteraction(left.id, { + interactionId: 'question-runtime', + optionId: 'question-runtime:option:0', + }); + expect(workers.get(left.id)!.requests.at(-1)).toEqual({ + type: 'extension_ui_response', id: 'question-runtime', value: 'Small seam', + }); + expect((await runtime.getSnapshot(left.id)).pendingInteractions).toContainEqual( + expect.objectContaining({ id: 'question-runtime', status: 'answered' }), + ); const durable = { state: { sessionId: `session-${left.id}`, diff --git a/tests/unit/pi-extension-bundle.test.ts b/tests/unit/pi-extension-bundle.test.ts new file mode 100644 index 0000000..cb7c6ac --- /dev/null +++ b/tests/unit/pi-extension-bundle.test.ts @@ -0,0 +1,95 @@ +// @vitest-environment node + +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import path from 'node:path'; +import { pathToFileURL } from 'node:url'; +import { afterEach, describe, expect, it } from 'vitest'; +import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host'; + +type ExtensionHandler = (...arguments_: unknown[]) => Promise | unknown; + +const roots: string[] = []; +const hosts: PiManagedExtensionHost[] = []; + +afterEach(async () => { + await Promise.all(hosts.splice(0).map((host) => host.close())); + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); +}); + +describe('Makelore Pi extension bundle', () => { + it('loads the real bundle and releases its project lease on tool_result', async () => { + const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-bundle-')); + roots.push(root); + const host = new PiManagedExtensionHost(); + hosts.push(host); + const extensionWorker = await host.registerWorker({ + conversationId: 'conversation-a1', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + const waitingWorker = await host.registerWorker({ + conversationId: 'conversation-a2', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + await Promise.all([ + host.bindRun('conversation-a1', 1, 'run-a1'), + host.bindRun('conversation-a2', 1, 'run-a2'), + ]); + + const previousEnvironment = { + bridge: process.env.MAKELORE_PI_BRIDGE_URL, + token: process.env.MAKELORE_PI_WORKER_TOKEN, + context: process.env.MAKELORE_PI_CONTEXT_FILE, + }; + Object.assign(process.env, extensionWorker.env); + try { + const module = await import(/* @vite-ignore */ pathToFileURL(extensionWorker.extensionPath).href) as { + default(factory: { + registerTool(tool: { name: string }): void; + on(event: string, handler: ExtensionHandler): void; + }): void; + }; + const handlers = new Map(); + const tools: string[] = []; + module.default({ + registerTool: (tool) => tools.push(tool.name), + on: (event, handler) => handlers.set(event, handler), + }); + expect(tools).toEqual(['ask_user']); + + const statuses: Array = []; + const context = { + signal: new AbortController().signal, + ui: { setStatus: (_key: string, text: string | undefined) => statuses.push(text) }, + }; + await handlers.get('tool_call')?.({ toolName: 'read', toolCallId: 'read-1' }, context); + await handlers.get('tool_call')?.({ toolName: 'write', toolCallId: 'write-1' }, context); + expect(statuses).toEqual(['等待项目写入', undefined]); + + let waiterSettled = false; + const waiting = fetch(waitingWorker.env.MAKELORE_PI_BRIDGE_URL as string, { + method: 'POST', + headers: { + authorization: `Bearer ${waitingWorker.env.MAKELORE_PI_WORKER_TOKEN}`, + 'content-type': 'application/json', + }, + body: JSON.stringify({ + action: 'lease.acquire', conversationId: 'conversation-a2', workerGeneration: 1, + runId: 'run-a2', resourceId: 'write-2', + }), + }).then((response) => { + waiterSettled = true; + return response; + }); + await Promise.resolve(); + expect(waiterSettled).toBe(false); + await handlers.get('tool_result')?.({ toolCallId: 'write-1' }); + expect((await waiting).status).toBe(200); + } finally { + if (previousEnvironment.bridge === undefined) delete process.env.MAKELORE_PI_BRIDGE_URL; + else process.env.MAKELORE_PI_BRIDGE_URL = previousEnvironment.bridge; + if (previousEnvironment.token === undefined) delete process.env.MAKELORE_PI_WORKER_TOKEN; + else process.env.MAKELORE_PI_WORKER_TOKEN = previousEnvironment.token; + if (previousEnvironment.context === undefined) delete process.env.MAKELORE_PI_CONTEXT_FILE; + else process.env.MAKELORE_PI_CONTEXT_FILE = previousEnvironment.context; + } + }); +}); diff --git a/tests/unit/pi-extension-host.test.ts b/tests/unit/pi-extension-host.test.ts new file mode 100644 index 0000000..2a6f619 --- /dev/null +++ b/tests/unit/pi-extension-host.test.ts @@ -0,0 +1,120 @@ +// @vitest-environment node + +import { mkdtemp, readFile, rm } from 'node:fs/promises'; +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'; + +const roots: string[] = []; +const hosts: PiManagedExtensionHost[] = []; + +afterEach(async () => { + await Promise.all(hosts.splice(0).map((host) => host.close())); + await Promise.all(roots.splice(0).map((root) => rm(root, { recursive: true, force: true }))); +}); + +async function post( + registration: Awaited>, + body: Record, +): Promise { + return await fetch(registration.env.MAKELORE_PI_BRIDGE_URL as string, { + method: 'POST', + headers: { + authorization: `Bearer ${registration.env.MAKELORE_PI_WORKER_TOKEN}`, + 'content-type': 'application/json', + }, + body: JSON.stringify(body), + }); +} + +describe('managed Pi extension bridge', () => { + 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); + const host = new PiManagedExtensionHost(); + hosts.push(host); + const first = await host.registerWorker({ + conversationId: 'conversation-a', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + await host.bindRun('conversation-a', 1, 'run-a'); + const replacement = await host.registerWorker({ + conversationId: 'conversation-a', generation: 2, projectId: 'project-a', extensionsDir: root, + }); + const context = JSON.parse(await readFile( + replacement.env.MAKELORE_PI_CONTEXT_FILE as string, + 'utf8', + )) as Record; + expect(context).toEqual({ + conversationId: 'conversation-a', workerGeneration: 2, runId: 'run-a', + }); + await first.dispose(); + const response = await post(replacement, { + action: 'lease.acquire', conversationId: 'conversation-a', workerGeneration: 2, + runId: 'run-a', resourceId: 'replacement-tool', + }); + expect(response.status).toBe(200); + }); + + it('validates worker identity and enforces project-scoped leases over loopback HTTP', async () => { + const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-')); + roots.push(root); + const host = new PiManagedExtensionHost(); + hosts.push(host); + const first = await host.registerWorker({ + conversationId: 'conversation-a1', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + const second = await host.registerWorker({ + conversationId: 'conversation-a2', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + const other = await host.registerWorker({ + conversationId: 'conversation-b1', generation: 1, projectId: 'project-b', extensionsDir: root, + }); + await Promise.all([ + host.bindRun('conversation-a1', 1, 'run-a1'), + host.bindRun('conversation-a2', 1, 'run-a2'), + host.bindRun('conversation-b1', 1, 'run-b1'), + ]); + const identity = (conversationId: string, runId: string, resourceId: string) => ({ + action: 'lease.acquire', conversationId, workerGeneration: 1, runId, resourceId, + }); + + const firstResponse = await post(first, identity('conversation-a1', 'run-a1', 'tool-a1')); + expect(firstResponse.status).toBe(200); + const firstLease = await firstResponse.json() as { leaseId: string }; + let sameProjectSettled = false; + const sameProjectFlight = post(second, identity('conversation-a2', 'run-a2', 'tool-a2')) + .then((response) => { + sameProjectSettled = true; + return response; + }); + const otherResponse = await post(other, identity('conversation-b1', 'run-b1', 'tool-b1')); + expect(otherResponse.status).toBe(200); + await Promise.resolve(); + expect(sameProjectSettled).toBe(false); + + const releaseResponse = await post(first, { + action: 'lease.release', + conversationId: 'conversation-a1', + workerGeneration: 1, + runId: 'run-a1', + resourceId: 'tool-a1', + leaseId: firstLease.leaseId, + }); + expect(releaseResponse.status).toBe(200); + expect((await sameProjectFlight).status).toBe(200); + + const forged = await post(second, identity('conversation-a2', 'old-run', 'forged')); + expect(forged.status).toBe(409); + await first.dispose(); + const staleToken = await post(first, identity('conversation-a1', 'run-a1', 'stale')); + expect(staleToken.status).toBe(401); + const currentWorker = await post(other, { + action: 'lease.release', + conversationId: 'conversation-b1', workerGeneration: 1, runId: 'run-b1', + resourceId: 'tool-b1', + leaseId: (await otherResponse.clone().json() as { leaseId: string }).leaseId, + }); + expect(currentWorker.status).toBe(200); + }); +}); diff --git a/tests/unit/pi-extension-ui-projector.test.ts b/tests/unit/pi-extension-ui-projector.test.ts new file mode 100644 index 0000000..e45a027 --- /dev/null +++ b/tests/unit/pi-extension-ui-projector.test.ts @@ -0,0 +1,44 @@ +// @vitest-environment node + +import { describe, expect, it } from 'vitest'; +import { PiExtensionUiProjector } from '../../electron/coding-runtime/pi/extension-ui-projector'; + +describe('Pi extension UI projector', () => { + it('projects registered UI safely and refuses stale editor revisions', () => { + let draftRevision = 4; + const projector = new PiExtensionUiProjector({ + getDraftRevision: () => draftRevision, + knownWidgetKeys: ['makelore.runtime'], + }); + projector.beginRun('conversation-a', 1, 'run-1'); + expect(projector.project('conversation-a', 1, 'run-1', { + type: 'extension_ui_request', id: 'status', method: 'setStatus', + statusKey: 'makelore.write-lease', statusText: '等待项目写入', + })).toMatchObject({ kind: 'status', key: 'makelore.write-lease' }); + expect(projector.project('conversation-a', 1, 'run-1', { + type: 'extension_ui_request', id: 'draft', method: 'set_editor_text', text: 'new text', + })).toEqual({ + kind: 'editor-text', conversationId: 'conversation-a', text: 'new text', draftRevision: 4, + }); + + draftRevision = 5; + expect(projector.project('conversation-a', 1, 'run-1', { + type: 'extension_ui_request', id: 'stale-draft', method: 'set_editor_text', text: 'overwrite', + })).toBeNull(); + expect(projector.getDiagnostics()).toContainEqual({ + method: 'set_editor_text', reason: 'stale-draft-revision', + }); + }); + + it('records bounded diagnostics instead of projecting unknown widget payloads', () => { + const projector = new PiExtensionUiProjector({ getDraftRevision: () => 0 }); + projector.beginRun('conversation-a', 1, 'run-1'); + expect(projector.project('conversation-a', 1, 'run-1', { + type: 'extension_ui_request', id: 'unknown', method: 'setWidget', + widgetKey: 'third-party-widget', widgetLines: ['raw detail'], + })).toBeNull(); + expect(projector.getDiagnostics()).toEqual([{ + method: 'setWidget', reason: 'unsupported-ui-method', + }]); + }); +}); diff --git a/tests/unit/pi-interaction.test.ts b/tests/unit/pi-interaction.test.ts new file mode 100644 index 0000000..b9dba50 --- /dev/null +++ b/tests/unit/pi-interaction.test.ts @@ -0,0 +1,78 @@ +// @vitest-environment node + +import { describe, expect, it } from 'vitest'; +import type { ConversationInteraction } from '../../electron/coding-runtime/contracts'; +import { PiInteractionStore } from '../../electron/coding-runtime/pi/interaction'; +import type { PiRpcCommand } from '../../electron/coding-runtime/pi/rpc-client'; +import type { PiGenerationResourceInput } from '../../electron/coding-runtime/pi/worker-pool'; + +describe('Pi interaction store', () => { + it('responds by exact id and option, then rejects stale generation responses', async () => { + const sent: PiRpcCommand[] = []; + const changes: ConversationInteraction[] = []; + const resources = new Map void>(); + let generation = 1; + let runId = 'run-1'; + const store = new PiInteractionStore({ + getState: () => ({ + conversationId: 'conversation-a', workerId: 'worker-a', state: 'running', generation, + session: { piSessionId: 'session-a', sessionKey: 'key-a' }, + }), + getActiveRun: () => ({ generation, runId }), + send: async (_conversationId, command) => { sent.push(command); }, + trackGenerationResource: (input: PiGenerationResourceInput) => { + resources.set(input.id, input.cancel); + return () => resources.delete(input.id); + }, + }, (interaction) => changes.push(interaction)); + + const opened = store.open('conversation-a', 1, 'run-1', { + type: 'extension_ui_request', id: 'question-1', method: 'select', title: 'Choose', + options: ['Alpha', 'Beta'], + }); + expect(opened?.status).toBe('pending'); + await store.respond('conversation-a', { + interactionId: 'question-1', optionId: 'question-1:option:1', + }); + expect(sent).toEqual([{ + type: 'extension_ui_response', id: 'question-1', value: 'Beta', + }]); + expect(changes.at(-1)?.status).toBe('answered'); + await expect(store.respond('conversation-a', { + interactionId: 'question-1', optionId: 'question-1:option:0', + })).rejects.toThrow('not pending'); + + store.open('conversation-a', 1, 'run-1', { + type: 'extension_ui_request', id: 'question-2', method: 'confirm', title: 'Continue?', message: 'Proceed', + }); + generation = 2; + runId = 'run-2'; + await expect(store.respond('conversation-a', { + interactionId: 'question-2', confirmed: true, + })).rejects.toThrow('stale'); + expect(changes.at(-1)?.status).toBe('cancelled'); + }); + + it('cancels every pending dialog on abort', async () => { + const sent: PiRpcCommand[] = []; + const changes: ConversationInteraction[] = []; + 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); }, + trackGenerationResource: () => () => undefined, + }, (interaction) => changes.push(interaction)); + for (const id of ['question-1', 'question-2']) { + store.open('conversation-a', 1, 'run-1', { + type: 'extension_ui_request', id, method: 'input', title: id, + }); + } + await store.cancelRun('conversation-a', 'run-1', true); + expect(sent).toHaveLength(2); + expect(changes.map(({ status }) => status)).toEqual(['cancelled', 'cancelled']); + expect(store.list()).toEqual([]); + }); +}); diff --git a/tests/unit/pi-managed-worker-opener.test.ts b/tests/unit/pi-managed-worker-opener.test.ts index 7d61a08..ea93a26 100644 --- a/tests/unit/pi-managed-worker-opener.test.ts +++ b/tests/unit/pi-managed-worker-opener.test.ts @@ -26,6 +26,7 @@ import type { } from '../../electron/coding-runtime/pi/rpc-client'; import type { PiWorkerProcessOptions } from '../../electron/coding-runtime/pi/worker-process'; import type { PiRuntimeTelemetryEvent } from '../../electron/coding-runtime/pi/telemetry'; +import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host'; const roots: string[] = []; const NOW = '2026-08-22T16:00:00.000Z'; @@ -54,6 +55,8 @@ class OpenerFakeProcess implements PiWorkerProcessAdapter { }; } + async send(_command: PiRpcCommand): Promise {} + subscribe(_listener: (event: PiRpcEvent) => void): () => void { return () => undefined; } subscribeInvalidation(_listener: (error: PiProcessError) => void): () => void { return () => undefined; } async stop() { return { mode: 'stdin-close' as const, code: 0, signal: null }; } @@ -118,12 +121,14 @@ describe('managed Pi worker opener', () => { const processOptions: PiWorkerProcessOptions[] = []; const telemetry: PiRuntimeTelemetryEvent[] = []; const registry = new PiSessionRegistry({ projectStore }); + const extensionHost = new PiManagedExtensionHost(); const opener = createPiManagedWorkerOpener({ registry, executablePath: 'electron.exe', cliPath: 'pi-cli.js', userDataDir, bundledSkillsDir: path.resolve('resources/coding-skills'), + extensionHost, loadProviderInput: async () => ({ accounts: [account], modelSummaries: [] }), resolveCredential: async () => 'provider-secret-value', createSessionKey: () => 'session-key-a', @@ -155,10 +160,15 @@ describe('managed Pi worker opener', () => { expect(argv).toContain('--system-prompt'); expect(argv).toContain('grilling'); expect(argv).toContain('--session-id'); + expect(argv).toContain('--extension'); + expect(argv).toContain('makelore-runtime-v1.mjs'); + expect(options.additionalArgs?.filter((argument) => argument === '--extension')).toHaveLength(1); expect(argv).not.toContain('PRIVATE MANAGED PROMPT'); expect(argv).not.toContain('provider-secret-value'); expect(Object.values(options.env ?? {})).toContain('provider-secret-value'); expect(options.sensitiveValues).toContain('provider-secret-value'); + expect(options.env?.MAKELORE_PI_BRIDGE_URL).toMatch(/^http:\/\/127\.0\.0\.1:/); + expect(options.env?.MAKELORE_PI_CONTEXT_FILE).toContain('worker-'); } const modelsFile = path.join(userDataDir, 'coding-runtime', 'pi', 'config', 'models.json'); expect(await readFile(modelsFile, 'utf8')).not.toContain('provider-secret-value'); @@ -169,5 +179,8 @@ describe('managed Pi worker opener', () => { expect(JSON.stringify(telemetry)).not.toContain(created.id); expect(JSON.stringify(telemetry)).not.toContain('PRIVATE MANAGED PROMPT'); expect(JSON.stringify(telemetry)).not.toContain('provider-secret-value'); + await first.worker.stop(); + await reopened.worker.stop(); + await extensionHost.close(); }); }); diff --git a/tests/unit/pi-rpc-foundation.test.ts b/tests/unit/pi-rpc-foundation.test.ts index cbdcbab..c39cdae 100644 --- a/tests/unit/pi-rpc-foundation.test.ts +++ b/tests/unit/pi-rpc-foundation.test.ts @@ -166,7 +166,8 @@ describe('Pi worker process', () => { '--no-themes', '--no-context-files', '--no-approve', - '--no-tools', + '--tools', + 'read,bash,edit,write,grep,find,ls,ask_user', '--model', 'model-a', ]); }); diff --git a/tests/unit/pi-runtime-auth-recovery.test.ts b/tests/unit/pi-runtime-auth-recovery.test.ts index 38e2e9c..578a45b 100644 --- a/tests/unit/pi-runtime-auth-recovery.test.ts +++ b/tests/unit/pi-runtime-auth-recovery.test.ts @@ -53,6 +53,8 @@ class AuthFailureWorker implements PiConversationWorker { }; } + async send(_command: PiRpcCommand): Promise {} + subscribe(_listener: (event: PiRpcEvent) => void): () => void { return () => undefined; } subscribeInvalidation(_listener: (error: PiProcessError) => void): () => void { return () => undefined; } async stop() { return { mode: 'stdin-close' as const, code: 0, signal: null }; } diff --git a/tests/unit/pi-worker-pool-process-integration.test.ts b/tests/unit/pi-worker-pool-process-integration.test.ts index 1ec0ba0..6125cf4 100644 --- a/tests/unit/pi-worker-pool-process-integration.test.ts +++ b/tests/unit/pi-worker-pool-process-integration.test.ts @@ -48,6 +48,10 @@ class ProcessBackedWorker implements PiConversationWorker { return this.process.request(command, options); } + send(command: PiRpcCommand): Promise { + return this.process.send(command); + } + subscribe(listener: (event: PiRpcEvent) => void): () => void { return this.process.subscribe(listener); } diff --git a/tests/unit/pi-worker-pool.test.ts b/tests/unit/pi-worker-pool.test.ts index 15088a1..4228b50 100644 --- a/tests/unit/pi-worker-pool.test.ts +++ b/tests/unit/pi-worker-pool.test.ts @@ -52,6 +52,10 @@ class FakeWorker implements PiConversationWorker { return { type: 'response' as const, id: 'fake', success: true }; } + async send(command: PiRpcCommand): Promise { + this.requests.push(command); + } + subscribe(listener: (event: PiRpcEvent) => void): () => void { this.eventListeners.add(listener); return () => this.eventListeners.delete(listener); diff --git a/tests/unit/pi-worker-process-real.test.ts b/tests/unit/pi-worker-process-real.test.ts index e16d188..daa513a 100644 --- a/tests/unit/pi-worker-process-real.test.ts +++ b/tests/unit/pi-worker-process-real.test.ts @@ -7,6 +7,7 @@ import { createRequire } from 'node:module'; import { tmpdir } from 'node:os'; import { join, resolve } from 'node:path'; import { PiWorkerProcess } from '../../electron/coding-runtime/pi/worker-process'; +import { PiManagedExtensionHost } from '../../electron/coding-runtime/pi/extension-host'; const scratchRoots: string[] = []; @@ -39,12 +40,23 @@ describe('locked Pi worker process smoke', () => { const cwd = join(root, 'project'); await Promise.all([mkdir(configDir), mkdir(sessionDir), mkdir(cwd)]); + const extensionHost = new PiManagedExtensionHost(); + const extension = await extensionHost.registerWorker({ + conversationId: 'real-conversation', + generation: 1, + projectId: 'real-project', + extensionsDir: join(root, 'extensions'), + }); + await extensionHost.bindRun('real-conversation', 1, 'real-run'); const worker = await new PiWorkerProcess({ executablePath: electronExecutable, cliPath: join(packageRoot, 'dist', 'cli.js'), cwd, configDir, sessionDir, + additionalArgs: ['--extension', extension.extensionPath], + env: extension.env, + sensitiveValues: extension.sensitiveValues, commandTimeoutMs: 5_000, }).start(); try { @@ -53,9 +65,12 @@ describe('locked Pi worker process smoke', () => { command: 'get_state', success: true, }); + expect(worker.stderrDiagnostic).not.toContain('Failed to load extension'); await expect(worker.stop()).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); } finally { await worker.stop().catch(() => undefined); + await extension.dispose(); + await extensionHost.close(); } }, 15_000); }); diff --git a/tests/unit/pi-write-lease.test.ts b/tests/unit/pi-write-lease.test.ts new file mode 100644 index 0000000..226567b --- /dev/null +++ b/tests/unit/pi-write-lease.test.ts @@ -0,0 +1,42 @@ +// @vitest-environment node + +import { describe, expect, it } from 'vitest'; +import { PiProjectWriteLeaseCoordinator } from '../../electron/coding-runtime/pi/write-lease'; + +describe('Pi project write lease', () => { + it('serializes mutations in one project while allowing other projects to proceed', async () => { + const coordinator = new PiProjectWriteLeaseCoordinator(); + const first = await coordinator.acquire('project-a', 'write-a1'); + let secondSettled = false; + const secondFlight = coordinator.acquire('project-a', 'write-a2').then((lease) => { + secondSettled = true; + return lease; + }); + const otherProject = await coordinator.acquire('project-b', 'write-b1'); + + await Promise.resolve(); + expect(secondSettled).toBe(false); + expect(coordinator.activeCount).toBe(2); + expect(coordinator.waitingCount('project-a')).toBe(1); + + first.release(); + const second = await secondFlight; + expect(second.holderId).toBe('write-a2'); + second.release(); + otherProject.release(); + expect(coordinator.activeCount).toBe(0); + }); + + it('removes a cancelled waiter without disturbing the active lease', async () => { + const coordinator = new PiProjectWriteLeaseCoordinator(); + const active = await coordinator.acquire('project-a', 'active'); + const controller = new AbortController(); + const waiting = coordinator.acquire('project-a', 'waiting', controller.signal); + controller.abort(); + + await expect(waiting).rejects.toThrow('cancelled'); + expect(coordinator.activeCount).toBe(1); + expect(coordinator.waitingCount()).toBe(0); + active.release(); + }); +});