579 lines
17 KiB
TypeScript
579 lines
17 KiB
TypeScript
import { readdir, readFile } from "node:fs/promises";
|
|
import { basename, extname, join } from "node:path";
|
|
|
|
import { config } from "../src/config.js";
|
|
import { AgentDatabase } from "../src/db/index.js";
|
|
import {
|
|
RESULT_REFERENCE_KIND,
|
|
RESULT_REFERENCE_SOURCE,
|
|
RESULT_REF_PATTERN,
|
|
type ResultPreview,
|
|
} from "../src/results/store.js";
|
|
import { sanitizePersistentDocument, sanitizePersistentLine } from "../src/utils/persistencePolicy.js";
|
|
import { toProjectKey, toStableId } from "../src/utils/fileStore.js";
|
|
|
|
type JsonRecord = Record<string, unknown>;
|
|
|
|
const db = new AgentDatabase();
|
|
const runtimeConversationMap = new Map<string, string>();
|
|
|
|
const counters = {
|
|
conversations: 0,
|
|
states: 0,
|
|
turns: 0,
|
|
runtimeSessions: 0,
|
|
learningStates: 0,
|
|
resultRefs: 0,
|
|
memories: 0,
|
|
skippedResultRefs: 0,
|
|
};
|
|
|
|
await db.initialize();
|
|
await migrateConversations();
|
|
await migrateConversationStates();
|
|
await migrateConversationTurns();
|
|
await migrateRuntimeSessions();
|
|
await migrateLearningStates();
|
|
await migrateResultRefs();
|
|
await migrateMemories();
|
|
await db.close();
|
|
|
|
console.log(JSON.stringify(counters, null, 2));
|
|
|
|
async function migrateConversations() {
|
|
for (const filePath of await listFiles(config.CONVERSATION_STORAGE_DIR, ".json")) {
|
|
const row = await readJson(filePath);
|
|
if (!row || typeof row.sessionId !== "string" || typeof row.actorKey !== "string") {
|
|
continue;
|
|
}
|
|
await upsertConversation({
|
|
sessionId: row.sessionId,
|
|
actorKey: row.actorKey,
|
|
ownerUserId: stringOrUndefined(row.ownerUserId),
|
|
projectId: stringOrUndefined(row.projectId),
|
|
projectKey:
|
|
stringOrUndefined(row.projectKey) ??
|
|
toProjectKey(stringOrUndefined(row.projectId)),
|
|
parentSessionId: stringOrUndefined(row.parentSessionId),
|
|
title: stringOrUndefined(row.title),
|
|
status: row.status === "archived" ? "archived" : "active",
|
|
createdAt: stringOrUndefined(row.createdAt),
|
|
updatedAt: stringOrUndefined(row.updatedAt),
|
|
});
|
|
counters.conversations += 1;
|
|
}
|
|
}
|
|
|
|
async function migrateConversationStates() {
|
|
for (const filePath of await listFiles(config.CONVERSATION_STATE_STORAGE_DIR, ".json")) {
|
|
const row = await readJson(filePath);
|
|
if (!row || typeof row.sessionId !== "string") {
|
|
continue;
|
|
}
|
|
await db.query(
|
|
`
|
|
INSERT INTO ${db.table("conversation_states")} (
|
|
session_id,
|
|
is_title_manually_edited,
|
|
messages,
|
|
branch_groups
|
|
)
|
|
VALUES ($1, $2, $3::jsonb, $4::jsonb)
|
|
ON CONFLICT (session_id)
|
|
DO UPDATE SET
|
|
is_title_manually_edited = EXCLUDED.is_title_manually_edited,
|
|
messages = EXCLUDED.messages,
|
|
branch_groups = EXCLUDED.branch_groups
|
|
`,
|
|
[
|
|
row.sessionId,
|
|
Boolean(row.isTitleManuallyEdited),
|
|
JSON.stringify(Array.isArray(row.messages) ? row.messages : []),
|
|
JSON.stringify(Array.isArray(row.branchGroups) ? row.branchGroups : []),
|
|
],
|
|
);
|
|
counters.states += 1;
|
|
}
|
|
}
|
|
|
|
async function migrateConversationTurns() {
|
|
for (const filePath of await listFiles(config.SESSION_HISTORY_STORAGE_DIR, ".json")) {
|
|
const row = await readJson(filePath);
|
|
if (!row || typeof row.actorKey !== "string" || typeof row.projectKey !== "string") {
|
|
continue;
|
|
}
|
|
const sessionId =
|
|
stringOrUndefined(row.clientSessionId) ?? stringOrUndefined(row.sessionId);
|
|
if (!sessionId) {
|
|
continue;
|
|
}
|
|
const turns = Array.isArray(row.turns) ? row.turns : [];
|
|
const lastTimestamp = stringOrUndefined(row.updatedAt);
|
|
await upsertConversation({
|
|
sessionId,
|
|
actorKey: row.actorKey,
|
|
projectKey: row.projectKey,
|
|
projectId: undefined,
|
|
createdAt:
|
|
turns.length > 0 && isRecord(turns[0]) ? stringOrUndefined(turns[0].timestamp) : undefined,
|
|
updatedAt:
|
|
lastTimestamp ??
|
|
(turns.length > 0 && isRecord(turns[turns.length - 1])
|
|
? stringOrUndefined(turns[turns.length - 1].timestamp)
|
|
: undefined),
|
|
});
|
|
for (const [turnIndex, turn] of turns.entries()) {
|
|
if (!isRecord(turn)) {
|
|
continue;
|
|
}
|
|
const userMessage = sanitizePersistentDocument(String(turn.userMessage ?? ""), 4000);
|
|
const assistantMessage = sanitizePersistentDocument(
|
|
String(turn.assistantMessage ?? ""),
|
|
4000,
|
|
);
|
|
if (!userMessage || !assistantMessage) {
|
|
continue;
|
|
}
|
|
const timestamp = stringOrUndefined(turn.timestamp) ?? new Date().toISOString();
|
|
await db.query(
|
|
`
|
|
INSERT INTO ${db.table("conversation_turns")} (
|
|
turn_id,
|
|
session_id,
|
|
turn_index,
|
|
actor_key,
|
|
project_key,
|
|
user_message,
|
|
assistant_message,
|
|
tool_call_count,
|
|
turn_timestamp
|
|
)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
|
|
ON CONFLICT (turn_id) DO NOTHING
|
|
`,
|
|
[
|
|
typeof turn.id === "string"
|
|
? turn.id
|
|
: toStableId(sessionId, String(turnIndex), timestamp, userMessage, assistantMessage),
|
|
sessionId,
|
|
turnIndex,
|
|
row.actorKey,
|
|
row.projectKey,
|
|
userMessage,
|
|
assistantMessage,
|
|
Math.max(0, numberOrZero(turn.toolCallCount)),
|
|
timestamp,
|
|
],
|
|
);
|
|
counters.turns += 1;
|
|
}
|
|
}
|
|
}
|
|
|
|
async function migrateRuntimeSessions() {
|
|
const seen = new Set<string>();
|
|
for (const filePath of await listFiles(config.SESSION_CONTEXT_STORAGE_DIR, ".json")) {
|
|
const row = await readJson(filePath);
|
|
if (
|
|
!row ||
|
|
typeof row.sessionId !== "string" ||
|
|
typeof row.clientSessionId !== "string" ||
|
|
typeof row.actorKey !== "string" ||
|
|
typeof row.projectKey !== "string"
|
|
) {
|
|
continue;
|
|
}
|
|
if (seen.has(row.sessionId)) {
|
|
continue;
|
|
}
|
|
seen.add(row.sessionId);
|
|
runtimeConversationMap.set(row.sessionId, row.clientSessionId);
|
|
await upsertConversation({
|
|
sessionId: row.clientSessionId,
|
|
actorKey: row.actorKey,
|
|
projectId: stringOrUndefined(row.projectId),
|
|
projectKey: row.projectKey,
|
|
});
|
|
await db.query(
|
|
`
|
|
INSERT INTO ${db.table("runtime_sessions")} (
|
|
runtime_session_id,
|
|
session_id,
|
|
actor_key,
|
|
allow_learning_write,
|
|
learning_mode,
|
|
project_id,
|
|
project_key,
|
|
trace_id,
|
|
released_at
|
|
)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, NOW())
|
|
ON CONFLICT (runtime_session_id)
|
|
DO UPDATE SET
|
|
session_id = EXCLUDED.session_id,
|
|
actor_key = EXCLUDED.actor_key,
|
|
allow_learning_write = EXCLUDED.allow_learning_write,
|
|
learning_mode = EXCLUDED.learning_mode,
|
|
project_id = EXCLUDED.project_id,
|
|
project_key = EXCLUDED.project_key,
|
|
trace_id = EXCLUDED.trace_id,
|
|
released_at = NOW()
|
|
`,
|
|
[
|
|
row.sessionId,
|
|
row.clientSessionId,
|
|
row.actorKey,
|
|
Boolean(row.allowLearningWrite),
|
|
row.learningMode === "review" ? "review" : row.learningMode === "interactive" ? "interactive" : null,
|
|
stringOrUndefined(row.projectId) ?? null,
|
|
row.projectKey,
|
|
typeof row.traceId === "string" ? row.traceId : `trace-${row.sessionId}`,
|
|
],
|
|
);
|
|
counters.runtimeSessions += 1;
|
|
}
|
|
}
|
|
|
|
async function migrateLearningStates() {
|
|
for (const filePath of await listFiles(config.LEARNING_STATE_STORAGE_DIR, ".json")) {
|
|
const row = await readJson(filePath);
|
|
if (!row || typeof row.sessionId !== "string") {
|
|
continue;
|
|
}
|
|
const sessionId = runtimeConversationMap.get(row.sessionId) ?? row.sessionId;
|
|
const existingConversation = await db.query(
|
|
`SELECT 1 FROM ${db.table("conversations")} WHERE session_id = $1 LIMIT 1`,
|
|
[sessionId],
|
|
);
|
|
if (existingConversation.rowCount === 0) {
|
|
continue;
|
|
}
|
|
await db.query(
|
|
`
|
|
INSERT INTO ${db.table("learning_states")} (
|
|
session_id,
|
|
last_gated_turn,
|
|
last_reviewed_turn,
|
|
pending_review
|
|
)
|
|
VALUES ($1, $2, $3, $4)
|
|
ON CONFLICT (session_id)
|
|
DO UPDATE SET
|
|
last_gated_turn = EXCLUDED.last_gated_turn,
|
|
last_reviewed_turn = EXCLUDED.last_reviewed_turn,
|
|
pending_review = EXCLUDED.pending_review
|
|
`,
|
|
[
|
|
sessionId,
|
|
numberOrZero(row.lastGatedTurn),
|
|
numberOrZero(row.lastReviewedTurn),
|
|
Boolean(row.pendingReview),
|
|
],
|
|
);
|
|
counters.learningStates += 1;
|
|
}
|
|
}
|
|
|
|
async function migrateResultRefs() {
|
|
for (const filePath of await listFiles(config.RESULT_REF_STORAGE_DIR, ".json")) {
|
|
const row = await readJson(filePath);
|
|
if (!row) {
|
|
continue;
|
|
}
|
|
const resultRef = normalizeResultRef(stringOrUndefined(row.resultRef) ?? basename(filePath, extname(filePath)));
|
|
const actorKey = stringOrUndefined(row.actorKey);
|
|
const sessionId =
|
|
stringOrUndefined(row.clientSessionId) ?? stringOrUndefined(row.sessionId);
|
|
if (!resultRef || !actorKey || !sessionId) {
|
|
counters.skippedResultRefs += 1;
|
|
continue;
|
|
}
|
|
const projectId = stringOrUndefined(row.projectId);
|
|
const projectKey = stringOrUndefined(row.projectKey) ?? toProjectKey(projectId);
|
|
await upsertConversation({
|
|
sessionId,
|
|
actorKey,
|
|
projectId,
|
|
projectKey,
|
|
});
|
|
const payload = unwrapPayload(row);
|
|
await db.query(
|
|
`
|
|
INSERT INTO ${db.table("result_refs")} (
|
|
result_ref,
|
|
actor_key,
|
|
session_id,
|
|
created_at,
|
|
kind,
|
|
preview,
|
|
project_id,
|
|
project_key,
|
|
schema_version,
|
|
size_bytes,
|
|
source,
|
|
trace_id,
|
|
payload_path,
|
|
object_key
|
|
)
|
|
VALUES ($1, $2, $3, $4, $5, $6::jsonb, $7, $8, $9, $10, $11, $12, $13, NULL)
|
|
ON CONFLICT (result_ref)
|
|
DO UPDATE SET
|
|
actor_key = EXCLUDED.actor_key,
|
|
session_id = EXCLUDED.session_id,
|
|
created_at = EXCLUDED.created_at,
|
|
kind = EXCLUDED.kind,
|
|
preview = EXCLUDED.preview,
|
|
project_id = EXCLUDED.project_id,
|
|
project_key = EXCLUDED.project_key,
|
|
schema_version = EXCLUDED.schema_version,
|
|
size_bytes = EXCLUDED.size_bytes,
|
|
source = EXCLUDED.source,
|
|
trace_id = EXCLUDED.trace_id,
|
|
payload_path = EXCLUDED.payload_path
|
|
`,
|
|
[
|
|
resultRef,
|
|
actorKey,
|
|
sessionId,
|
|
stringOrUndefined(row.createdAt) ?? new Date().toISOString(),
|
|
inferResultKind(row, payload),
|
|
JSON.stringify(isResultPreview(row.preview) ? row.preview : buildPreview(payload)),
|
|
projectId ?? null,
|
|
projectKey,
|
|
positiveIntegerOrOne(row.schemaVersion),
|
|
estimateBytes(payload),
|
|
inferResultSource(row.source),
|
|
stringOrUndefined(row.traceId) ?? `trace-${resultRef}`,
|
|
filePath,
|
|
],
|
|
);
|
|
counters.resultRefs += 1;
|
|
}
|
|
}
|
|
|
|
async function migrateMemories() {
|
|
await migrateMemoryScope("user", join(config.MEMORY_STORAGE_DIR, "users"));
|
|
await migrateMemoryScope("workspace", join(config.MEMORY_STORAGE_DIR, "workspaces"));
|
|
}
|
|
|
|
async function migrateMemoryScope(scope: "user" | "workspace", dir: string) {
|
|
for (const filePath of await listFiles(dir, ".md")) {
|
|
const markdown = await readText(filePath);
|
|
if (!markdown) {
|
|
continue;
|
|
}
|
|
const scopeKey = basename(filePath, ".md");
|
|
for (const entry of parseMemoryMarkdown(markdown)) {
|
|
await db.query(
|
|
`
|
|
INSERT INTO ${db.table("memories")} (
|
|
memory_id,
|
|
scope,
|
|
scope_key,
|
|
content,
|
|
source
|
|
)
|
|
VALUES ($1, $2, $3, $4, 'tool')
|
|
ON CONFLICT (memory_id)
|
|
DO UPDATE SET content = EXCLUDED.content
|
|
`,
|
|
[entry.id, scope, scopeKey, entry.content],
|
|
);
|
|
counters.memories += 1;
|
|
}
|
|
}
|
|
}
|
|
|
|
async function upsertConversation(input: {
|
|
sessionId: string;
|
|
actorKey: string;
|
|
projectKey: string;
|
|
projectId?: string;
|
|
ownerUserId?: string;
|
|
parentSessionId?: string;
|
|
title?: string;
|
|
status?: "active" | "archived";
|
|
createdAt?: string;
|
|
updatedAt?: string;
|
|
}) {
|
|
await db.query(
|
|
`
|
|
INSERT INTO ${db.table("conversations")} (
|
|
session_id,
|
|
actor_key,
|
|
owner_user_id,
|
|
project_id,
|
|
project_key,
|
|
parent_session_id,
|
|
title,
|
|
status,
|
|
created_at,
|
|
updated_at
|
|
)
|
|
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, COALESCE($9, NOW()), COALESCE($10, COALESCE($9, NOW())))
|
|
ON CONFLICT (session_id)
|
|
DO UPDATE SET
|
|
actor_key = EXCLUDED.actor_key,
|
|
owner_user_id = COALESCE(EXCLUDED.owner_user_id, ${db.table("conversations")}.owner_user_id),
|
|
project_id = COALESCE(EXCLUDED.project_id, ${db.table("conversations")}.project_id),
|
|
project_key = EXCLUDED.project_key,
|
|
parent_session_id = COALESCE(EXCLUDED.parent_session_id, ${db.table("conversations")}.parent_session_id),
|
|
title = COALESCE(EXCLUDED.title, ${db.table("conversations")}.title),
|
|
status = EXCLUDED.status,
|
|
created_at = LEAST(${db.table("conversations")}.created_at, EXCLUDED.created_at),
|
|
updated_at = GREATEST(${db.table("conversations")}.updated_at, EXCLUDED.updated_at)
|
|
`,
|
|
[
|
|
input.sessionId,
|
|
input.actorKey,
|
|
input.ownerUserId ?? null,
|
|
input.projectId ?? null,
|
|
input.projectKey,
|
|
input.parentSessionId ?? null,
|
|
input.title ?? null,
|
|
input.status ?? "active",
|
|
input.createdAt ?? null,
|
|
input.updatedAt ?? null,
|
|
],
|
|
);
|
|
}
|
|
|
|
async function listFiles(dir: string, extension: string) {
|
|
try {
|
|
const names = await readdir(dir);
|
|
return names
|
|
.filter((name) => name.endsWith(extension))
|
|
.map((name) => join(dir, name));
|
|
} catch {
|
|
return [];
|
|
}
|
|
}
|
|
|
|
async function readJson(path: string): Promise<JsonRecord | null> {
|
|
try {
|
|
return JSON.parse(await readFile(path, "utf8")) as JsonRecord;
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
async function readText(path: string): Promise<string | null> {
|
|
try {
|
|
return await readFile(path, "utf8");
|
|
} catch {
|
|
return null;
|
|
}
|
|
}
|
|
|
|
function parseMemoryMarkdown(content: string) {
|
|
return content
|
|
.split("\n")
|
|
.map((line) => line.trim())
|
|
.filter((line) => line.startsWith("- "))
|
|
.map((line) => line.slice(2).trim())
|
|
.map((line) => {
|
|
const match = line.match(/^\[([a-z0-9]{8,})\]\s+(.*)$/i);
|
|
const entryContent = sanitizePersistentLine(match?.[2] ?? line, 240);
|
|
return {
|
|
id: match?.[1] ?? toStableId("memory-entry", entryContent.toLowerCase()),
|
|
content: entryContent,
|
|
};
|
|
})
|
|
.filter((entry) => entry.content);
|
|
}
|
|
|
|
function inferResultKind(row: JsonRecord, payload: unknown) {
|
|
if (row.kind === RESULT_REFERENCE_KIND.renderJunctionsPayload) {
|
|
return RESULT_REFERENCE_KIND.renderJunctionsPayload;
|
|
}
|
|
if (isRecord(payload) && isRecord(payload.node_area_map)) {
|
|
return RESULT_REFERENCE_KIND.renderJunctionsPayload;
|
|
}
|
|
return RESULT_REFERENCE_KIND.dynamicHttpResult;
|
|
}
|
|
|
|
function inferResultSource(value: unknown) {
|
|
return value === RESULT_REFERENCE_SOURCE.dynamicHttp ||
|
|
value === RESULT_REFERENCE_SOURCE.agentGenerated ||
|
|
value === RESULT_REFERENCE_SOURCE.migration
|
|
? value
|
|
: RESULT_REFERENCE_SOURCE.legacy;
|
|
}
|
|
|
|
function unwrapPayload(row: JsonRecord) {
|
|
if ("data" in row) {
|
|
return row.data;
|
|
}
|
|
return row;
|
|
}
|
|
|
|
function isRecord(value: unknown): value is JsonRecord {
|
|
return typeof value === "object" && value !== null && !Array.isArray(value);
|
|
}
|
|
|
|
function isResultPreview(value: unknown): value is ResultPreview {
|
|
return (
|
|
isRecord(value) &&
|
|
typeof value.count === "number" &&
|
|
Array.isArray(value.fields) &&
|
|
typeof value.summary === "string" &&
|
|
"sample" in value
|
|
);
|
|
}
|
|
|
|
function buildPreview(data: unknown): ResultPreview {
|
|
if (Array.isArray(data)) {
|
|
const sample = data.slice(0, config.MAX_PREVIEW_SAMPLE_ITEMS);
|
|
const fields =
|
|
sample.length > 0 && isRecord(sample[0])
|
|
? Object.keys(sample[0]).slice(0, 30)
|
|
: [];
|
|
return {
|
|
count: data.length,
|
|
fields,
|
|
sample,
|
|
summary: `list[${data.length}]`,
|
|
};
|
|
}
|
|
if (isRecord(data)) {
|
|
const fields = Object.keys(data).slice(0, 30);
|
|
const sample = Object.fromEntries(
|
|
fields
|
|
.slice(0, config.MAX_PREVIEW_SAMPLE_ITEMS)
|
|
.map((field) => [field, data[field]]),
|
|
);
|
|
return {
|
|
count: fields.length,
|
|
fields,
|
|
sample,
|
|
summary: `object<${fields.length} fields>`,
|
|
};
|
|
}
|
|
return {
|
|
count: 1,
|
|
fields: [],
|
|
sample: String(data).slice(0, 300),
|
|
summary: `scalar<${typeof data}>`,
|
|
};
|
|
}
|
|
|
|
function estimateBytes(data: unknown) {
|
|
return Buffer.byteLength(JSON.stringify(data));
|
|
}
|
|
|
|
function normalizeResultRef(value?: string) {
|
|
return value && RESULT_REF_PATTERN.test(value) ? value : null;
|
|
}
|
|
|
|
function stringOrUndefined(value: unknown) {
|
|
return typeof value === "string" && value.trim().length > 0 ? value.trim() : undefined;
|
|
}
|
|
|
|
function numberOrZero(value: unknown) {
|
|
return typeof value === "number" && Number.isFinite(value) ? value : 0;
|
|
}
|
|
|
|
function positiveIntegerOrOne(value: unknown) {
|
|
return typeof value === "number" && Number.isInteger(value) && value > 0 ? value : 1;
|
|
}
|