diff --git a/.project-docs/30-worklog/tasks/20260825-pi-worker-exit-thinking-5e7c2a91.md b/.project-docs/30-worklog/tasks/20260825-pi-worker-exit-thinking-5e7c2a91.md new file mode 100644 index 0000000..89a6b50 --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260825-pi-worker-exit-thinking-5e7c2a91.md @@ -0,0 +1,178 @@ +# Task: Fix Pi worker exit terminalization and thinking contract + +## Identity + +- Task ID: 20260825-pi-worker-exit-thinking-5e7c2a91 +- Mode: Feature +- Branch: codex/20260825-pi-worker-exit-thinking-5e7c2a91-pi-worker-exit-thinking +- Worktree: D:\Datas\OthersProjects\makelore-pi-worker-exit-thinking-5e7c2a91 +- Base commit: 274187e3cfe2ae6cb97f2b4bcb422b3fdf06fbf2 +- Owner: codex-root +- Status: In Progress + +## Scope + +- Fix the installed Windows Pi Conversation lifecycle so an accepted run whose + matching worker generation exits or is protocol-invalidated converges within + a bounded time to a recoverable terminal failure instead of remaining + `running` or `aborting`. +- Make abort-after-exit and crash/abort/`agent_settled` ordering monotonic on + the target Conversation projection chain while preserving target isolation, + the original session binding, and no replay of an uncertain prompt. +- Add Main-only worker lifecycle forensics that distinguish unexpected exit, + protocol invalidation, and intentional stop with a required finite reason, + generation, stage, code/signal, and bounded redacted stderr. +- Persist and display only the thinking level Pi actually supports and accepts, + and derive managed custom-model reasoning compatibility from the authoritative + capability data without implicitly replacing a worker generation. +- Add focused, Host/SSE/Renderer, Windows Electron, and final packaged fault- + injection regressions; build and verify a new Windows x64 NSIS installer. + +## Intent And Constraints + +- Keep the Pi hard cutover. Do not add an OpenCode fallback, compatibility + layer, generic Provider watchdog, feature flag, or uncertain prompt replay. +- Serialize crash, abort, and settled handling by Conversation + generation. + A terminal run is monotonic; a stale generation cannot modify the active + Conversation; one crash cannot affect another Conversation. +- On crash/protocol invalidation, finally settle the target run and converge + its queue permit, process lease, pending interaction, extension run, child + worker, and project write lease. Normal live-worker abort continues to wait + for authoritative `agent_settled`. +- Thinking changes and same-account model changes must not restart a normally + running worker. Provider/resource/credential revisions rebuild only after the + accepted run settles, with an explicit generation-replacement reason. +- Every Main-initiated stop must use a bounded reason enum covering application + shutdown, idle eviction, stale-resource rebuild, recover, dispose, fork + replacement, protocol invalidation, and test injection or their exact current + implementation equivalents. Do not log prompt/response/tool/header/token, + expose stderr to Renderer, or write credentials into diagnostics. +- Preserve direct API-key, OAuth, local Provider, existing session persistence, + fork/recover behavior, and the user's installed application under + `D:\Tools\泥土\niancode\Makelore`. +- Use pinned pnpm `10.33.4`, frozen install semantics, focused/full verification, + formal Windows packaging, final artifact/runtime closure checks, and isolated + packaged product fault injection with zero residual proof processes. +- Real external Provider verification remains `Explicitly Waived / Accepted + Risk` with `realTurnVerified=false`; loopback/fault injection is not a real + Provider Pass. macOS and native non-WSL Linux status is unchanged. + +## Project Context Loaded + +- Concurrent Task Gate passed in the isolated worktree owned by `codex-root`; + task identity, branch, worktree, base, and task record match the registry. +- Planning Gate passed after reading project entry/memory/current-state, + architecture/domain/decision/evidence/reflection/commitment/stale records, + the Pi hard-cutover runtime/ticket constraints, both cumulative Windows Pi + hotfix records, the read-only stuck-processing diagnosis, and all active peer + task scopes. +- Base `274187e3cfe2ae6cb97f2b4bcb422b3fdf06fbf2` contains the earlier proxy-token + and persisted-session reopen delivery. The diagnosis worktree is read-only; + dirty `main` and all other task worktrees remain untouched. +- Shared canonical architecture/current-state files predate the Pi delivery. + The current source, Pi cutover contract, cumulative task records, and the + accepted diagnostic evidence are authoritative for this repair. +- The active peer tasks are older, unrelated, test-only, or read-only. The one + main-worktree OpenCode integration task explicitly does not package; no + unresolved semantic conflict changes this plan. +- Eleven copied-session packaged loopback cases settled normally, so Provider + SSE, session corruption, and `task_state` are not established causes. The + incident trigger remains unknown until the replacement artifact records the + bounded exit/stop classification. +- No subagents are used, per user direction and repository policy. +- The committed planning supplement + `2d9d8274d77986fe96e88f02ccba2b5389d2e696` is a direct child of this + task's base and identifies this executor task as `PI-R001`. Its + `RES-INV-001..009`, `RES-ACC-001..016`, release blockers, Main-only privacy + boundary, and final packaged fault-injection matrix are additional acceptance + contracts for this implementation. The pure-document commit will be brought + into the implementation branch before final delivery so planning and code + share one history; it is deferred until the current red-test/code work is at + a safe commit boundary. + +## Plan + +1. Inspect runtime, pool, process/RPC, extension/child/lease, model capability, + Host/SSE, Renderer, E2E, and packaged proof seams; map all worker stop callers + and generation-replacement reasons before changing contracts. +2. Add focused red regressions for active accepted-run exit, abort-after-exit, + exit/abort/settled ordering, target isolation and resource convergence; + unexpected/protocol/intentional diagnostics and redaction; effective + thinking persistence and no worker replacement. +3. Implement the smallest lifecycle, diagnostics, and capability changes that + satisfy the existing Pi contracts, retaining the original error category + and session binding through recover. +4. Extend Host SSE/Renderer unlock-and-recover and Windows packaged deterministic + fault-injection proof for running exit, abort-after-exit, settled-before-close, + strict JSONL invalidation, deliberate stop, isolation, no prompt replay, no + unexplained generation replacement, and zero residual processes. +5. Run frozen install, focused tests, typecheck, lint, full unit, `build:vite`, + Windows Electron E2E, formal `package:win`, artifact/runtime/closure + verification, and final packaged proofs from a clean committed candidate. +6. Record exact commits, changed files, verification results, structured + packaged evidence, installer size/SHA-256, accepted/deferred risks, clean git + state, and complete the Task Documentation Gate. + +## Outcome + +- Implemented matching-generation worker crash/protocol terminalization on the + per-Conversation projection chain. An accepted run now reaches recoverable + `error` without overriding an earlier terminal `agent_settled`; abort after + exit cannot restore `running`. +- Generation invalidation now cancels command, interaction, child/dispatch, + extension-run, process-permit, and write-lease resources for only the target + Conversation. Recover reopens the same persisted Pi binding and never + replays the uncertain accepted prompt. +- Added finite, required Main stop/replacement reasons and separate structured + lifecycle classification for unexpected exit, strict protocol invalidation, + and intentional stop. `PI_RPC_EXITED` retains bounded stderr after token, + header, credential, and managed-path redaction; prompt/response/tool content + is not copied into lifecycle diagnostics or Renderer state. +- Same-account model and thinking changes no longer replace the worker. + Hydration and mutations query Pi capabilities/state and persist/project only + the effective level. The managed `deepseek-v4-pro` profile now derives its + reasoning, DeepSeek compatibility, context/output limits, and supported + product thinking level from the imported authoritative capability profile; + unsupported levels reject instead of presenting false success. +- Renderer error state displays `本地 Agent 已中断，原请求未自动重发。`, + leaves the Composer editable, exposes recovery, and does not remain in + processing/recovering. The packaged proof now drives the real final-ASAR Main + composition through target exit, abort-after-exit, protocol invalidation, + settled-before-close, target recovery, sibling isolation, child/write-lease + cleanup, explicit deliberate stop, redacted lifecycle evidence, and clean + process exit. +- Formal NSIS packaging and final packaged execution are still pending; no + installer or real-Provider result is claimed at this checkpoint. + +## Verification + +- `corepack pnpm install --frozen-lockfile` — pass with pinned pnpm `10.33.4`. +- Focused Pi/runtime/pool/RPC/extension/subagent/UI regressions — pass, latest + combined run: 6 files / 60 tests. +- `corepack pnpm run typecheck` — pass. +- `corepack pnpm run lint:check` — pass with 0 errors and 5 pre-existing + warnings in unrelated `Home`/`Makelore` files. +- `corepack pnpm test` — pass: 178 files / 1519 passed / 2 skipped, plus the + isolated pressure file 1/1 passed. +- `corepack pnpm run build:vite` — pass for Renderer, Main, Preload, and utility + bundles; existing dynamic-import and chunk-size warnings remain non-blocking. +- `corepack pnpm run test:electron:windows` — pass: 2 files / 4 tests. +- Formal `package:win`, artifact/runtime closure verifiers, final packaged + resilience proof, installer byte size/SHA-256, and zero-residue evidence — + pending the clean committed candidate. + +## Follow-ups + +- Later soak and any new real-external-Provider occurrence may use the new + bounded diagnostics to identify the initiating exit class. They are not + substitutes for this task's required deterministic convergence proof. +- Real Provider protocol/concurrency/credential-isolation verification remains + explicitly waived accepted risk and must continue to report + `realTurnVerified=false`. +- `PI-R002` loopback soak and field incident classification start only after + `PI-R001` is delivered. They do not block this repair or broaden its current + implementation scope. + +## Promotion Candidates + +- None recorded. diff --git a/electron/coding-runtime/pi/extension-host.ts b/electron/coding-runtime/pi/extension-host.ts index 0c2f303..8eb1740 100644 --- a/electron/coding-runtime/pi/extension-host.ts +++ b/electron/coding-runtime/pi/extension-host.ts @@ -158,6 +158,25 @@ export class PiManagedExtensionHost { this.productTools = productTools; } + getDiagnostics(): { + registrations: { parent: number; child: number }; + writeLeases: { active: number; waiting: number }; + bridgeRequests: number; + } { + const records = [...this.registrations.values()]; + return { + registrations: { + parent: records.filter(({ role }) => role === 'parent').length, + child: records.filter(({ role }) => role === 'child').length, + }, + writeLeases: { + active: records.reduce((total, { leases }) => total + leases.size, 0), + waiting: records.reduce((total, { waiters }) => total + waiters.size, 0), + }, + bridgeRequests: this.requestFlights.size, + }; + } + async registerWorker(input: RegisterPiExtensionWorkerInput): Promise { if (!Number.isSafeInteger(input.generation) || input.generation <= 0) { throw new Error('Worker generation must be a positive safe integer'); diff --git a/electron/coding-runtime/pi/process-errors.ts b/electron/coding-runtime/pi/process-errors.ts index fd9ca10..78cb396 100644 --- a/electron/coding-runtime/pi/process-errors.ts +++ b/electron/coding-runtime/pi/process-errors.ts @@ -13,12 +13,16 @@ export type PiProcessErrorDetails = { cause?: unknown; generation?: number; diagnostic?: string; + exitCode?: number | null; + signal?: NodeJS.Signals | null; }; export class PiProcessError extends Error { readonly code: PiProcessErrorCode; readonly generation?: number; readonly diagnostic?: string; + readonly exitCode?: number | null; + readonly signal?: NodeJS.Signals | null; constructor( code: PiProcessErrorCode, @@ -30,6 +34,8 @@ export class PiProcessError extends Error { this.code = code; this.generation = details.generation; this.diagnostic = details.diagnostic; + this.exitCode = details.exitCode; + this.signal = details.signal; } } diff --git a/electron/coding-runtime/pi/provider-config.ts b/electron/coding-runtime/pi/provider-config.ts index 6b11ad8..ed199b0 100644 --- a/electron/coding-runtime/pi/provider-config.ts +++ b/electron/coding-runtime/pi/provider-config.ts @@ -41,9 +41,11 @@ export interface PiProviderModelDescriptor { contextWindow?: number; maxOutputTokens?: number; compat?: { - thinkingFormat: 'openrouter'; - sessionAffinityFormat: 'openrouter'; + thinkingFormat?: 'deepseek' | 'openrouter'; + sessionAffinityFormat?: 'openrouter'; + requiresReasoningContentOnAssistantMessages?: boolean; }; + thinkingLevelMap?: Partial>; } export interface PiProviderDescriptor { @@ -71,6 +73,7 @@ export interface PiModelsFile { maxTokens?: number; cost: { input: number; output: number; cacheRead: number; cacheWrite: number }; compat?: PiProviderModelDescriptor['compat']; + thinkingLevelMap?: PiProviderModelDescriptor['thinkingLevelMap']; }>; }>; } @@ -312,10 +315,17 @@ function modelDescriptor( id: modelId, name: summary?.name || (typeof backend?.name === 'string' && backend.name.trim()) || modelId, input: supportsImage ? ['text', 'image'] : ['text'], - reasoning: summary?.supportsReasoning === true || backend?.reasoning === true, + reasoning: summary?.supportsReasoning === true + || profile?.pi?.reasoning === true + || backend?.reasoning === true, ...(contextWindow ? { contextWindow } : {}), ...(maxOutputTokens ? { maxOutputTokens } : {}), - ...(compat ? { compat } : {}), + ...(compat || profile?.pi?.compat + ? { compat: { ...compat, ...profile?.pi?.compat } } + : {}), + ...(profile?.pi?.thinkingLevelMap + ? { thinkingLevelMap: { ...profile.pi.thinkingLevelMap } } + : {}), }; } @@ -403,6 +413,9 @@ function modelsFileForDescriptors(descriptors: readonly PiProviderDescriptor[]): ...(model.maxOutputTokens ? { maxTokens: model.maxOutputTokens } : {}), cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0 }, ...(model.compat ? { compat: { ...model.compat } } : {}), + ...(model.thinkingLevelMap + ? { thinkingLevelMap: { ...model.thinkingLevelMap } } + : {}), })), }, ])), diff --git a/electron/coding-runtime/pi/release-proof.ts b/electron/coding-runtime/pi/release-proof.ts index db0992e..4b3338b 100644 --- a/electron/coding-runtime/pi/release-proof.ts +++ b/electron/coding-runtime/pi/release-proof.ts @@ -6,6 +6,7 @@ import path from 'node:path'; import { promisify } from 'node:util'; import type { CodingProductComposition } from '../../api/coding-product-services'; +import { getRecentLogs } from '../../utils/logger'; import { createCodingConversationStore } from '../../coding-projects/conversation-store'; import { createCodingProjectAgent } from '../../coding-projects/project-config'; import { @@ -25,7 +26,7 @@ import { PiManagedExtensionHost } from './extension-host'; import { PiManagedInputRevisionCoordinator } from './managed-input-revision'; import { runPiReleasePressureCleanup } from './release-proof-cleanup'; import type { PiRpcEvent } from './rpc-client'; -import { createPiManagedWorkerOpener } from './runtime'; +import { createPiManagedWorkerOpener, PiConversationRuntime } from './runtime'; import { PiSessionRegistry } from './session-registry'; import { createPiManagedSubagentChildOpener } from './subagent-child'; import { PiSubagentScheduler } from './subagent'; @@ -35,7 +36,7 @@ import { PiProcessBudget, PiWorkerPool, type PiWorkerPoolEvent } from './worker- import { PiProjectWriteLeaseCoordinator, type PiProjectWriteLease } from './write-lease'; type ProofWorkerRole = 'parent' | 'child'; -type ProofProviderMode = 'subagent' | 'pressure'; +type ProofProviderMode = 'subagent' | 'pressure' | 'resilience'; type ProofMilestone = PiRuntimeTelemetryEvent['milestone'] | 'agent.start' | 'provider.first_event'; type ProofMilestoneSource = 'main.telemetry' | 'pi.agent_start' | 'pi.assistant_message_start'; @@ -67,6 +68,7 @@ type LocalProofProvider = { requests: ProviderRequest[]; activeCounts(): { parent: number; child: number }; releaseChildren(): void; + releaseParents(): void; releaseAll(): void; close(): Promise; }; @@ -102,6 +104,18 @@ type ProxyCompositionRun = { activeStatus?: PiReleaseProxyCompositionStatus; }; +type ResilienceCompositionRun = { + composition: CodingProductComposition; + provider: LocalProofProvider; + projectId: string; + projectPath: string; + targetConversationId: string; + otherConversationId: string; + targetBinding: { piSessionId: string; sessionKey: string }; + otherBinding: { piSessionId: string; sessionKey: string }; + hostToken: string; +}; + type InspectedPiProcess = { processId: number; role: ProofWorkerRole; @@ -187,6 +201,49 @@ export interface PiReleaseProxyCompositionProof extends PiReleaseProxyCompositio released: { workers: number }; } +export interface PiReleaseResilienceStatus { + target: { + conversationId: string; + workerStatus: string; + workerGeneration: number; + runStatus: string; + errorCode: string | null; + recoverable: boolean; + bindingPreserved: boolean; + }; + other: { + conversationId: string; + workerStatus: string; + workerGeneration: number; + runStatus: string; + bindingPreserved: boolean; + }; + providerRequests: { parent: number; child: number }; + activeProviderRequests: { parent: number; child: number }; + resources: ReturnType; + processes: { supported: boolean; parent: number[]; child: number[] }; + realTurnVerified: false; +} + +export interface PiReleaseResilienceProof extends PiReleaseResilienceStatus { + lifecycle: { + unexpectedExit: boolean; + protocolInvalidation: boolean; + intentionalStop: boolean; + everyStopHasReason: boolean; + everyReplacementHasReason: boolean; + diagnosticRedacted: boolean; + promptFree: boolean; + projectPathFree: boolean; + }; + released: { + workers: number; + processes: number; + resources: ReturnType; + }; + realTurnVerified: false; +} + const PROOF_ACCOUNT_ID = 'release-proof-account'; const PROOF_AGENT_ID = 'release-proof-agent'; const PROOF_MODEL_ID = 'release-proof-model'; @@ -205,6 +262,7 @@ const EXPECTED_TURN_MILESTONES: readonly ProofMilestone[] = [ let pressureRun: PressureRun | null = null; let proxyCompositionRun: ProxyCompositionRun | null = null; +let resilienceCompositionRun: ResilienceCompositionRun | null = null; const execFileAsync = promisify(execFile); async function waitFor(predicate: () => boolean, message: string): Promise { @@ -298,7 +356,11 @@ function respondWithText(response: ServerResponse, model: string, text: string): finishResponse(response, model); } -function respondWithSubagentCall(response: ServerResponse, model: string): void { +function respondWithSubagentCall( + response: ServerResponse, + model: string, + toolProfile: 'read-only' | 'coding' = 'read-only', +): void { response.writeHead(200, { 'content-type': 'text/event-stream' }); writeChunk(response, model, { role: 'assistant', @@ -313,7 +375,7 @@ function respondWithSubagentCall(response: ServerResponse, model: string): void tasks: [{ agentId: PROOF_AGENT_ID, task: 'Return REAL_CHILD_COMPLETE for final packaged qualification.', - toolProfile: 'read-only', + toolProfile, }], }), }, @@ -323,6 +385,30 @@ function respondWithSubagentCall(response: ServerResponse, model: string): void response.end('data: [DONE]\n\n'); } +function respondWithHoldingBashCall(response: ServerResponse, model: string): void { + response.writeHead(200, { 'content-type': 'text/event-stream' }); + writeChunk(response, model, { + role: 'assistant', + tool_calls: [{ + index: 0, + id: 'release-proof-holding-bash', + type: 'function', + function: { + name: 'bash', + arguments: JSON.stringify({ + command: 'powershell.exe -NoProfile -NonInteractive -Command "Start-Sleep -Seconds 120"', + }), + }, + }], + }, null); + writeChunk(response, model, {}, 'tool_calls'); + response.end('data: [DONE]\n\n'); +} + +function requestContains(body: Record, marker: string): boolean { + return JSON.stringify(body.messages ?? []).includes(marker); +} + async function startLocalProofProvider(mode: ProofProviderMode): Promise { const requests: ProviderRequest[] = []; const held = new Set(); @@ -342,6 +428,32 @@ async function startLocalProofProvider(mode: ProofProviderMode): Promise held.delete(entry)); + return; + } + if (mode === 'subagent' && role === 'parent' && !toolResult) { await delay(PROOF_PROVIDER_FIRST_EVENT_DELAY_MS); respondWithSubagentCall(response, model); @@ -393,6 +505,7 @@ async function startLocalProofProvider(mode: ProofProviderMode): Promise role === 'child').length, }), releaseChildren: () => release('child'), + releaseParents: () => release('parent'), releaseAll: () => release(), close: async () => { if (closed) return; @@ -1276,6 +1389,296 @@ export async function finishFinalAsarProxyCompositionProof(): Promise { + const binding = await run.composition.projects.conversationStore(run.projectPath).get(conversationId); + return binding?.piSessionId === expected.piSessionId && binding.sessionKey === expected.sessionKey; +} + +export async function getFinalAsarResilienceStatus(): Promise { + const run = resilienceCompositionRun; + if (!run) throw new Error('PI resilience proof is not running'); + const runtime = resilienceRuntime(run); + const [target, other, processInspection, targetBindingPreserved, otherBindingPreserved] = await Promise.all([ + run.composition.conversations.getSnapshot(run.targetConversationId), + run.composition.conversations.getSnapshot(run.otherConversationId), + inspectWindowsPiProcesses(run.hostToken), + resilienceBindingPreserved(run, run.targetConversationId, run.targetBinding), + resilienceBindingPreserved(run, run.otherConversationId, run.otherBinding), + ]); + const targetError = target.run.error ?? target.worker.error; + return { + target: { + conversationId: run.targetConversationId, + workerStatus: target.worker.status, + workerGeneration: target.cursor.workerGeneration, + runStatus: target.run.status, + errorCode: targetError?.code ?? null, + recoverable: targetError?.recoverable ?? false, + bindingPreserved: targetBindingPreserved, + }, + other: { + conversationId: run.otherConversationId, + workerStatus: other.worker.status, + workerGeneration: other.cursor.workerGeneration, + runStatus: other.run.status, + bindingPreserved: otherBindingPreserved, + }, + providerRequests: providerRequestCounts(run.provider), + activeProviderRequests: run.provider.activeCounts(), + resources: runtime.getResilienceProofDiagnostics(), + processes: { + supported: processInspection.supported, + parent: processInspection.processes + .filter(({ role }) => role === 'parent') + .map(({ processId }) => processId) + .sort((left, right) => left - right), + child: processInspection.processes + .filter(({ role }) => role === 'child') + .map(({ processId }) => processId) + .sort((left, right) => left - right), + }, + realTurnVerified: false, + }; +} + +async function waitForResilienceStatus( + predicate: (status: PiReleaseResilienceStatus) => boolean, + message: string, +): Promise { + const deadline = Date.now() + 30_000; + let latest: PiReleaseResilienceStatus | null = null; + while (Date.now() < deadline) { + latest = await getFinalAsarResilienceStatus(); + if (predicate(latest)) return latest; + await delay(20); + } + throw new Error(`${message}: ${JSON.stringify(latest)}`); +} + +export async function startFinalAsarResilienceProof(input: { + composition: CodingProductComposition; + projectPath: string; + hostProxyBaseUrl: string; + hostToken: string; +}): Promise<{ + projectId: string; + targetConversationId: string; + otherConversationId: string; + bindingsEstablished: true; + otherRunAccepted: true; + realTurnVerified: false; +}> { + if (resilienceCompositionRun) throw new Error('PI resilience proof is already running'); + const hostToken = input.hostToken.trim(); + if (!hostToken) throw new Error('Current Main Host token is unavailable'); + const provider = await startLocalProofProvider('resilience'); + let projectId: string | null = null; + try { + seedWorksSquareAIGatewayCredential({ + accessToken: 'release-proof-upstream-only', + oneApiBaseUrl: provider.baseUrl, + }); + const providerService = getProviderService(); + await providerService.createAccount(proxyProviderAccount(input.hostProxyBaseUrl)); + await providerService.setDefaultAccount(PROOF_ACCOUNT_ID); + const project = await input.composition.projects.createProject({ projectPath: input.projectPath }); + projectId = project.project.id; + await createCodingProjectAgent(input.projectPath, { + id: PROOF_AGENT_ID, + avatarId: 'avatar-01', + roleName: 'Packaged resilience proof', + name: 'Packaged resilience proof agent', + model: { + accountId: PROOF_ACCOUNT_ID, + modelId: PROOF_MODEL_ID, + thinkingLevel: 'medium', + }, + modelResolution: 'resolved', + responsibility: { + mission: 'Exercise packaged worker failure convergence', + owns: [], + boundaries: [], + collaborators: [], + principles: [], + }, + prompt: 'Follow the controlled loopback resilience qualification Provider.', + skillIds: [], + }); + const other = await input.composition.conversations.createConversation({ + projectId, + agentId: PROOF_AGENT_ID, + title: 'Resilience isolation control', + }); + const target = await input.composition.conversations.createConversation({ + projectId, + agentId: PROOF_AGENT_ID, + title: 'Resilience fault target', + }); + await Promise.all([ + input.composition.conversations.getSnapshot(other.id), + input.composition.conversations.getSnapshot(target.id), + ]); + const store = input.composition.projects.conversationStore(input.projectPath); + const [targetStored, otherStored] = await Promise.all([store.get(target.id), store.get(other.id)]); + if (!targetStored?.piSessionId || !targetStored.sessionKey + || !otherStored?.piSessionId || !otherStored.sessionKey) { + throw new Error('Resilience proof did not establish both Pi session bindings'); + } + resilienceCompositionRun = { + composition: input.composition, + provider, + projectId, + projectPath: input.projectPath, + targetConversationId: target.id, + otherConversationId: other.id, + targetBinding: { piSessionId: targetStored.piSessionId, sessionKey: targetStored.sessionKey }, + otherBinding: { piSessionId: otherStored.piSessionId, sessionKey: otherStored.sessionKey }, + hostToken, + }; + const accepted = await input.composition.conversations.acceptPrompt({ + conversationId: other.id, + clientRequestId: 'release-proof-other-active', + mode: 'prompt', + text: 'RESILIENCE_OTHER_ACTIVE', + attachments: [], + }); + if (!accepted.accepted) throw new Error('Resilience isolation run was not accepted'); + await waitForResilienceStatus( + (status) => status.other.runStatus === 'running' + && status.activeProviderRequests.parent === 1, + 'Resilience isolation run did not become active', + ); + return { + projectId, + targetConversationId: target.id, + otherConversationId: other.id, + bindingsEstablished: true, + otherRunAccepted: true, + realTurnVerified: false, + }; + } catch (error) { + provider.releaseAll(); + if (projectId) await input.composition.projects.removeProject(projectId).catch(() => undefined); + await provider.close().catch(() => undefined); + clearWorksSquareAIGatewayCredential(); + await getProviderService().deleteAccount(PROOF_ACCOUNT_ID).catch(() => undefined); + resilienceCompositionRun = null; + throw error; + } +} + +export async function injectFinalAsarResilienceFailure( + failure: 'unexpected_exit' | 'protocol_invalidation', +): Promise<{ generation: number; terminalizationMs: number; status: PiReleaseResilienceStatus }> { + const run = resilienceCompositionRun; + if (!run) throw new Error('PI resilience proof is not running'); + const startedAt = Date.now(); + const injected = await resilienceRuntime(run).injectWorkerFailureForProof( + run.targetConversationId, + failure, + ); + const status = await waitForResilienceStatus( + (candidate) => candidate.target.workerStatus === 'error' + && candidate.target.workerGeneration === injected.generation, + 'Injected Pi worker failure did not reach the target Conversation', + ); + const terminalizationMs = Date.now() - startedAt; + if (terminalizationMs > 2_000) { + throw new Error(`Injected Pi worker failure exceeded the 2s convergence bound: ${terminalizationMs}ms`); + } + return { ...injected, terminalizationMs, status }; +} + +export async function abortFinalAsarResilienceTarget(): Promise { + const run = resilienceCompositionRun; + if (!run) throw new Error('PI resilience proof is not running'); + await run.composition.conversations.abort(run.targetConversationId); + return await getFinalAsarResilienceStatus(); +} + +export function releaseFinalAsarResilienceParents(): void { + const run = resilienceCompositionRun; + if (!run) throw new Error('PI resilience proof is not running'); + run.provider.releaseParents(); +} + +export async function finishFinalAsarResilienceProof(): Promise { + const run = resilienceCompositionRun; + if (!run) throw new Error('PI resilience proof is not running'); + try { + const status = await getFinalAsarResilienceStatus(); + run.provider.releaseAll(); + await run.composition.projects.removeProject(run.projectId); + const processInspection = await inspectWindowsPiProcesses(run.hostToken); + const resources = resilienceRuntime(run).getResilienceProofDiagnostics(); + const lifecycleLogs = getRecentLogs().filter((line) => ( + line.includes('[PiWorkerLifecycle]') + && (line.includes(run.targetConversationId) || line.includes(run.otherConversationId)) + )); + const stopLogs = lifecycleLogs.filter((line) => line.includes('"stage": "stop_requested"')); + const replacementLogs = lifecycleLogs.filter((line) => line.includes('worker.replacement_')); + const lifecycle = { + unexpectedExit: lifecycleLogs.some((line) => line.includes('"classification": "unexpected_exit"')), + protocolInvalidation: lifecycleLogs.some((line) => line.includes('"classification": "protocol_invalidation"')), + intentionalStop: lifecycleLogs.some((line) => line.includes('"classification": "intentional_stop"')), + everyStopHasReason: stopLogs.length > 0 && stopLogs.every((line) => line.includes('"reason":')), + everyReplacementHasReason: replacementLogs.every((line) => line.includes('"reason":')), + diagnosticRedacted: lifecycleLogs.some((line) => line.includes('[REDACTED]')) + && lifecycleLogs.every((line) => !line.includes('packaged-proof-secret')), + promptFree: lifecycleLogs.every((line) => !line.includes('RESILIENCE_')), + projectPathFree: lifecycleLogs.every((line) => !line.includes(run.projectPath)), + }; + if (Object.values(lifecycle).some((value) => value !== true)) { + throw new Error(`Packaged Pi lifecycle evidence is incomplete: ${JSON.stringify(lifecycle)}`); + } + const released = { + workers: run.composition.runtime.getDiagnostics().workers.length, + processes: processInspection.processes.length, + resources, + }; + if (released.workers !== 0 + || released.processes !== 0 + || released.resources.pool.processBudget.active !== 0 + || released.resources.pool.processBudget.waiting !== 0 + || released.resources.pool.runs.active !== 0 + || released.resources.pool.runs.waiting !== 0 + || released.resources.subagents?.activeChildPermits !== 0 + || released.resources.subagents?.waitingChildPermits !== 0 + || released.resources.subagents?.activeDispatches !== 0 + || released.resources.subagents?.activeParents !== 0 + || released.resources.extension?.registrations.parent !== 0 + || released.resources.extension?.registrations.child !== 0 + || released.resources.extension?.writeLeases.active !== 0 + || released.resources.extension?.writeLeases.waiting !== 0) { + throw new Error(`Packaged Pi resources were not fully released: ${JSON.stringify(released)}`); + } + await run.provider.close(); + clearWorksSquareAIGatewayCredential(); + await getProviderService().deleteAccount(PROOF_ACCOUNT_ID).catch(() => undefined); + resilienceCompositionRun = null; + return { ...status, lifecycle, released, realTurnVerified: false }; + } catch (error) { + run.provider.releaseAll(); + await run.composition.projects.removeProject(run.projectId).catch(() => undefined); + await run.provider.close().catch(() => undefined); + clearWorksSquareAIGatewayCredential(); + await getProviderService().deleteAccount(PROOF_ACCOUNT_ID).catch(() => undefined); + resilienceCompositionRun = null; + throw error; + } +} + export async function startFinalAsarPressureProof(): Promise { if (pressureRun) throw new Error('PI release pressure proof is already running'); pressureRun = await startPressureRun(); diff --git a/electron/coding-runtime/pi/runtime.ts b/electron/coding-runtime/pi/runtime.ts index fd33150..26b3dd9 100644 --- a/electron/coding-runtime/pi/runtime.ts +++ b/electron/coding-runtime/pi/runtime.ts @@ -1,5 +1,6 @@ import { randomUUID } from 'node:crypto'; import path from 'node:path'; +import { logger } from '../../utils/logger'; import type { ModelSummary, ProviderAccount } from '../../shared/providers/types'; import { validateSessionKey } from '../../coding-projects/conversation-store'; import { @@ -59,6 +60,8 @@ import type { import { PiWorkerProcess, type PiWorkerProcessOptions, + type PiWorkerProofFailure, + type PiWorkerStopReason, type PiWorkerStopResult, } from './worker-process'; import { @@ -118,7 +121,8 @@ export interface PiWorkerProcessAdapter { send(command: PiRpcCommand): Promise; subscribe(listener: (event: PiRpcEvent) => void): () => void; subscribeInvalidation(listener: (error: PiProcessError) => void): () => void; - stop(): Promise; + stop(reason: PiWorkerStopReason): Promise; + injectFailureForProof?(failure: PiWorkerProofFailure): Promise; } export interface PiManagedProviderInput { @@ -172,14 +176,21 @@ class ManagedPiConversationWorker implements PiConversationWorker { return this.process.subscribeInvalidation(listener); } - async stop(): Promise { + async stop(reason: PiWorkerStopReason): Promise { this.unsubscribeExtensionInvalidation(); try { - return await this.process.stop(); + return await this.process.stop(reason); } finally { await this.disposeExtension(); } } + + async injectFailureForProof(failure: PiWorkerProofFailure): Promise { + if (!this.process.injectFailureForProof) { + throw new Error('Managed Pi process does not support proof failure injection'); + } + await this.process.injectFailureForProof(failure); + } } export function createPiManagedWorkerOpener( @@ -261,6 +272,8 @@ export function createPiManagedWorkerOpener( ], env: { ...credential.env, ...extension.env }, sensitiveValues: [...credential.sensitiveValues, ...extension.sensitiveValues], + conversationId: input.conversation.conversationId, + workerGeneration: input.generation, }); let unsubscribeExtensionInvalidation = process.subscribeInvalidation(() => { void extension.dispose(); @@ -348,7 +361,7 @@ export function createPiManagedWorkerOpener( }; } catch (error) { unsubscribeExtensionInvalidation(); - await process.stop().catch(() => undefined); + await process.stop('open_failure').catch(() => undefined); await extension.dispose(); throw error; } @@ -377,6 +390,40 @@ function clone(value: T): T { return structuredClone(value); } +const PRODUCT_THINKING_LEVELS = new Set([ + 'off', + 'minimal', + 'low', + 'medium', + 'high', +]); + +function productThinkingLevel(value: unknown): ProductModelRef['thinkingLevel'] | null { + return typeof value === 'string' + && PRODUCT_THINKING_LEVELS.has(value as ProductModelRef['thinkingLevel']) + ? value as ProductModelRef['thinkingLevel'] + : null; +} + +function availableThinkingLevels(value: unknown): ProductModelRef['thinkingLevel'][] { + if (!value || typeof value !== 'object' || Array.isArray(value)) return []; + const levels = (value as { levels?: unknown }).levels; + if (!Array.isArray(levels)) return []; + return [...new Set(levels.flatMap((level) => { + const normalized = productThinkingLevel(level); + return normalized ? [normalized] : []; + }))]; +} + +function effectiveThinkingLevel(value: unknown): ProductModelRef['thinkingLevel'] | null { + if (!value || typeof value !== 'object' || Array.isArray(value)) return null; + return productThinkingLevel((value as { thinkingLevel?: unknown }).thinkingLevel); +} + +function runIsTerminal(status: ConversationSnapshot['run']['status']): boolean { + return status === 'idle' || status === 'error'; +} + function publicWorkerState(state: PiWorkerPoolState): ConversationSnapshot['worker'] { if (state.state === 'spawning') return { status: 'starting', generation: state.generation }; if (state.state === 'crashed') { @@ -415,6 +462,13 @@ function runtimeFailure(error: unknown): CodingRuntimePublicError { recoverable: true, }; } + if (error.code === 'PI_RPC_EXITED') { + return { + code: 'CODING_RUNTIME_START_FAILED', + message: '本地 Agent 已中断，原请求未自动重发。', + recoverable: true, + }; + } } return { code: 'CODING_RUNTIME_START_FAILED', @@ -458,6 +512,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { private readonly isAuthenticationError: ((error: unknown) => boolean) | undefined; private readonly refreshCredential: ((accountId: string) => Promise) | undefined; private readonly extensionHost: PiManagedExtensionHost | undefined; + private readonly subagentScheduler: PiSubagentScheduler | undefined; private readonly interactions: PiInteractionStore; private readonly extensionUi: PiExtensionUiProjector; private readonly onExtensionUiProjection: ((projection: PiExtensionUiProjection) => void) | undefined; @@ -493,6 +548,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.isAuthenticationError = options.isAuthenticationError; this.refreshCredential = options.refreshCredential; this.extensionHost = options.extensionHost; + this.subagentScheduler = options.subagentScheduler; if (options.subagentScheduler && !this.extensionHost) { throw new Error('Subagent scheduler requires the managed extension host'); } @@ -657,6 +713,8 @@ export class PiConversationRuntime implements CodingConversationRuntime { async abort(conversationId: string): Promise { await this.waitForProjection(conversationId); const current = this.snapshot(conversationId).run; + if (runIsTerminal(current.status)) return; + const generation = this.snapshot(conversationId).cursor.workerGeneration; this.emit(conversationId, { op: 'run.state', run: { ...current, status: 'aborting' }, @@ -670,8 +728,14 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.extensionUi.endRun(conversationId, current.runId); } } catch (error) { + await this.waitForProjection(conversationId); const latest = this.snapshot(conversationId).run; - if (latest.runId === current.runId && latest.status === 'aborting') { + const worker = this.pool.getState(conversationId); + if (latest.runId === current.runId + && latest.status === 'aborting' + && (!worker || worker.state === 'crashed' || worker.generation !== generation)) { + this.failRun(conversationId, current.runId!, error, generation); + } else if (latest.runId === current.runId && latest.status === 'aborting') { this.emit(conversationId, { op: 'run.state', run: current }, current.runId); } throw error; @@ -720,17 +784,31 @@ export class PiConversationRuntime implements CodingConversationRuntime { } throw error; } - return clone(persisted); + return clone(this.snapshot(input.conversationId).conversation.model); } await this.pool.request(input.conversationId, { type: 'set_model', provider: selection.runtimeProviderId, modelId: selection.modelId, }); - const persisted = await this.registry.setModel(input.conversationId, model); - this.pool.updateConversationModel(input.conversationId, persisted); - this.replaceModel(input.conversationId, persisted); - return clone(persisted); + const capabilities = await this.pool.request<{ levels?: unknown }>( + input.conversationId, + { type: 'get_available_thinking_levels' }, + { retry: 'read-only-once' }, + ); + const state = await this.pool.request( + input.conversationId, + { type: 'get_state' }, + { retry: 'read-only-once' }, + ); + return await this.persistEffectiveThinking( + input.conversationId, + model, + state.data, + capabilities.data, + true, + true, + ); } async setThinking(input: SetThinkingLevelInput): Promise { @@ -743,17 +821,44 @@ export class PiConversationRuntime implements CodingConversationRuntime { true, ); } + const capabilities = await this.pool.request<{ levels?: unknown }>( + input.conversationId, + { type: 'get_available_thinking_levels' }, + { retry: 'read-only-once' }, + ); + const available = availableThinkingLevels(capabilities.data); + const withCapabilities: ConversationModelState = { + ...clone(current), + ...(available.length > 0 ? { availableThinkingLevels: available } : {}), + }; + this.replaceModel(input.conversationId, withCapabilities); + if (!available.includes(input.thinkingLevel)) { + throw new CodingRuntimeContractError( + 'CODING_MODEL_UNAVAILABLE', + 'The selected thinking level is not supported by this model', + true, + ); + } await this.pool.request(input.conversationId, { type: 'set_thinking_level', level: input.thinkingLevel, }); - const model: ConversationModelState = { + const state = await this.pool.request( + input.conversationId, + { type: 'get_state' }, + { retry: 'read-only-once' }, + ); + const persisted = await this.persistEffectiveThinking(input.conversationId, { model: { ...current.model, thinkingLevel: input.thinkingLevel }, modelResolution: 'resolved', - }; - const persisted = await this.registry.setModel(input.conversationId, model); - this.pool.updateConversationModel(input.conversationId, persisted); - this.replaceModel(input.conversationId, persisted); + }, state.data, capabilities.data, true, true); + if (persisted.model?.thinkingLevel !== input.thinkingLevel) { + throw new CodingRuntimeContractError( + 'CODING_MODEL_UNAVAILABLE', + 'Pi did not accept the selected thinking level', + true, + ); + } return clone(persisted); } @@ -892,6 +997,25 @@ export class PiConversationRuntime implements CodingConversationRuntime { return this.pool.getDiagnostics(); } + async injectWorkerFailureForProof( + conversationId: string, + failure: PiWorkerProofFailure, + ): Promise<{ generation: number }> { + return await this.pool.injectFailureForProof(conversationId, failure); + } + + getResilienceProofDiagnostics(): { + pool: ReturnType; + subagents: ReturnType | null; + extension: ReturnType | null; + } { + return { + pool: this.pool.getResilienceProofDiagnostics(), + subagents: this.subagentScheduler?.getDiagnostics() ?? null, + extension: this.extensionHost?.getDiagnostics() ?? null, + }; + } + markProviderStale(): void { this.pool.markProviderStale(); } @@ -1121,14 +1245,29 @@ export class PiConversationRuntime implements CodingConversationRuntime { 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) }); + void this.enqueueProjection(event.conversationId, async () => { + const snapshot = this.states.get(event.conversationId)?.snapshot; + if (!snapshot || snapshot.cursor.workerGeneration !== event.generation) return; + await this.interactions.cancelGeneration(event.conversationId, event.generation); + const current = this.snapshot(event.conversationId).run; + if (current.runId && !runIsTerminal(current.status)) { + this.failRun( + event.conversationId, + current.runId, + event.error, + event.generation, + ); + } + const state = this.pool.getState(event.conversationId); + if (state?.generation === event.generation) { + this.emit(event.conversationId, { + op: 'worker.state', + state: publicWorkerState(state), + }); + } + }).catch((error) => { + this.recordProjectionFailure(event.conversationId, event.generation, error); + }); return; } void this.enqueueProjection(event.conversationId, async () => { @@ -1161,7 +1300,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { } if (event.event.type === 'agent_end' || event.event.type === 'agent_settled') { const worker = this.pool.getState(event.conversationId); - if (worker?.generation === event.generation) { + if (worker?.generation === event.generation && worker.state !== 'crashed') { await this.hydrateGenerationNow( event.conversationId, worker, @@ -1183,9 +1322,18 @@ export class PiConversationRuntime implements CodingConversationRuntime { }); } - private failRun(conversationId: string, runId: string, error: unknown): void { + private failRun( + conversationId: string, + runId: string, + error: unknown, + generation?: number, + ): void { const current = this.states.get(conversationId)?.snapshot?.run; - if (current?.runId !== runId) return; + const snapshotGeneration = this.states.get(conversationId)?.snapshot?.cursor.workerGeneration; + if (current?.runId !== runId + || runIsTerminal(current.status) + || (generation !== undefined && snapshotGeneration !== generation)) return; + const publicError = runtimeFailure(error); this.emit(conversationId, { op: 'run.state', run: { @@ -1193,13 +1341,20 @@ export class PiConversationRuntime implements CodingConversationRuntime { runId, settledAt: this.now(), terminalReason: 'failed', - error: runtimeFailure(error), + error: publicError, }, }, runId); + logger.warn('[PiWorkerLifecycle]', { + event: 'run.failed', + conversationId, + runId, + generation: generation ?? snapshotGeneration, + code: publicError.code, + }); 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); + const runGeneration = generation ?? this.pool.getState(conversationId)?.generation; + if (runGeneration) void this.extensionHost?.clearRun(conversationId, runGeneration, runId); } private requestHydration( @@ -1232,10 +1387,13 @@ export class PiConversationRuntime implements CodingConversationRuntime { this.replaceWorkerGeneration(conversationId, workerState, preserveRun); this.resetProjector(conversationId); } - const [stateResponse, entriesResponse, statsResponse] = await Promise.all([ + const [stateResponse, entriesResponse, statsResponse, capabilitiesResponse] = await Promise.all([ this.pool.request(conversationId, { type: 'get_state' }, { retry: 'read-only-once' }), this.pool.request(conversationId, { type: 'get_entries' }, { retry: 'read-only-once' }), this.pool.request(conversationId, { type: 'get_session_stats' }, { retry: 'read-only-once' }), + this.pool.request(conversationId, { type: 'get_available_thinking_levels' }, { + retry: 'read-only-once', + }), ]); const before = this.snapshot(conversationId); let projected = await projectPiSessionSnapshot({ @@ -1253,9 +1411,54 @@ export class PiConversationRuntime implements CodingConversationRuntime { queue: clone(before.queue), }; } + projected = { + ...projected, + conversation: { + ...projected.conversation, + model: await this.persistEffectiveThinking( + conversationId, + projected.conversation.model, + stateResponse.data, + capabilitiesResponse.data, + false, + ), + }, + }; this.states.set(conversationId, createConversationReducerState(projected)); } + private async persistEffectiveThinking( + conversationId: string, + requested: ConversationModelState, + stateValue: unknown, + capabilitiesValue: unknown, + replaceSnapshot = true, + forcePersist = false, + ): Promise { + if (!requested.model) return clone(requested); + const effective = effectiveThinkingLevel(stateValue) ?? requested.model.thinkingLevel; + const available = availableThinkingLevels(capabilitiesValue); + if (!available.includes(effective)) available.push(effective); + const durable: ConversationModelState = { + model: { ...requested.model, thinkingLevel: effective }, + modelResolution: 'resolved', + }; + const persisted = !forcePersist && requested.model.thinkingLevel === effective + ? durable + : await this.registry.setModel(conversationId, durable); + this.pool.updateConversationModel(conversationId, persisted); + const result: ConversationModelState = { + ...persisted, + ...(available.length > 0 ? { availableThinkingLevels: available } : {}), + }; + if (replaceSnapshot) this.replaceModel(conversationId, result); + else { + const input = this.inputs.get(conversationId); + if (input) input.model = clone(persisted); + } + return clone(result); + } + private enqueueProjection(conversationId: string, action: () => Promise): Promise { const previous = this.projectionChains.get(conversationId) ?? Promise.resolve(); const flight = previous.then(action); diff --git a/electron/coding-runtime/pi/subagent-child.ts b/electron/coding-runtime/pi/subagent-child.ts index de521b7..54472cd 100644 --- a/electron/coding-runtime/pi/subagent-child.ts +++ b/electron/coding-runtime/pi/subagent-child.ts @@ -26,7 +26,12 @@ import type { PiSubagentChildResult, } from './subagent'; import { PiSubagentChildError } from './subagent'; -import { PiWorkerProcess, type PiWorkerProcessOptions, type PiWorkerStopResult } from './worker-process'; +import { + PiWorkerProcess, + type PiWorkerProcessOptions, + type PiWorkerStopReason, + type PiWorkerStopResult, +} from './worker-process'; import type { PiProcessError } from './process-errors'; import type { PiManagedExtensionHost } from './extension-host'; @@ -41,7 +46,7 @@ export interface PiSubagentProcessAdapter { ): Promise>; subscribe(listener: (event: PiRpcEvent) => void): () => void; subscribeInvalidation(listener: (error: PiProcessError) => void): () => void; - stop(): Promise; + stop(reason: PiWorkerStopReason): Promise; } export interface PiManagedSubagentChildOpenerOptions { @@ -134,11 +139,11 @@ class ManagedPiSubagentChild implements PiSubagentChild { } } - async stop(): Promise { + async stop(reason: PiWorkerStopReason): Promise { if (this.stopped) return; this.stopped = true; try { - await this.process.stop(); + await this.process.stop(reason); } finally { await this.disposeExtension(); } @@ -214,6 +219,8 @@ export function createPiManagedSubagentChildOpener( ], env: { ...credential.env, ...extension.env }, sensitiveValues: [...credential.sensitiveValues, ...extension.sensitiveValues], + conversationId: input.conversationId, + workerGeneration: input.workerGeneration, }); try { await process.start(); @@ -221,7 +228,7 @@ export function createPiManagedSubagentChildOpener( if (!ready.success) throw new PiSubagentChildError('SUBAGENT_CHILD_START_FAILED'); return new ManagedPiSubagentChild(input.taskId, process, extension.dispose); } catch (error) { - await process.stop().catch(() => undefined); + await process.stop('subagent_open_failure').catch(() => undefined); await extension.dispose(); throw error; } diff --git a/electron/coding-runtime/pi/subagent.ts b/electron/coding-runtime/pi/subagent.ts index 35d4f4e..e422947 100644 --- a/electron/coding-runtime/pi/subagent.ts +++ b/electron/coding-runtime/pi/subagent.ts @@ -2,6 +2,7 @@ import { randomUUID } from 'node:crypto'; import type { PublicUsage, SubagentDetailsV1 } from '../contracts'; import { PiProcessBudget, type PiProcessLease } from './worker-pool'; +import type { PiWorkerStopReason } from './worker-process'; export type PiSubagentMode = SubagentDetailsV1['mode']; export type PiSubagentToolProfile = SubagentDetailsV1['tasks'][number]['toolProfile']; @@ -43,7 +44,7 @@ export interface PiSubagentChild { signal: AbortSignal, onProgress?: (summary: string) => void, ): Promise; - stop(): Promise; + stop(reason: PiWorkerStopReason): Promise; } export interface PiSubagentDispatchResult { @@ -368,6 +369,7 @@ export class PiSubagentScheduler { let releaseChild: (() => void) | undefined; let processLease: PiProcessLease | undefined; let child: PiSubagentChild | undefined; + let stopReason: PiWorkerStopReason = 'subagent_complete'; try { releaseChild = await this.childPermits.acquire(record.controller.signal); processLease = await this.acquireProcessLease(record.controller.signal); @@ -394,12 +396,13 @@ export class PiSubagentScheduler { } catch (error) { const aborted = record.controller.signal.aborted || (error instanceof PiSubagentChildError && error.code === 'SUBAGENT_ABORTED'); + if (aborted) stopReason = 'subagent_abort'; projected.status = aborted ? 'aborted' : 'error'; projected.errorCode = publicErrorCode(error, aborted); } finally { if (child) { record.children.delete(child); - await child.stop().catch(() => undefined); + await child.stop(stopReason).catch(() => undefined); } processLease?.release(); releaseChild?.(); diff --git a/electron/coding-runtime/pi/worker-pool.ts b/electron/coding-runtime/pi/worker-pool.ts index cb5674e..febc8df 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -10,7 +10,11 @@ import type { PiRpcRequestOptions, PiRpcResponse, } from './rpc-client'; -import type { PiWorkerStopResult } from './worker-process'; +import type { + PiWorkerProofFailure, + PiWorkerStopReason, + PiWorkerStopResult, +} from './worker-process'; import { PiManagedInputRevisionCoordinator, type PiManagedInputRevision, @@ -19,6 +23,7 @@ import { createPiRuntimeTelemetryEvent, type PiRuntimeTelemetryEvent, } from './telemetry'; +import { logger } from '../../utils/logger'; export interface PiConversationWorker { readonly id: string; @@ -30,7 +35,8 @@ export interface PiConversationWorker { send(command: PiRpcCommand): Promise; subscribe(listener: (event: PiRpcEvent) => void): () => void; subscribeInvalidation(listener: (error: PiProcessError) => void): () => void; - stop(): Promise; + stop(reason: PiWorkerStopReason): Promise; + injectFailureForProof?(failure: PiWorkerProofFailure): Promise; } export interface PiWorkerSessionBinding { @@ -100,9 +106,17 @@ export type PiWorkerPoolEvent = type: 'worker.replaced'; conversationId: string; generation: number; + reason: PiWorkerReplacementReason; state: PiWorkerPoolState; }; +export type PiWorkerReplacementReason = + | 'stale_resource_rebuild' + | 'recover' + | 'model_reconfiguration' + | 'process_capacity_reopen' + | 'fork_replacement'; + export interface PiWorkerPoolOptions { openWorker(input: PiWorkerOpenInput): Promise; maxRunning?: number; @@ -365,6 +379,22 @@ export class PiWorkerPool { return record ? this.publicState(record) : null; } + async injectFailureForProof( + conversationId: string, + failure: PiWorkerProofFailure, + ): Promise<{ generation: number }> { + const record = this.workers.get(conversationId); + if (!record || record.state === 'crashed') { + throw new Error('Conversation worker is not available for proof failure injection'); + } + if (!record.worker.injectFailureForProof) { + throw new Error('Conversation worker does not support proof failure injection'); + } + const generation = record.generation; + await record.worker.injectFailureForProof(failure); + return { generation }; + } + getDiagnostics(): CodingRuntimeDiagnostics { const stage = ( state: PiWorkerPoolState['state'], @@ -386,6 +416,22 @@ export class PiWorkerPool { }; } + getResilienceProofDiagnostics(): { + processBudget: { active: number; waiting: number }; + runs: { active: number; waiting: number }; + } { + return { + processBudget: { + active: this.processBudget.activeCount, + waiting: this.processBudget.waitingCount, + }, + runs: { + active: this.activeRuns.size, + waiting: this.waitingRuns.length, + }, + }; + } + async reclaimIdleWorker(signal?: AbortSignal): Promise { while (true) { if (signal?.aborted) throw new Error('Pi idle worker reclaim cancelled'); @@ -505,7 +551,11 @@ export class PiWorkerPool { record.reconfigureAfterSettled = true; return null; } - record.rebuildFlight = this.beginRebuild(record, this.revisions.current); + record.rebuildFlight = this.beginRebuild( + record, + this.revisions.current, + 'model_reconfiguration', + ); return this.publicState(await record.rebuildFlight); } @@ -581,7 +631,7 @@ export class PiWorkerPool { break; } } - record.rebuildFlight = this.beginRebuild(record, this.revisions.current); + record.rebuildFlight = this.beginRebuild(record, this.revisions.current, 'recover'); return this.publicState(await record.rebuildFlight); } @@ -599,7 +649,7 @@ export class PiWorkerPool { this.cancelGenerationResources(record); this.revisions.removeWorker(record.revisionWorkerId); if (this.workers.get(conversationId) === record) this.workers.delete(conversationId); - await this.ensureStoppedAndReleased(record); + await this.ensureStoppedAndReleased(record, 'dispose'); } private async performShutdown(): Promise { @@ -618,7 +668,7 @@ export class PiWorkerPool { record.unsubscribeInvalidation(); this.cancelGenerationResources(record); this.revisions.removeWorker(record.revisionWorkerId); - await this.ensureStoppedAndReleased(record); + await this.ensureStoppedAndReleased(record, 'app_shutdown'); })); } @@ -710,6 +760,9 @@ export class PiWorkerPool { revisionAction.action === 'rebuild-after-settled' ? revisionAction.revision : this.revisions.current, + revisionAction.action === 'rebuild-after-settled' + ? 'stale_resource_rebuild' + : 'model_reconfiguration', ); void record.rebuildFlight.catch(() => undefined); } @@ -731,20 +784,31 @@ export class PiWorkerPool { if (record.rebuildFlight) return await record.rebuildFlight; } if (!record.processLease) { - record.rebuildFlight = this.beginRebuild(record, this.revisions.current); + record.rebuildFlight = this.beginRebuild( + record, + this.revisions.current, + 'process_capacity_reopen', + ); return await record.rebuildFlight; } const action = this.revisions.beforePrompt(record.revisionWorkerId); if (action.action !== 'rebuild-before-prompt') return record; - record.rebuildFlight = this.beginRebuild(record, action.revision); + record.rebuildFlight = this.beginRebuild(record, action.revision, 'stale_resource_rebuild'); return await record.rebuildFlight; } private beginRebuild( record: WorkerRecord, revision: PiManagedInputRevision, + reason: PiWorkerReplacementReason, ): Promise { - const flight = this.rebuild(record, revision).finally(() => { + logger.info('[PiWorkerLifecycle]', { + event: 'worker.replacement_started', + conversationId: record.conversation.conversationId, + generation: record.generation, + reason, + }); + const flight = this.rebuild(record, revision, reason).finally(() => { this.rebuildFlights.delete(flight); }); this.rebuildFlights.add(flight); @@ -754,6 +818,7 @@ export class PiWorkerPool { private async rebuild( record: WorkerRecord, revision: PiManagedInputRevision, + reason: PiWorkerReplacementReason, ): Promise { const conversationId = record.conversation.conversationId; record.state = 'spawning'; @@ -763,7 +828,7 @@ export class PiWorkerPool { this.revisions.removeWorker(record.revisionWorkerId); try { if (record.processStopFlight) await record.processStopFlight; - else await record.worker.stop(); + else await record.worker.stop(reason); let lease = record.processLease; if (!lease) lease = await this.processBudget.acquire(this.shutdownController.signal); if (this.shuttingDown) { @@ -781,7 +846,7 @@ export class PiWorkerPool { existingSession: structuredClone(record.session), }); if (this.shuttingDown) { - await opened.worker.stop().catch(() => undefined); + await opened.worker.stop('app_shutdown').catch(() => undefined); lease.release(); if (record.processLease === lease) record.processLease = null; throw new Error('Pi worker pool is shutting down'); @@ -793,7 +858,7 @@ export class PiWorkerPool { } if (opened.session.piSessionId !== record.session.piSessionId || opened.session.sessionKey !== record.session.sessionKey) { - await opened.worker.stop().catch(() => undefined); + await opened.worker.stop('session_binding_mismatch').catch(() => undefined); lease.release(); if (record.processLease === lease) record.processLease = null; throw new Error('Reopened Pi worker returned a different session binding'); @@ -817,8 +882,15 @@ export class PiWorkerPool { type: 'worker.replaced', conversationId, generation, + reason, state: this.publicState(replacement), }); + logger.info('[PiWorkerLifecycle]', { + event: reason === 'recover' ? 'worker.recovered' : 'worker.replacement_ready', + conversationId, + generation, + reason, + }); await this.trimIdleWorkers(); return replacement; } catch (error) { @@ -903,7 +975,7 @@ export class PiWorkerPool { this.cancelGenerationResources(record); this.revisions.removeWorker(record.revisionWorkerId); this.workers.delete(conversationId); - await this.ensureStoppedAndReleased(record); + await this.ensureStoppedAndReleased(record, 'idle_eviction'); return true; } @@ -918,7 +990,7 @@ export class PiWorkerPool { record.unsubscribeEvent(); record.unsubscribeInvalidation(); this.cancelGenerationResources(record); - record.processStopFlight = this.stopAndRelease(record); + record.processStopFlight = this.stopAndRelease(record, 'queued_suspension'); await record.processStopFlight; return true; } @@ -944,7 +1016,12 @@ export class PiWorkerPool { this.revisions.removeWorker(record.revisionWorkerId); this.cancelGenerationResources(record); if (!record.processStopFlight) { - record.processStopFlight = this.stopAndRelease(record).catch(() => undefined); + record.processStopFlight = this.stopAndRelease( + record, + error.code === 'PI_RPC_PROTOCOL_ERROR' + ? 'protocol_invalidation' + : 'unexpected_exit_cleanup', + ).catch(() => undefined); } const active = this.activeRuns.get(conversationId); @@ -959,6 +1036,12 @@ export class PiWorkerPool { pending.reject(error); } this.launchWaitingRuns(); + logger.warn('[PiWorkerLifecycle]', { + event: 'worker.crashed', + conversationId, + generation: record.generation, + code: error.code, + }); this.emit({ type: 'worker.crashed', conversationId, @@ -1006,7 +1089,7 @@ export class PiWorkerPool { try { const opened = await this.openWorker(input); if (this.shuttingDown) { - await opened.worker.stop().catch(() => undefined); + await opened.worker.stop('app_shutdown').catch(() => undefined); lease.release(); throw new Error('Pi worker pool is shutting down'); } @@ -1017,18 +1100,21 @@ export class PiWorkerPool { } } - private async stopAndRelease(record: WorkerRecord): Promise { + private async stopAndRelease(record: WorkerRecord, reason: PiWorkerStopReason): Promise { try { - await record.worker.stop(); + await record.worker.stop(reason); } finally { record.processLease?.release(); record.processLease = null; } } - private async ensureStoppedAndReleased(record: WorkerRecord): Promise { + private async ensureStoppedAndReleased( + record: WorkerRecord, + reason: PiWorkerStopReason, + ): Promise { if (!record.processStopFlight) { - record.processStopFlight = this.stopAndRelease(record); + record.processStopFlight = this.stopAndRelease(record, reason); } await record.processStopFlight; } diff --git a/electron/coding-runtime/pi/worker-process.ts b/electron/coding-runtime/pi/worker-process.ts index 7cbd069..00647a8 100644 --- a/electron/coding-runtime/pi/worker-process.ts +++ b/electron/coding-runtime/pi/worker-process.ts @@ -4,7 +4,8 @@ import { type ChildProcessWithoutNullStreams, } from 'node:child_process'; import { platform } from 'node:os'; -import { PiProcessError } from './process-errors'; +import { logger } from '../../utils/logger'; +import { PiProcessError, type PiProcessErrorCode } from './process-errors'; import { PiRpcClient, type PiRpcCommand, @@ -46,6 +47,59 @@ export type PiWorkerStopResult = { signal: NodeJS.Signals | null; }; +export type PiWorkerProofFailure = 'unexpected_exit' | 'protocol_invalidation'; + +export type PiWorkerStopReason = + | 'app_shutdown' + | 'idle_eviction' + | 'queued_suspension' + | 'stale_resource_rebuild' + | 'recover' + | 'dispose' + | 'model_reconfiguration' + | 'process_capacity_reopen' + | 'fork_replacement' + | 'protocol_invalidation' + | 'unexpected_exit_cleanup' + | 'open_failure' + | 'session_binding_mismatch' + | 'subagent_complete' + | 'subagent_abort' + | 'subagent_open_failure' + | 'test_injection'; + +const PI_WORKER_STOP_REASONS = new Set([ + 'app_shutdown', + 'idle_eviction', + 'queued_suspension', + 'stale_resource_rebuild', + 'recover', + 'dispose', + 'model_reconfiguration', + 'process_capacity_reopen', + 'fork_replacement', + 'protocol_invalidation', + 'unexpected_exit_cleanup', + 'open_failure', + 'session_binding_mismatch', + 'subagent_complete', + 'subagent_abort', + 'subagent_open_failure', + 'test_injection', +]); + +export type PiWorkerLifecycleEvent = { + classification: 'unexpected_exit' | 'protocol_invalidation' | 'intentional_stop'; + stage: 'protocol' | 'close' | 'stop_requested' | 'stop_completed' | 'stop_failed'; + conversationId?: string; + generation: number; + code: PiProcessErrorCode; + reason?: PiWorkerStopReason; + exitCode: number | null; + signal: NodeJS.Signals | null; + diagnostic?: string; +}; + export type PiWorkerProcessOptions = { executablePath: string; cliPath: string; @@ -60,6 +114,9 @@ export type PiWorkerProcessOptions = { shutdownGraceMs?: number; maxLineBytes?: number; diagnosticBytes?: number; + conversationId?: string; + workerGeneration?: number; + onLifecycleEvent?(event: PiWorkerLifecycleEvent): void; }; export function buildPiRpcArgs( @@ -174,11 +231,11 @@ export class PiWorkerProcess { private readonly commandTimeoutMs: number; private readonly shutdownGraceMs: number; private readonly diagnosticBytes: number; + private readonly diagnosticSensitiveValues: string[]; private child: ChildProcessWithoutNullStreams | null = null; private rpc: PiRpcClient | null = null; - private generationValue = 1; + private generationValue: number; private diagnostic = ''; - private stdoutTail = ''; private invalidation: PiProcessError | null = null; private stopping = false; private stopPromise: Promise | null = null; @@ -187,6 +244,7 @@ export class PiWorkerProcess { constructor(options: PiWorkerProcessOptions) { this.options = options; + this.generationValue = positiveInteger(options.workerGeneration, 1, 'workerGeneration'); this.commandTimeoutMs = positiveInteger( options.commandTimeoutMs, DEFAULT_COMMAND_TIMEOUT_MS, @@ -202,6 +260,17 @@ export class PiWorkerProcess { DEFAULT_DIAGNOSTIC_BYTES, 'diagnosticBytes', ); + this.diagnosticSensitiveValues = [...new Set([ + ...(options.sensitiveValues ?? []), + options.cwd, + options.cliPath, + options.configDir, + options.sessionDir, + options.env?.HOME, + options.env?.USERPROFILE, + options.env?.APPDATA, + options.env?.LOCALAPPDATA, + ].filter((value): value is string => Boolean(value)))]; } get generation(): number { @@ -267,11 +336,26 @@ export class PiWorkerProcess { }); child.once('close', (code, signal) => { if (!this.stopping && !this.invalidation) { - this.invalidate(new PiProcessError( + const failure = new PiProcessError( 'PI_RPC_EXITED', `Pi worker exited unexpectedly (code=${code ?? 'null'}, signal=${signal ?? 'none'})`, - { generation }, - )); + { + generation, + diagnostic: this.diagnostic || undefined, + exitCode: code, + signal, + }, + ); + this.recordLifecycle({ + classification: 'unexpected_exit', + stage: 'close', + generation, + code: failure.code, + exitCode: code, + signal, + ...(this.diagnostic ? { diagnostic: this.diagnostic } : {}), + }); + this.invalidate(failure); } }); @@ -280,10 +364,6 @@ export class PiWorkerProcess { onRecord: (record) => this.acceptRecord(record), }); child.stdout.on('data', (chunk: Buffer) => { - this.stdoutTail = boundedUtf8Tail( - `${this.stdoutTail}${sanitizePiDiagnostic(chunk.toString('utf8'), this.options.sensitiveValues)}`, - Math.min(this.diagnosticBytes, 2_048), - ); try { framer.push(chunk); } catch (error) { @@ -351,11 +431,26 @@ export class PiWorkerProcess { return () => this.invalidationListeners.delete(listener); } - stop(): Promise { - if (!this.stopPromise) this.stopPromise = this.performStop(); + stop(reason: PiWorkerStopReason): Promise { + if (!PI_WORKER_STOP_REASONS.has(reason)) { + return Promise.reject(new Error('Pi worker stop reason is required')); + } + if (!this.stopPromise) this.stopPromise = this.performStop(reason); return this.stopPromise; } + async injectFailureForProof(failure: PiWorkerProofFailure): Promise { + if (!this.isRunning) throw new Error('Pi worker proof failure requires a running process'); + this.appendDiagnostic( + '[release-proof] worker failure Authorization: Bearer packaged-proof-secret\n', + ); + if (failure === 'protocol_invalidation') { + this.handleProtocolFailure(new Error('Injected strict JSONL protocol failure')); + return; + } + await this.forceKillTree(); + } + private acceptRecord(record: unknown): void { try { this.rpc?.accept(record); @@ -367,12 +462,21 @@ export class PiWorkerProcess { private handleProtocolFailure(error: unknown): void { if (this.invalidation) return; const message = error instanceof Error ? error.message : 'Pi RPC protocol failure'; - this.appendDiagnostic(`[stdout-protocol] ${this.stdoutTail}\n`); + this.appendDiagnostic('[stdout-protocol]\n'); const failure = new PiProcessError('PI_RPC_PROTOCOL_ERROR', message, { cause: error, generation: this.generationValue, diagnostic: this.diagnostic, }); + this.recordLifecycle({ + classification: 'protocol_invalidation', + stage: 'protocol', + generation: this.generationValue, + code: failure.code, + exitCode: this.child?.exitCode ?? null, + signal: this.child?.signalCode ?? null, + ...(this.diagnostic ? { diagnostic: this.diagnostic } : {}), + }); this.invalidate(failure); void this.forceKillTree(); } @@ -394,14 +498,40 @@ export class PiWorkerProcess { } private appendDiagnostic(source: string): void { - const sanitized = sanitizePiDiagnostic(source, this.options.sensitiveValues); + const sanitized = sanitizePiDiagnostic(source, this.diagnosticSensitiveValues); this.diagnostic = boundedUtf8Tail(`${this.diagnostic}${sanitized}`, this.diagnosticBytes); } - private async performStop(): Promise { + private async performStop(reason: PiWorkerStopReason): Promise { const child = this.child; const exitResult = this.exitResult; + const generation = this.invalidation?.generation ?? this.generationValue; + const classification = this.invalidation?.code === 'PI_RPC_PROTOCOL_ERROR' + ? 'protocol_invalidation' + : this.invalidation?.code === 'PI_RPC_EXITED' + ? 'unexpected_exit' + : 'intentional_stop'; + this.recordLifecycle({ + classification, + stage: 'stop_requested', + generation, + code: this.invalidation?.code ?? 'PI_WORKER_STOPPED', + reason, + exitCode: child?.exitCode ?? null, + signal: child?.signalCode ?? null, + ...(this.diagnostic ? { diagnostic: this.diagnostic } : {}), + }); if (!child || !exitResult) { + this.recordLifecycle({ + classification, + stage: 'stop_completed', + generation, + code: this.invalidation?.code ?? 'PI_WORKER_STOPPED', + reason, + exitCode: null, + signal: null, + ...(this.diagnostic ? { diagnostic: this.diagnostic } : {}), + }); return { mode: 'not-started', code: null, signal: null }; } @@ -411,17 +541,71 @@ export class PiWorkerProcess { })); child.stdin.end(); const graceful = await this.waitForExit(this.shutdownGraceMs); - if (graceful) return { mode: 'stdin-close', ...graceful }; + if (graceful) { + this.recordLifecycle({ + classification, + stage: 'stop_completed', + generation, + code: this.invalidation?.code ?? 'PI_WORKER_STOPPED', + reason, + exitCode: graceful.code, + signal: graceful.signal, + ...(this.diagnostic ? { diagnostic: this.diagnostic } : {}), + }); + return { mode: 'stdin-close', ...graceful }; + } await this.forceKillTree(); const forced = await this.waitForExit(this.shutdownGraceMs); - if (forced) return { mode: 'forced-tree-kill', ...forced }; + if (forced) { + this.recordLifecycle({ + classification, + stage: 'stop_completed', + generation, + code: this.invalidation?.code ?? 'PI_WORKER_STOPPED', + reason, + exitCode: forced.code, + signal: forced.signal, + ...(this.diagnostic ? { diagnostic: this.diagnostic } : {}), + }); + return { mode: 'forced-tree-kill', ...forced }; + } + this.recordLifecycle({ + classification, + stage: 'stop_failed', + generation, + code: 'PI_WORKER_STOP_FAILED', + reason, + exitCode: child.exitCode, + signal: child.signalCode, + ...(this.diagnostic ? { diagnostic: this.diagnostic } : {}), + }); throw new PiProcessError('PI_WORKER_STOP_FAILED', 'Pi worker did not exit after forced tree kill', { generation: this.generationValue, diagnostic: this.diagnostic, }); } + private recordLifecycle(event: PiWorkerLifecycleEvent): void { + const diagnostic = event.diagnostic + ? boundedUtf8Tail( + sanitizePiDiagnostic(event.diagnostic, this.diagnosticSensitiveValues), + this.diagnosticBytes, + ) + : undefined; + const safeEvent: PiWorkerLifecycleEvent = { + ...event, + ...(this.options.conversationId ? { conversationId: this.options.conversationId } : {}), + ...(diagnostic ? { diagnostic } : {}), + }; + try { + this.options.onLifecycleEvent?.(structuredClone(safeEvent)); + } catch { + // Lifecycle observers must not affect process cleanup. + } + logger.warn('[PiWorkerLifecycle]', safeEvent); + } + private async waitForExit( timeoutMs: number, ): Promise<{ code: number | null; signal: NodeJS.Signals | null } | null> { diff --git a/electron/main/index.ts b/electron/main/index.ts index 6993faf..49d7c11 100644 --- a/electron/main/index.ts +++ b/electron/main/index.ts @@ -89,13 +89,19 @@ import { resolveCodingPiRuntimePaths, } from '../api/coding-composition'; import { + abortFinalAsarResilienceTarget, finishFinalAsarProxyCompositionProof, finishFinalAsarPressureProof, + finishFinalAsarResilienceProof, getFinalAsarProxyCompositionStatus, + getFinalAsarResilienceStatus, + injectFinalAsarResilienceFailure, releaseFinalAsarProxyCompositionChild, + releaseFinalAsarResilienceParents, runFinalAsarExtensionProof, startFinalAsarProxyCompositionProof, startFinalAsarPressureProof, + startFinalAsarResilienceProof, } from '../coding-runtime/pi/release-proof'; // Diagnostic package: force Chromium networking onto HTTP/1.1 for transport A/B testing. @@ -890,6 +896,13 @@ type PiReleaseProofAction = | 'proxy.status' | 'proxy.release-child' | 'proxy.finish' + | 'resilience.start' + | 'resilience.status' + | 'resilience.inject-exit' + | 'resilience.inject-protocol' + | 'resilience.abort-after-exit' + | 'resilience.release-parents' + | 'resilience.finish' | 'pressure.start' | 'pressure.finish' | 'pressure.finish.inject-failure'; @@ -930,6 +943,43 @@ export async function runPiReleaseProofE2E(action: PiReleaseProofAction) { if (action === 'proxy.finish') { return { action, packagedMain, proxy: await finishFinalAsarProxyCompositionProof() }; } + if (action === 'resilience.start') { + if (!codingProducts) throw new Error('Packaged Main coding composition is unavailable'); + const address = hostApiServer?.address(); + if (!address || typeof address === 'string') throw new Error('Packaged Main Host API is unavailable'); + return { + action, + packagedMain, + resilience: await startFinalAsarResilienceProof({ + composition: codingProducts, + projectPath: join(app.getPath('userData'), 'pi-resilience-proof-project'), + hostProxyBaseUrl: `http://127.0.0.1:${address.port}/api/ai-proxy/v1`, + hostToken: getHostApiToken(), + }), + }; + } + if (action === 'resilience.status') { + return { action, packagedMain, resilience: await getFinalAsarResilienceStatus() }; + } + if (action === 'resilience.inject-exit' || action === 'resilience.inject-protocol') { + return { + action, + packagedMain, + resilience: await injectFinalAsarResilienceFailure( + action === 'resilience.inject-exit' ? 'unexpected_exit' : 'protocol_invalidation', + ), + }; + } + if (action === 'resilience.abort-after-exit') { + return { action, packagedMain, resilience: await abortFinalAsarResilienceTarget() }; + } + if (action === 'resilience.release-parents') { + releaseFinalAsarResilienceParents(); + return { action, packagedMain, resilience: { released: 'parents' } }; + } + if (action === 'resilience.finish') { + return { action, packagedMain, resilience: await finishFinalAsarResilienceProof() }; + } if (action === 'pressure.start') { return { action, packagedMain, pressure: await startFinalAsarPressureProof() }; } diff --git a/scripts/run-pi-subagent-packaged-smoke.mjs b/scripts/run-pi-subagent-packaged-smoke.mjs index 57b9238..bda432e 100644 --- a/scripts/run-pi-subagent-packaged-smoke.mjs +++ b/scripts/run-pi-subagent-packaged-smoke.mjs @@ -1,15 +1,19 @@ #!/usr/bin/env node import { _electron as electron } from '@playwright/test'; +import { execFile } from 'node:child_process'; import { mkdir, mkdtemp, rm, writeFile } from 'node:fs/promises'; import { createServer } from 'node:net'; import { tmpdir } from 'node:os'; import { dirname, join, resolve } from 'node:path'; import { pathToFileURL } from 'node:url'; import { performance } from 'node:perf_hooks'; +import { promisify } from 'node:util'; import { defaultProductExecutable } from './lib/pi-product-artifact.mjs'; +const execFileAsync = promisify(execFile); + function parseArgs(argv, projectRoot = process.cwd()) { const options = { projectRoot: resolve(projectRoot), @@ -175,6 +179,41 @@ async function waitForActiveProxyProof(electronApplication) { throw new Error(`Packaged Main proxy child did not become active: ${JSON.stringify(latest?.proxy)}`); } +async function waitForResilienceProof(electronApplication, predicate, message) { + const deadline = Date.now() + 30_000; + let latest; + while (Date.now() < deadline) { + latest = await evaluateProof(electronApplication, 'resilience.status'); + if (predicate(latest?.resilience)) return latest; + await new Promise((resolveWait) => setTimeout(resolveWait, 100)); + } + throw new Error(`${message}: ${JSON.stringify(latest?.resilience)}`); +} + +function assertResilienceTerminal(status, expectedRunStatus = 'error') { + if (status?.target?.workerStatus !== 'error' + || status?.target?.runStatus !== expectedRunStatus + || status?.target?.errorCode !== 'CODING_RUNTIME_START_FAILED' + || status?.target?.recoverable !== true + || status?.target?.bindingPreserved !== true + || status?.other?.bindingPreserved !== true) { + throw new Error(`Packaged resilience target did not converge: ${JSON.stringify(status)}`); + } +} + +async function recoverSelectedConversation(page, electronApplication) { + await page.getByRole('button', { name: '恢复' }).first().click({ timeout: 10_000 }); + const recovered = await waitForResilienceProof( + electronApplication, + (status) => status?.target?.workerStatus === 'ready' && status?.target?.runStatus === 'idle', + 'Packaged resilience target did not recover', + ); + const composer = page.getByRole('textbox'); + await composer.waitFor({ state: 'visible', timeout: 10_000 }); + if (!await composer.isEnabled()) throw new Error('Recovered resilience composer is disabled'); + return recovered; +} + async function evaluateProof(electronApplication, action) { return await electronApplication.evaluate(async (_electron, requestedAction) => { const proof = globalThis.__niancodeRunPiReleaseProofE2E; @@ -190,6 +229,42 @@ async function closeApplication(electronApplication) { ]); } +async function windowsDescendantProcessIds(rootPid) { + if (process.platform !== 'win32') return []; + const script = 'Get-CimInstance Win32_Process | Select-Object ProcessId, ParentProcessId | ConvertTo-Json -Compress'; + const { stdout } = await execFileAsync( + 'powershell.exe', + ['-NoProfile', '-NonInteractive', '-Command', script], + { windowsHide: true, maxBuffer: 4 * 1024 * 1024 }, + ); + const parsed = stdout.trim() ? JSON.parse(stdout) : []; + const processes = (Array.isArray(parsed) ? parsed : [parsed]).flatMap((entry) => { + const pid = Number(entry?.ProcessId); + const parentPid = Number(entry?.ParentProcessId); + return Number.isSafeInteger(pid) && Number.isSafeInteger(parentPid) ? [{ pid, parentPid }] : []; + }); + const descendants = new Set(); + const frontier = [rootPid]; + while (frontier.length > 0) { + const parentPid = frontier.shift(); + for (const processEntry of processes) { + if (processEntry.parentPid !== parentPid || descendants.has(processEntry.pid)) continue; + descendants.add(processEntry.pid); + frontier.push(processEntry.pid); + } + } + return [...descendants]; +} + +function processIsAlive(pid) { + try { + process.kill(pid, 0); + return true; + } catch { + return false; + } +} + export async function runPackagedProductProof(options) { const scratchRoot = await mkdtemp(join(tmpdir(), 'makelore-pi-final-product-proof-')); const homeDir = join(scratchRoot, 'home'); @@ -204,6 +279,7 @@ export async function runPackagedProductProof(options) { let electronApplication; let pressureActive = false; let proxyActive = false; + let resilienceActive = false; try { electronApplication = await electron.launch({ executablePath: options.electronExecutable, @@ -332,6 +408,143 @@ export async function runPackagedProductProof(options) { assertPackagedMain(proxyFinish); assertProxyProof(proxyFinish.proxy); + const resilienceStart = await evaluateProof(electronApplication, 'resilience.start'); + resilienceActive = true; + assertPackagedMain(resilienceStart); + if (resilienceStart?.resilience?.bindingsEstablished !== true + || resilienceStart?.resilience?.otherRunAccepted !== true + || resilienceStart?.resilience?.realTurnVerified !== false) { + throw new Error(`Packaged resilience setup failed: ${JSON.stringify(resilienceStart?.resilience)}`); + } + await page.reload(); + await page.waitForLoadState('domcontentloaded'); + if (await page.getByTestId('ai-module-selection-page').count()) { + await page.getByTestId('ai-module-option-programming').click(); + await page.getByTestId('main-layout').waitFor({ state: 'visible', timeout: 10_000 }); + } + await page.evaluate(() => { window.location.hash = '/chat'; }); + const resilienceComposer = page.getByRole('textbox'); + await resilienceComposer.waitFor({ state: 'visible', timeout: 30_000 }); + await resilienceComposer.fill('RESILIENCE_TARGET_ACTIVE'); + await page.getByRole('button', { name: '发送' }).click({ timeout: 30_000 }); + + const activeResilience = await waitForResilienceProof( + electronApplication, + (status) => status?.target?.runStatus === 'running' + && status?.other?.runStatus === 'running' + && status?.resources?.subagents?.activeChildPermits === 1 + && status?.resources?.subagents?.activeDispatches === 1 + && status?.resources?.extension?.registrations?.child === 1 + && status?.resources?.extension?.writeLeases?.active === 1 + && status?.processes?.parent?.length === 2 + && status?.processes?.child?.length === 1, + 'Packaged resilience parent/child/write-lease fault state did not become active', + ); + const requestsBeforeExit = activeResilience.resilience.providerRequests; + const exitFailure = await evaluateProof(electronApplication, 'resilience.inject-exit'); + assertPackagedMain(exitFailure); + assertResilienceTerminal(exitFailure.resilience.status); + if (exitFailure.resilience.terminalizationMs > 2_000) { + throw new Error(`Packaged worker exit convergence exceeded 2s: ${exitFailure.resilience.terminalizationMs}`); + } + const releasedAfterExit = await waitForResilienceProof( + electronApplication, + (status) => status?.other?.runStatus === 'running' + && status?.resources?.subagents?.activeChildPermits === 0 + && status?.resources?.subagents?.activeDispatches === 0 + && status?.resources?.extension?.registrations?.child === 0 + && status?.resources?.extension?.writeLeases?.active === 0 + && status?.processes?.child?.length === 0, + 'Target crash did not release child and write-lease resources or preserve the other run', + ); + await page.getByText('本地 Agent 已中断，原请求未自动重发。').waitFor({ state: 'visible', timeout: 10_000 }); + const composerAfterExit = await page.getByTestId('coding-message-composer').innerText(); + if (composerAfterExit.includes('当前对话正在处理') || composerAfterExit.includes('生成中')) { + throw new Error(`Packaged resilience UI remained busy after worker exit: ${composerAfterExit}`); + } + if (!await resilienceComposer.isEnabled()) { + throw new Error('Packaged resilience composer stayed disabled after worker exit'); + } + const abortAfterExit = await evaluateProof(electronApplication, 'resilience.abort-after-exit'); + assertPackagedMain(abortAfterExit); + assertResilienceTerminal(abortAfterExit.resilience); + + const recoveredAfterExit = await recoverSelectedConversation(page, electronApplication); + if (JSON.stringify(recoveredAfterExit.resilience.providerRequests) !== JSON.stringify(requestsBeforeExit)) { + throw new Error('Recover replayed the uncertain accepted prompt after worker exit'); + } + await resilienceComposer.fill('RESILIENCE_RECOVERED_TURN'); + await page.getByRole('button', { name: '发送' }).click({ timeout: 30_000 }); + await page.getByText('RESILIENCE_RECOVERED_COMPLETE').waitFor({ state: 'visible', timeout: 30_000 }); + + await evaluateProof(electronApplication, 'resilience.release-parents'); + await waitForResilienceProof( + electronApplication, + (status) => status?.other?.runStatus === 'idle' && status?.target?.runStatus === 'idle', + 'Isolation control Conversation did not settle after release', + ); + + await resilienceComposer.fill('RESILIENCE_PROTOCOL_ACTIVE'); + await page.getByRole('button', { name: '发送' }).click({ timeout: 30_000 }); + const protocolActive = await waitForResilienceProof( + electronApplication, + (status) => status?.target?.runStatus === 'running' + && status?.activeProviderRequests?.parent === 1, + 'Protocol invalidation target did not become active', + ); + const protocolFailure = await evaluateProof(electronApplication, 'resilience.inject-protocol'); + assertPackagedMain(protocolFailure); + assertResilienceTerminal(protocolFailure.resilience.status); + if (protocolFailure.resilience.terminalizationMs > 2_000) { + throw new Error(`Packaged protocol convergence exceeded 2s: ${protocolFailure.resilience.terminalizationMs}`); + } + await page.getByText('本地 Agent 已中断，原请求未自动重发。').waitFor({ state: 'visible', timeout: 10_000 }); + const recoveredAfterProtocol = await recoverSelectedConversation(page, electronApplication); + if (JSON.stringify(recoveredAfterProtocol.resilience.providerRequests) + !== JSON.stringify(protocolFailure.resilience.status.providerRequests)) { + throw new Error('Recover replayed the uncertain prompt after protocol invalidation'); + } + + await resilienceComposer.fill('RESILIENCE_SETTLED_BEFORE_CLOSE'); + await page.getByRole('button', { name: '发送' }).click({ timeout: 30_000 }); + await page.getByText('RESILIENCE_SETTLED_COMPLETE').waitFor({ state: 'visible', timeout: 30_000 }); + const settledBeforeClose = await waitForResilienceProof( + electronApplication, + (status) => status?.target?.runStatus === 'idle' && status?.target?.workerStatus === 'ready', + 'Settled-before-close target did not reach its authoritative terminal event', + ); + const settledClose = await evaluateProof(electronApplication, 'resilience.inject-exit'); + assertPackagedMain(settledClose); + assertResilienceTerminal(settledClose.resilience.status, 'idle'); + await recoverSelectedConversation(page, electronApplication); + + const resilienceFinish = await evaluateProof(electronApplication, 'resilience.finish'); + resilienceActive = false; + assertPackagedMain(resilienceFinish); + if (resilienceFinish?.resilience?.released?.workers !== 0 + || resilienceFinish?.resilience?.released?.processes !== 0 + || Object.values(resilienceFinish?.resilience?.lifecycle ?? {}).some((value) => value !== true) + || resilienceFinish?.resilience?.realTurnVerified !== false) { + throw new Error(`Packaged resilience proof failed: ${JSON.stringify(resilienceFinish?.resilience)}`); + } + + const mainProcessId = electronApplication.process().pid; + const proofProcessIds = [ + mainProcessId, + ...await windowsDescendantProcessIds(mainProcessId), + ]; + await closeApplication(electronApplication); + electronApplication = undefined; + const exitDeadline = Date.now() + 5_000; + let lingeringProcessIds = proofProcessIds.filter(processIsAlive); + while (lingeringProcessIds.length > 0 && Date.now() < exitDeadline) { + await new Promise((resolveWait) => setTimeout(resolveWait, 100)); + lingeringProcessIds = proofProcessIds.filter(processIsAlive); + } + if (lingeringProcessIds.length > 0) { + throw new Error(`Packaged proof left Electron/Pi processes alive: ${lingeringProcessIds.join(',')}`); + } + const report = { schemaVersion: 1, generatedAt: new Date().toISOString(), @@ -362,6 +575,33 @@ export async function runPackagedProductProof(options) { }, realTurnVerified: false, }, + resilience: { + setup: resilienceStart.resilience, + active: activeResilience.resilience, + exit: exitFailure.resilience, + abortAfterExit: abortAfterExit.resilience, + releasedAfterExit: releasedAfterExit.resilience, + protocol: { + active: protocolActive.resilience, + failure: protocolFailure.resilience, + }, + settledBeforeClose: { + settled: settledBeforeClose.resilience, + close: settledClose.resilience, + }, + completed: resilienceFinish.resilience, + ui: { + terminalErrorVisible: true, + composerUnlocked: true, + recoverAvailable: true, + uncertainPromptReplayed: false, + }, + realTurnVerified: false, + }, + cleanExit: { + trackedProcessIds: proofProcessIds, + lingeringProcessIds, + }, result: 'pass', }; if (options.reportPath) { @@ -377,6 +617,10 @@ export async function runPackagedProductProof(options) { await evaluateProof(electronApplication, 'proxy.release-child').catch(() => undefined); await evaluateProof(electronApplication, 'proxy.finish').catch(() => undefined); } + if (electronApplication && resilienceActive) { + await evaluateProof(electronApplication, 'resilience.release-parents').catch(() => undefined); + await evaluateProof(electronApplication, 'resilience.finish').catch(() => undefined); + } if (electronApplication) await closeApplication(electronApplication); await rm(scratchRoot, { recursive: true, force: true, maxRetries: 5, retryDelay: 200 }); } diff --git a/shared/coding-conversation-contracts.ts b/shared/coding-conversation-contracts.ts index 84013d9..fb0a86a 100644 --- a/shared/coding-conversation-contracts.ts +++ b/shared/coding-conversation-contracts.ts @@ -9,6 +9,7 @@ export interface ProductModelRef { export interface ConversationModelState { model: ProductModelRef | null; modelResolution: 'resolved' | 'required'; + availableThinkingLevels?: ConversationThinkingLevel[]; } export interface PublicUsage { diff --git a/shared/coding-conversation-reducer.ts b/shared/coding-conversation-reducer.ts index c56c345..985ba23 100644 --- a/shared/coding-conversation-reducer.ts +++ b/shared/coding-conversation-reducer.ts @@ -54,6 +54,7 @@ const RUN_STATUSES = new Set([ ]); const WORKER_STATUSES = new Set(['stopped', 'starting', 'ready', 'recovering', 'error']); +const THINKING_LEVELS = new Set(['off', 'minimal', 'low', 'medium', 'high']); const ERROR_CODES = new Set([ 'CODING_RUNTIME_START_FAILED', 'CODING_RUNTIME_READY_TIMEOUT', @@ -115,12 +116,18 @@ function isPublicError(value: unknown): value is CodingRuntimePublicError { function isModelState(value: unknown): boolean { const record = asRecord(value); if (!record || !['resolved', 'required'].includes(String(record.modelResolution))) return false; - if (record.model === null) return record.modelResolution === 'required'; + if (record.model === null) { + return record.modelResolution === 'required' && record.availableThinkingLevels === undefined; + } const model = asRecord(record.model); return model !== null && isNonEmptyString(model.accountId) && isNonEmptyString(model.modelId) - && ['off', 'minimal', 'low', 'medium', 'high'].includes(String(model.thinkingLevel)) + && THINKING_LEVELS.has(String(model.thinkingLevel)) + && (record.availableThinkingLevels === undefined + || (Array.isArray(record.availableThinkingLevels) + && record.availableThinkingLevels.length > 0 + && record.availableThinkingLevels.every((level) => THINKING_LEVELS.has(String(level))))) && record.modelResolution === 'resolved'; } diff --git a/shared/imported-model-profile.ts b/shared/imported-model-profile.ts index 4f7916c..3e8e0b4 100644 --- a/shared/imported-model-profile.ts +++ b/shared/imported-model-profile.ts @@ -1,5 +1,15 @@ export type ImportedModelModality = 'text' | 'audio' | 'image' | 'pdf'; export type ImportedVisionTokenEstimator = 'qwen-32px-grid'; +export type ImportedThinkingLevel = 'off' | 'minimal' | 'low' | 'medium' | 'high'; + +export interface ImportedPiModelProfile { + reasoning: boolean; + thinkingLevelMap?: Partial>; + compat?: { + thinkingFormat?: 'deepseek' | 'openrouter'; + requiresReasoningContentOnAssistantMessages?: boolean; + }; +} export interface ImportedModelProfile { modalities: { @@ -11,6 +21,7 @@ export interface ImportedModelProfile { output: number; }; visionTokenEstimator?: ImportedVisionTokenEstimator; + pi?: ImportedPiModelProfile; } const VERIFIED_QWEN_PLUS_MODEL_IDS = new Set([ @@ -31,6 +42,32 @@ function imageModalities(): ImportedModelProfile['modalities'] { export function getImportedModelProfile(rawModelId: string): ImportedModelProfile | null { const modelId = rawModelId.trim(); + if (modelId.toLowerCase() === 'deepseek-v4-pro') { + return { + modalities: { + input: ['text'], + output: ['text'], + }, + limit: { + context: 1_000_000, + output: 384_000, + }, + pi: { + reasoning: true, + thinkingLevelMap: { + off: null, + minimal: null, + low: null, + medium: null, + high: 'high', + }, + compat: { + thinkingFormat: 'deepseek', + requiresReasoningContentOnAssistantMessages: true, + }, + }, + }; + } if (VERIFIED_QWEN_PLUS_MODEL_IDS.has(modelId.toLowerCase())) { return { modalities: imageModalities(), diff --git a/src/pages/Chat/CodingConversationHeader.tsx b/src/pages/Chat/CodingConversationHeader.tsx index b813ff5..1ecac76 100644 --- a/src/pages/Chat/CodingConversationHeader.tsx +++ b/src/pages/Chat/CodingConversationHeader.tsx @@ -98,6 +98,10 @@ export function CodingConversationHeader({ const modelValue = model ? codingModelKey(model) : ''; const hasCurrentOption = options.some((option) => option.key === modelValue); const thinkingLevel = model?.thinkingLevel ?? 'off'; + const availableThinkingLevels = snapshot?.conversation.model.availableThinkingLevels; + const thinkingOptions = availableThinkingLevels + ? THINKING_OPTIONS.filter((option) => availableThinkingLevels.includes(option.value)) + : THINKING_OPTIONS; const runStatus = snapshot?.run.status ?? 'preparing'; const running = ['queued', 'running', 'retrying', 'compacting', 'aborting'].includes(runStatus); const queueCount = snapshot?.queue.items.length ?? 0; @@ -196,7 +200,7 @@ export function CodingConversationHeader({ await onRefresh(); })} > - {THINKING_OPTIONS.map((option) => )} + {thinkingOptions.map((option) => )}
上下文 {context?.usedTokens ?? 0}/{context?.contextWindow ?? 0} diff --git a/tests/fixtures/fake-pi-rpc-child.mjs b/tests/fixtures/fake-pi-rpc-child.mjs index e8ed268..7dc7583 100644 --- a/tests/fixtures/fake-pi-rpc-child.mjs +++ b/tests/fixtures/fake-pi-rpc-child.mjs @@ -100,6 +100,14 @@ function handle(command) { if (command.type === 'crash') { process.exit(7); } + if (command.type === 'crash_with_stderr') { + const secret = process.env.FAKE_PI_SECRET ?? 'missing-secret'; + process.stderr.write(`Authorization: Bearer ${secret}\n`); + process.stderr.write(`token=${secret}\n`); + process.stderr.write(`cwd=${process.cwd()}\n`); + process.stderr.write('worker-exit-marker\n', () => process.exit(9)); + return; + } writeRecord({ type: 'response', id: command.id, diff --git a/tests/unit/coding-feature-ui.test.tsx b/tests/unit/coding-feature-ui.test.tsx index c8c6bad..5f9a551 100644 --- a/tests/unit/coding-feature-ui.test.tsx +++ b/tests/unit/coding-feature-ui.test.tsx @@ -1,4 +1,4 @@ -import { fireEvent, render, screen, waitFor } from '@testing-library/react'; +import { fireEvent, render, screen, waitFor, within } from '@testing-library/react'; import { describe, expect, it, vi } from 'vitest'; import type { ProviderAccount, ProviderVendorInfo } from '@/lib/providers'; @@ -102,6 +102,41 @@ describe('PI-130 feature-complete Coding UI', () => { expect(onSubmit).toHaveBeenCalledOnce(); }); + it('unlocks editing and offers recovery after the local Agent exits', async () => { + const onRecover = vi.fn(); + const { CodingComposer } = await import('@/pages/Chat/CodingComposer'); + render( + , + ); + + expect(screen.getByRole('textbox')).toBeEnabled(); + expect(screen.getByRole('alert')).toHaveTextContent('本地 Agent 已中断，原请求未自动重发。'); + expect(screen.queryByText('当前对话正在处理。')).not.toBeInTheDocument(); + fireEvent.click(screen.getByRole('button', { name: '重试' })); + expect(onRecover).toHaveBeenCalledOnce(); + }); + it('answers product interactions and explains stale responses', async () => { interactionApi.respond.mockResolvedValueOnce(undefined).mockRejectedValueOnce(new Error('409 stale')); const { CodingInteractionPanel } = await import('@/pages/Chat/CodingInteractionPanel'); @@ -198,8 +233,9 @@ describe('PI-130 feature-complete Coding UI', () => { agentId: 'agent-1', title: 'Feature UI', model: { - model: { accountId: 'account-1', modelId: 'model-a', thinkingLevel: 'medium' }, + model: { accountId: 'account-1', modelId: 'model-a', thinkingLevel: 'high' }, modelResolution: 'resolved', + availableThinkingLevels: ['high'], }, }, nodes: [], @@ -227,9 +263,13 @@ describe('PI-130 feature-complete Coding UI', () => { await waitFor(() => expect(interactionApi.model).toHaveBeenCalledWith('conversation-1', { accountId: 'account-1', modelId: 'model-b', - thinkingLevel: 'medium', + thinkingLevel: 'high', })); await waitFor(() => expect(callbacks.refresh).toHaveBeenCalledOnce()); + const thinkingSelect = screen.getByRole('combobox', { name: '当前对话思考级别' }); + expect(within(thinkingSelect).getAllByRole('option')).toHaveLength(1); + expect(within(thinkingSelect).getByRole('option', { name: '高思考' })).toBeInTheDocument(); + expect(within(thinkingSelect).queryByRole('option', { name: '关闭思考' })).not.toBeInTheDocument(); fireEvent.click(screen.getByRole('button', { name: '中止' })); await waitFor(() => expect(interactionApi.abort).toHaveBeenCalledWith('conversation-1')); fireEvent.click(screen.getByRole('button', { name: '创建分支' })); diff --git a/tests/unit/pi-conversation-runtime.test.ts b/tests/unit/pi-conversation-runtime.test.ts index da77fec..3cfefc9 100644 --- a/tests/unit/pi-conversation-runtime.test.ts +++ b/tests/unit/pi-conversation-runtime.test.ts @@ -19,7 +19,7 @@ import { PiWorkerPool, type PiConversationWorker, } from '../../electron/coding-runtime/pi/worker-pool'; -import type { PiProcessError } from '../../electron/coding-runtime/pi/process-errors'; +import { PiProcessError } from '../../electron/coding-runtime/pi/process-errors'; import type { PiRpcCommand, PiRpcEvent, @@ -40,6 +40,9 @@ class RuntimeFakeWorker implements PiConversationWorker { tokens: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }; private commandsData: unknown = { commands: [] }; + private thinkingLevelsData: unknown = { + levels: ['off', 'minimal', 'low', 'medium', 'high'], + }; private failType: string | null = null; private readonly responseGates = new Map>(); private readonly events = new Set<(event: PiRpcEvent) => void>(); @@ -66,6 +69,15 @@ class RuntimeFakeWorker implements PiConversationWorker { this.failType = null; throw new Error(`fake ${command.type} rejection`); } + if (command.type === 'set_thinking_level') { + const levels = (this.thinkingLevelsData as { levels?: unknown }).levels; + if (Array.isArray(levels) && levels.includes(command.level)) { + this.stateData = { + ...(this.stateData as Record), + thinkingLevel: command.level, + }; + } + } const data = command.type === 'get_state' ? this.stateData : command.type === 'get_entries' @@ -74,7 +86,9 @@ class RuntimeFakeWorker implements PiConversationWorker { ? this.statsData : command.type === 'get_commands' ? this.commandsData - : undefined; + : command.type === 'get_available_thinking_levels' + ? this.thinkingLevelsData + : undefined; return { type: 'response', id: `${this.id}-${this.requests.length}`, @@ -87,11 +101,20 @@ class RuntimeFakeWorker implements PiConversationWorker { this.requests.push(structuredClone(command)); } - setSessionData(input: { state?: unknown; entries?: unknown; stats?: unknown; commands?: unknown }): void { + setSessionData(input: { + state?: unknown; + entries?: unknown; + stats?: unknown; + commands?: unknown; + thinkingLevels?: unknown; + }): void { if (input.state !== undefined) this.stateData = structuredClone(input.state); if (input.entries !== undefined) this.entriesData = structuredClone(input.entries); if (input.stats !== undefined) this.statsData = structuredClone(input.stats); if (input.commands !== undefined) this.commandsData = structuredClone(input.commands); + if (input.thinkingLevels !== undefined) { + this.thinkingLevelsData = structuredClone(input.thinkingLevels); + } } failNext(type: string): void { this.failType = type; } @@ -116,6 +139,10 @@ class RuntimeFakeWorker implements PiConversationWorker { for (const listener of this.events) listener(event); } + invalidate(error = new PiProcessError('PI_RPC_EXITED', 'fake worker crashed')): void { + for (const listener of this.invalidations) listener(error); + } + async stop() { return { mode: 'stdin-close' as const, code: 0, signal: null }; } @@ -215,7 +242,8 @@ describe('Pi Conversation runtime', () => { createId: (kind) => `${kind}-fixed`, extensionHost: trackingHost, resolveModel: async (candidate) => { - if (candidate.accountId !== 'account-b' || candidate.modelId !== 'model-b') { + if (candidate.accountId !== 'account-b' + || !['model-b', 'model-c'].includes(candidate.modelId)) { throw new Error('model unavailable'); } return { @@ -239,10 +267,16 @@ describe('Pi Conversation runtime', () => { ]); expect(await runtime.listCommands('00000000-0000-4000-8000-000000000000')).toEqual([]); const leftGenerationOneSeqs: number[] = []; + const crashPatchOps: string[] = []; const unsubscribe = runtime.subscribe((envelope) => { if (envelope.conversationId === left.id && envelope.workerGeneration === 1) { leftGenerationOneSeqs.push(envelope.seq); } + if (envelope.conversationId === left.id + && envelope.workerGeneration === 3 + && (envelope.patch.op === 'run.state' || envelope.patch.op === 'worker.state')) { + crashPatchOps.push(envelope.patch.op); + } }); const releasePromptAcceptance = workers.get(left.id)!.holdNext('prompt'); @@ -258,7 +292,7 @@ describe('Pi Conversation runtime', () => { return value; }); await expect.poll(() => workers.get(left.id)!.requests.at(-1)?.type).toBe('prompt'); - expect(workers.get(left.id)!.requests).toHaveLength(5); + expect(workers.get(left.id)!.requests).toHaveLength(6); expect(acceptanceResolved).toBe(false); releasePromptAcceptance(); const accepted = await acceptance; @@ -354,6 +388,7 @@ describe('Pi Conversation runtime', () => { type: 'set_model', })); expect(workers.get(left.id)!.requests.map(({ type }) => type).sort()).toEqual([ + 'get_available_thinking_levels', 'get_entries', 'get_session_stats', 'get_state', @@ -363,12 +398,31 @@ describe('Pi Conversation runtime', () => { expect((await runtime.getSnapshot(right.id)).conversation.model.model).toEqual(model); expect((await runtime.getSnapshot(left.id)).nodes).toEqual(settledNodes); + const generationBeforeSameAccountModelChange = pool.getState(left.id)!.generation; + const sameAccountChanged = await runtime.setModel({ + conversationId: left.id, + accountId: 'account-b', + modelId: 'model-c', + }); + expect(sameAccountChanged.model).toEqual({ + accountId: 'account-b', + modelId: 'model-c', + thinkingLevel: 'medium', + }); + expect(pool.getState(left.id)!.generation).toBe(generationBeforeSameAccountModelChange); + expect(workerHistory.get(left.id)).toHaveLength(2); + expect(workers.get(left.id)!.requests.slice(-3).map(({ type }) => type)).toEqual([ + 'set_model', + 'get_available_thinking_levels', + 'get_state', + ]); + await expect(runtime.setModel({ conversationId: left.id, accountId: 'account-missing', modelId: 'model-missing', })).rejects.toThrow('model unavailable'); - expect((await runtime.getSnapshot(left.id)).conversation.model).toEqual(changed); + expect((await runtime.getSnapshot(left.id)).conversation.model).toEqual(sameAccountChanged); await runtime.prompt({ clientRequestId: 'request-left-active', @@ -398,22 +452,26 @@ describe('Pi Conversation runtime', () => { text: 'Then run the focused test', attachments: [], })).resolves.toMatchObject({ accepted: true, mode: 'follow-up', queuePosition: 2 }); + const generationBeforeThinking = pool.getState(left.id)!.generation; await expect(runtime.setThinking({ conversationId: left.id, thinkingLevel: 'high', })).resolves.toMatchObject({ - model: { accountId: 'account-b', modelId: 'model-b', thinkingLevel: 'high' }, + model: { accountId: 'account-b', modelId: 'model-c', thinkingLevel: 'high' }, }); + expect(pool.getState(left.id)!.generation).toBe(generationBeforeThinking); workers.get(left.id)!.failNext('abort'); await expect(runtime.abort(left.id)).rejects.toThrow('fake abort rejection'); expect((await runtime.getSnapshot(left.id)).run.status).toBe('running'); await runtime.abort(left.id); expect((await runtime.getSnapshot(left.id)).run.status).toBe('aborting'); - expect(workers.get(left.id)!.requests.slice(-6).map(({ type }) => type)).toEqual([ + expect(workers.get(left.id)!.requests.slice(-8).map(({ type }) => type)).toEqual([ 'steer', 'steer', 'follow_up', + 'get_available_thinking_levels', 'set_thinking_level', + 'get_state', 'abort', 'abort', ]); @@ -421,6 +479,7 @@ describe('Pi Conversation runtime', () => { 'get_state', 'get_entries', 'get_session_stats', + 'get_available_thinking_levels', ]); workers.get(left.id)!.emit({ type: 'agent_settled' }); await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('idle'); @@ -487,6 +546,7 @@ describe('Pi Conversation runtime', () => { seq: 0, }); expect(workers.get(right.id)!.requests.map(({ type }) => type).sort()).toEqual([ + 'get_available_thinking_levels', 'get_entries', 'get_session_stats', 'get_state', @@ -525,6 +585,97 @@ describe('Pi Conversation runtime', () => { await runtime.recover(left.id); expect(trackingHost.runs.has(left.id)).toBe(false); expect((await runtime.getSnapshot(left.id)).run).toEqual({ status: 'idle' }); + + workers.get(left.id)!.setSessionData({ + thinkingLevels: { levels: ['high'] }, + state: { + sessionId: `session-${left.id}`, + thinkingLevel: 'high', + isStreaming: false, + isCompacting: false, + pendingMessageCount: 0, + }, + }); + await expect(runtime.setThinking({ + conversationId: left.id, + thinkingLevel: 'off', + })).rejects.toMatchObject({ + publicError: { code: 'CODING_MODEL_UNAVAILABLE', recoverable: true }, + }); + expect((await runtime.getSnapshot(left.id)).conversation.model).toMatchObject({ + model: { thinkingLevel: 'medium' }, + availableThinkingLevels: ['high'], + }); + + const rightBeforeCrash = await runtime.getSnapshot(right.id); + const requestsBeforeCrash = workers.get(left.id)!.requests.length; + await runtime.prompt({ + clientRequestId: 'request-worker-exit', + conversationId: left.id, + mode: 'prompt', + text: 'This accepted prompt must not be replayed', + attachments: [], + }); + expect(trackingHost.runs.has(left.id)).toBe(true); + crashPatchOps.length = 0; + workers.get(left.id)!.invalidate(); + await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('error'); + expect((await runtime.getSnapshot(left.id))).toMatchObject({ + run: { + status: 'error', + terminalReason: 'failed', + error: { + code: 'CODING_RUNTIME_START_FAILED', + message: '本地 Agent 已中断，原请求未自动重发。', + recoverable: true, + }, + }, + worker: { status: 'error', generation: 3 }, + }); + expect(trackingHost.runs.has(left.id)).toBe(false); + expect(crashPatchOps).toEqual(['run.state', 'worker.state']); + await expect(runtime.abort(left.id)).resolves.toBeUndefined(); + expect((await runtime.getSnapshot(left.id)).run.status).toBe('error'); + expect(await runtime.getSnapshot(right.id)).toEqual(rightBeforeCrash); + + await runtime.recover(left.id); + expect(workers.get(left.id)!.requests.slice(0, requestsBeforeCrash)).not.toContainEqual({ + type: 'prompt', + message: 'This accepted prompt must not be replayed', + }); + + await runtime.prompt({ + clientRequestId: 'request-settled-before-close', + conversationId: left.id, + mode: 'prompt', + text: 'Settle before close', + attachments: [], + }); + workers.get(left.id)!.emit({ type: 'agent_settled' }); + workers.get(left.id)!.invalidate(); + await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('idle'); + expect((await runtime.getSnapshot(left.id)).run.terminalReason).toBe('completed'); + + await runtime.recover(left.id); + await runtime.prompt({ + clientRequestId: 'request-protocol-invalidation', + conversationId: left.id, + mode: 'prompt', + text: 'Fail this generation without replay', + attachments: [], + }); + workers.get(left.id)!.invalidate(new PiProcessError( + 'PI_RPC_PROTOCOL_ERROR', + 'strict JSONL protocol failure', + )); + await expect.poll(async () => (await runtime.getSnapshot(left.id)).run.status).toBe('error'); + expect((await runtime.getSnapshot(left.id)).run).toMatchObject({ + terminalReason: 'failed', + error: { code: 'CODING_RUNTIME_PROTOCOL_ERROR', recoverable: true }, + }); + workers.get(left.id)!.emit({ type: 'agent_settled' }); + await new Promise((resolve) => setImmediate(resolve)); + expect((await runtime.getSnapshot(left.id)).run.status).toBe('error'); unsubscribe(); }); }); diff --git a/tests/unit/pi-extension-host.test.ts b/tests/unit/pi-extension-host.test.ts index 76dd789..63b0470 100644 --- a/tests/unit/pi-extension-host.test.ts +++ b/tests/unit/pi-extension-host.test.ts @@ -269,6 +269,49 @@ describe('managed Pi extension bridge', () => { expect(currentWorker.status).toBe(200); }); + it('releases a crashed worker write lease so only its same-project waiter advances', async () => { + const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-crash-lease-')); + roots.push(root); + const leases = new PiProjectWriteLeaseCoordinator(); + const host = new PiManagedExtensionHost(leases); + hosts.push(host); + const crashed = await host.registerWorker({ + conversationId: 'conversation-crashed', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + const waiter = await host.registerWorker({ + conversationId: 'conversation-waiter', generation: 1, projectId: 'project-a', extensionsDir: root, + }); + const other = await host.registerWorker({ + conversationId: 'conversation-other', generation: 1, projectId: 'project-b', extensionsDir: root, + }); + await Promise.all([ + host.bindRun('conversation-crashed', 1, 'run-crashed'), + host.bindRun('conversation-waiter', 1, 'run-waiter'), + host.bindRun('conversation-other', 1, 'run-other'), + ]); + expect((await post(crashed, { + action: 'lease.acquire', conversationId: 'conversation-crashed', workerGeneration: 1, + runId: 'run-crashed', resourceId: 'crashed-write', + })).status).toBe(200); + const waiting = post(waiter, { + action: 'lease.acquire', conversationId: 'conversation-waiter', workerGeneration: 1, + runId: 'run-waiter', resourceId: 'waiting-write', + }); + expect((await post(other, { + action: 'lease.acquire', conversationId: 'conversation-other', workerGeneration: 1, + runId: 'run-other', resourceId: 'other-write', + })).status).toBe(200); + await expect.poll(() => leases.waitingCount('project-a')).toBe(1); + + await crashed.dispose(); + + await expect(waiting).resolves.toMatchObject({ status: 200 }); + expect(leases.waitingCount()).toBe(0); + expect(leases.activeCount).toBe(2); + await Promise.all([waiter.dispose(), other.dispose()]); + expect(leases.activeCount).toBe(0); + }); + it('joins a coding child to the same project write lease as its parent', async () => { const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-extension-child-lease-')); roots.push(root); diff --git a/tests/unit/pi-managed-worker-opener.test.ts b/tests/unit/pi-managed-worker-opener.test.ts index 740dd76..cf0457c 100644 --- a/tests/unit/pi-managed-worker-opener.test.ts +++ b/tests/unit/pi-managed-worker-opener.test.ts @@ -200,9 +200,9 @@ 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 restarted.worker.stop(); + await first.worker.stop('test_injection'); + await reopened.worker.stop('test_injection'); + await restarted.worker.stop('test_injection'); await extensionHost.close(); }); diff --git a/tests/unit/pi-provider-config.test.ts b/tests/unit/pi-provider-config.test.ts index 0e14df2..60a5011 100644 --- a/tests/unit/pi-provider-config.test.ts +++ b/tests/unit/pi-provider-config.test.ts @@ -179,6 +179,47 @@ describe('Pi Provider catalog', () => { })).toThrowError(PiProviderConfigError); }); + it('projects the managed DeepSeek capability contract into Pi models.json', () => { + const provider = account({ + id: 'niancode-user-models', + vendorId: 'custom', + apiProtocol: 'openai-completions', + baseUrl: 'http://127.0.0.1:54321/api/ai-proxy/v1', + model: 'deepseek/deepseek-v4-pro', + metadata: { + worksSquareCredentialMode: 'works_square_ai_gateway_proxy', + customModels: ['deepseek/deepseek-v4-pro'], + }, + }); + + const catalog = buildPiProviderCatalog({ accounts: [provider] }); + const descriptor = catalog.descriptors[0]!.models[0]!; + const written = catalog.modelsFile.providers[resolvePiRuntimeProviderId(provider.id)]!.models[0]!; + + expect(descriptor).toMatchObject({ + id: 'deepseek-v4-pro', + reasoning: true, + contextWindow: 1_000_000, + maxOutputTokens: 384_000, + compat: { + thinkingFormat: 'deepseek', + requiresReasoningContentOnAssistantMessages: true, + }, + thinkingLevelMap: { + off: null, + minimal: null, + low: null, + medium: null, + high: 'high', + }, + }); + expect(written).toMatchObject({ + reasoning: true, + thinkingLevelMap: descriptor.thinkingLevelMap, + compat: descriptor.compat, + }); + }); + it('does not replace an existing catalog when model selection is unavailable', async () => { const root = await mkdtemp(path.join(tmpdir(), 'makelore-pi-provider-')); temporaryRoots.push(root); diff --git a/tests/unit/pi-rpc-foundation.test.ts b/tests/unit/pi-rpc-foundation.test.ts index 1d8656b..8ba3d12 100644 --- a/tests/unit/pi-rpc-foundation.test.ts +++ b/tests/unit/pi-rpc-foundation.test.ts @@ -13,6 +13,7 @@ import { buildPiRpcArgs, sanitizePiDiagnostic, buildPiWorkerEnvironment, + type PiWorkerLifecycleEvent, } from '../../electron/coding-runtime/pi/worker-process'; const fakeChildPath = resolve('tests/fixtures/fake-pi-rpc-child.mjs'); @@ -56,7 +57,7 @@ async function processAlive(pid: number): Promise { } afterEach(async () => { - await Promise.all(workers.splice(0).map((worker) => worker.stop().catch(() => undefined))); + await Promise.all(workers.splice(0).map((worker) => worker.stop('test_injection').catch(() => undefined))); await Promise.all(scratchRoots.splice(0).map((root) => rm(root, { recursive: true, force: true, @@ -259,6 +260,87 @@ describe('Pi worker process', () => { expect(invalidations).toEqual(['PI_RPC_EXITED']); }); + it('classifies unexpected exit with bounded redacted stderr and generation correlation', async () => { + const secret = 'unexpected-exit-secret'; + const lifecycle: PiWorkerLifecycleEvent[] = []; + const worker = await makeWorker({ + conversationId: 'conversation-exit', + workerGeneration: 7, + env: { FAKE_PI_SECRET: secret }, + sensitiveValues: [secret], + diagnosticBytes: 160, + onLifecycleEvent: (event) => lifecycle.push(event), + }); + + await expect(worker.request({ type: 'crash_with_stderr' })).rejects.toMatchObject({ + code: 'PI_RPC_EXITED', + generation: 7, + exitCode: 9, + signal: null, + diagnostic: expect.stringContaining('worker-exit-marker'), + }); + const close = lifecycle.find((event) => ( + event.classification === 'unexpected_exit' && event.stage === 'close' + )); + expect(close).toMatchObject({ + conversationId: 'conversation-exit', + generation: 7, + code: 'PI_RPC_EXITED', + exitCode: 9, + signal: null, + }); + expect(JSON.stringify(close)).not.toContain(secret); + expect(JSON.stringify(close)).not.toContain(scratchRoots.at(-1)); + expect(close?.diagnostic).toContain('cwd=[REDACTED]'); + expect(Buffer.byteLength(close?.diagnostic ?? '')).toBeLessThanOrEqual(160); + }); + + it('keeps protocol invalidation and deliberate stop as distinct reasoned lifecycle events', async () => { + const protocolLifecycle: PiWorkerLifecycleEvent[] = []; + const protocolWorker = await makeWorker({ + conversationId: 'conversation-protocol', + workerGeneration: 3, + onLifecycleEvent: (event) => protocolLifecycle.push(event), + }); + await expect(protocolWorker.request({ type: 'malformed' })).rejects.toMatchObject({ + code: 'PI_RPC_PROTOCOL_ERROR', + generation: 3, + }); + expect(protocolLifecycle).toContainEqual(expect.objectContaining({ + classification: 'protocol_invalidation', + stage: 'protocol', + conversationId: 'conversation-protocol', + generation: 3, + code: 'PI_RPC_PROTOCOL_ERROR', + })); + expect(protocolLifecycle).not.toContainEqual(expect.objectContaining({ + classification: 'unexpected_exit', + })); + + const stopLifecycle: PiWorkerLifecycleEvent[] = []; + const stoppedWorker = await makeWorker({ + conversationId: 'conversation-stop', + workerGeneration: 5, + onLifecycleEvent: (event) => stopLifecycle.push(event), + }); + await expect(stoppedWorker.stop('test_injection')).resolves.toMatchObject({ + mode: 'stdin-close', + }); + expect(stopLifecycle).toContainEqual(expect.objectContaining({ + classification: 'intentional_stop', + stage: 'stop_completed', + reason: 'test_injection', + conversationId: 'conversation-stop', + generation: 5, + code: 'PI_WORKER_STOPPED', + exitCode: 0, + signal: null, + })); + expect(stopLifecycle).not.toContainEqual(expect.objectContaining({ + classification: 'unexpected_exit', + })); + }); + it('keeps only bounded redacted stderr diagnostics', async () => { const secret = 'credential-that-must-not-leak'; const worker = await makeWorker({ @@ -314,7 +396,7 @@ describe('Pi worker process', () => { const descendantPid = response.data?.pid; expect(descendantPid).toBeTypeOf('number'); - await expect(worker.stop()).resolves.toMatchObject({ mode: 'forced-tree-kill' }); + await expect(worker.stop('test_injection')).resolves.toMatchObject({ mode: 'forced-tree-kill' }); for (let attempt = 0; attempt < 20 && await processAlive(descendantPid!); attempt += 1) { await new Promise((resolvePromise) => setTimeout(resolvePromise, 25)); } diff --git a/tests/unit/pi-subagent-child.test.ts b/tests/unit/pi-subagent-child.test.ts index 067a62a..155b9d9 100644 --- a/tests/unit/pi-subagent-child.test.ts +++ b/tests/unit/pi-subagent-child.test.ts @@ -147,8 +147,8 @@ describe('managed Pi subagent child opener', () => { summary: 'managed child summary', usage: { inputTokens: 7, outputTokens: 11, cacheReadTokens: 2 }, }); - await readOnly.stop(); - await coding.stop(); + await readOnly.stop('test_injection'); + await coding.stop('test_injection'); expect(processes.every(({ stopped }) => stopped)).toBe(true); await host.close(); }); @@ -210,8 +210,8 @@ describe('managed Pi subagent child opener', () => { expect(await readFile(path.join(userDataDir, 'coding-runtime', 'pi', 'config', 'models.json'), 'utf8')) .not.toContain('host-token-'); - await first.stop(); - await second.stop(); + await first.stop('test_injection'); + await second.stop('test_injection'); await host.close(); }); }); diff --git a/tests/unit/pi-subagent.test.ts b/tests/unit/pi-subagent.test.ts index e0b0028..678d30c 100644 --- a/tests/unit/pi-subagent.test.ts +++ b/tests/unit/pi-subagent.test.ts @@ -160,7 +160,7 @@ describe('Pi subagent scheduler', () => { it('aborts every unfinished child with the parent and leaves no permit or process lease', async () => { const processBudget = new PiProcessBudget(8); - const stopped: string[] = []; + const stopped: Array<{ taskId: string; reason: string }> = []; const scheduler = new PiSubagentScheduler({ processBudget, openChild: async (input) => ({ @@ -173,7 +173,7 @@ describe('Pi subagent scheduler', () => { }); return { summary: 'unreachable' }; }, - async stop() { stopped.push(input.taskId); }, + async stop(reason) { stopped.push({ taskId: input.taskId, reason }); }, }), }); const controller = new AbortController(); @@ -186,6 +186,7 @@ describe('Pi subagent scheduler', () => { const result = await flight; expect(result.details.tasks.map(({ status }) => status)).toEqual(['aborted', 'aborted']); expect(stopped).toHaveLength(2); + expect(stopped.map(({ reason }) => reason)).toEqual(['subagent_abort', 'subagent_abort']); expect(processBudget.activeCount).toBe(0); await scheduler.close(); }); diff --git a/tests/unit/pi-worker-pool-process-integration.test.ts b/tests/unit/pi-worker-pool-process-integration.test.ts index 6125cf4..48da2f4 100644 --- a/tests/unit/pi-worker-pool-process-integration.test.ts +++ b/tests/unit/pi-worker-pool-process-integration.test.ts @@ -19,6 +19,7 @@ import { } from '../../electron/coding-runtime/pi/worker-pool'; import { PiWorkerProcess, + type PiWorkerStopReason, type PiWorkerStopResult, } from '../../electron/coding-runtime/pi/worker-process'; @@ -60,8 +61,8 @@ class ProcessBackedWorker implements PiConversationWorker { return this.process.subscribeInvalidation(listener); } - stop(): Promise { - return this.process.stop(); + stop(reason: PiWorkerStopReason): Promise { + return this.process.stop(reason); } } diff --git a/tests/unit/pi-worker-pool.test.ts b/tests/unit/pi-worker-pool.test.ts index 23e124c..7c1c05c 100644 --- a/tests/unit/pi-worker-pool.test.ts +++ b/tests/unit/pi-worker-pool.test.ts @@ -13,6 +13,10 @@ import { PiProcessError as PiProcessFailure } from '../../electron/coding-runtim import type { PiRpcCommand, PiRpcEvent } from '../../electron/coding-runtime/pi/rpc-client'; import type { PiRuntimeTelemetryEvent } from '../../electron/coding-runtime/pi/telemetry'; import { PiSubagentScheduler } from '../../electron/coding-runtime/pi/subagent'; +import type { + PiWorkerProofFailure, + PiWorkerStopReason, +} from '../../electron/coding-runtime/pi/worker-process'; const MODEL = { model: { @@ -43,6 +47,7 @@ class FakeWorker implements PiConversationWorker { readonly generation = 1; readonly requests: PiRpcCommand[] = []; stopped = false; + readonly stopReasons: PiWorkerStopReason[] = []; private readonly eventListeners = new Set<(event: PiRpcEvent) => void>(); private readonly invalidationListeners = new Set<(error: PiProcessError) => void>(); @@ -75,13 +80,50 @@ class FakeWorker implements PiConversationWorker { for (const listener of this.invalidationListeners) listener(error); } - async stop() { + async injectFailureForProof(failure: PiWorkerProofFailure): Promise { + this.invalidate(new PiProcessFailure( + failure === 'protocol_invalidation' ? 'PI_RPC_PROTOCOL_ERROR' : 'PI_RPC_EXITED', + 'injected proof failure', + )); + } + + async stop(reason: PiWorkerStopReason) { this.stopped = true; + this.stopReasons.push(reason); return { mode: 'stdin-close' as const, code: 0, signal: null }; } } describe('Pi worker pool', () => { + it('injects a proof failure only into the requested current generation', async () => { + const workers = new Map(); + const pool = new PiWorkerPool({ + maxIdle: 4, + openWorker: async ({ conversation: input }) => { + const worker = new FakeWorker(`worker-${input.conversationId}`); + workers.set(input.conversationId, worker); + return { + worker, + session: { + piSessionId: `session-${input.conversationId}`, + sessionKey: `key-${input.conversationId}`, + }, + }; + }, + }); + await Promise.all([ + pool.prepare(conversation('conversation-target')), + pool.prepare(conversation('conversation-other')), + ]); + + await expect(pool.injectFailureForProof('conversation-target', 'unexpected_exit')) + .resolves.toEqual({ generation: 1 }); + + expect(pool.getState('conversation-target')).toMatchObject({ state: 'crashed', generation: 1 }); + expect(pool.getState('conversation-other')).toMatchObject({ state: 'ready', generation: 1 }); + expect(workers.get('conversation-other')?.stopped).toBe(false); + }); + it('single-flights prepare per Conversation and never shares its worker with another Conversation', async () => { const gate = deferred(); const opened: string[] = []; @@ -418,9 +460,11 @@ describe('Pi worker pool', () => { it('cleans only the crashed generation and releases its permit for the next Conversation', async () => { const workers = new Map(); + const processBudget = new PiProcessBudget(8); const pool = new PiWorkerPool({ maxRunning: 2, maxIdle: 3, + processBudget, openWorker: async ({ conversation: input }) => { const worker = new FakeWorker(`worker-${input.conversationId}`); workers.set(input.conversationId, worker); @@ -461,6 +505,7 @@ describe('Pi worker pool', () => { expect(pool.getState('conversation-a')).toMatchObject({ state: 'crashed', generation: 1 }); expect(pool.getState('conversation-b')).toMatchObject({ state: 'running', generation: 1 }); await expect.poll(() => workers.get('conversation-c')!.requests.length).toBe(1); + await expect.poll(() => processBudget.activeCount).toBe(2); expect(cancelled).not.toContain('other'); }); @@ -468,6 +513,7 @@ describe('Pi worker pool', () => { const workers = new Map(); const revisions: Array<{ conversationId: string; provider: number; resources: number }> = []; const telemetry: PiRuntimeTelemetryEvent[] = []; + const replacementReasons: string[] = []; const pool = new PiWorkerPool({ maxIdle: 4, onTelemetry: (event) => telemetry.push(event), @@ -484,6 +530,9 @@ describe('Pi worker pool', () => { }; }, }); + pool.subscribe((event) => { + if (event.type === 'worker.replaced') replacementReasons.push(event.reason); + }); await Promise.all([ pool.prepare(conversation('conversation-running')), pool.prepare(conversation('conversation-idle')), @@ -520,6 +569,46 @@ describe('Pi worker pool', () => { { conversationId: 'conversation-idle', provider: 2, resources: 1 }, { conversationId: 'conversation-running', provider: 2, resources: 1 }, ])); + expect(replacementReasons).toEqual([ + 'stale_resource_rebuild', + 'stale_resource_rebuild', + ]); + expect(workers.get('conversation-idle')![0]!.stopReasons).toEqual([ + 'stale_resource_rebuild', + ]); + expect(workers.get('conversation-running')![0]!.stopReasons).toEqual([ + 'stale_resource_rebuild', + ]); + }); + + it('records an explicit recover replacement reason without changing the session binding', async () => { + const workers: FakeWorker[] = []; + const replacements: Array<{ reason: string; generation: number }> = []; + const pool = new PiWorkerPool({ + openWorker: async ({ conversation: input, generation, existingSession }) => { + const worker = new FakeWorker(`worker-${generation}`); + workers.push(worker); + return { + worker, + session: existingSession ?? { + piSessionId: `session-${input.conversationId}`, + sessionKey: `key-${input.conversationId}`, + }, + }; + }, + }); + pool.subscribe((event) => { + if (event.type === 'worker.replaced') { + replacements.push({ reason: event.reason, generation: event.generation }); + } + }); + + const before = await pool.prepare(conversation('conversation-recover')); + const recovered = await pool.recover('conversation-recover'); + + expect(recovered.session).toEqual(before.session); + expect(replacements).toEqual([{ reason: 'recover', generation: 2 }]); + expect(workers[0]!.stopReasons).toEqual(['recover']); }); it('re-applies the idle LRU after a running stale worker rebuilds on settle', async () => { diff --git a/tests/unit/pi-worker-process-real.test.ts b/tests/unit/pi-worker-process-real.test.ts index 74bd43d..897b598 100644 --- a/tests/unit/pi-worker-process-real.test.ts +++ b/tests/unit/pi-worker-process-real.test.ts @@ -119,9 +119,9 @@ describe('locked Pi worker process smoke', () => { 'changed_file', 'runtime_context', ])); - await expect(worker.stop()).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); + await expect(worker.stop('test_injection')).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); } finally { - await worker.stop().catch(() => undefined); + await worker.stop('test_injection').catch(() => undefined); await extension.dispose(); await extensionHost.close(); } @@ -173,9 +173,9 @@ describe('locked Pi worker process smoke', () => { }); expect(worker.stderrDiagnostic).not.toContain('Failed to load extension'); expect(await readActiveTools(probe.resultPath)).toEqual(['read', 'grep', 'find', 'ls']); - await expect(worker.stop()).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); + await expect(worker.stop('test_injection')).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); } finally { - await worker.stop().catch(() => undefined); + await worker.stop('test_injection').catch(() => undefined); await extension.dispose(); await extensionHost.close(); } @@ -235,9 +235,9 @@ describe('locked Pi worker process smoke', () => { }); expect(await readActiveTools(probe.resultPath)).toEqual(['read', 'grep', 'find', 'ls']); expect(worker.stderrDiagnostic).not.toContain('Failed to load extension'); - await expect(worker.stop()).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); + await expect(worker.stop('test_injection')).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); } finally { - await worker.stop().catch(() => undefined); + await worker.stop('test_injection').catch(() => undefined); await extension.dispose(); await extensionHost.close(); } @@ -294,9 +294,9 @@ describe('locked Pi worker process smoke', () => { 'runtime_context', ])); expect(worker.stderrDiagnostic).not.toContain('Failed to load extension'); - await expect(worker.stop()).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); + await expect(worker.stop('test_injection')).resolves.toMatchObject({ mode: 'stdin-close', code: 0 }); } finally { - await worker.stop().catch(() => undefined); + await worker.stop('test_injection').catch(() => undefined); await extension.dispose(); await extensionHost.close(); }