jiang
fbe273e90a
Use verified task names until Agent generates a specific conversation title. Repair legacy names without changing manual titles or history ordering. Bound page repairs to 30 entries and a shared two-second deadline, and re-read titles after task access validation to avoid stale reuse receipts. Validation: bun run check; 40 focused tests across six files.
341 lines
13 KiB
TypeScript
341 lines
13 KiB
TypeScript
import { join } from "node:path";
|
|
|
|
import { SessionTopicStore } from "./topic-store.js";
|
|
import { config } from "../config.js";
|
|
import { type SessionWorkspaceStore } from "./workspace-store.js";
|
|
import {
|
|
atomicWriteJson,
|
|
ensureDirectory,
|
|
listJsonFiles,
|
|
readJsonFile,
|
|
slugify,
|
|
} from "../utils/fileStore.js";
|
|
|
|
export type SessionStatus = "active" | "archived";
|
|
export type SessionPurpose = "ordinary_chat" | "task_execution" | "task_collaboration";
|
|
|
|
export type SessionRecord = {
|
|
sessionId: string;
|
|
actorKey: string;
|
|
ownerUserId?: string;
|
|
projectId?: string;
|
|
projectKey: string;
|
|
parentSessionId?: string;
|
|
createdAt: string;
|
|
updatedAt: string;
|
|
status: SessionStatus;
|
|
deletedAt?: string;
|
|
purpose?: SessionPurpose;
|
|
taskId?: string;
|
|
runId?: string;
|
|
requestId?: string;
|
|
modelId?: string;
|
|
title?: string;
|
|
topicId?: string | null;
|
|
triggerId?: string;
|
|
triggerSource?: string;
|
|
triggerSubjectType?: string;
|
|
triggerSubjectId?: string;
|
|
triggerWorkflow?: string;
|
|
triggerOccurredAt?: string;
|
|
};
|
|
|
|
export type SessionTriggerMetadata = {
|
|
triggerId: string;
|
|
source: string;
|
|
subjectType: string;
|
|
subjectId: string;
|
|
workflow: string;
|
|
occurredAt: string;
|
|
};
|
|
|
|
type SessionMetadataContext = {
|
|
actorKey: string;
|
|
userId?: string;
|
|
projectId?: string;
|
|
projectKey: string;
|
|
};
|
|
|
|
type EnsureSessionMetadataInput = SessionMetadataContext & {
|
|
sessionId: string;
|
|
parentSessionId?: string;
|
|
trigger?: SessionTriggerMetadata;
|
|
purpose?: SessionPurpose;
|
|
taskId?: string;
|
|
runId?: string;
|
|
requestId?: string;
|
|
modelId?: string;
|
|
topicId?: string;
|
|
};
|
|
|
|
export class SessionMetadataStore {
|
|
readonly topics: SessionTopicStore;
|
|
private readonly ensureQueues = new Map<string, Promise<void>>();
|
|
private index?: Map<string, Map<string, SessionRecord>>;
|
|
private indexLoading?: Promise<void>;
|
|
private indexVersion = 0;
|
|
constructor(private readonly baseDir = config.SESSION_METADATA_STORAGE_DIR, private readonly workspaces?: SessionWorkspaceStore) {
|
|
this.topics = new SessionTopicStore(join(baseDir, "topics"));
|
|
workspaces?.setDeletionCheck(async (context) => Boolean((await this.get(context, context.sessionId, { includeDeleted: true }))?.deletedAt));
|
|
}
|
|
|
|
async initialize() {
|
|
await ensureDirectory(this.baseDir);
|
|
}
|
|
|
|
async ensure(input: EnsureSessionMetadataInput) {
|
|
return this.serialize(input.sessionId, () => this.ensureRecord(input));
|
|
}
|
|
|
|
private async serialize<T>(sessionId: string, operation: () => Promise<T>): Promise<T> {
|
|
const key = this.filePath(sessionId);
|
|
const previous = this.ensureQueues.get(key) ?? Promise.resolve();
|
|
const result = previous.catch(() => undefined).then(operation);
|
|
const pending = result.then(() => undefined, () => undefined);
|
|
this.ensureQueues.set(key, pending);
|
|
try { return await result; }
|
|
finally { if (this.ensureQueues.get(key) === pending) this.ensureQueues.delete(key); }
|
|
}
|
|
|
|
private async ensureRecord(input: EnsureSessionMetadataInput) {
|
|
const sessionId = normalizeSessionId(input.sessionId);
|
|
if (!sessionId) {
|
|
throw new Error("sessionId is required");
|
|
}
|
|
const existing = await readJsonFile<SessionRecord>(
|
|
this.filePath(sessionId),
|
|
);
|
|
if (existing) {
|
|
if (
|
|
existing.deletedAt || existing.sessionId !== sessionId || !matchesContext(existing, input) ||
|
|
(input.trigger && existing.triggerId !== input.trigger.triggerId) ||
|
|
(input.purpose && readPurpose(existing) !== input.purpose) ||
|
|
(input.taskId && existing.taskId !== input.taskId.trim()) ||
|
|
(input.runId && existing.runId !== input.runId.trim()) ||
|
|
(input.requestId && existing.requestId !== input.requestId.trim()) ||
|
|
(input.modelId && existing.modelId !== input.modelId.trim())
|
|
) {
|
|
throw new SessionContextMismatchError();
|
|
}
|
|
if (this.workspaces) await this.workspaces.ensure({ actorKey: existing.actorKey, projectKey: existing.projectKey, sessionId });
|
|
return { created: false, record: existing };
|
|
}
|
|
|
|
const now = new Date().toISOString();
|
|
if (this.workspaces) await this.workspaces.ensure({ actorKey: input.actorKey, projectKey: input.projectKey, sessionId });
|
|
const record: SessionRecord = {
|
|
sessionId,
|
|
actorKey: input.actorKey,
|
|
ownerUserId: input.userId?.trim(),
|
|
projectId: input.projectId,
|
|
projectKey: input.projectKey,
|
|
parentSessionId: normalizeSessionId(input.parentSessionId),
|
|
createdAt: now,
|
|
updatedAt: now,
|
|
status: "active",
|
|
purpose: input.purpose ?? (input.trigger ? "task_execution" : "ordinary_chat"),
|
|
taskId: input.taskId?.trim() || input.trigger?.subjectId,
|
|
runId: input.runId?.trim(),
|
|
requestId: input.requestId?.trim(),
|
|
modelId: input.modelId?.trim(),
|
|
topicId: input.topicId,
|
|
...(input.trigger ? {
|
|
triggerId: input.trigger.triggerId,
|
|
triggerSource: input.trigger.source,
|
|
triggerSubjectType: input.trigger.subjectType,
|
|
triggerSubjectId: input.trigger.subjectId,
|
|
triggerWorkflow: input.trigger.workflow,
|
|
triggerOccurredAt: input.trigger.occurredAt,
|
|
} : {}),
|
|
};
|
|
await atomicWriteJson(
|
|
this.filePath(record.sessionId),
|
|
record,
|
|
0o600,
|
|
);
|
|
this.updateIndex(record);
|
|
return { created: true, record };
|
|
}
|
|
|
|
async get(context: SessionMetadataContext, sessionId: string, options?: { includeDeleted?: boolean }) {
|
|
const normalizedSessionId = normalizeSessionId(sessionId);
|
|
if (!normalizedSessionId) {
|
|
return null;
|
|
}
|
|
const record = await readJsonFile<SessionRecord>(
|
|
this.filePath(normalizedSessionId),
|
|
);
|
|
return record && (!record.deletedAt || options?.includeDeleted) && record.sessionId === normalizedSessionId && matchesContext(record, context) ? record : null;
|
|
}
|
|
|
|
// Internal delivery only, after Server has authorized and resolved this run.
|
|
// Does not authorize ordinary chat access or expose a global session lookup.
|
|
async getTaskExecutionBinding(projectId: string, taskId: string, runId: string, sessionId: string) {
|
|
const id = normalizeSessionId(sessionId);
|
|
if (!id) return null;
|
|
const record = await readJsonFile<SessionRecord>(this.filePath(id));
|
|
return record && !record.deletedAt && record.sessionId === id
|
|
&& record.projectId === projectId && record.taskId === taskId && record.runId === runId
|
|
&& readPurpose(record) === "task_execution" ? record : null;
|
|
}
|
|
|
|
async touch(
|
|
record: SessionRecord,
|
|
updates: Partial<Pick<SessionRecord, "title" | "status" | "topicId">> = {},
|
|
options: { preserveUpdatedAt?: boolean; expectedTitle?: string } = {},
|
|
) {
|
|
return this.serialize(record.sessionId, async () => {
|
|
const current = await this.get({ ...record, userId: record.ownerUserId }, record.sessionId);
|
|
if (!current) throw new SessionContextMismatchError();
|
|
if (options.expectedTitle !== undefined && (current.title ?? "") !== options.expectedTitle) return current;
|
|
const next: SessionRecord = { ...current, ...normalizeSessionUpdates(updates),
|
|
updatedAt: options.preserveUpdatedAt ? current.updatedAt : new Date().toISOString() };
|
|
await atomicWriteJson(this.filePath(record.sessionId), next, 0o600);
|
|
this.updateIndex(next);
|
|
return next;
|
|
});
|
|
}
|
|
|
|
async list(
|
|
context: SessionMetadataContext,
|
|
filter?: { purpose?: SessionPurpose; taskId?: string },
|
|
) {
|
|
await this.loadIndex();
|
|
const records = [...(this.index?.get(this.scopeKey(context))?.values() ?? [])].map((record) => ({ ...record }));
|
|
return records
|
|
.filter((record): record is SessionRecord => Boolean(record))
|
|
.filter(
|
|
(record) =>
|
|
!record.deletedAt && matchesContext(record, context),
|
|
)
|
|
.filter((record) => !filter?.purpose || readPurpose(record) === filter.purpose)
|
|
.filter((record) => !filter?.taskId || record.taskId === filter.taskId.trim())
|
|
.sort((left, right) => right.updatedAt.localeCompare(left.updatedAt));
|
|
}
|
|
|
|
async findByTask(
|
|
context: SessionMetadataContext,
|
|
taskId: string,
|
|
purpose: SessionPurpose,
|
|
) {
|
|
const records = await this.list(context, { purpose, taskId });
|
|
return records[0] ?? null;
|
|
}
|
|
|
|
async findByTrigger(context: SessionMetadataContext, triggerId: string) {
|
|
const normalizedTriggerId = triggerId.trim();
|
|
if (!normalizedTriggerId) {
|
|
return null;
|
|
}
|
|
const records = await this.list(context);
|
|
return records.find((record) => record.triggerId === normalizedTriggerId) ?? null;
|
|
}
|
|
|
|
async findByRequest(context: SessionMetadataContext, requestId: string) {
|
|
const normalized = requestId.trim();
|
|
if (!normalized) return null;
|
|
const records = await this.list(context);
|
|
return records.find((record) => record.requestId === normalized) ?? null;
|
|
}
|
|
|
|
async remove(record: SessionRecord) {
|
|
return this.serialize(record.sessionId, async () => {
|
|
const current = await this.get({ ...record, userId: record.ownerUserId }, record.sessionId, { includeDeleted: true });
|
|
if (!current) throw new SessionContextMismatchError();
|
|
// Retain only identity and deletion state, so legacy imports and late writes cannot revive it.
|
|
const deleted: SessionRecord = {
|
|
sessionId: current.sessionId, actorKey: current.actorKey, projectKey: current.projectKey,
|
|
ownerUserId: current.ownerUserId, projectId: current.projectId,
|
|
createdAt: current.createdAt, updatedAt: new Date().toISOString(), status: "archived",
|
|
deletedAt: current.deletedAt ?? new Date().toISOString(),
|
|
};
|
|
await atomicWriteJson(this.filePath(record.sessionId), deleted, 0o600);
|
|
this.updateIndex(deleted);
|
|
await this.workspaces?.markDeleted(deleted);
|
|
});
|
|
}
|
|
|
|
async purge(record: SessionRecord) {
|
|
const current = await this.get({ ...record, userId: record.ownerUserId }, record.sessionId, { includeDeleted: true });
|
|
if (!current?.deletedAt) throw new SessionContextMismatchError();
|
|
await this.workspaces?.purge(current);
|
|
}
|
|
|
|
private scopeKey(context: { actorKey: string; projectKey: string }) {
|
|
return JSON.stringify([context.actorKey, context.projectKey]);
|
|
}
|
|
|
|
private updateIndex(record: SessionRecord) {
|
|
this.indexVersion++;
|
|
if (!this.index) return;
|
|
const key = this.scopeKey(record);
|
|
const records = this.index.get(key) ?? new Map<string, SessionRecord>();
|
|
records.set(record.sessionId, { ...record });
|
|
this.index.set(key, records);
|
|
}
|
|
|
|
private async loadIndex(): Promise<void> {
|
|
if (this.index) return;
|
|
if (this.indexLoading) { await this.indexLoading; return this.loadIndex(); }
|
|
const version = this.indexVersion;
|
|
const loading = (async () => {
|
|
const files = await listJsonFiles(this.baseDir);
|
|
const records = await Promise.all(files.map((file) => readJsonFile<SessionRecord>(file)));
|
|
const index = new Map<string, Map<string, SessionRecord>>();
|
|
for (const record of records) {
|
|
if (!record) continue;
|
|
const key = this.scopeKey(record);
|
|
const scoped = index.get(key) ?? new Map<string, SessionRecord>();
|
|
scoped.set(record.sessionId, record);
|
|
index.set(key, scoped);
|
|
}
|
|
if (version === this.indexVersion) this.index = index;
|
|
})();
|
|
this.indexLoading = loading;
|
|
try { await loading; }
|
|
finally { if (this.indexLoading === loading) this.indexLoading = undefined; }
|
|
if (!this.index) await this.loadIndex();
|
|
}
|
|
|
|
private filePath(sessionId: string) {
|
|
return join(this.baseDir, `${slugify(sessionId)}.json`);
|
|
}
|
|
}
|
|
|
|
const normalizeSessionId = (value?: string) => {
|
|
const normalized = value?.trim();
|
|
return normalized ? normalized.slice(0, 128) : undefined;
|
|
};
|
|
|
|
export const readPurpose = (record: SessionRecord): SessionPurpose =>
|
|
record.purpose ?? (record.triggerId ? "task_execution" : "ordinary_chat");
|
|
|
|
export class SessionContextMismatchError extends Error {
|
|
constructor() {
|
|
super("session not found");
|
|
}
|
|
}
|
|
|
|
const matchesContext = (record: SessionRecord, context: SessionMetadataContext) =>
|
|
record.actorKey === context.actorKey &&
|
|
record.projectKey === context.projectKey &&
|
|
(context.userId === undefined || record.ownerUserId === context.userId) &&
|
|
(context.projectId === undefined || record.projectId === context.projectId);
|
|
|
|
const normalizeSessionUpdates = (
|
|
updates: Partial<Pick<SessionRecord, "title" | "status" | "topicId">>,
|
|
) => {
|
|
const normalized: Partial<Pick<SessionRecord, "title" | "status" | "topicId">> = {};
|
|
if (updates.status === "active" || updates.status === "archived") {
|
|
normalized.status = updates.status;
|
|
}
|
|
if (updates.topicId === null || typeof updates.topicId === "string") normalized.topicId = updates.topicId;
|
|
if (typeof updates.title === "string") {
|
|
const trimmed = updates.title.trim();
|
|
if (trimmed) {
|
|
normalized.title = trimmed.slice(0, 120);
|
|
}
|
|
}
|
|
return normalized;
|
|
};
|