brother7
0f7093d173
Persist the selected model before opening dormant sessions, and reconfigure crashed workers on the same account. Preserve session history and report model removal through the model-unavailable contract instead of a generic runtime failure.
794 lines
31 KiB
TypeScript
794 lines
31 KiB
TypeScript
import type {
|
|
CodingConversationPatchBatchEvent,
|
|
CodingConversationRuntime,
|
|
CodingRuntimeCommand,
|
|
CodingRuntimeDiagnostics,
|
|
ConversationInteraction,
|
|
ConversationInteractionResponse,
|
|
ConversationModelState,
|
|
ConversationPatchEnvelope,
|
|
ConversationSnapshot,
|
|
PrepareConversationInput,
|
|
ProductModelRef,
|
|
PromptAcceptance,
|
|
PromptMode,
|
|
} from './contracts';
|
|
import type { CodingConversationV2 } from '../coding-projects/conversation-store';
|
|
import { firstMessageTitle } from './conversation-title';
|
|
import {
|
|
CodingProjectService,
|
|
CodingProjectServiceError,
|
|
} from '../coding-projects/project-service';
|
|
|
|
const MAX_TITLE_LENGTH = 200;
|
|
const MAX_PROMPT_LENGTH = 100_000;
|
|
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(
|
|
readonly status: 400 | 404 | 409 | 500 | 503,
|
|
readonly code: string,
|
|
message: string,
|
|
) {
|
|
super(message);
|
|
this.name = 'CodingConversationServiceError';
|
|
}
|
|
}
|
|
|
|
export type CodingPromptAcceptance = PromptAcceptance;
|
|
|
|
interface AcceptanceRecord {
|
|
fingerprint: string;
|
|
flight: Promise<CodingPromptAcceptance>;
|
|
state: { value: 'protected' | 'settled' };
|
|
}
|
|
|
|
export interface CodingConversationServiceOptions {
|
|
archiveSession?(input: {
|
|
projectId: string;
|
|
sessionKey: string;
|
|
}): Promise<void>;
|
|
deliveryBatchWindowMs?: number;
|
|
deliveryBatchMaxItems?: number;
|
|
deliveryBatchMaxBytes?: number;
|
|
}
|
|
|
|
export type CodingConversationStreamEvent =
|
|
| { type: 'conversation.metadata-changed'; projectId: string; conversationId: string }
|
|
| {
|
|
type: 'snapshot';
|
|
conversationId: string;
|
|
workerGeneration: number;
|
|
seq: number;
|
|
snapshot: ConversationSnapshot;
|
|
}
|
|
| CodingConversationPatchBatchEvent;
|
|
|
|
export interface CodingConversationEventStream {
|
|
snapshots: ConversationSnapshot[];
|
|
events: AsyncIterable<CodingConversationStreamEvent>;
|
|
close(): void;
|
|
}
|
|
|
|
function requiredString(value: unknown, label: string, maximum: number): string {
|
|
const normalized = typeof value === 'string' ? value.trim() : '';
|
|
if (!normalized || normalized.length > maximum) {
|
|
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', `${label} is invalid`);
|
|
}
|
|
return normalized;
|
|
}
|
|
|
|
function runtimeError(error: unknown): never {
|
|
if (error instanceof CodingConversationServiceError || error instanceof CodingProjectServiceError) {
|
|
throw error;
|
|
}
|
|
const publicError = error && typeof error === 'object'
|
|
&& 'publicError' in error
|
|
&& error.publicError
|
|
&& typeof error.publicError === 'object'
|
|
? error.publicError as { code?: unknown; message?: unknown }
|
|
: null;
|
|
if (publicError && typeof publicError.code === 'string') {
|
|
const status = publicError.code === 'CODING_CONVERSATION_NOT_FOUND'
|
|
? 404
|
|
: publicError.code === 'CODING_STORAGE_WRITE_FAILED'
|
|
? 500
|
|
: publicError.code === 'CODING_RUNTIME_START_FAILED'
|
|
|| publicError.code === 'CODING_RUNTIME_READY_TIMEOUT'
|
|
|| publicError.code === 'CODING_RUNTIME_PROTOCOL_ERROR'
|
|
? 503
|
|
: 409;
|
|
throw new CodingConversationServiceError(
|
|
status,
|
|
publicError.code,
|
|
typeof publicError.message === 'string' ? publicError.message : 'Coding runtime request failed',
|
|
);
|
|
}
|
|
throw new CodingConversationServiceError(503, 'CODING_RUNTIME_UNAVAILABLE', 'The local coding runtime is unavailable');
|
|
}
|
|
|
|
async function persist<T>(operation: () => Promise<T>): Promise<T> {
|
|
try {
|
|
return await operation();
|
|
} catch (error) {
|
|
if (error instanceof CodingConversationServiceError || error instanceof CodingProjectServiceError) {
|
|
throw error;
|
|
}
|
|
throw new CodingConversationServiceError(
|
|
500,
|
|
'CODING_STORAGE_WRITE_FAILED',
|
|
'Coding data could not be persisted',
|
|
);
|
|
}
|
|
}
|
|
|
|
function publicConversation(conversation: CodingConversationV2): CodingConversationV2 {
|
|
const { piSessionId: _piSessionId, sessionKey: _sessionKey,
|
|
titleMode: _titleMode, autoTitleSet: _autoTitleSet, ...safe } = conversation;
|
|
return safe;
|
|
}
|
|
|
|
class PatchQueue implements AsyncIterable<CodingConversationStreamEvent> {
|
|
private readonly values: CodingConversationStreamEvent[] = [];
|
|
private readonly waiters: Array<(result: IteratorResult<CodingConversationStreamEvent>) => void> = [];
|
|
private closed = false;
|
|
|
|
push(value: CodingConversationStreamEvent): void {
|
|
if (this.closed) return;
|
|
const waiter = this.waiters.shift();
|
|
if (waiter) waiter({ value: structuredClone(value), done: false });
|
|
else this.values.push(structuredClone(value));
|
|
}
|
|
|
|
close(): void {
|
|
if (this.closed) return;
|
|
this.closed = true;
|
|
for (const waiter of this.waiters.splice(0)) waiter({ value: undefined, done: true });
|
|
}
|
|
|
|
[Symbol.asyncIterator](): AsyncIterator<CodingConversationStreamEvent> {
|
|
return {
|
|
next: async () => {
|
|
const value = this.values.shift();
|
|
if (value) return { value, done: false };
|
|
if (this.closed) return { value: undefined, done: true };
|
|
return await new Promise<IteratorResult<CodingConversationStreamEvent>>((resolve) => {
|
|
this.waiters.push(resolve);
|
|
});
|
|
},
|
|
return: async () => {
|
|
this.close();
|
|
return { value: undefined, done: true };
|
|
},
|
|
};
|
|
}
|
|
}
|
|
|
|
interface PendingPatchBatch {
|
|
conversationId: string;
|
|
workerGeneration: number;
|
|
items: CodingConversationPatchBatchEvent['items'];
|
|
byteLength: number;
|
|
timer: ReturnType<typeof setTimeout>;
|
|
}
|
|
|
|
class PatchBatchScheduler {
|
|
private readonly pending = new Map<string, PendingPatchBatch>();
|
|
private readonly windowMs: number;
|
|
private readonly maxItems: number;
|
|
private readonly maxBytes: number;
|
|
private closed = false;
|
|
|
|
constructor(
|
|
private readonly publish: (event: CodingConversationPatchBatchEvent) => void,
|
|
options: CodingConversationServiceOptions,
|
|
) {
|
|
this.windowMs = Math.min(33, Math.max(16, options.deliveryBatchWindowMs ?? 24));
|
|
this.maxItems = Math.max(1, options.deliveryBatchMaxItems ?? 128);
|
|
this.maxBytes = Math.max(1, options.deliveryBatchMaxBytes ?? 256 * 1024);
|
|
}
|
|
|
|
push(envelope: ConversationPatchEnvelope): void {
|
|
if (this.closed) return;
|
|
const key = this.key(envelope.conversationId, envelope.workerGeneration);
|
|
const item = {
|
|
...(envelope.runId ? { runId: envelope.runId } : {}),
|
|
seq: envelope.seq,
|
|
at: envelope.at,
|
|
patch: envelope.patch,
|
|
};
|
|
const itemBytes = Buffer.byteLength(JSON.stringify(item), 'utf8');
|
|
let batch = this.pending.get(key);
|
|
if (batch && batch.items.length > 0
|
|
&& (batch.items.length >= this.maxItems || batch.byteLength + itemBytes > this.maxBytes)) {
|
|
this.flush(key);
|
|
batch = undefined;
|
|
}
|
|
if (!batch) {
|
|
batch = {
|
|
conversationId: envelope.conversationId,
|
|
workerGeneration: envelope.workerGeneration,
|
|
items: [],
|
|
byteLength: 0,
|
|
timer: setTimeout(() => this.flush(key), this.windowMs),
|
|
};
|
|
this.pending.set(key, batch);
|
|
}
|
|
batch.items.push(item);
|
|
batch.byteLength += itemBytes;
|
|
if (batch.items.length >= this.maxItems || batch.byteLength >= this.maxBytes) {
|
|
this.flush(key);
|
|
}
|
|
}
|
|
|
|
cancel(conversationId: string): void {
|
|
for (const [key, batch] of this.pending) {
|
|
if (batch.conversationId !== conversationId) continue;
|
|
clearTimeout(batch.timer);
|
|
this.pending.delete(key);
|
|
}
|
|
}
|
|
|
|
close(): void {
|
|
this.closed = true;
|
|
for (const batch of this.pending.values()) clearTimeout(batch.timer);
|
|
this.pending.clear();
|
|
}
|
|
|
|
private key(conversationId: string, workerGeneration: number): string {
|
|
return `${conversationId}\u0000${workerGeneration}`;
|
|
}
|
|
|
|
private flush(key: string): void {
|
|
const batch = this.pending.get(key);
|
|
if (!batch || this.closed) return;
|
|
this.pending.delete(key);
|
|
clearTimeout(batch.timer);
|
|
const first = batch.items[0];
|
|
const last = batch.items.at(-1);
|
|
if (!first || !last) return;
|
|
this.publish({
|
|
type: 'patch-batch',
|
|
conversationId: batch.conversationId,
|
|
workerGeneration: batch.workerGeneration,
|
|
fromSeq: first.seq,
|
|
toSeq: last.seq,
|
|
items: batch.items,
|
|
});
|
|
}
|
|
}
|
|
|
|
export class CodingConversationService {
|
|
private readonly prepareFlights = new Map<string, Promise<PrepareConversationInput>>();
|
|
private readonly acceptances = new Map<string, AcceptanceRecord>();
|
|
|
|
constructor(
|
|
readonly projects: CodingProjectService,
|
|
readonly runtime: CodingConversationRuntime,
|
|
private readonly options: CodingConversationServiceOptions = {},
|
|
) {
|
|
this.unsubscribeTitles = runtime.subscribe((event) => {
|
|
if (event.patch.op !== 'message.upsert') return;
|
|
const title = firstMessageTitle(event.patch.node);
|
|
if (!title) return;
|
|
const context = this.titleContexts.get(event.conversationId);
|
|
if (!context) return;
|
|
void this.setFirstMessageTitle(context, event.conversationId, title).catch(() => undefined);
|
|
});
|
|
}
|
|
|
|
private readonly titleContexts = new Map<string, { id: string; path: string }>();
|
|
private readonly metadataListeners = new Set<(event: CodingConversationStreamEvent) => void>();
|
|
private readonly unsubscribeTitles: () => void;
|
|
|
|
private async setFirstMessageTitle(project: { id: string; path: string }, id: string, title: string): Promise<void> {
|
|
if (await this.projects.conversationStore(project.path).setFirstMessageTitle(id, title)) {
|
|
this.metadataChanged(project.id, id);
|
|
}
|
|
this.titleContexts.delete(id);
|
|
}
|
|
|
|
dispose(): void {
|
|
this.unsubscribeTitles();
|
|
this.titleContexts.clear();
|
|
this.metadataListeners.clear();
|
|
}
|
|
|
|
private metadataChanged(projectId: string, conversationId: string): void {
|
|
for (const listener of this.metadataListeners) {
|
|
listener({ type: 'conversation.metadata-changed', projectId, conversationId });
|
|
}
|
|
}
|
|
|
|
private async assertUnarchived(conversationId: string): Promise<void> {
|
|
const { conversation } = await this.projects.findActiveConversation(conversationId);
|
|
if (conversation.archivedAt) {
|
|
throw new CodingConversationServiceError(409, 'CODING_CONVERSATION_ARCHIVED', '请先恢复已归档的对话，再继续操作。');
|
|
}
|
|
}
|
|
|
|
async listConversations(projectId?: string): Promise<CodingConversationV2[]> {
|
|
const project = projectId
|
|
? await this.projects.getProject(projectId)
|
|
: await this.projects.requireActiveProject();
|
|
return (await this.projects.conversationStore(project.path).read()).conversations.map(publicConversation);
|
|
}
|
|
|
|
async createConversation(input: {
|
|
projectId?: string;
|
|
agentId: unknown;
|
|
title: unknown;
|
|
}): Promise<CodingConversationV2> {
|
|
const project = input.projectId
|
|
? await this.projects.getProject(input.projectId)
|
|
: await this.projects.requireActiveProject();
|
|
const config = await this.projects.getConfig(project.id);
|
|
const agentId = requiredString(input.agentId, 'Agent id', 64);
|
|
const agent = config.config.agents.find((candidate) => (
|
|
candidate.id === agentId && candidate.enabled && !candidate.archivedAt
|
|
));
|
|
if (!agent) {
|
|
throw new CodingConversationServiceError(404, 'CODING_AGENT_NOT_FOUND', 'Coding project Agent does not exist');
|
|
}
|
|
const created = await persist(() => this.projects.conversationStore(project.path).create({
|
|
agentId,
|
|
title: requiredString(input.title, 'Conversation title', MAX_TITLE_LENGTH),
|
|
titleMode: input.title === '新对话' ? 'automatic' : 'manual',
|
|
model: agent.model,
|
|
modelResolution: agent.modelResolution,
|
|
}));
|
|
return publicConversation(created);
|
|
}
|
|
|
|
async getConversation(conversationId: string): Promise<CodingConversationV2> {
|
|
const { conversation } = await this.projects.findActiveConversation(conversationId);
|
|
return publicConversation(conversation);
|
|
}
|
|
|
|
async patchConversation(conversationId: string, patch: {
|
|
title?: unknown;
|
|
archived?: unknown;
|
|
unread?: unknown;
|
|
}): Promise<CodingConversationV2> {
|
|
const { project } = await this.projects.findActiveConversation(conversationId);
|
|
if (patch.title === undefined && patch.archived === undefined && patch.unread === undefined) {
|
|
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', 'Conversation patch is empty');
|
|
}
|
|
if (patch.archived !== undefined && typeof patch.archived !== 'boolean') {
|
|
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', 'Conversation archive state is invalid');
|
|
}
|
|
if (patch.unread !== undefined && typeof patch.unread !== 'boolean') {
|
|
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', 'Conversation unread state is invalid');
|
|
}
|
|
const updated = await persist(() => this.projects.conversationStore(project.path).patchMetadata(conversationId, {
|
|
...(patch.title !== undefined
|
|
? { title: requiredString(patch.title, 'Conversation title', MAX_TITLE_LENGTH) }
|
|
: {}),
|
|
...(typeof patch.archived === 'boolean'
|
|
? { archivedAt: patch.archived ? new Date().toISOString() : null }
|
|
: {}),
|
|
...(typeof patch.unread === 'boolean' ? { unread: patch.unread } : {}),
|
|
}));
|
|
this.metadataChanged(project.id, conversationId);
|
|
return publicConversation(updated);
|
|
}
|
|
|
|
async deleteConversation(conversationId: string): Promise<void> {
|
|
const { project, conversation } = await this.projects.findActiveConversation(conversationId);
|
|
try {
|
|
await this.runtime.dispose(conversationId, 'conversation_deleted');
|
|
} catch (error) {
|
|
runtimeError(error);
|
|
}
|
|
await this.archiveSession(project.id, conversation.sessionKey);
|
|
await persist(() => this.projects.conversationStore(project.path).delete(conversationId));
|
|
this.titleContexts.delete(conversationId);
|
|
this.prepareFlights.delete(conversationId);
|
|
for (const key of [...this.acceptances.keys()]) {
|
|
if (key.startsWith(`${conversationId}\u0000`)) this.acceptances.delete(key);
|
|
}
|
|
}
|
|
|
|
async getSnapshot(conversationId: string): Promise<ConversationSnapshot> {
|
|
await this.ensurePrepared(conversationId);
|
|
try {
|
|
return await this.runtime.getSnapshot(conversationId);
|
|
} catch (error) {
|
|
runtimeError(error);
|
|
}
|
|
}
|
|
|
|
async acceptPrompt(input: {
|
|
conversationId: string;
|
|
clientRequestId: unknown;
|
|
mode: unknown;
|
|
text: unknown;
|
|
attachments?: unknown;
|
|
}): Promise<CodingPromptAcceptance> {
|
|
const conversationId = requiredString(input.conversationId, 'Conversation id', 128);
|
|
const clientRequestId = typeof input.clientRequestId === 'string' ? input.clientRequestId : '';
|
|
if (!REQUEST_ID_PATTERN.test(clientRequestId)) {
|
|
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', 'Client request id is invalid');
|
|
}
|
|
if (!['prompt', 'steer', 'follow-up'].includes(String(input.mode))) {
|
|
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', 'Prompt mode is invalid');
|
|
}
|
|
if (typeof input.text !== 'string' || input.text.length > MAX_PROMPT_LENGTH) {
|
|
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', 'Prompt text is invalid');
|
|
}
|
|
const attachments = Array.isArray(input.attachments) ? input.attachments : [];
|
|
if (attachments.length > MAX_ATTACHMENTS || attachments.some((item) => (
|
|
!item || typeof item !== 'object' || Array.isArray(item)
|
|
|| !ATTACHMENT_ID_PATTERN.test(String((item as { attachmentId?: unknown }).attachmentId ?? ''))
|
|
))) {
|
|
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', 'Prompt attachments are invalid');
|
|
}
|
|
if (!input.text.trim() && attachments.length === 0) {
|
|
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', 'Prompt content is empty');
|
|
}
|
|
const mode = input.mode as PromptMode;
|
|
const normalizedAttachments = attachments.map((item) => ({
|
|
attachmentId: String((item as { attachmentId: unknown }).attachmentId),
|
|
}));
|
|
const fingerprint = JSON.stringify({ mode, text: input.text, attachments: normalizedAttachments });
|
|
const key = `${conversationId}\u0000${clientRequestId}`;
|
|
const prior = this.acceptances.get(key);
|
|
if (prior) {
|
|
if (prior.fingerprint !== fingerprint) {
|
|
throw new CodingConversationServiceError(409, 'CODING_REQUEST_ID_CONFLICT', 'Client request id was already used');
|
|
}
|
|
if (prior.state.value === 'settled') {
|
|
this.acceptances.delete(key);
|
|
this.acceptances.set(key, prior);
|
|
}
|
|
return await prior.flight;
|
|
}
|
|
this.reserveAcceptanceSlot();
|
|
const acceptanceState: AcceptanceRecord['state'] = { value: 'protected' };
|
|
const flight = (async (): Promise<CodingPromptAcceptance> => {
|
|
try {
|
|
await this.assertUnarchived(conversationId);
|
|
await this.ensurePrepared(conversationId);
|
|
const acceptance = await this.runtime.prompt({
|
|
conversationId,
|
|
clientRequestId,
|
|
mode,
|
|
text: input.text as string,
|
|
attachments: normalizedAttachments,
|
|
});
|
|
acceptanceState.value = 'settled';
|
|
return acceptance;
|
|
} catch (error) {
|
|
const publicCode = error && typeof error === 'object'
|
|
&& 'publicError' in error
|
|
&& error.publicError
|
|
&& typeof error.publicError === 'object'
|
|
&& 'code' in error.publicError
|
|
? String(error.publicError.code)
|
|
: '';
|
|
if (publicCode !== 'CODING_REQUEST_UNCERTAIN') this.acceptances.delete(key);
|
|
runtimeError(error);
|
|
}
|
|
})();
|
|
this.acceptances.set(key, { fingerprint, flight, state: acceptanceState });
|
|
return await flight;
|
|
}
|
|
|
|
async abort(conversationId: string): Promise<void> {
|
|
await this.ensurePrepared(conversationId);
|
|
try { await this.runtime.abort(conversationId); } catch (error) { runtimeError(error); }
|
|
}
|
|
|
|
async setModel(conversationId: string, model: ProductModelRef): Promise<ConversationModelState> {
|
|
const { project, conversation } = await this.projects.findActiveConversation(conversationId);
|
|
let selected: ProductModelRef;
|
|
try {
|
|
selected = await this.runtime.validateModel(model);
|
|
} catch (error) { runtimeError(error); }
|
|
// A dormant Conversation must not need its previous model to start in order
|
|
// to select a replacement (the previous model may have been removed).
|
|
await this.prepareFlights.get(conversationId)?.catch(() => undefined);
|
|
const worker = this.runtime.getDiagnostics().workers.find((item) => item.conversationId === conversationId);
|
|
if (conversation.modelResolution === 'resolved' && conversation.model && worker) {
|
|
await this.ensurePrepared(conversationId);
|
|
let snapshot: ConversationSnapshot;
|
|
try {
|
|
snapshot = await this.runtime.getSnapshot(conversationId);
|
|
} catch (error) { runtimeError(error); }
|
|
this.assertSnapshotAllowsMutation(snapshot);
|
|
let state: ConversationModelState;
|
|
try {
|
|
state = await this.runtime.setModel({
|
|
conversationId,
|
|
accountId: selected.accountId,
|
|
modelId: selected.modelId,
|
|
});
|
|
} catch (error) { runtimeError(error); }
|
|
await persist(() => this.projects.conversationStore(project.path).setModelState(conversationId, state));
|
|
return state;
|
|
}
|
|
const state: ConversationModelState = {
|
|
model: selected,
|
|
modelResolution: 'resolved',
|
|
};
|
|
await persist(() => this.projects.conversationStore(project.path).setModelState(conversationId, state));
|
|
const preparing = this.prepareFlights.get(conversationId);
|
|
if (preparing) await preparing.catch(() => undefined);
|
|
try {
|
|
await this.runtime.dispose(conversationId, 'model_reconfiguration');
|
|
this.prepareFlights.delete(conversationId);
|
|
await this.ensurePrepared(conversationId);
|
|
return state;
|
|
} catch (error) { runtimeError(error); }
|
|
}
|
|
|
|
async setThinking(conversationId: string, thinkingLevel: ProductModelRef['thinkingLevel'], reasoningChoice?: ProductModelRef['reasoningChoice']): Promise<ConversationModelState> {
|
|
await this.ensurePrepared(conversationId);
|
|
const { project } = await this.projects.findActiveConversation(conversationId);
|
|
try {
|
|
const state = await this.runtime.setThinking({ conversationId, thinkingLevel, reasoningChoice });
|
|
await persist(() => this.projects.conversationStore(project.path).setModelState(conversationId, state));
|
|
return state;
|
|
} catch (error) { runtimeError(error); }
|
|
}
|
|
|
|
async compact(conversationId: string): Promise<void> {
|
|
await this.ensurePrepared(conversationId);
|
|
try { await this.runtime.compact(conversationId); } catch (error) { runtimeError(error); }
|
|
}
|
|
|
|
async recover(conversationId: string): Promise<void> {
|
|
await this.ensurePrepared(conversationId);
|
|
try { await this.runtime.recover(conversationId); } catch (error) { runtimeError(error); }
|
|
}
|
|
|
|
async fork(sourceConversationId: string, sourceEntryId?: string): Promise<CodingConversationV2> {
|
|
await this.assertUnarchived(sourceConversationId);
|
|
const entryId = requiredString(sourceEntryId, 'Fork source entry id', 256);
|
|
const prepared = await this.ensurePrepared(sourceConversationId);
|
|
let sourceSnapshot: ConversationSnapshot;
|
|
try {
|
|
sourceSnapshot = await this.runtime.getSnapshot(sourceConversationId);
|
|
} catch (error) {
|
|
runtimeError(error);
|
|
}
|
|
this.assertSnapshotAllowsMutation(sourceSnapshot);
|
|
const sourceNode = sourceSnapshot.nodes.find((node) => (
|
|
node.kind === 'message'
|
|
&& node.sourceEntryId === entryId
|
|
&& node.role === 'user'
|
|
&& node.status !== 'optimistic'
|
|
));
|
|
if (!sourceNode) {
|
|
throw new CodingConversationServiceError(
|
|
400,
|
|
'CODING_CONVERSATION_REQUEST_INVALID',
|
|
'Fork source must be a durable user message on the active Conversation path',
|
|
);
|
|
}
|
|
const source = await this.projects.findActiveConversation(sourceConversationId);
|
|
const store = this.projects.conversationStore(source.project.path);
|
|
const created = await persist(() => store.create({
|
|
agentId: source.conversation.agentId,
|
|
title: `${source.conversation.title} (fork)`,
|
|
titleMode: 'manual',
|
|
model: source.conversation.model,
|
|
modelResolution: source.conversation.modelResolution,
|
|
}));
|
|
try {
|
|
await this.runtime.fork({
|
|
sourceConversationId,
|
|
sourceEntryId: entryId,
|
|
conversation: {
|
|
...prepared,
|
|
conversationId: created.id,
|
|
title: created.title,
|
|
model: {
|
|
model: created.model,
|
|
modelResolution: created.modelResolution,
|
|
},
|
|
},
|
|
});
|
|
return publicConversation(created);
|
|
} catch (error) {
|
|
try {
|
|
await this.runtime.dispose(created.id, 'fork_replacement');
|
|
const latest = await store.get(created.id);
|
|
await this.archiveSession(source.project.id, latest?.sessionKey);
|
|
await persist(() => store.delete(created.id));
|
|
} catch {
|
|
// Keep product metadata when cleanup cannot complete so the partial
|
|
// runtime/session remains owned and can be diagnosed or deleted later.
|
|
}
|
|
runtimeError(error);
|
|
}
|
|
}
|
|
|
|
async listInteractions(conversationId?: string): Promise<ConversationInteraction[]> {
|
|
if (conversationId) await this.projects.findActiveConversation(conversationId);
|
|
try { return await this.runtime.listInteractions(conversationId); } catch (error) { runtimeError(error); }
|
|
}
|
|
|
|
async respondInteraction(conversationId: string, response: ConversationInteractionResponse): Promise<void> {
|
|
const id = requiredString(conversationId, 'Conversation id', 128);
|
|
if ('optionId' in response && (response.optionId.length === 0 || response.optionId.length > 256)) {
|
|
throw new CodingConversationServiceError(400, 'CODING_INTERACTION_REQUEST_INVALID', 'Interaction option is invalid');
|
|
}
|
|
if ('value' in response && response.value.length > MAX_PROMPT_LENGTH) {
|
|
throw new CodingConversationServiceError(400, 'CODING_INTERACTION_REQUEST_INVALID', 'Interaction value is invalid');
|
|
}
|
|
await this.projects.findActiveConversation(id);
|
|
try { await this.runtime.respondInteraction(id, response); } catch (error) { runtimeError(error); }
|
|
}
|
|
|
|
async listLiveCommands(conversationId: string): Promise<CodingRuntimeCommand[]> {
|
|
await this.projects.findActiveConversation(conversationId);
|
|
try { return await this.runtime.listCommands(conversationId); } catch { return []; }
|
|
}
|
|
|
|
getDiagnostics(): CodingRuntimeDiagnostics {
|
|
return this.runtime.getDiagnostics();
|
|
}
|
|
|
|
markProviderStale(): void { this.runtime.markProviderStale(); }
|
|
markResourcesStale(): void { this.runtime.markResourcesStale(); }
|
|
|
|
async openEventStream(conversationId?: string): Promise<CodingConversationEventStream> {
|
|
const id = conversationId === undefined
|
|
? undefined
|
|
: requiredString(conversationId, 'Conversation id', 128);
|
|
if (id) await this.projects.findActiveConversation(id);
|
|
const queue = new PatchQueue();
|
|
const metadataListener = (event: CodingConversationStreamEvent) => {
|
|
if (!id || event.conversationId === id) queue.push(event);
|
|
};
|
|
this.metadataListeners.add(metadataListener);
|
|
const batches = new PatchBatchScheduler((event) => queue.push(event), this.options);
|
|
const cursors = new Map<string, ConversationSnapshot['cursor']>();
|
|
let initialize!: () => void;
|
|
const initialized = new Promise<void>((resolve) => { initialize = resolve; });
|
|
let eventTail = Promise.resolve();
|
|
const unsubscribe = this.runtime.subscribe((event) => {
|
|
if (id && event.conversationId !== id) return;
|
|
eventTail = eventTail.then(async () => {
|
|
await initialized;
|
|
const cursor = cursors.get(event.conversationId);
|
|
if (cursor && event.workerGeneration === cursor.workerGeneration && event.seq <= cursor.seq) return;
|
|
if (!cursor
|
|
|| event.workerGeneration !== cursor.workerGeneration
|
|
|| event.seq !== cursor.seq + 1) {
|
|
batches.cancel(event.conversationId);
|
|
try {
|
|
const snapshot = await this.runtime.getSnapshot(event.conversationId);
|
|
cursors.set(event.conversationId, snapshot.cursor);
|
|
queue.push({
|
|
type: 'snapshot',
|
|
conversationId: event.conversationId,
|
|
workerGeneration: snapshot.cursor.workerGeneration,
|
|
seq: snapshot.cursor.seq,
|
|
snapshot,
|
|
});
|
|
} catch {
|
|
// A disposed Conversation has no reconnect state to publish.
|
|
}
|
|
return;
|
|
}
|
|
cursors.set(event.conversationId, {
|
|
workerGeneration: event.workerGeneration,
|
|
seq: event.seq,
|
|
});
|
|
batches.push(event);
|
|
}).catch(() => undefined);
|
|
});
|
|
try {
|
|
const snapshots = id
|
|
? [await this.getSnapshot(id)]
|
|
: (await Promise.all(this.runtime.getDiagnostics().workers.map(async ({ conversationId: target }) => {
|
|
try { return await this.runtime.getSnapshot(target); } catch { return null; }
|
|
}))).filter((snapshot): snapshot is ConversationSnapshot => snapshot !== null);
|
|
for (const snapshot of snapshots) {
|
|
cursors.set(snapshot.conversation.id, snapshot.cursor);
|
|
}
|
|
initialize();
|
|
return {
|
|
snapshots,
|
|
events: queue,
|
|
close: () => {
|
|
this.metadataListeners.delete(metadataListener);
|
|
unsubscribe();
|
|
batches.close();
|
|
queue.close();
|
|
},
|
|
};
|
|
} catch (error) {
|
|
initialize();
|
|
this.metadataListeners.delete(metadataListener);
|
|
unsubscribe();
|
|
batches.close();
|
|
queue.close();
|
|
throw error;
|
|
}
|
|
}
|
|
|
|
private ensurePrepared(conversationId: string): Promise<PrepareConversationInput> {
|
|
const id = conversationId.trim();
|
|
const prior = this.prepareFlights.get(id);
|
|
if (prior) return prior;
|
|
const flight = (async () => {
|
|
const { project, conversation } = await this.projects.findActiveConversation(id);
|
|
if (conversation.titleMode === 'automatic' && !conversation.autoTitleSet) {
|
|
this.titleContexts.set(id, { id: project.id, path: project.path });
|
|
}
|
|
const config = await this.projects.getConfig(project.id);
|
|
const agent = config.config.agents.find((candidate) => (
|
|
candidate.id === conversation.agentId && candidate.enabled && !candidate.archivedAt
|
|
));
|
|
if (!agent) {
|
|
throw new CodingConversationServiceError(409, 'CODING_AGENT_NOT_FOUND', 'Conversation Agent is unavailable');
|
|
}
|
|
const prepared: PrepareConversationInput = {
|
|
conversationId: conversation.id,
|
|
projectId: project.id,
|
|
agentId: conversation.agentId,
|
|
title: conversation.title,
|
|
model: {
|
|
model: conversation.model,
|
|
modelResolution: conversation.modelResolution,
|
|
},
|
|
};
|
|
try { await this.runtime.prepare(prepared); } catch (error) { runtimeError(error); }
|
|
if (this.titleContexts.has(id)) {
|
|
const snapshot = await this.runtime.getSnapshot(id);
|
|
const title = snapshot.nodes.map(firstMessageTitle).find((candidate) => candidate !== null);
|
|
if (title) await this.setFirstMessageTitle(project, id, title).catch(() => undefined);
|
|
}
|
|
return prepared;
|
|
})().finally(() => {
|
|
if (this.prepareFlights.get(id) === flight) this.prepareFlights.delete(id);
|
|
});
|
|
this.prepareFlights.set(id, flight);
|
|
return flight;
|
|
}
|
|
|
|
private reserveAcceptanceSlot(): void {
|
|
while (this.acceptances.size >= MAX_ACCEPTANCES) {
|
|
const settled = [...this.acceptances.entries()]
|
|
.find(([, record]) => record.state.value === 'settled');
|
|
if (!settled) {
|
|
throw new CodingConversationServiceError(
|
|
503,
|
|
'CODING_REQUEST_CAPACITY_EXCEEDED',
|
|
'The local request registry is full',
|
|
);
|
|
}
|
|
this.acceptances.delete(settled[0]);
|
|
}
|
|
}
|
|
|
|
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<void> {
|
|
if (!sessionKey) return;
|
|
if (!this.options.archiveSession) {
|
|
throw new CodingConversationServiceError(
|
|
500,
|
|
'CODING_STORAGE_WRITE_FAILED',
|
|
'Coding session archival is unavailable',
|
|
);
|
|
}
|
|
await persist(() => this.options.archiveSession!({ projectId, sessionKey }));
|
|
}
|
|
}
|