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

feat: project Pi conversation events

Browse Source
This commit is contained in:
brother7
2026-08-23 08:07:33 +08:00
parent e79aeffffa
commit 3599064bc4
11 changed files with 2428 additions and 32 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -35,6 +35,10 @@ import {
type ConversationReducerState,
} from '../conversation-reducer';
import { CodingRuntimeContractError } from '../in-memory-conversation-runtime';
import {
PiEventProjector,
type PiEventProjectorOptions,
} from './event-projector';
import { PiProcessError } from './process-errors';
import type { PiSessionRegistry } from './session-registry';
import {
@@ -66,6 +70,10 @@ import {
type PiWorkerPoolState,
PiWorkerPool,
} from './worker-pool';
import {
PiSessionProjectionError,
projectPiSessionSnapshot,
} from './session-projector';
type RuntimeIdKind = 'run' | 'queue';
@@ -74,6 +82,7 @@ export interface PiConversationRuntimeOptions {
registry: PiSessionRegistry;
resolveModel(model: ProductModelRef): Promise<PiProviderSelection>;
resolveImages?(attachments: Array<{ attachmentId: string }>): Promise<unknown[]>;
projectImage?: PiEventProjectorOptions['projectImage'];
createId?(kind: RuntimeIdKind): string;
now?: () => number;
providerRefreshCoordinator?: PiProviderRefreshCoordinator;
@@ -332,6 +341,9 @@ function publicWorkerState(state: PiWorkerPoolState): ConversationSnapshot['work
function runtimeFailure(error: unknown): CodingRuntimePublicError {
if (error instanceof CodingRuntimeContractError) return clone(error.publicError);
if (error instanceof PiSessionProjectionError) {
return { code: error.code, message: error.message, recoverable: error.recoverable };
}
if (error instanceof PiProcessError) {
if (error.code === 'PI_RPC_PROTOCOL_ERROR') {
return {
@@ -383,6 +395,7 @@ export class PiConversationRuntime implements CodingConversationRuntime {
private readonly registry: PiSessionRegistry;
private readonly resolveModel: PiConversationRuntimeOptions['resolveModel'];
private readonly resolveImages: NonNullable<PiConversationRuntimeOptions['resolveImages']>;
private readonly projectImage: PiEventProjectorOptions['projectImage'];
private readonly createRuntimeId: (kind: RuntimeIdKind) => string;
private readonly now: () => number;
private readonly providerRefresh: PiProviderRefreshCoordinator;
@@ -391,6 +404,12 @@ export class PiConversationRuntime implements CodingConversationRuntime {
private readonly states = new Map<string, ConversationReducerState>();
private readonly inputs = new Map<string, PrepareConversationInput>();
private readonly listeners = new Set<(patch: ConversationPatchEnvelope) => void>();
private readonly projectors = new Map<string, PiEventProjector>();
private readonly projectionChains = new Map<string, Promise<void>>();
private readonly hydrationFlights = new Map<
string,
{ generation: number; flight: Promise<void> }
>();
private readonly unsubscribePool: () => void;
constructor(options: PiConversationRuntimeOptions) {
@@ -407,6 +426,7 @@ export class PiConversationRuntime implements CodingConversationRuntime {
}
return [];
});
this.projectImage = options.projectImage;
this.createRuntimeId = options.createId ?? (() => randomUUID());
this.now = options.now ?? Date.now;
this.providerRefresh = options.providerRefreshCoordinator ?? new PiProviderRefreshCoordinator();
@@ -433,21 +453,33 @@ export class PiConversationRuntime implements CodingConversationRuntime {
const worker = await this.prepareWorker(canonicalInput);
await this.registry.ensureBinding(canonicalInput, async () => clone(worker.session));
this.inputs.set(input.conversationId, clone(canonicalInput));
if (!this.states.has(input.conversationId)) {
const isNewState = !this.states.has(input.conversationId);
if (isNewState) {
this.states.set(input.conversationId, createConversationReducerState(
emptySnapshot(canonicalInput, worker),
));
this.resetProjector(input.conversationId);
} else {
this.emit(input.conversationId, { op: 'worker.state', state: publicWorkerState(worker) });
}
if (isNewState) {
try {
await this.requestHydration(input.conversationId, worker, false);
} catch (error) {
this.recordProjectionFailure(input.conversationId, worker.generation, error);
throw error;
}
}
return this.runtimeState(input.conversationId);
}
async getSnapshot(conversationId: string): Promise<ConversationSnapshot> {
await this.waitForProjection(conversationId);
return clone(this.snapshot(conversationId));
}
async prompt(input: PromptConversationInput): Promise<PromptAcceptance> {
await this.waitForProjection(input.conversationId);
if (input.mode === 'steer') {
const acceptance = await this.steer(input);
return {
@@ -474,6 +506,20 @@ export class PiConversationRuntime implements CodingConversationRuntime {
this.snapshot(input.conversationId);
const images = await this.resolveImages(input.attachments);
const runId = this.id('run');
const messageId = `client:${input.clientRequestId}`;
this.emit(input.conversationId, {
op: 'message.upsert',
node: {
kind: 'message',
id: messageId,
clientRequestId: input.clientRequestId,
role: 'user',
status: 'optimistic',
blocks: input.text.length > 0
? [{ kind: 'text', id: `${messageId}:content:0`, text: input.text, status: 'complete' }]
: [],
},
});
const command: PiRpcCommand = {
type: 'prompt',
message: input.text,
@@ -515,6 +561,7 @@ export class PiConversationRuntime implements CodingConversationRuntime {
}
async abort(conversationId: string): Promise<void> {
await this.waitForProjection(conversationId);
const current = this.snapshot(conversationId).run;
this.emit(conversationId, {
op: 'run.state',
@@ -532,6 +579,7 @@ export class PiConversationRuntime implements CodingConversationRuntime {
}
async setModel(input: SetConversationModelInput): Promise<ConversationModelState> {
await this.waitForProjection(input.conversationId);
const snapshot = this.snapshot(input.conversationId);
const thinkingLevel = snapshot.conversation.model.model?.thinkingLevel ?? 'off';
const selection = await this.resolveModel({
@@ -552,10 +600,15 @@ export class PiConversationRuntime implements CodingConversationRuntime {
this.replaceModel(input.conversationId, persisted);
try {
const worker = await this.pool.reconfigureConversationModel(input.conversationId, persisted);
if (worker) await this.hydrateRecoveredGeneration(input.conversationId, worker, true);
if (worker) await this.requestHydration(input.conversationId, worker, true);
} catch (error) {
const worker = this.pool.getState(input.conversationId);
if (worker) this.replaceWorkerGeneration(input.conversationId, worker, true);
if (worker) {
if (this.snapshot(input.conversationId).cursor.workerGeneration !== worker.generation) {
this.replaceWorkerGeneration(input.conversationId, worker, true);
}
this.recordProjectionFailure(input.conversationId, worker.generation, error);
}
throw error;
}
return clone(persisted);
@@ -572,6 +625,7 @@ export class PiConversationRuntime implements CodingConversationRuntime {
}
async setThinking(input: SetThinkingLevelInput): Promise<ConversationModelState> {
await this.waitForProjection(input.conversationId);
const current = this.snapshot(input.conversationId).conversation.model;
if (!current.model) {
throw new CodingRuntimeContractError(
@@ -595,6 +649,7 @@ export class PiConversationRuntime implements CodingConversationRuntime {
}
async compact(conversationId: string): Promise<void> {
await this.waitForProjection(conversationId);
const runId = this.id('run');
const ticket = this.pool.startTopLevel({
conversationId,
@@ -614,6 +669,7 @@ export class PiConversationRuntime implements CodingConversationRuntime {
}
async fork(input: ForkConversationInput): Promise<ForkResult> {
await this.waitForProjection(input.sourceConversationId);
this.snapshot(input.sourceConversationId);
const registered = await this.registry.prepare(input.conversation);
const canonicalInput: PrepareConversationInput = {
@@ -635,12 +691,24 @@ export class PiConversationRuntime implements CodingConversationRuntime {
this.inputs.set(canonicalInput.conversationId, clone(canonicalInput));
const snapshot = emptySnapshot(canonicalInput, worker);
this.states.set(canonicalInput.conversationId, createConversationReducerState(snapshot));
return { conversationId: canonicalInput.conversationId, snapshot: clone(snapshot) };
this.resetProjector(canonicalInput.conversationId);
try {
await this.requestHydration(canonicalInput.conversationId, worker, false);
} catch (error) {
this.recordProjectionFailure(canonicalInput.conversationId, worker.generation, error);
throw error;
}
return {
conversationId: canonicalInput.conversationId,
snapshot: clone(this.snapshot(canonicalInput.conversationId)),
};
}
async recover(conversationId: string): Promise<ConversationRuntimeState> {
await this.waitForProjection(conversationId);
const state = await this.pool.recover(conversationId);
await this.hydrateRecoveredGeneration(conversationId, state, false);
await this.requestHydration(conversationId, state, false);
this.settleRecoveredRun(conversationId);
return this.runtimeState(conversationId);
}
@@ -649,6 +717,9 @@ export class PiConversationRuntime implements CodingConversationRuntime {
this.registry.forget(conversationId);
this.inputs.delete(conversationId);
this.states.delete(conversationId);
this.projectors.delete(conversationId);
this.projectionChains.delete(conversationId);
this.hydrationFlights.delete(conversationId);
}
subscribe(listener: (patch: ConversationPatchEnvelope) => void): () => void {
@@ -659,12 +730,16 @@ export class PiConversationRuntime implements CodingConversationRuntime {
async shutdown(): Promise<void> {
this.unsubscribePool();
await this.pool.shutdown();
this.projectors.clear();
this.projectionChains.clear();
this.hydrationFlights.clear();
}
private async queue(
mode: 'steer' | 'follow-up',
input: QueueMessageInput,
): Promise<QueueAcceptance> {
await this.waitForProjection(input.conversationId);
const snapshot = this.snapshot(input.conversationId);
const images = await this.resolveImages(input.attachments);
const queuePosition = snapshot.queue.items.length + 1;
@@ -720,7 +795,7 @@ export class PiConversationRuntime implements CodingConversationRuntime {
if (!this.pool.getState(input.conversationId)) return;
const recovered = await this.pool.recover(input.conversationId);
if (this.states.has(input.conversationId)) {
await this.hydrateRecoveredGeneration(input.conversationId, recovered, false);
await this.requestHydration(input.conversationId, recovered, false);
}
},
});
@@ -748,7 +823,7 @@ export class PiConversationRuntime implements CodingConversationRuntime {
refreshCredential: async () => await this.refreshCredential!(accountId),
reopenWorker: async () => {
const recovered = await this.pool.recover(conversationId);
await this.hydrateRecoveredGeneration(conversationId, recovered, true);
await this.requestHydration(conversationId, recovered, true);
},
});
} else {
@@ -822,7 +897,12 @@ export class PiConversationRuntime implements CodingConversationRuntime {
private onPoolEvent(event: PiWorkerPoolEvent): void {
if (event.type === 'worker.replaced') {
if (!this.states.has(event.conversationId)) return;
this.replaceWorkerGeneration(event.conversationId, event.state, true);
this.resetProjector(event.conversationId);
void this.requestHydration(event.conversationId, event.state, true).catch((error) => {
this.recordProjectionFailure(event.conversationId, event.generation, error);
});
return;
}
if (event.type === 'worker.crashed') {
@@ -830,19 +910,27 @@ export class PiConversationRuntime implements CodingConversationRuntime {
if (state) this.emit(event.conversationId, { op: 'worker.state', state: publicWorkerState(state) });
return;
}
if (event.event.type !== 'agent_settled') return;
const current = this.states.get(event.conversationId)?.snapshot?.run;
if (!current || current.status === 'idle') return;
this.emit(event.conversationId, {
op: 'run.state',
run: {
status: 'idle',
...(current.runId ? { runId: current.runId } : {}),
settledAt: this.now(),
terminalReason: current.status === 'aborting' ? 'aborted' : 'completed',
},
}, current.runId);
this.emit(event.conversationId, { op: 'queue.replace', queue: { items: [] } }, current.runId);
void this.enqueueProjection(event.conversationId, async () => {
const snapshot = this.states.get(event.conversationId)?.snapshot;
if (!snapshot || snapshot.cursor.workerGeneration !== event.generation) return;
const projector = this.projector(event.conversationId);
const patches = await projector.project(snapshot, event.event);
for (const patch of patches) {
this.emit(event.conversationId, patch, this.snapshot(event.conversationId).run.runId);
}
if (event.event.type === 'agent_end' || event.event.type === 'agent_settled') {
const worker = this.pool.getState(event.conversationId);
if (worker?.generation === event.generation) {
await this.hydrateGenerationNow(
event.conversationId,
worker,
event.event.type === 'agent_end',
);
}
}
}).catch((error) => {
this.recordProjectionFailure(event.conversationId, event.generation, error);
});
}
private failRun(conversationId: string, runId: string, error: unknown): void {
@@ -860,16 +948,121 @@ export class PiConversationRuntime implements CodingConversationRuntime {
}, runId);
}
private async hydrateRecoveredGeneration(
private requestHydration(
conversationId: string,
workerState: PiWorkerPoolState,
preserveRun: boolean,
): Promise<void> {
await Promise.all([
const existing = this.hydrationFlights.get(conversationId);
if (existing?.generation === workerState.generation) return existing.flight;
const hydration = this.enqueueProjection(conversationId, async () => {
await this.hydrateGenerationNow(conversationId, workerState, preserveRun);
});
const flight = hydration.finally(() => {
if (this.hydrationFlights.get(conversationId)?.flight === flight) {
this.hydrationFlights.delete(conversationId);
}
});
this.hydrationFlights.set(conversationId, { generation: workerState.generation, flight });
return flight;
}
private async hydrateGenerationNow(
conversationId: string,
workerState: PiWorkerPoolState,
preserveRun: boolean,
): Promise<void> {
const currentWorker = this.pool.getState(conversationId);
if (!currentWorker || currentWorker.generation !== workerState.generation) return;
if (this.snapshot(conversationId).cursor.workerGeneration !== workerState.generation) {
this.replaceWorkerGeneration(conversationId, workerState, preserveRun);
this.resetProjector(conversationId);
}
const [stateResponse, entriesResponse, statsResponse] = 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.replaceWorkerGeneration(conversationId, workerState, preserveRun);
const before = this.snapshot(conversationId);
let projected = await projectPiSessionSnapshot({
snapshot: before,
workerGeneration: workerState.generation,
state: stateResponse.data,
entries: entriesResponse.data,
stats: statsResponse.data,
...(this.projectImage ? { projectImage: this.projectImage } : {}),
});
if (preserveRun) {
projected = {
...projected,
run: clone(before.run),
queue: clone(before.queue),
};
}
this.states.set(conversationId, createConversationReducerState(projected));
}
private enqueueProjection(conversationId: string, action: () => Promise<void>): Promise<void> {
const previous = this.projectionChains.get(conversationId) ?? Promise.resolve();
const flight = previous.then(action);
const tail = flight.catch(() => undefined);
this.projectionChains.set(conversationId, tail);
void tail.finally(() => {
if (this.projectionChains.get(conversationId) === tail) {
this.projectionChains.delete(conversationId);
}
});
return flight;
}
private async waitForProjection(conversationId: string): Promise<void> {
await this.projectionChains.get(conversationId);
}
private projector(conversationId: string): PiEventProjector {
const existing = this.projectors.get(conversationId);
if (existing) return existing;
return this.resetProjector(conversationId);
}
private resetProjector(conversationId: string): PiEventProjector {
const projector = new PiEventProjector({
createId: randomUUID,
now: this.now,
...(this.projectImage ? { projectImage: this.projectImage } : {}),
});
this.projectors.set(conversationId, projector);
return projector;
}
private recordProjectionFailure(
conversationId: string,
generation: number,
error: unknown,
): void {
const state = this.states.get(conversationId);
const snapshot = state?.snapshot;
if (!state || !snapshot || snapshot.cursor.workerGeneration !== generation) return;
const publicError = runtimeFailure(error);
this.states.set(conversationId, createConversationReducerState({
...clone(snapshot),
worker: { status: 'error', generation, error: publicError },
run: {
...clone(snapshot.run),
status: 'error',
error: publicError,
},
}));
}
private settleRecoveredRun(conversationId: string): void {
const current = this.snapshot(conversationId);
this.states.set(conversationId, createConversationReducerState({
...clone(current),
run: { status: 'idle' },
queue: { items: [] },
pendingInteractions: [],
}));
}
private replaceWorkerGeneration(