feat: add conversation runtime contracts
This commit is contained in:
451
electron/coding-runtime/in-memory-conversation-runtime.ts
Normal file
451
electron/coding-runtime/in-memory-conversation-runtime.ts
Normal file
@@ -0,0 +1,451 @@
|
||||
import type {
|
||||
CodingConversationRuntime,
|
||||
CodingRuntimeErrorCode,
|
||||
CodingRuntimePublicError,
|
||||
ConversationModelState,
|
||||
ConversationPatch,
|
||||
ConversationPatchEnvelope,
|
||||
ConversationRuntimeState,
|
||||
ConversationSnapshot,
|
||||
ForkConversationInput,
|
||||
ForkResult,
|
||||
PrepareConversationInput,
|
||||
PromptAcceptance,
|
||||
PromptConversationInput,
|
||||
QueueAcceptance,
|
||||
QueueMessageInput,
|
||||
SetConversationModelInput,
|
||||
SetThinkingLevelInput,
|
||||
} from './contracts';
|
||||
import {
|
||||
createConversationReducerState,
|
||||
reduceConversationPatch,
|
||||
type ConversationReducerState,
|
||||
} from './conversation-reducer';
|
||||
|
||||
export class CodingRuntimeContractError extends Error {
|
||||
readonly publicError: CodingRuntimePublicError;
|
||||
|
||||
constructor(code: CodingRuntimeErrorCode, message: string, recoverable: boolean) {
|
||||
super(message);
|
||||
this.name = 'CodingRuntimeContractError';
|
||||
this.publicError = { code, message, recoverable };
|
||||
}
|
||||
}
|
||||
|
||||
export interface InMemoryConversationRuntimeOptions {
|
||||
snapshots?: ConversationSnapshot[];
|
||||
now?: () => number;
|
||||
createId?: (kind: 'run' | 'node' | 'queue' | 'compaction') => string;
|
||||
}
|
||||
|
||||
function clone<T>(value: T): T {
|
||||
return structuredClone(value);
|
||||
}
|
||||
|
||||
function emptySnapshot(input: PrepareConversationInput): ConversationSnapshot {
|
||||
return {
|
||||
schemaVersion: 1,
|
||||
conversation: {
|
||||
id: input.conversationId,
|
||||
projectId: input.projectId,
|
||||
agentId: input.agentId,
|
||||
title: input.title,
|
||||
model: clone(input.model),
|
||||
},
|
||||
nodes: [],
|
||||
run: { status: 'idle' },
|
||||
queue: { items: [] },
|
||||
context: {
|
||||
usedTokens: 0,
|
||||
contextWindow: 0,
|
||||
compaction: 'idle',
|
||||
},
|
||||
pendingInteractions: [],
|
||||
worker: {
|
||||
status: 'ready',
|
||||
generation: 0,
|
||||
},
|
||||
cursor: {
|
||||
workerGeneration: 0,
|
||||
seq: 0,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export class InMemoryConversationRuntime implements CodingConversationRuntime {
|
||||
private readonly states = new Map<string, ConversationReducerState>();
|
||||
private readonly listeners = new Set<(patch: ConversationPatchEnvelope) => void>();
|
||||
private readonly promptAcceptances = new Map<string, PromptAcceptance>();
|
||||
private readonly queueAcceptances = new Map<string, QueueAcceptance>();
|
||||
private readonly now: () => number;
|
||||
private readonly createId: InMemoryConversationRuntimeOptions['createId'];
|
||||
private nextId = 0;
|
||||
|
||||
constructor(options: InMemoryConversationRuntimeOptions = {}) {
|
||||
this.now = options.now ?? Date.now;
|
||||
this.createId = options.createId;
|
||||
for (const snapshot of options.snapshots ?? []) {
|
||||
const state = createConversationReducerState(snapshot);
|
||||
if (!state.snapshot || state.invalidation) {
|
||||
throw new CodingRuntimeContractError(
|
||||
'CODING_SESSION_UNREADABLE',
|
||||
'Initial Conversation snapshot is invalid',
|
||||
true,
|
||||
);
|
||||
}
|
||||
this.states.set(snapshot.conversation.id, state);
|
||||
}
|
||||
}
|
||||
|
||||
private id(kind: 'run' | 'node' | 'queue' | 'compaction'): string {
|
||||
if (this.createId) return this.createId(kind);
|
||||
this.nextId += 1;
|
||||
return `${kind}-${this.nextId}`;
|
||||
}
|
||||
|
||||
private state(conversationId: string): ConversationReducerState {
|
||||
const state = this.states.get(conversationId);
|
||||
if (!state?.snapshot) {
|
||||
throw new CodingRuntimeContractError(
|
||||
'CODING_CONVERSATION_NOT_FOUND',
|
||||
'Conversation is not prepared',
|
||||
true,
|
||||
);
|
||||
}
|
||||
return state;
|
||||
}
|
||||
|
||||
private snapshot(conversationId: string): ConversationSnapshot {
|
||||
return this.state(conversationId).snapshot as ConversationSnapshot;
|
||||
}
|
||||
|
||||
private replaceSnapshot(snapshot: ConversationSnapshot): void {
|
||||
const state = createConversationReducerState(snapshot);
|
||||
if (!state.snapshot || state.invalidation) {
|
||||
throw new CodingRuntimeContractError(
|
||||
'CODING_SESSION_UNREADABLE',
|
||||
'Conversation snapshot replacement is invalid',
|
||||
true,
|
||||
);
|
||||
}
|
||||
this.states.set(snapshot.conversation.id, state);
|
||||
}
|
||||
|
||||
private emit(conversationId: string, patch: ConversationPatch, runId?: string): void {
|
||||
const state = this.state(conversationId);
|
||||
const snapshot = state.snapshot as ConversationSnapshot;
|
||||
const envelope: ConversationPatchEnvelope = {
|
||||
conversationId,
|
||||
workerGeneration: snapshot.cursor.workerGeneration,
|
||||
...(runId ? { runId } : {}),
|
||||
seq: snapshot.cursor.seq + 1,
|
||||
at: this.now(),
|
||||
patch: clone(patch),
|
||||
};
|
||||
const next = reduceConversationPatch(state, envelope);
|
||||
if (next.invalidation) {
|
||||
throw new CodingRuntimeContractError(
|
||||
'CODING_RUNTIME_PROTOCOL_ERROR',
|
||||
next.invalidation.reason,
|
||||
true,
|
||||
);
|
||||
}
|
||||
this.states.set(conversationId, next);
|
||||
for (const listener of this.listeners) listener(clone(envelope));
|
||||
}
|
||||
|
||||
private runtimeState(conversationId: string): ConversationRuntimeState {
|
||||
const snapshot = this.snapshot(conversationId);
|
||||
return {
|
||||
conversationId,
|
||||
status: snapshot.worker.status,
|
||||
workerGeneration: snapshot.worker.generation,
|
||||
...(snapshot.worker.error ? { error: clone(snapshot.worker.error) } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
async prepare(input: PrepareConversationInput): Promise<ConversationRuntimeState> {
|
||||
const existing = this.states.get(input.conversationId)?.snapshot;
|
||||
if (!existing) this.replaceSnapshot(emptySnapshot(input));
|
||||
return this.runtimeState(input.conversationId);
|
||||
}
|
||||
|
||||
async getSnapshot(conversationId: string): Promise<ConversationSnapshot> {
|
||||
return clone(this.snapshot(conversationId));
|
||||
}
|
||||
|
||||
async prompt(input: PromptConversationInput): Promise<PromptAcceptance> {
|
||||
const key = `${input.conversationId}:${input.clientRequestId}`;
|
||||
const prior = this.promptAcceptances.get(key);
|
||||
if (prior) return clone(prior);
|
||||
|
||||
const snapshot = this.snapshot(input.conversationId);
|
||||
if (snapshot.worker.status !== 'ready') {
|
||||
throw new CodingRuntimeContractError(
|
||||
'CODING_RUNTIME_START_FAILED',
|
||||
'Conversation worker is not ready',
|
||||
true,
|
||||
);
|
||||
}
|
||||
|
||||
const runId = this.id('run');
|
||||
const messageId = this.id('node');
|
||||
this.emit(input.conversationId, {
|
||||
op: 'message.upsert',
|
||||
node: {
|
||||
kind: 'message',
|
||||
id: messageId,
|
||||
clientRequestId: input.clientRequestId,
|
||||
role: 'user',
|
||||
status: 'optimistic',
|
||||
blocks: [
|
||||
...(input.text ? [{
|
||||
kind: 'text' as const,
|
||||
id: `${messageId}:text`,
|
||||
text: input.text,
|
||||
status: 'complete' as const,
|
||||
}] : []),
|
||||
...input.attachments.map(({ attachmentId }, index) => ({
|
||||
kind: 'image' as const,
|
||||
id: `${messageId}:attachment:${index}`,
|
||||
attachmentId,
|
||||
mime: 'application/octet-stream',
|
||||
})),
|
||||
],
|
||||
},
|
||||
}, runId);
|
||||
this.emit(input.conversationId, {
|
||||
op: 'run.state',
|
||||
run: {
|
||||
status: 'running',
|
||||
runId,
|
||||
mode: input.mode,
|
||||
startedAt: this.now(),
|
||||
},
|
||||
}, runId);
|
||||
|
||||
const acceptance: PromptAcceptance = {
|
||||
accepted: true,
|
||||
conversationId: input.conversationId,
|
||||
clientRequestId: input.clientRequestId,
|
||||
runId,
|
||||
mode: input.mode,
|
||||
};
|
||||
this.promptAcceptances.set(key, acceptance);
|
||||
return clone(acceptance);
|
||||
}
|
||||
|
||||
private async queue(
|
||||
mode: 'steer' | 'follow-up',
|
||||
input: QueueMessageInput,
|
||||
): Promise<QueueAcceptance> {
|
||||
const key = `${input.conversationId}:${mode}:${input.clientRequestId}`;
|
||||
const prior = this.queueAcceptances.get(key);
|
||||
if (prior) return clone(prior);
|
||||
|
||||
const snapshot = this.snapshot(input.conversationId);
|
||||
const queuePosition = snapshot.queue.items.length + 1;
|
||||
this.emit(input.conversationId, {
|
||||
op: 'queue.replace',
|
||||
queue: {
|
||||
items: [
|
||||
...snapshot.queue.items,
|
||||
{
|
||||
id: this.id('queue'),
|
||||
clientRequestId: input.clientRequestId,
|
||||
mode,
|
||||
text: input.text,
|
||||
attachmentIds: input.attachments.map(({ attachmentId }) => attachmentId),
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
const acceptance: QueueAcceptance = {
|
||||
accepted: true,
|
||||
conversationId: input.conversationId,
|
||||
clientRequestId: input.clientRequestId,
|
||||
mode,
|
||||
queuePosition,
|
||||
};
|
||||
this.queueAcceptances.set(key, acceptance);
|
||||
return clone(acceptance);
|
||||
}
|
||||
|
||||
async steer(input: QueueMessageInput): Promise<QueueAcceptance> {
|
||||
return this.queue('steer', input);
|
||||
}
|
||||
|
||||
async followUp(input: QueueMessageInput): Promise<QueueAcceptance> {
|
||||
return this.queue('follow-up', input);
|
||||
}
|
||||
|
||||
async abort(conversationId: string): Promise<void> {
|
||||
const current = this.snapshot(conversationId).run;
|
||||
this.emit(conversationId, {
|
||||
op: 'run.state',
|
||||
run: {
|
||||
status: 'idle',
|
||||
...(current.runId ? { runId: current.runId } : {}),
|
||||
settledAt: this.now(),
|
||||
terminalReason: 'aborted',
|
||||
},
|
||||
}, current.runId);
|
||||
}
|
||||
|
||||
async setModel(input: SetConversationModelInput): Promise<ConversationModelState> {
|
||||
const snapshot = this.snapshot(input.conversationId);
|
||||
const thinkingLevel = snapshot.conversation.model.model?.thinkingLevel ?? 'off';
|
||||
const model: ConversationModelState = {
|
||||
model: {
|
||||
accountId: input.accountId,
|
||||
modelId: input.modelId,
|
||||
thinkingLevel,
|
||||
},
|
||||
modelResolution: 'resolved',
|
||||
};
|
||||
this.replaceSnapshot({
|
||||
...snapshot,
|
||||
conversation: { ...snapshot.conversation, model },
|
||||
});
|
||||
return clone(model);
|
||||
}
|
||||
|
||||
async setThinking(input: SetThinkingLevelInput): Promise<ConversationModelState> {
|
||||
const snapshot = this.snapshot(input.conversationId);
|
||||
if (!snapshot.conversation.model.model) {
|
||||
throw new CodingRuntimeContractError(
|
||||
'CODING_MIGRATION_MODEL_REQUIRED',
|
||||
'Conversation model must be selected first',
|
||||
true,
|
||||
);
|
||||
}
|
||||
const model: ConversationModelState = {
|
||||
model: {
|
||||
...snapshot.conversation.model.model,
|
||||
thinkingLevel: input.thinkingLevel,
|
||||
},
|
||||
modelResolution: 'resolved',
|
||||
};
|
||||
this.replaceSnapshot({
|
||||
...snapshot,
|
||||
conversation: { ...snapshot.conversation, model },
|
||||
});
|
||||
return clone(model);
|
||||
}
|
||||
|
||||
async compact(conversationId: string): Promise<void> {
|
||||
const before = this.snapshot(conversationId);
|
||||
const compactionId = this.id('compaction');
|
||||
const runId = before.run.runId ?? this.id('run');
|
||||
const willRetry = before.run.status !== 'idle';
|
||||
this.emit(conversationId, {
|
||||
op: 'compaction.upsert',
|
||||
node: {
|
||||
kind: 'compaction',
|
||||
id: compactionId,
|
||||
runId,
|
||||
source: 'manual',
|
||||
status: 'running',
|
||||
willRetry,
|
||||
},
|
||||
}, runId);
|
||||
this.emit(conversationId, {
|
||||
op: 'context.replace',
|
||||
context: {
|
||||
...before.context,
|
||||
compaction: 'running',
|
||||
lastCompactionId: compactionId,
|
||||
},
|
||||
}, runId);
|
||||
this.emit(conversationId, {
|
||||
op: 'compaction.upsert',
|
||||
node: {
|
||||
kind: 'compaction',
|
||||
id: compactionId,
|
||||
runId,
|
||||
source: 'manual',
|
||||
status: 'complete',
|
||||
willRetry,
|
||||
},
|
||||
}, runId);
|
||||
this.emit(conversationId, {
|
||||
op: 'context.replace',
|
||||
context: {
|
||||
...before.context,
|
||||
compaction: 'idle',
|
||||
lastCompactionId: compactionId,
|
||||
},
|
||||
}, runId);
|
||||
}
|
||||
|
||||
async fork(input: ForkConversationInput): Promise<ForkResult> {
|
||||
const source = this.snapshot(input.sourceConversationId);
|
||||
let nodes = source.nodes;
|
||||
if (input.sourceEntryId) {
|
||||
const index = source.nodes.findIndex(
|
||||
(node) => node.kind === 'message' && node.sourceEntryId === input.sourceEntryId,
|
||||
);
|
||||
if (index < 0) {
|
||||
throw new CodingRuntimeContractError(
|
||||
'CODING_SESSION_UNREADABLE',
|
||||
'Fork source entry is not on the active product path',
|
||||
true,
|
||||
);
|
||||
}
|
||||
nodes = source.nodes.slice(0, index + 1);
|
||||
}
|
||||
const snapshot: ConversationSnapshot = {
|
||||
...emptySnapshot(input.conversation),
|
||||
nodes: clone(nodes),
|
||||
context: clone(source.context),
|
||||
cursor: {
|
||||
workerGeneration: 0,
|
||||
seq: 0,
|
||||
...(input.sourceEntryId ? { leafEntryId: input.sourceEntryId } : {}),
|
||||
},
|
||||
worker: {
|
||||
status: 'stopped',
|
||||
generation: 0,
|
||||
},
|
||||
};
|
||||
this.replaceSnapshot(snapshot);
|
||||
return { conversationId: input.conversation.conversationId, snapshot: clone(snapshot) };
|
||||
}
|
||||
|
||||
async recover(conversationId: string): Promise<ConversationRuntimeState> {
|
||||
const snapshot = this.snapshot(conversationId);
|
||||
const generation = snapshot.cursor.workerGeneration + 1;
|
||||
this.replaceSnapshot({
|
||||
...snapshot,
|
||||
worker: { status: 'ready', generation },
|
||||
run: { status: 'idle' },
|
||||
pendingInteractions: [],
|
||||
cursor: {
|
||||
...snapshot.cursor,
|
||||
workerGeneration: generation,
|
||||
seq: 0,
|
||||
},
|
||||
});
|
||||
return this.runtimeState(conversationId);
|
||||
}
|
||||
|
||||
async dispose(conversationId: string): Promise<void> {
|
||||
const snapshot = this.snapshot(conversationId);
|
||||
this.replaceSnapshot({
|
||||
...snapshot,
|
||||
worker: {
|
||||
status: 'stopped',
|
||||
generation: snapshot.cursor.workerGeneration,
|
||||
},
|
||||
run: { status: 'idle' },
|
||||
pendingInteractions: [],
|
||||
});
|
||||
}
|
||||
|
||||
subscribe(listener: (patch: ConversationPatchEnvelope) => void): () => void {
|
||||
this.listeners.add(listener);
|
||||
return () => this.listeners.delete(listener);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user