471 lines
20 KiB
TypeScript
471 lines
20 KiB
TypeScript
import { randomUUID } from 'node:crypto';
|
||
import { readdir, rm } from 'node:fs/promises';
|
||
import path from 'node:path';
|
||
import { EventEmitter } from 'node:events';
|
||
import type { CodingProjectService } from '../coding-projects/project-service';
|
||
import type { CodingConversationRuntime } from '../coding-runtime/contracts';
|
||
import type {
|
||
ConsultationRole,
|
||
TeacherDefinition,
|
||
TeacherReference,
|
||
TeacherSend,
|
||
TeacherSourceContext,
|
||
TeacherTopic,
|
||
} from '../../shared/coding-teacher';
|
||
import {
|
||
currentTeacherAccount,
|
||
assertTeacherAccount,
|
||
teacherAvailability,
|
||
teacherVersion,
|
||
teacherPreview,
|
||
TeacherError,
|
||
type TeacherAccount,
|
||
} from './config-client';
|
||
import { TeacherTopicStore, teacherTopicId } from './store';
|
||
import { compileTeacherContext } from './context';
|
||
import { prepareTeacherModel } from './model-runner';
|
||
import { consultationDefinition } from './consultation-role';
|
||
import { readTeacherSource } from './source-reader';
|
||
import { parseTeacherSuggestions } from './suggestions';
|
||
import { subscribeWorksSquareSession } from '../services/works-square-session';
|
||
|
||
export interface TeacherScope {
|
||
projectId: string;
|
||
sourceId: string;
|
||
role?: ConsultationRole;
|
||
}
|
||
interface PreviewTopic extends TeacherTopic {
|
||
sampleContext?: string;
|
||
}
|
||
export interface TeacherServiceOptions {
|
||
projects: CodingProjectService;
|
||
runtime: CodingConversationRuntime;
|
||
userDataDir: string;
|
||
acquireLease?(id: string): () => void;
|
||
account?: typeof currentTeacherAccount;
|
||
assertAccount?: typeof assertTeacherAccount;
|
||
availability?: typeof teacherAvailability;
|
||
version?: typeof teacherVersion;
|
||
preview?: typeof teacherPreview;
|
||
prepareModel?: typeof prepareTeacherModel;
|
||
readSource?(scope: TeacherScope): Promise<TeacherSourceContext>;
|
||
}
|
||
export class CodingTeacherService {
|
||
private readonly stores = new Map<string, TeacherTopicStore>();
|
||
private readonly tails = new Map<string, Promise<unknown>>();
|
||
private readonly active = new Map<
|
||
string,
|
||
{ account: TeacherAccount; controller: AbortController; sourceId?: string; projectId: string }
|
||
>();
|
||
private readonly finishes = new Map<string, Promise<void>>();
|
||
private readonly deletingSources = new Set<string>();
|
||
private readonly events = new EventEmitter();
|
||
private readonly unsubscribe: () => void;
|
||
private readonly account: typeof currentTeacherAccount;
|
||
private readonly assertAccount: typeof assertTeacherAccount;
|
||
constructor(private readonly options: TeacherServiceOptions) {
|
||
this.account = options.account ?? currentTeacherAccount;
|
||
this.assertAccount = options.assertAccount ?? assertTeacherAccount;
|
||
this.unsubscribe = subscribeWorksSquareSession(() => {
|
||
for (const run of this.active.values()) {
|
||
try {
|
||
this.assertAccount(run.account);
|
||
} catch {
|
||
run.controller.abort();
|
||
}
|
||
}
|
||
});
|
||
}
|
||
async previewDefinition(revision: number) {
|
||
const account = await this.account();
|
||
return await (this.options.preview ?? teacherPreview)(account, revision);
|
||
}
|
||
async definition(role: ConsultationRole = 'teacher') {
|
||
const account = await this.account();
|
||
const status = await (this.options.availability ?? teacherAvailability)(account);
|
||
const published = status.published_version
|
||
? await (this.options.version ?? teacherVersion)(account, status.published_version)
|
||
: null;
|
||
return { ...status, definition: published ? consultationDefinition(published.payload, role) : null };
|
||
}
|
||
private async scopedStore(
|
||
account: TeacherAccount,
|
||
scope: TeacherScope
|
||
): Promise<TeacherTopicStore> {
|
||
if (this.deletingSources.has(scope.projectId + ':' + scope.sourceId))
|
||
throw new TeacherError(404, 'teacher_source_not_found', '来源会话已删除。');
|
||
let directory: string;
|
||
if (scope.projectId === 'preview') {
|
||
directory = path.join(this.options.userDataDir, 'teacher-preview', account.id);
|
||
} else {
|
||
const project = await this.options.projects.getProject(scope.projectId);
|
||
if (scope.sourceId !== 'project' && !(await this.options.projects.conversationStore(project.path).get(scope.sourceId)))
|
||
throw new TeacherError(404, 'teacher_source_not_found', '来源会话已删除。');
|
||
directory = path.join(
|
||
project.path,
|
||
'.makelore',
|
||
scope.role === 'friend' ? 'friend-conversations' : 'teacher-conversations',
|
||
account.id,
|
||
scope.sourceId === 'project' ? 'project' : teacherTopicId(scope.sourceId)
|
||
);
|
||
}
|
||
let store = this.stores.get(directory);
|
||
if (!store) {
|
||
store = new TeacherTopicStore(directory);
|
||
this.stores.set(directory, store);
|
||
}
|
||
return store;
|
||
}
|
||
private key(account: TeacherAccount, scope: TeacherScope, id: string) {
|
||
return account.id + ':' + scope.projectId + ':' + scope.sourceId + ':' + (scope.role ?? 'teacher') + ':' + id;
|
||
}
|
||
private async serialize<T>(key: string, operation: () => Promise<T>): Promise<T> {
|
||
const previous = this.tails.get(key) ?? Promise.resolve();
|
||
const next = previous.catch(() => undefined).then(operation);
|
||
this.tails.set(key, next);
|
||
try {
|
||
return await next;
|
||
} finally {
|
||
if (this.tails.get(key) === next) this.tails.delete(key);
|
||
}
|
||
}
|
||
async list(scope: TeacherScope) {
|
||
const account = await this.account();
|
||
return await (await this.scopedStore(account, scope)).list();
|
||
}
|
||
async create(
|
||
scope: TeacherScope,
|
||
draftRevision?: number,
|
||
sampleContext = ''
|
||
): Promise<TeacherTopic> {
|
||
const account = await this.account();
|
||
const store = await this.scopedStore(account, scope);
|
||
let definition: TeacherDefinition, version: number;
|
||
if (scope.projectId === 'preview') {
|
||
if (
|
||
!Number.isSafeInteger(draftRevision) ||
|
||
draftRevision! < 1 ||
|
||
typeof sampleContext !== 'string' ||
|
||
sampleContext.length > 12000
|
||
)
|
||
throw new TeacherError(400, 'teacher_preview_invalid', '试聊参数无效。');
|
||
definition = (await (this.options.preview ?? teacherPreview)(account, draftRevision!))
|
||
.payload;
|
||
version = 0;
|
||
} else {
|
||
const status = await (this.options.availability ?? teacherAvailability)(account);
|
||
if (!status.enabled || !status.published_version)
|
||
throw new TeacherError(409, 'teacher_disabled', '老师暂未开放。');
|
||
const published = await (this.options.version ?? teacherVersion)(
|
||
account,
|
||
status.published_version
|
||
);
|
||
definition = consultationDefinition(published.payload, scope.role ?? 'teacher');
|
||
version = published.version;
|
||
}
|
||
this.assertAccount(account);
|
||
const now = new Date().toISOString();
|
||
const topic: PreviewTopic = {
|
||
schemaVersion: 1,
|
||
role: scope.role ?? 'teacher',
|
||
revision: 0,
|
||
id: randomUUID(),
|
||
accountId: account.id,
|
||
projectId: scope.projectId,
|
||
sourceConversationId: scope.sourceId,
|
||
definition,
|
||
version,
|
||
createdAt: now,
|
||
updatedAt: now,
|
||
requests: [],
|
||
...(scope.projectId === 'preview' ? { draftRevision, sampleContext } : {}),
|
||
};
|
||
await store.save(topic);
|
||
await store.select(topic.id);
|
||
return structuredClone(topic);
|
||
}
|
||
private async readOwned(account: TeacherAccount, scope: TeacherScope, id: string) {
|
||
const store = await this.scopedStore(account, scope);
|
||
const topic = (await store.read(id)) as PreviewTopic;
|
||
if (
|
||
topic.accountId !== account.id ||
|
||
topic.projectId !== scope.projectId ||
|
||
topic.sourceConversationId !== scope.sourceId ||
|
||
(topic.role ?? 'teacher') !== (scope.role ?? 'teacher')
|
||
)
|
||
throw new TeacherError(404, 'teacher_topic_not_found', '老师话题不存在。');
|
||
return { store, topic };
|
||
}
|
||
async read(scope: TeacherScope, id: string) {
|
||
const account = await this.account();
|
||
const { store, topic } = await this.readOwned(account, scope, id);
|
||
await store.select(id);
|
||
return structuredClone(topic);
|
||
}
|
||
async send(scope: TeacherScope, id: string, input: TeacherSend): Promise<TeacherTopic> {
|
||
teacherTopicId(input.requestId);
|
||
const intent = input.intent === undefined ? 'question' : input.intent;
|
||
if (!['question', 'suggestions', 'guided-help'].includes(intent))
|
||
throw new TeacherError(422, 'teacher_intent_invalid', '提问方式无效，请重新打开老师后再试。');
|
||
if (intent !== 'question' && (
|
||
scope.projectId === 'preview' || scope.sourceId !== 'project' || (scope.role ?? 'teacher') !== 'teacher'
|
||
))
|
||
throw new TeacherError(422, 'teacher_intent_invalid', '这种提问方式只适用于项目里的老师。');
|
||
if (input.sourceConversationId !== undefined) teacherTopicId(input.sourceConversationId);
|
||
if (typeof input.text !== 'string' || !input.text.trim() || input.text.length > 6000)
|
||
throw new TeacherError(422, 'teacher_question_invalid', '请输入 1–6000 字的问题。');
|
||
const refs = input.references ?? [];
|
||
if (
|
||
!Array.isArray(refs) ||
|
||
refs.length > 20 ||
|
||
refs.some(
|
||
(ref) =>
|
||
!ref ||
|
||
!['message', 'code'].includes(ref.kind) ||
|
||
typeof ref.text !== 'string' ||
|
||
(ref.path !== undefined && typeof ref.path !== 'string')
|
||
) ||
|
||
refs.reduce((n, ref) => n + ref.text.length, 0) > 12000
|
||
)
|
||
throw new TeacherError(422, 'teacher_reference_invalid', '引用内容无效或超过 12000 字。');
|
||
const account = await this.account(),
|
||
key = this.key(account, scope, id);
|
||
return await this.serialize(key, async () => {
|
||
const { store, topic } = await this.readOwned(account, scope, id);
|
||
const existing = topic.requests.find((request) => request.id === input.requestId);
|
||
if (existing) {
|
||
if (
|
||
existing.text !== input.text ||
|
||
(existing.intent ?? 'question') !== intent ||
|
||
JSON.stringify(existing.references) !== JSON.stringify(refs) ||
|
||
(existing.sourceConversationId ?? undefined) !== (input.sourceConversationId ?? undefined)
|
||
)
|
||
throw new TeacherError(409, 'teacher_request_conflict', '同一请求标识不能用于不同问题。');
|
||
return structuredClone(topic);
|
||
}
|
||
if (
|
||
this.active.has(key) ||
|
||
topic.requests.some(
|
||
(request) => request.status === 'preparing' || request.status === 'running'
|
||
)
|
||
)
|
||
throw new TeacherError(409, 'teacher_topic_busy', '请等待当前回复完成，或先停止。');
|
||
if (topic.draftRevision) {
|
||
await (this.options.preview ?? teacherPreview)(account, topic.draftRevision);
|
||
} else {
|
||
const available = await (this.options.availability ?? teacherAvailability)(account);
|
||
if (!available.enabled)
|
||
throw new TeacherError(409, 'teacher_disabled', '老师已停用，历史仍可查看。');
|
||
}
|
||
const sourceId = scope.sourceId === 'project' ? input.sourceConversationId : scope.sourceId;
|
||
if (sourceId && this.deletingSources.has(scope.projectId + ':' + sourceId))
|
||
throw new TeacherError(404, 'teacher_source_not_found', '来源会话已删除。');
|
||
if (scope.sourceId === 'project' && sourceId) {
|
||
const project = await this.options.projects.getProject(scope.projectId);
|
||
if (!(await this.options.projects.conversationStore(project.path).get(sourceId)))
|
||
throw new TeacherError(404, 'teacher_source_not_found', '来源会话不属于当前项目。');
|
||
}
|
||
const source: TeacherSourceContext =
|
||
scope.projectId === 'preview'
|
||
? {
|
||
messages: topic.sampleContext
|
||
? [{ id: 'preview', role: 'user', text: topic.sampleContext }]
|
||
: [],
|
||
cursor: { workerGeneration: 0, seq: 0 },
|
||
capturedAt: new Date().toISOString(),
|
||
}
|
||
: !sourceId
|
||
? { messages: [], cursor: { workerGeneration: 0, seq: 0 }, capturedAt: new Date().toISOString() }
|
||
: await (this.options.readSource?.({ ...scope, sourceId }) ??
|
||
readTeacherSource(
|
||
this.options.projects,
|
||
this.options.runtime,
|
||
this.options.userDataDir,
|
||
scope.projectId,
|
||
sourceId
|
||
));
|
||
const references: TeacherReference[] = refs.map((ref) => {
|
||
if (
|
||
ref.kind === 'message' &&
|
||
!source.messages.some(
|
||
(message) => message.id === ref.messageId && message.text.includes(ref.text)
|
||
)
|
||
)
|
||
throw new TeacherError(
|
||
422,
|
||
'teacher_reference_invalid',
|
||
'选中的消息不属于当前完整会话，请重新引用。'
|
||
);
|
||
return structuredClone(ref);
|
||
});
|
||
if (!topic.definition.system_prompt.trim())
|
||
throw new TeacherError(422, 'teacher_definition_invalid', '请先配置老师的系统提示词。');
|
||
const model = await (this.options.prepareModel ?? prepareTeacherModel)(
|
||
account,
|
||
topic.definition
|
||
);
|
||
const compiled = compileTeacherContext(
|
||
topic.definition,
|
||
source,
|
||
topic.requests,
|
||
input.text,
|
||
references,
|
||
model.inputLimit,
|
||
intent
|
||
);
|
||
// Reading context and resolving model credentials can yield while a source
|
||
// is being deleted. Project consultations must recheck the actual source.
|
||
if (sourceId && this.deletingSources.has(scope.projectId + ':' + sourceId))
|
||
throw new TeacherError(404, 'teacher_source_not_found', '来源会话已删除。');
|
||
const request = {
|
||
id: input.requestId,
|
||
intent,
|
||
...(input.sourceConversationId ? { sourceConversationId: input.sourceConversationId } : {}),
|
||
text: input.text,
|
||
references,
|
||
createdAt: new Date().toISOString(),
|
||
sourceCursor: source.cursor,
|
||
sourceCapturedAt: source.capturedAt,
|
||
includedSourceMessageIds: compiled.includedSourceMessageIds,
|
||
omittedMessages: compiled.omittedMessages,
|
||
status: 'preparing' as const,
|
||
response: '',
|
||
};
|
||
topic.requests.push(request);
|
||
topic.updatedAt = request.createdAt;
|
||
topic.revision++;
|
||
try {
|
||
await store.save(topic);
|
||
} catch (error) {
|
||
topic.requests.pop();
|
||
throw error;
|
||
}
|
||
const controller = new AbortController();
|
||
// If deletion began during persistence, retain a cancelled request rather
|
||
// than leaving a permanently preparing request in the project topic.
|
||
if (sourceId && this.deletingSources.has(scope.projectId + ':' + sourceId))
|
||
controller.abort();
|
||
this.active.set(key, { account, controller, sourceId, projectId: scope.projectId });
|
||
const release = this.options.acquireLease?.(key) ?? (() => undefined);
|
||
const finish = async () => {
|
||
const current = topic.requests.at(-1)!;
|
||
try {
|
||
this.assertAccount(account);
|
||
if (controller.signal.aborted) throw controller.signal.reason;
|
||
current.status = 'running';
|
||
topic.revision++;
|
||
this.events.emit(key, structuredClone(topic));
|
||
current.usage = await model.run(compiled.messages, controller.signal, (delta) => {
|
||
current.response += delta;
|
||
topic.revision++;
|
||
this.events.emit(key, structuredClone(topic));
|
||
});
|
||
this.assertAccount(account);
|
||
if (!controller.signal.aborted && intent === 'suggestions') {
|
||
const suggestions = parseTeacherSuggestions(current.response);
|
||
current.response = suggestions.intro;
|
||
current.suggestedQuestions = suggestions.questions;
|
||
}
|
||
current.status = controller.signal.aborted ? 'cancelled' : 'completed';
|
||
} catch (error) {
|
||
current.status = controller.signal.aborted ? 'cancelled' : 'failed';
|
||
current.error = controller.signal.aborted
|
||
? '已停止回复，部分内容可能不完整。'
|
||
: error instanceof TeacherError
|
||
? error.message
|
||
: '老师回复失败，已保留本次问题与收到的内容。';
|
||
} finally {
|
||
if (intent === 'suggestions' && current.status !== 'completed') {
|
||
current.response = '';
|
||
delete current.suggestedQuestions;
|
||
}
|
||
topic.updatedAt = new Date().toISOString();
|
||
topic.revision++;
|
||
try {
|
||
await store.save(topic);
|
||
} catch {
|
||
topic.unsaved = true;
|
||
}
|
||
this.active.delete(key);
|
||
release();
|
||
this.events.emit(key, structuredClone(topic));
|
||
}
|
||
};
|
||
const completion = finish();
|
||
this.finishes.set(key, completion);
|
||
void completion.finally(() => this.finishes.delete(key));
|
||
return structuredClone(topic);
|
||
});
|
||
}
|
||
async cancel(scope: TeacherScope, id: string, requestId: string) {
|
||
const account = await this.account();
|
||
const { topic } = await this.readOwned(account, scope, id);
|
||
if (topic.requests.at(-1)?.id === requestId)
|
||
this.active.get(this.key(account, scope, id))?.controller.abort();
|
||
return structuredClone(topic);
|
||
}
|
||
async save(scope: TeacherScope, id: string) {
|
||
const account = await this.account();
|
||
const { store, topic } = await this.readOwned(account, scope, id);
|
||
if (this.active.has(this.key(account, scope, id)))
|
||
throw new TeacherError(409, 'teacher_topic_busy', '回复结束后再保存。');
|
||
await this.serialize(this.key(account, scope, id), () => store.save(topic));
|
||
return structuredClone(topic);
|
||
}
|
||
async subscribe(scope: TeacherScope, id: string, onTopic: (topic: TeacherTopic) => void) {
|
||
const account = await this.account();
|
||
const key = this.key(account, scope, id);
|
||
const { topic } = await this.readOwned(account, scope, id);
|
||
const listener = (next: TeacherTopic) => {
|
||
try {
|
||
this.assertAccount(account);
|
||
onTopic(next);
|
||
} catch {
|
||
this.events.off(key, listener);
|
||
}
|
||
};
|
||
this.events.on(key, listener);
|
||
onTopic(structuredClone(topic));
|
||
return () => this.events.off(key, listener);
|
||
}
|
||
async removeSource(projectId: string, sourceId: string) {
|
||
const project = await this.options.projects.getProject(projectId);
|
||
const root = path.join(project.path, '.makelore', 'teacher-conversations');
|
||
this.deletingSources.add(projectId + ':' + sourceId);
|
||
await Promise.allSettled(
|
||
[...this.tails.entries()]
|
||
// A project-level topic can be preparing a request for this source even
|
||
// though its serialization key uses "project" instead of the source id.
|
||
.filter(([key]) => key.includes(':' + projectId + ':'))
|
||
.map(([, promise]) => promise)
|
||
);
|
||
const affectedKeys = [...this.active.entries()]
|
||
.filter(([, run]) => run.projectId === projectId && run.sourceId === sourceId)
|
||
.map(([key, run]) => { run.controller.abort(); return key; });
|
||
await Promise.allSettled(affectedKeys.map((key) => this.finishes.get(key)));
|
||
await Promise.allSettled(
|
||
[...this.finishes.entries()]
|
||
.filter(([key]) => key.includes(':' + projectId + ':' + sourceId + ':'))
|
||
.map(([, promise]) => promise)
|
||
);
|
||
const accounts = await readdir(root, { withFileTypes: true }).catch(() => []);
|
||
// Account directories are verified Works UUIDs; the source is a stored conversation UUID.
|
||
for (const entry of accounts)
|
||
if (entry.isDirectory() && /^[0-9a-f-]{36}$/i.test(entry.name)) {
|
||
const directory = path.join(root, entry.name, teacherTopicId(sourceId));
|
||
const pending = [...this.tails.entries()]
|
||
.filter(([key]) => key.includes(':' + projectId + ':' + sourceId + ':'))
|
||
.map(([, promise]) => promise);
|
||
await Promise.allSettled(pending);
|
||
await rm(directory, { recursive: true, force: true });
|
||
this.stores.delete(directory);
|
||
}
|
||
}
|
||
async dispose() {
|
||
this.unsubscribe();
|
||
for (const run of this.active.values()) run.controller.abort();
|
||
await Promise.allSettled(this.finishes.values());
|
||
this.events.removeAllListeners();
|
||
}
|
||
}
|