diff --git a/.project-docs/30-worklog/tasks/20260825-pi-background-run-lease-6a4e2c91.md b/.project-docs/30-worklog/tasks/20260825-pi-background-run-lease-6a4e2c91.md index adf22e2..7627df0 100644 --- a/.project-docs/30-worklog/tasks/20260825-pi-background-run-lease-6a4e2c91.md +++ b/.project-docs/30-worklog/tasks/20260825-pi-background-run-lease-6a4e2c91.md @@ -8,10 +8,23 @@ - Worktree: D:\Datas\OthersProjects\makelore-pi-background-run-lease-6a4e2c91 - Base commit: 92f4c91088e79252aca76af3279b184da68e1eb7 - Owner: codex-root -- Status: Ready for integration +- Status: Implementing follow-up ## Scope +- Follow up the installed `621ebb1` Windows artifact: treat an RPC + confirmation timeout for an already-written Pi top-level mutation as + uncertain ownership rather than a definite terminal failure. Keep the + target run permit, process ownership, and Main background lease until a + correlated late result/event, worker failure, abort, recover, or other + authoritative cleanup converges it. +- Prevent prompt, follow-up, steer, compact, model reconfiguration, and fork + mutations from overlapping an unresolved uncertain mutation in the same + Conversation, while preserving sibling Conversation independence. +- Give manual compact a long-mutation lifecycle through compaction completion + and authoritative settle, and keep `CODING_REQUEST_UNCERTAIN` distinct from + local-runtime-unavailable UI semantics. + - Add a Main-owned authoritative background lease for every queued or accepted top-level Pi Conversation run, acquired before mutation queueing and held through parent/child work until authoritative terminal cleanup. @@ -71,6 +84,18 @@ ## Plan +Follow-up plan from cumulative HEAD `621ebb17810394f6f7b97154cb01217bc9112857`: + +1. Add deterministic red tests for delayed Pi prompt preflight confirmation, + delayed compact confirmation, late failure/exit/abort convergence, + same-Conversation mutation exclusion, sibling isolation, and Renderer + uncertainty unlock. +2. Make RPC confirmation timeout preserve authoritative uncertain ownership + without replay, then converge exactly once from late response/events or + explicit cleanup. +3. Extend final-ASAR packaged proof beyond the old ten-second threshold and + re-run the full Windows release gate from a clean follow-up commit. + 1. Trace actual background lifecycle, Pi mutation/dispose, composition, and stop-reason callers; build a deterministic combined red test for the exact hidden-active-run failure and the in-flight sleep race. @@ -90,6 +115,34 @@ ## Outcome +### Installed-package P1 follow-up + +- Pi RPC mutation confirmation now has an explicit retained-correlation mode. + Crossing the former 10-second Host confirmation threshold returns the safe + recoverable `CODING_REQUEST_UNCERTAIN` contract but does not retire the RPC + id or release the pool's active run, revision run, process permit, or Main + background lease. +- A late correlated success clears the provisional run error without ending + the run; a late explicit RPC failure terminalizes only the target; worker + invalidation, abort/recover cleanup, or authoritative `agent_settled` + converge the same ownership exactly once. `agent_settled` also confirms a + mutation when it arrives before the RPC success response, avoiding a hung + Host acceptance promise or terminal-state regression. +- Prompt, queued follow-up/steer, compact, model reconfiguration, and fork are + blocked for the same Conversation while uncertainty is unresolved. The + service performs model/fork rejection before metadata persistence or fork + target creation; sibling Conversations retain independent permits. +- Manual compact uses the same retained ownership through + `compaction_start`, `compaction_end`, authoritative settlement, and hydrate. + The final-ASAR proof now arms a controlled 12-second compact response and + records its running/uncertain and completed/idle states. +- Renderer submission errors retain their stable backend code. An uncertain + submission keeps the draft and prevents overlap, shows the actionable delay + message instead of a runtime-outage banner, and automatically removes the + transient message when the authoritative run reaches a terminal state. + Model, thinking, compact, and fork controls are disabled while the run is + active; abort and recover remain available. + - Implemented a Main-owned per-Conversation top-level run lease in the Pi runtime. Prompt and compact acquire before optimistic/queue acceptance; steer/follow-up reuse the parent run lease. Settle, definite rejection, @@ -121,6 +174,30 @@ ## Verification +### Installed-package P1 follow-up candidate + +- Red phase — 5 expected failures across RPC correlation, pool ownership, + Main background lease, compact overlap, and Renderer reconciliation: the + old implementation retired the correlation and released active ownership at + 10 seconds, returned a raw timeout, and retained the UI uncertainty after a + completed run. +- Focused green gate — passed: 8 files / 115 tests, including delayed prompt + and compact, late success/failure, settled-before-response, same-target + mutation exclusion with zero fork/model persistence, sibling isolation, + safe Host error mapping, Renderer automatic unlock, and final-ASAR wiring. +- `pnpm run typecheck` — passed after the final runtime/Renderer/proof changes. +- `pnpm run lint:check` — passed with zero errors and the same 5 pre-existing + warnings in `src/pages/Home/index.tsx` and `src/pages/Makelore/index.tsx`. +- `pnpm test` — passed: 181 regular files / 1535 tests passed / 2 skipped, + followed by the isolated pressure file / 1 test passed. +- `pnpm run build:vite` — passed for Renderer, Main, Preload, and utility + bundles; only existing dynamic-import and chunk-size warnings were emitted. +- `pnpm run test:electron:windows` — passed: 2 files / 4 tests. +- Final amended-HEAD NSIS, artifact closure, 12-second packaged UI/Main proof, + Authenticode, and zero-residual-process results are completed after the task + documentation/registry commit so the returned artifact binds to a clean + cumulative HEAD. + - `pnpm install --frozen-lockfile` — passed with package-manager-pinned pnpm `10.33.4` and locked Pi `0.84.2`. - Red phase: the new background lifecycle in-flight test observed one unwanted diff --git a/README.md b/README.md index 148c5db..26e85f9 100644 --- a/README.md +++ b/README.md @@ -18,7 +18,7 @@ Makelore 是一个面向软件、视觉创作、互动学习与智能机器人 - 桌面技术栈：Electron、React 19、Vite、TypeScript、Zustand、Tailwind CSS。 - AI 编程核心对话运行时：Electron Main 管理固定版本的 Pi worker、会话绑定、Provider/凭证、恢复与事件投影；Renderer 不直接启动或调用 Pi，也不读取其 wire 类型。项目与 Conversation 元数据先从本地读取，选中对话后才按需准备对应 worker；输入框不等待运行时准备完成。 - 桌面性能策略：应用窗口使用不透明浅色表面并默认保留硬件合成；仅在显式安全模式或短时间内重复 GPU 进程崩溃时启用软件渲染，并把故障原因保存在用户数据目录。启动关键路径只创建本地应用壳，认证、代理、同步、更新和遥测在首帧后延迟初始化；开发版可通过 `app:performance` 快照观察 GPU、进程、WebContents、事件循环与 Renderer Long Task 聚合指标。 -- 后台生命周期：Main 统一维护模块活动状态与任务租约。隐藏窗口、离开模块和关闭开发浏览器会释放非必要连接；生成、Code 执行、下载与发布构建持有租约并在完成后释放。各模块的后台连接、轮询和子进程必须通过同一生命周期入口登记。 +- 后台生命周期：Main 统一维护模块活动状态与任务租约。隐藏窗口、离开模块和关闭开发浏览器会释放非必要连接；生成、Code 执行、下载与发布构建持有租约并在完成后释放。已写入 Pi 的 prompt 或上下文整理即使确认超过 10 秒，也会继续持有运行所有权与后台租约，直到迟到响应、权威事件、明确失败或用户清理使其收敛；此时同一 Conversation 不接受重叠 mutation，其他 Conversation 不受影响。各模块的后台连接、轮询和子进程必须通过同一生命周期入口登记。 - 共享开发浏览器：AI 编程右侧提供项目级浏览器，用户与 Agent 查看并调试同一实时页面、Console 和 Network，支持本地与公网开发地址。 - 后端边界：Renderer 通过 Main 所有的 Host API 访问认证、模型、同步、更新、语音、图像与运行时能力。 - 客户端更新：Electron Main 按平台与架构选择更新源并保留原始诊断；设置页只显示一条脱敏后的中文状态。正式源缺少对应安装包时保持错误并允许重试，不会误报为已是最新版。 diff --git a/electron/api/routes/coding-route-errors.ts b/electron/api/routes/coding-route-errors.ts index afc503e..0cf112a 100644 --- a/electron/api/routes/coding-route-errors.ts +++ b/electron/api/routes/coding-route-errors.ts @@ -27,7 +27,7 @@ const FIXED_CODING_ERROR_MESSAGES: Readonly> = { CODING_PROVIDER_AUTH_REQUIRED: 'Provider 凭证不可用，请修复账号后重试。', CODING_REQUEST_CAPACITY_EXCEEDED: '本地请求队列已满，请稍后重试。', CODING_REQUEST_ID_CONFLICT: '该请求标识已用于不同内容，请使用新的标识。', - CODING_REQUEST_UNCERTAIN: '请求状态无法确认，请先核对对话后再决定是否重试。', + CODING_REQUEST_UNCERTAIN: '请求确认延迟，可能仍在执行。请等待结果，或中止/恢复后再重试。', CODING_RUNTIME_PROTOCOL_ERROR: '本地 Agent 通信异常，请执行恢复。', CODING_RUNTIME_READY_TIMEOUT: '本地 Agent 启动超时，请执行恢复。', CODING_RUNTIME_START_FAILED: '本地 Agent 启动失败，请执行恢复。', diff --git a/electron/coding-runtime/conversation-service.ts b/electron/coding-runtime/conversation-service.ts index 364f0ea..15e6dfd 100644 --- a/electron/coding-runtime/conversation-service.ts +++ b/electron/coding-runtime/conversation-service.ts @@ -25,6 +25,7 @@ const MAX_ATTACHMENTS = 16; const MAX_ACCEPTANCES = 512; const REQUEST_ID_PATTERN = /^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$/; const ATTACHMENT_ID_PATTERN = /^[A-Za-z0-9-]{1,64}$/; +const REQUEST_UNCERTAIN_MESSAGE = '请求确认延迟，可能仍在执行。请等待结果，或中止/恢复后再重试。'; export class CodingConversationServiceError extends Error { constructor( @@ -442,8 +443,13 @@ export class CodingConversationService { try { selected = await this.runtime.validateModel(model); } catch (error) { runtimeError(error); } + await this.ensurePrepared(conversationId); + let snapshot: ConversationSnapshot; + try { + snapshot = await this.runtime.getSnapshot(conversationId); + } catch (error) { runtimeError(error); } + this.assertSnapshotAllowsMutation(snapshot); if (conversation.modelResolution === 'resolved' && conversation.model) { - await this.ensurePrepared(conversationId); let state: ConversationModelState; try { state = await this.runtime.setModel({ @@ -499,6 +505,7 @@ export class CodingConversationService { } catch (error) { runtimeError(error); } + this.assertSnapshotAllowsMutation(sourceSnapshot); const sourceNode = sourceSnapshot.nodes.find((node) => ( node.kind === 'message' && node.sourceEntryId === entryId @@ -696,6 +703,16 @@ export class CodingConversationService { } } + private assertSnapshotAllowsMutation(snapshot: ConversationSnapshot): void { + if (snapshot.run.error?.code !== 'CODING_REQUEST_UNCERTAIN') return; + if (!['queued', 'running', 'retrying', 'compacting', 'aborting'].includes(snapshot.run.status)) return; + throw new CodingConversationServiceError( + 409, + 'CODING_REQUEST_UNCERTAIN', + REQUEST_UNCERTAIN_MESSAGE, + ); + } + private async archiveSession(projectId: string, sessionKey: string | undefined): Promise { if (!sessionKey) return; if (!this.options.archiveSession) { diff --git a/electron/coding-runtime/pi/release-proof.ts b/electron/coding-runtime/pi/release-proof.ts index 0c74eb9..0b80cdb 100644 --- a/electron/coding-runtime/pi/release-proof.ts +++ b/electron/coding-runtime/pi/release-proof.ts @@ -74,6 +74,7 @@ type LocalProofProvider = { releaseChildren(): void; releaseParents(): void; releaseAll(): void; + armDelayedCompaction(): void; close(): Promise; }; @@ -239,6 +240,8 @@ export interface PiReleaseResilienceStatus { errorCode: string | null; recoverable: boolean; bindingPreserved: boolean; + contextCompaction: 'idle' | 'running'; + completedCompactions: number; }; other: { conversationId: string; @@ -295,6 +298,7 @@ const PROOF_AGENT_ID = 'release-proof-agent'; const PROOF_MODEL_ID = 'release-proof-model'; const PROXY_PROOF_MODEL_ID = 'deepseek-v4-pro'; const PROOF_PROVIDER_FIRST_EVENT_DELAY_MS = 75; +const PROOF_MUTATION_CONFIRMATION_DELAY_MS = 12_000; const EXPECTED_TURN_MILESTONES: readonly ProofMilestone[] = [ 'worker.queue_wait', 'resources.ready', @@ -482,6 +486,7 @@ async function startLocalProofProvider(mode: ProofProviderMode): Promise(); let closed = false; let closeFlight: Promise | null = null; + let delayedCompactionArmed = false; const server: Server = createServer(async (request, response) => { response.once('error', () => undefined); try { @@ -545,6 +550,26 @@ async function startLocalProofProvider(mode: ProofProviderMode): Promise held.delete(entry)); + await delay(PROOF_MUTATION_CONFIRMATION_DELAY_MS); + if (!held.delete(entry)) return; + respondWithText(response, model, 'RESILIENCE_COMPACTION_SUMMARY'); + return; + } + if (role === 'parent' && latestUserMessageContains(body, 'BACKGROUND_ACTIVE')) { + const entry = { role, response } satisfies HeldProviderResponse; + held.add(entry); + response.once('close', () => held.delete(entry)); + await delay(PROOF_MUTATION_CONFIRMATION_DELAY_MS); + if (!held.has(entry)) return; + response.writeHead(200, { 'content-type': 'text/event-stream' }); + writeChunk(response, model, { role: 'assistant', content: 'RESILIENCE_ACTIVE' }, null); + return; + } response.writeHead(200, { 'content-type': 'text/event-stream' }); writeChunk(response, model, { role: 'assistant', content: 'RESILIENCE_ACTIVE' }, null); const entry = { role, response }; @@ -634,6 +659,7 @@ async function startLocalProofProvider(mode: ProofProviderMode): Promise release('child'), releaseParents: () => release('parent'), releaseAll: () => release(), + armDelayedCompaction: () => { delayedCompactionArmed = true; }, close: async () => { if (closed) return; if (!closeFlight) { @@ -1666,6 +1692,10 @@ export async function getFinalAsarResilienceStatus(): Promise ( + node.kind === 'compaction' && node.status === 'complete' + )).length, }, other: { conversationId: run.otherConversationId, @@ -1860,6 +1890,13 @@ export function releaseFinalAsarResilienceParents(): void { run.provider.releaseParents(); } +export function armFinalAsarResilienceCompactDelay(): { delayMs: number } { + const run = resilienceCompositionRun; + if (!run) throw new Error('PI resilience proof is not running'); + run.provider.armDelayedCompaction(); + return { delayMs: PROOF_MUTATION_CONFIRMATION_DELAY_MS }; +} + export async function restartFinalAsarResilienceOther(): Promise { const run = resilienceCompositionRun; if (!run) throw new Error('PI resilience proof is not running'); diff --git a/electron/coding-runtime/pi/rpc-client.ts b/electron/coding-runtime/pi/rpc-client.ts index 2563943..ec3ce82 100644 --- a/electron/coding-runtime/pi/rpc-client.ts +++ b/electron/coding-runtime/pi/rpc-client.ts @@ -18,16 +18,24 @@ export type PiRpcResponse = { export type PiRpcEvent = Record & { type: string }; export type PiRpcRetryPolicy = 'none' | 'read-only-once'; +export type PiRpcLateResult = + | { response: PiRpcResponse; error?: never } + | { response?: never; error: PiProcessError }; + export type PiRpcRequestOptions = { signal?: AbortSignal; timeoutMs?: number; retry?: PiRpcRetryPolicy; + retainAfterTimeout?: boolean; + onLateResult?(result: PiRpcLateResult): void; }; type PendingRequest = { commandType: string; + timedOut: boolean; resolve(response: PiRpcResponse): void; reject(error: PiProcessError): void; + notifyLate(result: PiRpcLateResult): void; cancel(): void; }; @@ -155,17 +163,30 @@ export class PiRpcClient { } pending.cancel(); this.pending.delete(value.id); - if (value.success) pending.resolve(value); - else { - pending.reject(new PiProcessError( + if (value.success) { + if (pending.timedOut) pending.notifyLate({ response: value }); + else pending.resolve(value); + } else { + const error = new PiProcessError( 'PI_RPC_RESPONSE_ERROR', `Pi RPC ${value.command ?? pending.commandType} failed: ${value.error ?? 'unknown error'}`, { generation: this.generation }, - )); + ); + if (pending.timedOut) pending.notifyLate({ error }); + else pending.reject(error); } return; } + if (value.type === 'agent_settled') { + for (const [id, pending] of this.pending) { + if (!pending.timedOut) continue; + pending.cancel(); + this.pending.delete(id); + this.retire(id); + } + } + for (const listener of this.listeners) { try { listener(value as PiRpcEvent); @@ -185,7 +206,8 @@ export class PiRpcClient { for (const [id, pending] of this.pending) { pending.cancel(); this.retire(id); - pending.reject(error); + if (pending.timedOut) pending.notifyLate({ error }); + else pending.reject(error); } this.pending.clear(); } @@ -209,8 +231,12 @@ export class PiRpcClient { const response = new Promise>((resolve, reject) => { const timeout = setTimeout(() => { - this.pending.delete(id); - this.retire(id); + const pending = this.pending.get(id); + if (options.retainAfterTimeout && pending) pending.timedOut = true; + else { + this.pending.delete(id); + this.retire(id); + } reject(new PiProcessError( 'PI_RPC_TIMEOUT', `Pi RPC ${command.type} timed out after ${timeoutMs}ms`, @@ -218,18 +244,33 @@ export class PiRpcClient { )); }, timeoutMs); const abort = (): void => { + const pending = this.pending.get(id); this.pending.delete(id); this.retire(id); clearTimeout(timeout); - reject(new PiProcessError('PI_RPC_ABORTED', `Pi RPC ${command.type} was aborted`, { + const error = new PiProcessError('PI_RPC_ABORTED', `Pi RPC ${command.type} was aborted`, { generation: this.generation, - })); + }); + if (pending?.timedOut) pending.notifyLate({ error }); + else reject(error); }; options.signal?.addEventListener('abort', abort, { once: true }); this.pending.set(id, { commandType: command.type, + timedOut: false, resolve: (value) => resolve(value as PiRpcResponse), reject, + notifyLate: (result) => { + try { + options.onLateResult?.(result); + } catch (error) { + try { + this.onEventListenerError?.(error); + } catch { + // Diagnostics must not affect transport state. + } + } + }, cancel: () => { clearTimeout(timeout); options.signal?.removeEventListener('abort', abort); @@ -248,6 +289,12 @@ export class PiRpcClient { return record; } catch (error) { const pending = this.pending.get(id); + if (error instanceof PiProcessError + && error.code === 'PI_RPC_TIMEOUT' + && options.retainAfterTimeout + && pending?.timedOut) { + throw error; + } if (pending) { pending.cancel(); this.pending.delete(id); diff --git a/electron/coding-runtime/pi/runtime.ts b/electron/coding-runtime/pi/runtime.ts index 19e2c3f..eb0bf11 100644 --- a/electron/coding-runtime/pi/runtime.ts +++ b/electron/coding-runtime/pi/runtime.ts @@ -426,6 +426,16 @@ function runIsTerminal(status: ConversationSnapshot['run']['status']): boolean { return status === 'idle' || status === 'error'; } +const REQUEST_UNCERTAIN_MESSAGE = '请求确认延迟，可能仍在执行。请等待结果，或中止/恢复后再重试。'; + +function requestUncertainError(): CodingRuntimeContractError { + return new CodingRuntimeContractError( + 'CODING_REQUEST_UNCERTAIN', + REQUEST_UNCERTAIN_MESSAGE, + true, + ); +} + function publicWorkerState(state: PiWorkerPoolState): ConversationSnapshot['worker'] { if (state.state === 'spawning') return { status: 'starting', generation: state.generation }; if (state.state === 'crashed') { @@ -460,7 +470,7 @@ function runtimeFailure(error: unknown): CodingRuntimePublicError { if (error.code === 'PI_RPC_TIMEOUT') { return { code: 'CODING_REQUEST_UNCERTAIN', - message: 'The local Agent did not confirm the request', + message: REQUEST_UNCERTAIN_MESSAGE, recoverable: true, }; } @@ -626,6 +636,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { async prompt(input: PromptConversationInput): Promise { await this.waitForProjection(input.conversationId); + this.assertNoUncertainMutation(input.conversationId); if (input.mode === 'steer') { const acceptance = await this.steer(input); return { @@ -765,6 +776,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { async setModel(input: SetConversationModelInput): Promise { await this.waitForProjection(input.conversationId); + this.assertNoUncertainMutation(input.conversationId); const snapshot = this.snapshot(input.conversationId); const thinkingLevel = snapshot.conversation.model.model?.thinkingLevel ?? 'off'; const selection = await this.resolveModel({ @@ -825,6 +837,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { async setThinking(input: SetThinkingLevelInput): Promise { await this.waitForProjection(input.conversationId); + this.assertNoUncertainMutation(input.conversationId); const current = this.snapshot(input.conversationId).conversation.model; if (!current.model) { throw new CodingRuntimeContractError( @@ -876,6 +889,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { async compact(conversationId: string): Promise { await this.waitForProjection(conversationId); + this.assertNoUncertainMutation(conversationId); const runId = this.id('run'); this.acquireRunBackgroundLease(conversationId, runId); const generation = this.pool.getState(conversationId)?.generation; @@ -905,6 +919,11 @@ export class PiConversationRuntime implements CodingConversationRuntime { try { await ticket.accepted; } catch (error) { + if (error instanceof PiProcessError && error.code === 'PI_RPC_TIMEOUT') { + const failure = requestUncertainError(); + this.markRunUncertain(conversationId, runId, failure.publicError); + throw failure; + } await this.failRun(conversationId, runId, error); throw error; } @@ -912,6 +931,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { async fork(input: ForkConversationInput): Promise { await this.waitForProjection(input.sourceConversationId); + this.assertNoUncertainMutation(input.sourceConversationId); this.snapshot(input.sourceConversationId); const registered = await this.registry.prepare(input.conversation); const canonicalInput: PrepareConversationInput = { @@ -1091,6 +1111,14 @@ export class PiConversationRuntime implements CodingConversationRuntime { return Boolean(snapshot && !runIsTerminal(snapshot.run.status)); } + private assertNoUncertainMutation(conversationId: string): void { + const run = this.states.get(conversationId)?.snapshot?.run; + if (!runIsTerminal(run?.status ?? 'idle') + && run?.error?.code === 'CODING_REQUEST_UNCERTAIN') { + throw requestUncertainError(); + } + } + markProviderStale(): void { this.pool.markProviderStale(); } @@ -1137,6 +1165,7 @@ export class PiConversationRuntime implements CodingConversationRuntime { input: QueueMessageInput, ): Promise { await this.waitForProjection(input.conversationId); + this.assertNoUncertainMutation(input.conversationId); const snapshot = this.snapshot(input.conversationId); if (snapshot.run.runId) { this.acquireRunBackgroundLease(input.conversationId, snapshot.run.runId); @@ -1255,6 +1284,11 @@ export class PiConversationRuntime implements CodingConversationRuntime { true, ) : error; + if (failure instanceof PiProcessError && failure.code === 'PI_RPC_TIMEOUT') { + const uncertain = requestUncertainError(); + this.markRunUncertain(conversationId, runId, uncertain.publicError); + throw uncertain; + } this.pool.failTopLevel( conversationId, runId, @@ -1265,6 +1299,19 @@ export class PiConversationRuntime implements CodingConversationRuntime { } } + private markRunUncertain( + conversationId: string, + runId: string, + error: CodingRuntimePublicError, + ): void { + const current = this.states.get(conversationId)?.snapshot?.run; + if (!current || current.runId !== runId || runIsTerminal(current.status)) return; + this.emit(conversationId, { + op: 'run.state', + run: { ...current, error: clone(error) }, + }, runId); + } + private snapshot(conversationId: string): ConversationSnapshot { const snapshot = this.states.get(conversationId)?.snapshot; if (!snapshot) { @@ -1314,6 +1361,40 @@ export class PiConversationRuntime implements CodingConversationRuntime { } private onPoolEvent(event: PiWorkerPoolEvent): void { + if (event.type === 'top-level.confirmed') { + void this.enqueueProjection(event.conversationId, async () => { + const snapshot = this.states.get(event.conversationId)?.snapshot; + const current = snapshot?.run; + if (!snapshot + || snapshot.cursor.workerGeneration !== event.generation + || current?.runId !== event.runId + || runIsTerminal(current.status)) return; + const { error: _error, ...confirmed } = current; + this.emit(event.conversationId, { + op: 'run.state', + run: { + ...confirmed, + status: confirmed.status === 'queued' ? 'running' : confirmed.status, + }, + }, event.runId); + }).catch((error) => { + this.recordProjectionFailure(event.conversationId, event.generation, error); + }); + return; + } + if (event.type === 'top-level.failed') { + void this.enqueueProjection(event.conversationId, async () => { + await this.failRun( + event.conversationId, + event.runId, + event.error, + event.generation, + ); + }).catch((error) => { + this.recordProjectionFailure(event.conversationId, event.generation, error); + }); + return; + } if (event.type === 'worker.replaced') { if (!this.states.has(event.conversationId)) return; this.replaceWorkerGeneration(event.conversationId, event.state, true); diff --git a/electron/coding-runtime/pi/worker-pool.ts b/electron/coding-runtime/pi/worker-pool.ts index fa960f8..0b81f51 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -3,10 +3,11 @@ import type { ConversationModelState, PrepareConversationInput, } from '../contracts'; -import type { PiProcessError, PiProcessErrorCode } from './process-errors'; +import { PiProcessError, type PiProcessErrorCode } from './process-errors'; import type { PiRpcCommand, PiRpcEvent, + PiRpcLateResult, PiRpcRequestOptions, PiRpcResponse, } from './rpc-client'; @@ -108,6 +109,19 @@ export type PiWorkerPoolEvent = generation: number; reason: PiWorkerReplacementReason; state: PiWorkerPoolState; + } + | { + type: 'top-level.confirmed'; + conversationId: string; + generation: number; + runId: string; + } + | { + type: 'top-level.failed'; + conversationId: string; + generation: number; + runId: string; + error: PiProcessError; }; export type PiWorkerReplacementReason = @@ -247,9 +261,13 @@ export class PiWorkerPool { private readonly waitingRuns: PendingTopLevelRun[] = []; private readonly activeRuns = new Map void; }>(); private readonly generations = new Map(); private readonly listeners = new Set<(event: PiWorkerPoolEvent) => void>(); @@ -498,13 +516,14 @@ export class PiWorkerPool { getActiveRun(conversationId: string): { runId: string; generation: number } | null { const active = this.activeRuns.get(conversationId); - return active ? { ...active } : null; + return active ? { runId: active.runId, generation: active.generation } : null; } failTopLevel(conversationId: string, runId: string, error: Error): void { const active = this.activeRuns.get(conversationId); if (active?.runId === runId) { this.activeRuns.delete(conversationId); + active.cancelConfirmation?.(); this.runningCount -= 1; const record = this.workers.get(conversationId); if (record && record.state !== 'crashed') { @@ -667,8 +686,10 @@ export class PiWorkerPool { await Promise.allSettled([...this.rebuildFlights]); const records = [...this.workers.values()]; this.workers.clear(); + const activeRuns = [...this.activeRuns.values()]; this.activeRuns.clear(); this.runningCount = 0; + for (const active of activeRuns) active.cancelConfirmation?.(); await Promise.all(records.map(async (record) => { record.unsubscribeEvent(); record.unsubscribeInvalidation(); @@ -684,8 +705,10 @@ export class PiWorkerPool { this.runningCount += 1; this.activeRuns.set(run.conversationId, { runId: run.runId, + run, generation: record.generation, cold: record.coldStart && record.acceptedPromptCount === 0, + confirmation: 'pending', }); void this.acceptTopLevel(record, run); } @@ -708,29 +731,34 @@ export class PiWorkerPool { 'worker.queue_wait', run.queuedAt === undefined ? 0 : this.now() - run.queuedAt, ); - const acceptedAt = this.now(); - const response = await current.worker.request(run.command); - if (run.command.type === 'prompt') { - const acceptedFinishedAt = this.now(); - active.acceptedAt = acceptedFinishedAt; - this.recordMilestone( - current, - run, - 'prompt.accepted', - acceptedFinishedAt - acceptedAt, - active.cold, - ); - current.acceptedPromptCount += 1; - } + active.confirmationStartedAt = this.now(); + const confirmationController = new AbortController(); + active.cancelConfirmation = () => confirmationController.abort(); + const response = await current.worker.request(run.command, { + signal: confirmationController.signal, + retainAfterTimeout: true, + onLateResult: (result) => this.handleLateTopLevelResult(current!, run, result), + }); + this.confirmTopLevel(current, run); run.resolve(response); } catch (error) { const active = this.activeRuns.get(run.conversationId); + if (error instanceof PiProcessError + && error.code === 'PI_RPC_TIMEOUT' + && active?.runId === run.runId + && current + && active.generation === current.generation) { + if (active.confirmation !== 'confirmed') active.confirmation = 'uncertain'; + run.reject(error); + return; + } if (active) { this.activeRuns.delete(run.conversationId); + active.cancelConfirmation?.(); this.runningCount -= 1; this.launchWaitingRuns(); } - if (current && this.workers.get(run.conversationId) === current + if (active && current && this.workers.get(run.conversationId) === current && current.state !== 'crashed') { current.state = 'idle'; this.notifyReclaimableWorker(); @@ -741,10 +769,84 @@ export class PiWorkerPool { } } + private confirmTopLevel(record: WorkerRecord, run: PendingTopLevelRun): boolean { + const active = this.activeRuns.get(run.conversationId); + if (!active + || active.runId !== run.runId + || active.generation !== record.generation + || active.confirmation === 'confirmed') return false; + active.confirmation = 'confirmed'; + active.cancelConfirmation = undefined; + if (run.command.type === 'prompt') { + const acceptedFinishedAt = this.now(); + active.acceptedAt = acceptedFinishedAt; + this.recordMilestone( + record, + run, + 'prompt.accepted', + acceptedFinishedAt - (active.confirmationStartedAt ?? acceptedFinishedAt), + active.cold, + ); + record.acceptedPromptCount += 1; + } + return true; + } + + private handleLateTopLevelResult( + record: WorkerRecord, + run: PendingTopLevelRun, + result: PiRpcLateResult, + ): void { + const active = this.activeRuns.get(run.conversationId); + if (!active || active.runId !== run.runId || active.generation !== record.generation) return; + if (result.response) { + if (!this.confirmTopLevel(record, run)) return; + this.emit({ + type: 'top-level.confirmed', + conversationId: run.conversationId, + generation: record.generation, + runId: run.runId, + }); + return; + } + if (result.error.code === 'PI_RPC_EXITED' + || result.error.code === 'PI_RPC_PROTOCOL_ERROR') return; + this.activeRuns.delete(run.conversationId); + active.cancelConfirmation = undefined; + this.runningCount -= 1; + if (this.workers.get(run.conversationId) === record && record.state !== 'crashed') { + record.state = 'idle'; + this.notifyReclaimableWorker(); + try { + this.revisions.settleRun(record.revisionWorkerId); + } catch { + // Recover/crash may already have removed this generation. + } + } + this.launchWaitingRuns(); + void this.trimIdleWorkers(); + this.emit({ + type: 'top-level.failed', + conversationId: run.conversationId, + generation: record.generation, + runId: run.runId, + error: result.error, + }); + } + private settleTopLevel(record: WorkerRecord): void { const conversationId = record.conversation.conversationId; const active = this.activeRuns.get(conversationId); if (!active || active.generation !== record.generation) return; + const cancelConfirmation = active.cancelConfirmation; + if (active.confirmation !== 'confirmed') { + this.confirmTopLevel(record, active.run); + active.run.resolve({ + type: 'response', + id: `agent-settled:${active.runId}`, + success: true, + }); + } if (active.acceptedAt !== undefined) { this.recordMilestone( record, @@ -755,6 +857,7 @@ export class PiWorkerPool { ); } this.activeRuns.delete(conversationId); + cancelConfirmation?.(); this.runningCount -= 1; record.state = 'idle'; this.notifyReclaimableWorker(); @@ -1033,6 +1136,7 @@ export class PiWorkerPool { const active = this.activeRuns.get(conversationId); if (active?.generation === record.generation) { this.activeRuns.delete(conversationId); + active.cancelConfirmation?.(); this.runningCount -= 1; } for (let index = this.waitingRuns.length - 1; index >= 0; index -= 1) { @@ -1060,6 +1164,7 @@ export class PiWorkerPool { const active = this.activeRuns.get(conversationId); if (active) { this.activeRuns.delete(conversationId); + active.cancelConfirmation?.(); this.runningCount -= 1; } for (let index = this.waitingRuns.length - 1; index >= 0; index -= 1) { diff --git a/electron/main/index.ts b/electron/main/index.ts index a34ac13..668eec6 100644 --- a/electron/main/index.ts +++ b/electron/main/index.ts @@ -89,6 +89,7 @@ import { resolveCodingPiRuntimePaths, } from '../api/coding-composition'; import { + armFinalAsarResilienceCompactDelay, abortFinalAsarResilienceTarget, disposeFinalAsarResilienceTarget, finishFinalAsarProxyCompositionProof, @@ -903,6 +904,7 @@ type PiReleaseProofAction = | 'proxy.finish' | 'resilience.start' | 'resilience.status' + | 'resilience.arm-compact-delay' | 'resilience.idle-status' | 'resilience.restart-other' | 'resilience.dispose-target' @@ -969,6 +971,9 @@ export async function runPiReleaseProofE2E(action: PiReleaseProofAction) { if (action === 'resilience.status') { return { action, packagedMain, resilience: await getFinalAsarResilienceStatus() }; } + if (action === 'resilience.arm-compact-delay') { + return { action, packagedMain, resilience: armFinalAsarResilienceCompactDelay() }; + } if (action === 'resilience.idle-status') { return { action, packagedMain, resilience: await getFinalAsarResilienceIdleStatus() }; } diff --git a/scripts/run-pi-subagent-packaged-smoke.mjs b/scripts/run-pi-subagent-packaged-smoke.mjs index 9004361..bddedaa 100644 --- a/scripts/run-pi-subagent-packaged-smoke.mjs +++ b/scripts/run-pi-subagent-packaged-smoke.mjs @@ -580,12 +580,22 @@ export async function runPackagedProductProof(options) { const backgroundActive = await waitForResilienceProof( electronApplication, (status) => status?.target?.runStatus === 'running' + && status?.target?.errorCode === 'CODING_REQUEST_UNCERTAIN' && status?.other?.runStatus === 'running' && status?.activeProviderRequests?.parent === 2 && status?.resources?.backgroundLeases?.active === 2 && status?.processes?.parent?.length === 2, 'Packaged background lease proof did not reach two active Main-owned runs', ); + const uncertaintyMessage = page.getByText(/请求确认延迟，可能仍在执行/).last(); + await uncertaintyMessage.waitFor({ state: 'visible', timeout: 15_000 }); + const compactButton = page.getByRole('button', { name: '整理上下文' }); + if (await compactButton.isEnabled()) { + throw new Error('Packaged UI allowed compact to overlap a confirmation-uncertain prompt'); + } + if (await page.getByText(/本地编程运行时暂时不可用/).count()) { + throw new Error('Confirmation uncertainty was presented as a local runtime outage'); + } await setMainWindowVisible(electronApplication, false); await new Promise((resolveWait) => setTimeout(resolveWait, 600)); const backgroundHidden = await evaluateProof(electronApplication, 'resilience.status'); @@ -609,6 +619,44 @@ export async function runPackagedProductProof(options) { if (!await resilienceComposer.isEnabled()) { throw new Error('Packaged composer stayed disabled after the hidden run settled'); } + await uncertaintyMessage.waitFor({ state: 'hidden', timeout: 10_000 }); + + const compactDelay = await evaluateProof(electronApplication, 'resilience.arm-compact-delay'); + assertPackagedMain(compactDelay); + if (!(compactDelay?.resilience?.delayMs > 10_000)) { + throw new Error(`Packaged compact proof did not exceed the former 10s threshold: ${JSON.stringify(compactDelay?.resilience)}`); + } + await compactButton.click(); + await uncertaintyMessage.waitFor({ state: 'visible', timeout: 15_000 }); + const compactUncertain = await waitForResilienceProof( + electronApplication, + (status) => status?.target?.runStatus === 'compacting' + && status?.target?.errorCode === 'CODING_REQUEST_UNCERTAIN' + && status?.target?.contextCompaction === 'running' + && status?.activeProviderRequests?.parent === 1 + && status?.resources?.backgroundLeases?.active === 1, + 'Packaged manual compact did not retain ownership after the 10s confirmation threshold', + ); + if (await compactButton.isEnabled()) { + throw new Error('Packaged UI enabled another compact while the first compact was uncertain'); + } + if (await page.getByText(/本地编程运行时暂时不可用/).count()) { + throw new Error('Packaged manual compact uncertainty was presented as a local runtime outage'); + } + const compactSettled = await waitForResilienceProof( + electronApplication, + (status) => status?.target?.runStatus === 'idle' + && status?.target?.errorCode === null + && status?.target?.contextCompaction === 'idle' + && status?.target?.completedCompactions >= 1 + && status?.activeProviderRequests?.parent === 0 + && status?.resources?.backgroundLeases?.active === 0, + 'Packaged manual compact did not settle and release its Main-owned lease', + ); + await uncertaintyMessage.waitFor({ state: 'hidden', timeout: 10_000 }); + if (!await resilienceComposer.isEnabled()) { + throw new Error('Packaged composer stayed disabled after manual compact settled'); + } await setMainWindowVisible(electronApplication, false); const backgroundIdle = await waitForResilienceIdleProof( electronApplication, @@ -849,10 +897,19 @@ export async function runPackagedProductProof(options) { active: backgroundActive.resilience, hiddenPastGrace: backgroundHidden.resilience, idleEvicted: backgroundIdle.resilience, - uiPromptAccepted: true, + uiPromptSubmitted: true, + hostConfirmationUncertain: true, answerVisibleAfterShow: true, composerUnlocked: true, }, + delayedCompact: { + configured: compactDelay.resilience, + uncertain: compactUncertain.resilience, + settled: compactSettled.resilience, + overlappingMutationBlocked: true, + runtimeUnavailableBanner: false, + composerUnlocked: true, + }, intentionalDispose: { active: intentionalActive.resilience, disposed: intentionalDispose.resilience, diff --git a/src/pages/Chat/CodingChatPanel.tsx b/src/pages/Chat/CodingChatPanel.tsx index f571f2e..1f58dd6 100644 --- a/src/pages/Chat/CodingChatPanel.tsx +++ b/src/pages/Chat/CodingChatPanel.tsx @@ -53,6 +53,25 @@ interface LocalComposerAttachment { previewUrl: string; } +interface LocalSubmissionError { + message: string; + backendCode?: string; +} + +function localSubmissionError(error: unknown): LocalSubmissionError { + const message = error instanceof Error ? error.message : String(error); + const details = error && typeof error === 'object' + && 'details' in error + && error.details + && typeof error.details === 'object' + ? error.details as { backendCode?: unknown } + : null; + return { + message, + ...(typeof details?.backendCode === 'string' ? { backendCode: details.backendCode } : {}), + }; +} + function newestConversation( conversations: CodingConversationMetadata[], agentId: string, @@ -124,7 +143,7 @@ export function CodingChatPanel({ const conversationSummaries = useCodingConversationStore((state) => state.summariesByConversationId); const [provisionalDrafts, setProvisionalDrafts] = useState>({}); - const [submissionErrors, setSubmissionErrors] = useState>({}); + const [submissionErrors, setSubmissionErrors] = useState>({}); const [modesByDraftKey, setModesByDraftKey] = useState>({}); const [inspectorOpen, setInspectorOpen] = useState(false); const [attachmentsByDraftKey, setAttachmentsByDraftKey] = useState< @@ -387,7 +406,7 @@ export function CodingChatPanel({ } setSubmissionErrors((current) => { const next = { ...current }; - if (validationError) next[draftKey] = validationError; + if (validationError) next[draftKey] = { message: validationError }; else delete next[draftKey]; return next; }); @@ -488,7 +507,7 @@ export function CodingChatPanel({ }).catch((error) => { setSubmissionErrors((current) => ({ ...current, - [draftKey]: error instanceof Error ? error.message : String(error), + [draftKey]: localSubmissionError(error), })); }).finally(() => { submissionFlightsRef.current.delete(conversationId); @@ -506,6 +525,17 @@ export function CodingChatPanel({ const workerStatus = snapshot?.worker.status ?? 'stopped'; const runtimeError = snapshot?.run.error ?? snapshot?.worker.error ?? null; const preparationError = entryError ?? runtimeError?.message ?? null; + useEffect(() => { + if (!draftKey + || uncertainRequestCount > 0 + || runtimeError?.code === 'CODING_REQUEST_UNCERTAIN') return; + setSubmissionErrors((current) => { + if (current[draftKey]?.backendCode !== 'CODING_REQUEST_UNCERTAIN') return current; + const next = { ...current }; + delete next[draftKey]; + return next; + }); + }, [draftKey, runtimeError?.code, uncertainRequestCount]); const preparing = entryLoadState === 'loading' || workerStatus === 'starting' || workerStatus === 'recovering' @@ -744,7 +774,7 @@ export function CodingChatPanel({ if (!draftKey) return; setSubmissionErrors((current) => ({ ...current, - [draftKey]: error instanceof Error ? error.message : String(error), + [draftKey]: localSubmissionError(error), })); }); }} @@ -772,7 +802,7 @@ export function CodingChatPanel({ runStatus={runStatus} mode={promptMode} queue={snapshot?.queue ?? { items: [] }} - error={submissionError ?? preparationError} + error={submissionError?.message ?? preparationError} recoverableError={!submissionError && Boolean(preparationError)} acceptedCount={acceptedRequestCount} attachments={localAttachments.map(({ id, name, previewUrl }) => ({ id, name, previewUrl }))} diff --git a/src/pages/Chat/CodingConversationHeader.tsx b/src/pages/Chat/CodingConversationHeader.tsx index 1ecac76..a9670a6 100644 --- a/src/pages/Chat/CodingConversationHeader.tsx +++ b/src/pages/Chat/CodingConversationHeader.tsx @@ -170,7 +170,7 @@ export function CodingConversationHeader({ value={modelValue} aria-label="当前对话模型" className="h-9 min-w-52 max-w-72 text-xs" - disabled={Boolean(busyAction)} + disabled={Boolean(busyAction) || running} onChange={(event) => { const selected = parseCodingModelKey(event.target.value); if (!selected) return; @@ -191,7 +191,7 @@ export function CodingConversationHeader({ value={thinkingLevel} aria-label="当前对话思考级别" className="h-9 w-36 shrink-0 text-xs" - disabled={Boolean(busyAction) || !model} + disabled={Boolean(busyAction) || !model || running} onChange={(event) => perform('thinking', async () => { await setCodingConversationThinking( conversation.id, @@ -215,7 +215,7 @@ export function CodingConversationHeader({