wangxuming/makelore
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(coding): bound missing Pi settlement

Browse Source
This commit is contained in:
inman
2026-09-01 12:40:08 +08:00
parent 8062b45103
commit f1fd13a70f
6 changed files with 478 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
electron/coding-runtime/pi/runtime.ts
View File

@@ -1474,19 +1474,20 @@ export class PiConversationRuntime implements CodingConversationRuntime {
const current = snapshot?.run;
if (!snapshot
|| snapshot.cursor.workerGeneration !== event.generation
|| current?.runId !== event.runId
|| runIsTerminal(current.status)) return;
this.emit(event.conversationId, {
op: 'run.state',
run: {
status: 'idle',
runId: event.runId,
...(current.mode ? { mode: current.mode } : {}),
...(current.startedAt !== undefined ? { startedAt: current.startedAt } : {}),
settledAt: this.now(),
terminalReason: 'completed',
},
}, event.runId);
|| current?.runId !== event.runId) return;
if (!runIsTerminal(current.status)) {
this.emit(event.conversationId, {
op: 'run.state',
run: {
status: 'idle',
runId: event.runId,
...(current.mode ? { mode: current.mode } : {}),
...(current.startedAt !== undefined ? { startedAt: current.startedAt } : {}),
settledAt: this.now(),
terminalReason: current.status === 'aborting' ? 'aborted' : 'completed',
},
}, event.runId);
}
this.extensionUi.endRun(event.conversationId, event.runId);
try {
await Promise.allSettled([
@@ -1500,6 +1501,12 @@ export class PiConversationRuntime implements CodingConversationRuntime {
} finally {
this.releaseRunBackgroundLease(event.conversationId, event.runId);
}
if (event.source === 'state_probe') {
const worker = this.pool.getState(event.conversationId);
if (worker?.generation === event.generation && worker.state !== 'crashed') {
await this.hydrateGenerationNow(event.conversationId, worker, false);
}
}
}).catch((error) => {
this.recordProjectionFailure(event.conversationId, event.generation, error);
});

237
electron/coding-runtime/pi/worker-pool.ts
View File

@@ -26,6 +26,45 @@ import {
} from './telemetry';
import { logger } from '../../utils/logger';
const SETTLEMENT_PROBE_INTERVAL_MS = 3_000;
const SETTLEMENT_PROBE_TIMEOUT_MS = 5_000;
const TERMINAL_SETTLEMENT_TIMEOUT_MS = 30_000;
function recordValue(value: unknown): Record<string, unknown> | null {
return value !== null && typeof value === 'object' && !Array.isArray(value)
? value as Record<string, unknown>
: null;
}
function sessionIsAuthoritativelyIdle(value: unknown): boolean {
const state = recordValue(value);
return state?.isStreaming === false
&& state.isCompacting === false
&& state.pendingMessageCount === 0;
}
function sessionIsTerminallyStalled(value: unknown): boolean {
const state = recordValue(value);
return state?.isStreaming === true
&& state.isCompacting === false
&& state.pendingMessageCount === 0
&& state.retryAttempt === 0;
}
function isTerminalAssistantMessage(event: PiRpcEvent): boolean {
if (event.type !== 'message_end') return false;
const message = recordValue(event.message);
return message?.role === 'assistant'
&& ['stop', 'length', 'error', 'aborted'].includes(String(message.stopReason));
}
function continuesAfterTerminalCandidate(event: PiRpcEvent): boolean {
return event.type === 'agent_start'
|| event.type === 'auto_retry_start'
|| event.type === 'compaction_start'
|| (event.type === 'message_start' && recordValue(event.message)?.role === 'user');
}
export interface PiConversationWorker {
readonly id: string;
readonly generation: number;
@@ -122,6 +161,7 @@ export type PiWorkerPoolEvent =
conversationId: string;
generation: number;
runId: string;
source: 'agent_settled' | 'compact_rpc' | 'state_probe';
}
| {
type: 'top-level.failed';
@@ -147,6 +187,9 @@ export interface PiWorkerPoolOptions {
revisionCoordinator?: PiManagedInputRevisionCoordinator;
now?: () => number;
onTelemetry?: (event: PiRuntimeTelemetryEvent) => void;
settlementProbeIntervalMs?: number;
settlementProbeTimeoutMs?: number;
terminalSettlementTimeoutMs?: number;
}
export interface PiProcessLease {
@@ -255,6 +298,20 @@ interface PendingTopLevelRun extends PiTopLevelRunInput {
queuedAt?: number;
}
interface ActiveTopLevelRun {
runId: string;
run: PendingTopLevelRun;
generation: number;
cold: boolean;
acceptedAt?: number;
confirmationStartedAt?: number;
confirmation: 'pending' | 'uncertain' | 'confirmed';
cancelConfirmation?: () => void;
settlementProbeTimer?: ReturnType<typeof setTimeout>;
settlementProbeInFlight?: boolean;
terminalObservedAt?: number;
}
export class PiWorkerPool {
private readonly openWorker: PiWorkerPoolOptions['openWorker'];
private readonly maxRunning: number;
@@ -264,20 +321,14 @@ export class PiWorkerPool {
private readonly revisions: PiManagedInputRevisionCoordinator;
private readonly now: () => number;
private readonly onTelemetry: ((event: PiRuntimeTelemetryEvent) => void) | undefined;
private readonly settlementProbeIntervalMs: number;
private readonly settlementProbeTimeoutMs: number;
private readonly terminalSettlementTimeoutMs: number;
private readonly workers = new Map<string, WorkerRecord>();
private readonly prepareFlights = new Map<string, Promise<PiWorkerPoolState>>();
private readonly rebuildFlights = new Set<Promise<WorkerRecord>>();
private readonly waitingRuns: PendingTopLevelRun[] = [];
private readonly activeRuns = new Map<string, {
runId: string;
run: PendingTopLevelRun;
generation: number;
cold: boolean;
acceptedAt?: number;
confirmationStartedAt?: number;
confirmation: 'pending' | 'uncertain' | 'confirmed';
cancelConfirmation?: () => void;
}>();
private readonly activeRuns = new Map<string, ActiveTopLevelRun>();
private readonly generations = new Map<string, number>();
private readonly listeners = new Set<(event: PiWorkerPoolEvent) => void>();
private readonly reclaimWaiters = new Set<() => void>();
@@ -297,12 +348,27 @@ export class PiWorkerPool {
this.revisions = options.revisionCoordinator ?? new PiManagedInputRevisionCoordinator();
this.now = options.now ?? Date.now;
this.onTelemetry = options.onTelemetry;
this.settlementProbeIntervalMs = options.settlementProbeIntervalMs
?? SETTLEMENT_PROBE_INTERVAL_MS;
this.settlementProbeTimeoutMs = options.settlementProbeTimeoutMs
?? SETTLEMENT_PROBE_TIMEOUT_MS;
this.terminalSettlementTimeoutMs = options.terminalSettlementTimeoutMs
?? TERMINAL_SETTLEMENT_TIMEOUT_MS;
if (!Number.isSafeInteger(this.maxRunning) || this.maxRunning <= 0) {
throw new Error('maxRunning must be a positive safe integer');
}
if (!Number.isSafeInteger(this.maxIdle) || this.maxIdle < 0) {
throw new Error('maxIdle must be a non-negative safe integer');
}
for (const [name, value] of [
['settlementProbeIntervalMs', this.settlementProbeIntervalMs],
['settlementProbeTimeoutMs', this.settlementProbeTimeoutMs],
['terminalSettlementTimeoutMs', this.terminalSettlementTimeoutMs],
] as const) {
if (!Number.isSafeInteger(value) || value <= 0) {
throw new Error(`${name} must be a positive safe integer`);
}
}
}
prepare(conversation: PrepareConversationInput): Promise<PiWorkerPoolState> {
@@ -532,7 +598,7 @@ export class PiWorkerPool {
failTopLevel(conversationId: string, runId: string, error: Error): void {
const active = this.activeRuns.get(conversationId);
if (active?.runId === runId) {
this.activeRuns.delete(conversationId);
this.removeActiveRun(conversationId);
active.cancelConfirmation?.();
this.runningCount -= 1;
const record = this.workers.get(conversationId);
@@ -712,7 +778,7 @@ export class PiWorkerPool {
const records = [...this.workers.values()];
this.workers.clear();
const activeRuns = [...this.activeRuns.values()];
this.activeRuns.clear();
for (const active of activeRuns) this.removeActiveRun(active.run.conversationId);
this.runningCount = 0;
for (const active of activeRuns) active.cancelConfirmation?.();
await Promise.all(records.map(async (record) => {
@@ -766,13 +832,7 @@ export class PiWorkerPool {
});
this.confirmTopLevel(current, run);
if (run.command.type === 'compact') {
this.settleTopLevel(current);
this.emit({
type: 'top-level.settled',
conversationId: run.conversationId,
generation: current.generation,
runId: run.runId,
});
this.settleTopLevelAndNotify(current, 'compact_rpc');
}
run.resolve(response);
} catch (error) {
@@ -787,7 +847,7 @@ export class PiWorkerPool {
return;
}
if (active) {
this.activeRuns.delete(run.conversationId);
this.removeActiveRun(run.conversationId);
active.cancelConfirmation?.();
this.runningCount -= 1;
this.launchWaitingRuns();
@@ -822,6 +882,7 @@ export class PiWorkerPool {
active.cold,
);
record.acceptedPromptCount += 1;
this.scheduleSettlementProbe(record, active);
}
return true;
}
@@ -842,19 +903,13 @@ export class PiWorkerPool {
runId: run.runId,
});
if (run.command.type === 'compact') {
this.settleTopLevel(record);
this.emit({
type: 'top-level.settled',
conversationId: run.conversationId,
generation: record.generation,
runId: run.runId,
});
this.settleTopLevelAndNotify(record, 'compact_rpc');
}
return;
}
if (result.error.code === 'PI_RPC_EXITED'
|| result.error.code === 'PI_RPC_PROTOCOL_ERROR') return;
this.activeRuns.delete(run.conversationId);
this.removeActiveRun(run.conversationId);
active.cancelConfirmation = undefined;
this.runningCount -= 1;
if (this.workers.get(run.conversationId) === record && record.state !== 'crashed') {
@@ -877,10 +932,10 @@ export class PiWorkerPool {
});
}
private settleTopLevel(record: WorkerRecord): void {
private settleTopLevel(record: WorkerRecord): ActiveTopLevelRun | null {
const conversationId = record.conversation.conversationId;
const active = this.activeRuns.get(conversationId);
if (!active || active.generation !== record.generation) return;
if (!active || active.generation !== record.generation) return null;
const cancelConfirmation = active.cancelConfirmation;
if (active.confirmation !== 'confirmed') {
this.confirmTopLevel(record, active.run);
@@ -899,7 +954,7 @@ export class PiWorkerPool {
active.cold,
);
}
this.activeRuns.delete(conversationId);
this.removeActiveRun(conversationId);
cancelConfirmation?.();
this.runningCount -= 1;
record.state = 'idle';
@@ -920,6 +975,110 @@ export class PiWorkerPool {
}
this.launchWaitingRuns();
void this.trimIdleWorkers();
return active;
}
private settleTopLevelAndNotify(
record: WorkerRecord,
source: Extract<PiWorkerPoolEvent, { type: 'top-level.settled' }>['source'],
): boolean {
const active = this.settleTopLevel(record);
if (!active) return false;
this.emit({
type: 'top-level.settled',
conversationId: record.conversation.conversationId,
generation: record.generation,
runId: active.runId,
source,
});
return true;
}
private observeTopLevelEvent(record: WorkerRecord, event: PiRpcEvent): void {
const conversationId = record.conversation.conversationId;
const active = this.activeRuns.get(conversationId);
if (!active || active.generation !== record.generation) return;
if (continuesAfterTerminalCandidate(event)) {
active.terminalObservedAt = undefined;
return;
}
if (isTerminalAssistantMessage(event)
|| (event.type === 'agent_end' && event.willRetry === false)) {
active.terminalObservedAt ??= this.now();
this.scheduleSettlementProbe(record, active);
}
}
private scheduleSettlementProbe(record: WorkerRecord, active: ActiveTopLevelRun): void {
if (this.shuttingDown
|| active.run.command.type !== 'prompt'
|| active.settlementProbeTimer
|| active.settlementProbeInFlight
|| this.activeRuns.get(record.conversation.conversationId) !== active) return;
active.settlementProbeTimer = setTimeout(() => {
active.settlementProbeTimer = undefined;
void this.probeTopLevelSettlement(record, active);
}, this.settlementProbeIntervalMs);
active.settlementProbeTimer.unref();
}
private async probeTopLevelSettlement(
record: WorkerRecord,
active: ActiveTopLevelRun,
): Promise<void> {
const conversationId = record.conversation.conversationId;
if (this.activeRuns.get(conversationId) !== active
|| active.generation !== record.generation
|| this.workers.get(conversationId) !== record
|| record.state === 'crashed') return;
active.settlementProbeInFlight = true;
try {
const response = await record.worker.request({ type: 'get_state' }, {
retry: 'read-only-once',
timeoutMs: this.settlementProbeTimeoutMs,
});
if (this.activeRuns.get(conversationId) !== active) return;
if (sessionIsAuthoritativelyIdle(response.data)) {
this.settleTopLevelAndNotify(record, 'state_probe');
return;
}
if (this.terminalSettlementExpired(active)
&& sessionIsTerminallyStalled(response.data)) {
this.handleInvalidation(record, this.settlementProtocolError(record));
}
} catch {
if (this.activeRuns.get(conversationId) === active
&& this.terminalSettlementExpired(active)) {
this.handleInvalidation(record, this.settlementProtocolError(record));
}
} finally {
active.settlementProbeInFlight = false;
if (this.activeRuns.get(conversationId) === active) {
this.scheduleSettlementProbe(record, active);
}
}
}
private terminalSettlementExpired(active: ActiveTopLevelRun): boolean {
return active.terminalObservedAt !== undefined
&& this.now() - active.terminalObservedAt >= this.terminalSettlementTimeoutMs;
}
private settlementProtocolError(record: WorkerRecord): PiProcessError {
return new PiProcessError(
'PI_RPC_PROTOCOL_ERROR',
'Pi prompt produced a terminal response but did not settle',
{ generation: record.generation },
);
}
private removeActiveRun(conversationId: string): ActiveTopLevelRun | undefined {
const active = this.activeRuns.get(conversationId);
if (!active) return undefined;
this.activeRuns.delete(conversationId);
if (active.settlementProbeTimer) clearTimeout(active.settlementProbeTimer);
active.settlementProbeTimer = undefined;
return active;
}
private async ensureFresh(record: WorkerRecord): Promise<WorkerRecord> {
@@ -1086,7 +1245,17 @@ export class PiWorkerPool {
reconfigureAfterSettled: false,
};
record.unsubscribeEvent = worker.subscribe((event) => {
if (event.type === 'agent_settled') this.settleTopLevel(record);
if (event.type === 'makelore_thread_error'
&& event.code === 'PROMPT_FAILED_AFTER_ACCEPTANCE') {
this.handleInvalidation(record, new PiProcessError(
'PI_RPC_PROTOCOL_ERROR',
'Pi prompt failed after it was accepted',
{ generation },
));
return;
}
this.observeTopLevelEvent(record, event);
if (event.type === 'agent_settled') this.settleTopLevelAndNotify(record, 'agent_settled');
this.emit({
type: 'worker.event',
conversationId: conversation.conversationId,
@@ -1178,7 +1347,7 @@ export class PiWorkerPool {
const active = this.activeRuns.get(conversationId);
if (active?.generation === record.generation) {
this.activeRuns.delete(conversationId);
this.removeActiveRun(conversationId);
active.cancelConfirmation?.();
this.runningCount -= 1;
}
@@ -1206,7 +1375,7 @@ export class PiWorkerPool {
private cancelConversationRuns(conversationId: string, error: Error): void {
const active = this.activeRuns.get(conversationId);
if (active) {
this.activeRuns.delete(conversationId);
this.removeActiveRun(conversationId);
active.cancelConfirmation?.();
this.runningCount -= 1;
}