From f1fd13a70fbf6ae3702a7c164fd42bce1a183a0d Mon Sep 17 00:00:00 2001 From: inman Date: Tue, 1 Sep 2026 12:40:08 +0800 Subject: [PATCH] fix(coding): bound missing Pi settlement --- ...0260901-fix-session-settlement-a47d2e91.md | 72 ++++++ electron/coding-runtime/pi/runtime.ts | 33 ++- electron/coding-runtime/pi/worker-pool.ts | 237 +++++++++++++++--- resources/pi-agent-server.mjs | 10 +- .../unit/pi-agent-server-process-real.test.ts | 40 ++- tests/unit/pi-worker-pool.test.ts | 138 +++++++++- 6 files changed, 478 insertions(+), 52 deletions(-) create mode 100644 .project-docs/30-worklog/tasks/20260901-fix-session-settlement-a47d2e91.md diff --git a/.project-docs/30-worklog/tasks/20260901-fix-session-settlement-a47d2e91.md b/.project-docs/30-worklog/tasks/20260901-fix-session-settlement-a47d2e91.md new file mode 100644 index 0000000..e5d6619 --- /dev/null +++ b/.project-docs/30-worklog/tasks/20260901-fix-session-settlement-a47d2e91.md @@ -0,0 +1,72 @@ +# Task: Fix stuck Coding session settlement + +## Identity + +- Task ID: 20260901-fix-session-settlement-a47d2e91 +- Mode: Feature +- Branch: codex/20260901-fix-session-settlement-a47d2e91-fix-session-settlement +- Worktree: /Users/inmanx/Documents/makelore-fix-session-settlement-a47d2e91 +- Base commit: 8062b45103be45ae036167d3e9890c3ca9ba3e80 +- Owner: codex +- Status: Ready for integration + +## Scope + +- Harden the shared Pi Agent Server/Main terminal handshake so a prompt cannot remain + visibly processing after Pi has produced its terminal assistant response. +- Preserve authoritative `agent_settled` semantics during normal retries, follow-ups, + compaction, and queued work; add a bounded, evidence-based terminal fallback only for + a logically idle target thread. +- Add focused regression coverage for post-accept prompt failure and missing/delayed + settlement, including exactly-once release of target run ownership. + +## Intent And Constraints + +- Do not automatically replay any accepted or uncertain Coding mutation. +- Do not settle directly on `message_end`; Pi may still retry, compact, or process a + queued continuation after that event. +- Keep Pi wire/state Main-private and preserve per-Conversation generation isolation, + run permits, background leases, and sibling Conversation availability. +- A user-visible wait must end in finite time as completed only when target runtime + state proves idle; otherwise it must converge to an explicit safe failure. +- Do not touch or terminate the currently running app/processes from the diagnostic + task while implementing this isolated source fix. + +## Outcome + +- Confirmed the visible indefinite wait was a missing terminal-handshake failure: + Pi had already persisted a final assistant message, but Main never received an + effective `agent_settled`, so run ownership and the processing timer remained live. +- The Agent Server now reports prompt failures that happen after RPC acceptance as a + private target-thread error instead of swallowing the rejected prompt promise. +- Main now probes only the accepted target thread's authoritative runtime state. Exact + idle evidence settles a missing handshake and hydrates the persisted final response; + contradictory terminal state converges to an explicit target-only protocol failure + after a bounded 30-second grace period. +- Preserved normal `agent_settled` authority, retry/compaction/queued-work semantics, + exactly-once cleanup, sibling Conversation availability, and the no-replay rule for + accepted or uncertain mutations. + +## Verification + +- Focused Pi/runtime unit and real-process integration tests: 68 passed across 8 files. +- Full unit suite: 1,732 passed and 3 skipped (including the pressure project). +- `pnpm run typecheck`: passed. +- Targeted ESLint for all changed source/test files: passed. +- `pnpm run lint:check`: passed with 0 errors and 5 pre-existing unrelated warnings. +- `pnpm run build:vite`: passed for Renderer, Main, Preload, and utility bundles; only + existing Browserslist/import/chunk-size warnings remained. +- `git diff --check`: passed. +- Electron E2E was not run: the repository has no shared fixture that can inject a + dropped Pi settlement handshake, and the task intentionally did not stop or restart + the user's currently running application. + +## Follow-ups + +- Integrate this feature branch, then restart/rebuild the desktop app so the currently + running process loads the corrected Agent Server and Main runtime code. + +## Promotion Candidates + +- None. This restores the existing ADR-006 terminal-settlement contract and does not + introduce a new product or architecture decision. diff --git a/electron/coding-runtime/pi/runtime.ts b/electron/coding-runtime/pi/runtime.ts index 56bc7f9..65f81ef 100644 --- a/electron/coding-runtime/pi/runtime.ts +++ b/electron/coding-runtime/pi/runtime.ts @@ -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); }); diff --git a/electron/coding-runtime/pi/worker-pool.ts b/electron/coding-runtime/pi/worker-pool.ts index 158059c..3ade412 100644 --- a/electron/coding-runtime/pi/worker-pool.ts +++ b/electron/coding-runtime/pi/worker-pool.ts @@ -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 | null { + return value !== null && typeof value === 'object' && !Array.isArray(value) + ? value as Record + : 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; + 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(); private readonly prepareFlights = new Map>(); private readonly rebuildFlights = new Set>(); private readonly waitingRuns: PendingTopLevelRun[] = []; - private readonly activeRuns = new Map void; - }>(); + private readonly activeRuns = new Map(); private readonly generations = new Map(); 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 { @@ -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['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 { + 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 { @@ -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; } diff --git a/resources/pi-agent-server.mjs b/resources/pi-agent-server.mjs index ffbb142..23963fb 100644 --- a/resources/pi-agent-server.mjs +++ b/resources/pi-agent-server.mjs @@ -426,7 +426,14 @@ class AgentThread { this.emit(success(id, 'prompt')); }, }).catch((error) => { - if (!accepted) this.emit(failure(id, 'prompt', error)); + if (!accepted) { + this.emit(failure(id, 'prompt', error)); + return; + } + this.emit({ + type: 'makelore_thread_error', + code: 'PROMPT_FAILED_AFTER_ACCEPTANCE', + }); }); return undefined; } @@ -452,6 +459,7 @@ class AgentThread { thinkingLevel: session.thinkingLevel, isStreaming: session.isStreaming, isCompacting: session.isCompacting, + retryAttempt: session.retryAttempt, steeringMode: session.steeringMode, followUpMode: session.followUpMode, sessionFile: session.sessionFile, diff --git a/tests/unit/pi-agent-server-process-real.test.ts b/tests/unit/pi-agent-server-process-real.test.ts index e5accb8..7cbc584 100644 --- a/tests/unit/pi-agent-server-process-real.test.ts +++ b/tests/unit/pi-agent-server-process-real.test.ts @@ -8,6 +8,7 @@ import { afterEach, describe, expect, it } from 'vitest'; import { PiAgentServerProcess } from '../../electron/coding-runtime/pi/agent-server-process'; import { materializeMakelorePiExtension } from '../../electron/coding-runtime/pi/extensions/makelore-runtime'; import { MAKELORE_DEFAULT_LANGUAGE_PROMPT } from '../../electron/coding-runtime/pi/resource-loader'; +import type { PiRpcEvent } from '../../electron/coding-runtime/pi/rpc-client'; import type { PiWorkerProcessOptions } from '../../electron/coding-runtime/pi/worker-process'; const roots: string[] = []; @@ -124,6 +125,16 @@ describe('Pi Agent Server real process', () => { writeFile(languagePromptPath, MAKELORE_DEFAULT_LANGUAGE_PROMPT), ]); const extensionPath = await materializeMakelorePiExtension(extensionDir); + const invalidJsonExtensionPath = path.join(extensionDir, 'invalid-json-extension.mjs'); + await writeFile(invalidJsonExtensionPath, ` +export function createMakeloreRuntime() { + return async function invalidJsonRuntime(pi) { + pi.on('message_end', async (event) => ({ + message: { ...event.message, makeloreInvalidJson: 1n }, + })); + }; +} +`); const runtimeRoot = path.resolve('node_modules/@earendil-works/pi-coding-agent'); const server = new PiAgentServerProcess({ executablePath: process.execPath, @@ -132,7 +143,11 @@ describe('Pi Agent Server real process', () => { configDir, }); - const workerOptions = (conversationId: string, generation: number): PiWorkerProcessOptions => { + const workerOptions = ( + conversationId: string, + generation: number, + selectedExtensionPath = extensionPath, + ): PiWorkerProcessOptions => { const contextFile = path.join(root, `${conversationId}.json`); return { executablePath: process.execPath, @@ -148,7 +163,7 @@ describe('Pi Agent Server real process', () => { '--thinking', 'medium', '--system-prompt', promptPath, '--append-system-prompt', languagePromptPath, - '--extension', extensionPath, + '--extension', selectedExtensionPath, '--session-id', `session-${conversationId}`, ], env: { @@ -163,7 +178,7 @@ describe('Pi Agent Server real process', () => { }; }; - await Promise.all(['left', 'right'].map(async (conversationId) => { + await Promise.all(['left', 'right', 'broken'].map(async (conversationId) => { await writeFile(path.join(root, `${conversationId}.json`), JSON.stringify({ conversationId, workerGeneration: 1, @@ -216,6 +231,25 @@ describe('Pi Agent Server real process', () => { provider.release(); await Promise.all([leftSettled, rightSettled]); + const broken = server.createWorker(workerOptions('broken', 1, invalidJsonExtensionPath)); + await broken.start(); + const postAcceptanceFailure = new Promise((resolve) => { + const unsubscribe = broken.subscribe((event) => { + if (event.type !== 'makelore_thread_error') return; + unsubscribe(); + resolve(event); + }); + }); + await expect(broken.request({ type: 'prompt', message: 'BREAK_AFTER_ACCEPTANCE' })) + .resolves.toMatchObject({ success: true }); + await expect(postAcceptanceFailure).resolves.toMatchObject({ + type: 'makelore_thread_error', + code: 'PROMPT_FAILED_AFTER_ACCEPTANCE', + }); + await expect(broken.request<{ isStreaming: boolean }>({ type: 'get_state' })) + .resolves.toMatchObject({ data: { isStreaming: false } }); + await broken.stop('test_injection'); + await left.stop('test_injection'); expect(server.processId).toBe(processId); expect(server.activeThreadCount).toBe(1); diff --git a/tests/unit/pi-worker-pool.test.ts b/tests/unit/pi-worker-pool.test.ts index 477ef53..ff3dd1c 100644 --- a/tests/unit/pi-worker-pool.test.ts +++ b/tests/unit/pi-worker-pool.test.ts @@ -58,6 +58,12 @@ class FakeWorker implements PiConversationWorker { private readonly invalidationListeners = new Set<(error: PiProcessError) => void>(); private timeoutType: string | null = null; private pendingType: string | null = null; + private stateData: unknown = { + isStreaming: true, + isCompacting: false, + pendingMessageCount: 0, + retryAttempt: 0, + }; private lateResult: ((result: { response?: PiRpcResponse; error?: PiProcessError; @@ -86,7 +92,18 @@ class FakeWorker implements PiConversationWorker { } | undefined)?.onLateResult; throw new PiProcessFailure('PI_RPC_TIMEOUT', `fake ${command.type} confirmation timeout`); } - return { type: 'response' as const, id: 'fake', success: true }; + return { + type: 'response' as const, + id: 'fake', + success: true, + ...(command.type === 'get_state' + ? { data: structuredClone(this.stateData) as T } + : {}), + }; + } + + setState(state: unknown): void { + this.stateData = structuredClone(state); } timeoutNext(type: string): void { @@ -310,6 +327,125 @@ describe('Pi worker pool', () => { expect(pool.getResilienceProofDiagnostics().runs).toEqual({ active: 0, waiting: 0 }); }); + it('settles from authoritative idle state when agent_settled is missing and ignores a late duplicate', async () => { + const worker = new FakeWorker('worker-target'); + const events: PiWorkerPoolEvent[] = []; + const pool = new PiWorkerPool({ + maxIdle: 2, + settlementProbeIntervalMs: 5, + settlementProbeTimeoutMs: 20, + terminalSettlementTimeoutMs: 100, + openWorker: async () => ({ + worker, + session: { piSessionId: 'session-target', sessionKey: 'key-target' }, + }), + }); + pool.subscribe((event) => events.push(event)); + await pool.prepare(conversation('conversation-target')); + + const ticket = pool.startTopLevel({ + conversationId: 'conversation-target', + runId: 'run-missing-settled', + command: { type: 'prompt', message: 'finish without settlement event' }, + }); + await expect(ticket.accepted).resolves.toMatchObject({ success: true }); + worker.setState({ + isStreaming: false, + isCompacting: false, + pendingMessageCount: 0, + retryAttempt: 0, + }); + + await expect.poll(() => pool.getActiveRun('conversation-target')).toBeNull(); + expect(pool.getState('conversation-target')?.state).toBe('idle'); + expect(events.filter((event) => event.type === 'top-level.settled')).toEqual([ + expect.objectContaining({ source: 'state_probe' }), + ]); + + worker.emit({ type: 'agent_settled' }); + expect(events.filter((event) => event.type === 'top-level.settled')).toHaveLength(1); + await pool.shutdown(); + }); + + it('fails only the target thread when an accepted prompt rejects during settlement', async () => { + const workers = new Map(); + const events: PiWorkerPoolEvent[] = []; + const pool = new PiWorkerPool({ + maxIdle: 2, + 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}`, + }, + }; + }, + }); + pool.subscribe((event) => events.push(event)); + await Promise.all([ + pool.prepare(conversation('conversation-target')), + pool.prepare(conversation('conversation-sibling')), + ]); + const ticket = pool.startTopLevel({ + conversationId: 'conversation-target', + runId: 'run-post-accept-failure', + command: { type: 'prompt', message: 'accepted then failed' }, + }); + await expect(ticket.accepted).resolves.toMatchObject({ success: true }); + + workers.get('conversation-target')!.emit({ + type: 'makelore_thread_error', + code: 'PROMPT_FAILED_AFTER_ACCEPTANCE', + }); + + expect(pool.getActiveRun('conversation-target')).toBeNull(); + expect(pool.getState('conversation-target')).toMatchObject({ + state: 'crashed', + failureCode: 'PI_RPC_PROTOCOL_ERROR', + }); + expect(pool.getState('conversation-sibling')).toMatchObject({ state: 'ready' }); + expect(events).toContainEqual(expect.objectContaining({ + type: 'worker.crashed', + conversationId: 'conversation-target', + })); + expect(workers.get('conversation-target')!.requests.filter(({ type }) => type === 'prompt')) + .toHaveLength(1); + await pool.shutdown(); + }); + + it('bounds a contradictory terminal phase without replaying the accepted prompt', async () => { + const worker = new FakeWorker('worker-target'); + const pool = new PiWorkerPool({ + maxIdle: 2, + settlementProbeIntervalMs: 5, + settlementProbeTimeoutMs: 20, + terminalSettlementTimeoutMs: 25, + openWorker: async () => ({ + worker, + session: { piSessionId: 'session-target', sessionKey: 'key-target' }, + }), + }); + await pool.prepare(conversation('conversation-target')); + const ticket = pool.startTopLevel({ + conversationId: 'conversation-target', + runId: 'run-terminal-stall', + command: { type: 'prompt', message: 'terminal stall' }, + }); + await expect(ticket.accepted).resolves.toMatchObject({ success: true }); + worker.emit({ + type: 'message_end', + message: { role: 'assistant', content: [], stopReason: 'stop' }, + }); + + await expect.poll(() => pool.getState('conversation-target')?.state).toBe('crashed'); + expect(pool.getActiveRun('conversation-target')).toBeNull(); + expect(worker.requests.filter(({ type }) => type === 'prompt')).toHaveLength(1); + await pool.shutdown(); + }); + it('injects a proof failure only into the requested current generation', async () => { const workers = new Map(); const pool = new PiWorkerPool({