fix(coding): close PI-100 review findings
This commit is contained in:
@@ -415,6 +415,7 @@ export interface CodingConversationRuntime {
|
||||
steer(input: QueueMessageInput): Promise<QueueAcceptance>;
|
||||
followUp(input: QueueMessageInput): Promise<QueueAcceptance>;
|
||||
abort(conversationId: string): Promise<void>;
|
||||
validateModel(model: ProductModelRef): Promise<ProductModelRef>;
|
||||
setModel(input: SetConversationModelInput): Promise<ConversationModelState>;
|
||||
setThinking(input: SetThinkingLevelInput): Promise<ConversationModelState>;
|
||||
compact(conversationId: string): Promise<void>;
|
||||
|
||||
@@ -27,7 +27,7 @@ const ATTACHMENT_ID_PATTERN = /^[A-Za-z0-9-]{1,64}$/;
|
||||
|
||||
export class CodingConversationServiceError extends Error {
|
||||
constructor(
|
||||
readonly status: 400 | 404 | 409 | 503,
|
||||
readonly status: 400 | 404 | 409 | 500 | 503,
|
||||
readonly code: string,
|
||||
message: string,
|
||||
) {
|
||||
@@ -41,6 +41,14 @@ 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>;
|
||||
}
|
||||
|
||||
export type CodingConversationStreamEvent =
|
||||
@@ -78,8 +86,17 @@ function runtimeError(error: unknown): never {
|
||||
? 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(
|
||||
publicError.code === 'CODING_CONVERSATION_NOT_FOUND' ? 404 : 409,
|
||||
status,
|
||||
publicError.code,
|
||||
typeof publicError.message === 'string' ? publicError.message : 'Coding runtime request failed',
|
||||
);
|
||||
@@ -87,6 +104,21 @@ function runtimeError(error: unknown): never {
|
||||
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, ...safe } = conversation;
|
||||
return safe;
|
||||
@@ -135,6 +167,7 @@ export class CodingConversationService {
|
||||
constructor(
|
||||
readonly projects: CodingProjectService,
|
||||
readonly runtime: CodingConversationRuntime,
|
||||
private readonly options: CodingConversationServiceOptions = {},
|
||||
) {}
|
||||
|
||||
async listConversations(projectId?: string): Promise<CodingConversationV2[]> {
|
||||
@@ -160,12 +193,12 @@ export class CodingConversationService {
|
||||
if (!agent) {
|
||||
throw new CodingConversationServiceError(404, 'CODING_AGENT_NOT_FOUND', 'Coding project Agent does not exist');
|
||||
}
|
||||
const created = await this.projects.conversationStore(project.path).create({
|
||||
const created = await persist(() => this.projects.conversationStore(project.path).create({
|
||||
agentId,
|
||||
title: requiredString(input.title, 'Conversation title', MAX_TITLE_LENGTH),
|
||||
model: agent.model,
|
||||
modelResolution: agent.modelResolution,
|
||||
});
|
||||
}));
|
||||
return publicConversation(created);
|
||||
}
|
||||
|
||||
@@ -189,7 +222,7 @@ export class CodingConversationService {
|
||||
if (patch.unread !== undefined && typeof patch.unread !== 'boolean') {
|
||||
throw new CodingConversationServiceError(400, 'CODING_CONVERSATION_REQUEST_INVALID', 'Conversation unread state is invalid');
|
||||
}
|
||||
const updated = await this.projects.conversationStore(project.path).patchMetadata(conversationId, {
|
||||
const updated = await persist(() => this.projects.conversationStore(project.path).patchMetadata(conversationId, {
|
||||
...(patch.title !== undefined
|
||||
? { title: requiredString(patch.title, 'Conversation title', MAX_TITLE_LENGTH) }
|
||||
: {}),
|
||||
@@ -197,18 +230,19 @@ export class CodingConversationService {
|
||||
? { archivedAt: patch.archived ? new Date().toISOString() : null }
|
||||
: {}),
|
||||
...(typeof patch.unread === 'boolean' ? { unread: patch.unread } : {}),
|
||||
});
|
||||
}));
|
||||
return publicConversation(updated);
|
||||
}
|
||||
|
||||
async deleteConversation(conversationId: string): Promise<void> {
|
||||
const { project } = await this.projects.findActiveConversation(conversationId);
|
||||
const { project, conversation } = await this.projects.findActiveConversation(conversationId);
|
||||
try {
|
||||
await this.runtime.dispose(conversationId);
|
||||
} catch (error) {
|
||||
runtimeError(error);
|
||||
}
|
||||
await this.projects.conversationStore(project.path).delete(conversationId);
|
||||
await this.archiveSession(project.id, conversation.sessionKey);
|
||||
await persist(() => this.projects.conversationStore(project.path).delete(conversationId));
|
||||
this.prepareFlights.delete(conversationId);
|
||||
for (const key of [...this.acceptances.keys()]) {
|
||||
if (key.startsWith(`${conversationId}\u0000`)) this.acceptances.delete(key);
|
||||
@@ -263,18 +297,26 @@ export class CodingConversationService {
|
||||
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.ensurePrepared(conversationId);
|
||||
return await this.runtime.prompt({
|
||||
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
|
||||
@@ -287,8 +329,7 @@ export class CodingConversationService {
|
||||
runtimeError(error);
|
||||
}
|
||||
})();
|
||||
this.acceptances.set(key, { fingerprint, flight });
|
||||
this.trimAcceptances();
|
||||
this.acceptances.set(key, { fingerprint, flight, state: acceptanceState });
|
||||
return await flight;
|
||||
}
|
||||
|
||||
@@ -298,20 +339,23 @@ export class CodingConversationService {
|
||||
}
|
||||
|
||||
async setModel(conversationId: string, model: ProductModelRef): Promise<ConversationModelState> {
|
||||
await this.ensurePrepared(conversationId);
|
||||
const { project } = await this.projects.findActiveConversation(conversationId);
|
||||
let selected: ProductModelRef;
|
||||
try {
|
||||
const state = await this.runtime.setModel({
|
||||
conversationId,
|
||||
accountId: model.accountId,
|
||||
modelId: model.modelId,
|
||||
});
|
||||
const withThinking = state.model
|
||||
? { model: { ...state.model, thinkingLevel: model.thinkingLevel }, modelResolution: 'resolved' as const }
|
||||
: state;
|
||||
if (withThinking.model) await this.runtime.setThinking({ conversationId, thinkingLevel: model.thinkingLevel });
|
||||
await this.projects.conversationStore(project.path).setModelState(conversationId, withThinking);
|
||||
return withThinking;
|
||||
selected = await this.runtime.validateModel(model);
|
||||
} catch (error) { runtimeError(error); }
|
||||
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);
|
||||
this.prepareFlights.delete(conversationId);
|
||||
await this.ensurePrepared(conversationId);
|
||||
return state;
|
||||
} catch (error) { runtimeError(error); }
|
||||
}
|
||||
|
||||
@@ -338,12 +382,13 @@ export class CodingConversationService {
|
||||
async fork(sourceConversationId: string, sourceEntryId?: string): Promise<CodingConversationV2> {
|
||||
const prepared = await this.ensurePrepared(sourceConversationId);
|
||||
const source = await this.projects.findActiveConversation(sourceConversationId);
|
||||
const created = await this.projects.conversationStore(source.project.path).create({
|
||||
const store = this.projects.conversationStore(source.project.path);
|
||||
const created = await persist(() => store.create({
|
||||
agentId: source.conversation.agentId,
|
||||
title: `${source.conversation.title} (fork)`,
|
||||
model: source.conversation.model,
|
||||
modelResolution: source.conversation.modelResolution,
|
||||
});
|
||||
}));
|
||||
try {
|
||||
await this.runtime.fork({
|
||||
sourceConversationId,
|
||||
@@ -360,7 +405,15 @@ export class CodingConversationService {
|
||||
});
|
||||
return publicConversation(created);
|
||||
} catch (error) {
|
||||
await this.projects.conversationStore(source.project.path).delete(created.id).catch(() => undefined);
|
||||
try {
|
||||
await this.runtime.dispose(created.id);
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -493,11 +546,30 @@ export class CodingConversationService {
|
||||
return flight;
|
||||
}
|
||||
|
||||
private trimAcceptances(): void {
|
||||
while (this.acceptances.size > MAX_ACCEPTANCES) {
|
||||
const key = this.acceptances.keys().next().value as string | undefined;
|
||||
if (!key) return;
|
||||
this.acceptances.delete(key);
|
||||
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 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 }));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -14,6 +14,7 @@ import type {
|
||||
ForkConversationInput,
|
||||
ForkResult,
|
||||
PrepareConversationInput,
|
||||
ProductModelRef,
|
||||
PromptAcceptance,
|
||||
PromptConversationInput,
|
||||
QueueAcceptance,
|
||||
@@ -316,6 +317,17 @@ export class InMemoryConversationRuntime implements CodingConversationRuntime {
|
||||
}, current.runId);
|
||||
}
|
||||
|
||||
async validateModel(model: ProductModelRef): Promise<ProductModelRef> {
|
||||
if (!model.accountId.trim() || !model.modelId.trim()) {
|
||||
throw new CodingRuntimeContractError(
|
||||
'CODING_MODEL_UNAVAILABLE',
|
||||
'The selected model is unavailable',
|
||||
true,
|
||||
);
|
||||
}
|
||||
return clone(model);
|
||||
}
|
||||
|
||||
async setModel(input: SetConversationModelInput): Promise<ConversationModelState> {
|
||||
const snapshot = this.snapshot(input.conversationId);
|
||||
const thinkingLevel = snapshot.conversation.model.model?.thinkingLevel ?? 'off';
|
||||
|
||||
@@ -1,10 +1,11 @@
|
||||
import { mkdir, stat } from 'node:fs/promises';
|
||||
import { mkdir, rename, stat } from 'node:fs/promises';
|
||||
import path from 'node:path';
|
||||
import {
|
||||
BUNDLED_CODING_SKILL_IDS,
|
||||
type BundledCodingSkillId,
|
||||
} from '../../../shared/coding-skills';
|
||||
import { atomicWriteJson, atomicWriteText } from '../../coding-projects/atomic-json';
|
||||
import { validateSessionKey } from '../../coding-projects/conversation-store';
|
||||
import type { PiProviderSelection } from './provider-config';
|
||||
import type { PiManagedInputRevision } from './managed-input-revision';
|
||||
|
||||
@@ -104,6 +105,31 @@ export async function ensurePiManagedPaths(userDataDir: string): Promise<PiManag
|
||||
return paths;
|
||||
}
|
||||
|
||||
export async function archivePiConversationSession(input: {
|
||||
userDataDir: string;
|
||||
projectId: string;
|
||||
sessionKey: string;
|
||||
}): Promise<string | null> {
|
||||
const projectId = managedSegment(input.projectId, 'Project id');
|
||||
const sessionKey = validateSessionKey(input.sessionKey);
|
||||
const paths = getPiManagedPaths(input.userDataDir);
|
||||
const source = path.join(paths.sessionsDir, projectId, `${sessionKey}.jsonl`);
|
||||
const targetDirectory = path.join(paths.trashDir, projectId);
|
||||
const target = path.join(targetDirectory, `${sessionKey}.jsonl`);
|
||||
await mkdir(targetDirectory, { recursive: true });
|
||||
try {
|
||||
await rename(source, target);
|
||||
return target;
|
||||
} catch (error) {
|
||||
if ((error as NodeJS.ErrnoException).code !== 'ENOENT') throw error;
|
||||
const alreadyArchived = await stat(target).catch((targetError: NodeJS.ErrnoException) => {
|
||||
if (targetError.code === 'ENOENT') return null;
|
||||
throw targetError;
|
||||
});
|
||||
return alreadyArchived?.isFile() ? target : null;
|
||||
}
|
||||
}
|
||||
|
||||
function normalizeSkillIds(skillIds: readonly string[]): BundledCodingSkillId[] {
|
||||
const result: BundledCodingSkillId[] = [];
|
||||
const seen = new Set<string>();
|
||||
|
||||
@@ -672,6 +672,15 @@ export class PiConversationRuntime implements CodingConversationRuntime {
|
||||
}
|
||||
}
|
||||
|
||||
async validateModel(model: ProductModelRef): Promise<ProductModelRef> {
|
||||
const selection = await this.resolveModel(model);
|
||||
return {
|
||||
accountId: selection.accountId,
|
||||
modelId: selection.modelId,
|
||||
thinkingLevel: model.thinkingLevel,
|
||||
};
|
||||
}
|
||||
|
||||
async setModel(input: SetConversationModelInput): Promise<ConversationModelState> {
|
||||
await this.waitForProjection(input.conversationId);
|
||||
const snapshot = this.snapshot(input.conversationId);
|
||||
@@ -957,19 +966,30 @@ export class PiConversationRuntime implements CodingConversationRuntime {
|
||||
if (!this.isAuthenticationError || !this.refreshCredential || !input.model.model) {
|
||||
return await this.pool.prepare(input);
|
||||
}
|
||||
return await this.providerRefresh.withSingleAuthRecovery({
|
||||
accountId: input.model.model.accountId,
|
||||
operation: async () => await this.pool.prepare(input),
|
||||
isAuthenticationError: this.isAuthenticationError,
|
||||
refreshCredential: async () => await this.refreshCredential!(input.model.model!.accountId),
|
||||
reopenWorker: async () => {
|
||||
if (!this.pool.getState(input.conversationId)) return;
|
||||
const recovered = await this.pool.recover(input.conversationId);
|
||||
if (this.states.has(input.conversationId)) {
|
||||
await this.requestHydration(input.conversationId, recovered, false);
|
||||
}
|
||||
},
|
||||
});
|
||||
try {
|
||||
return await this.providerRefresh.withSingleAuthRecovery({
|
||||
accountId: input.model.model.accountId,
|
||||
operation: async () => await this.pool.prepare(input),
|
||||
isAuthenticationError: this.isAuthenticationError,
|
||||
refreshCredential: async () => await this.refreshCredential!(input.model.model!.accountId),
|
||||
reopenWorker: async () => {
|
||||
if (!this.pool.getState(input.conversationId)) return;
|
||||
const recovered = await this.pool.recover(input.conversationId);
|
||||
if (this.states.has(input.conversationId)) {
|
||||
await this.requestHydration(input.conversationId, recovered, false);
|
||||
}
|
||||
},
|
||||
});
|
||||
} catch (error) {
|
||||
if (this.isAuthenticationError(error)) {
|
||||
throw new CodingRuntimeContractError(
|
||||
'CODING_PROVIDER_AUTH_REQUIRED',
|
||||
'Provider authentication failed after one recovery attempt',
|
||||
true,
|
||||
);
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
private async acceptPrompt(
|
||||
@@ -1008,13 +1028,20 @@ export class PiConversationRuntime implements CodingConversationRuntime {
|
||||
}, runId);
|
||||
}
|
||||
} catch (error) {
|
||||
const failure = this.isAuthenticationError?.(error)
|
||||
? new CodingRuntimeContractError(
|
||||
'CODING_PROVIDER_AUTH_REQUIRED',
|
||||
'Provider authentication failed after one recovery attempt',
|
||||
true,
|
||||
)
|
||||
: error;
|
||||
this.pool.failTopLevel(
|
||||
conversationId,
|
||||
runId,
|
||||
error instanceof Error ? error : new Error('Prompt acceptance failed'),
|
||||
failure instanceof Error ? failure : new Error('Prompt acceptance failed'),
|
||||
);
|
||||
this.failRun(conversationId, runId, error);
|
||||
throw error;
|
||||
this.failRun(conversationId, runId, failure);
|
||||
throw failure;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user