import { mutation, query } from "./_generated/server"; import { v } from "convex/values"; const runtimeRunArgs = { schema: v.literal("mnote.acp_runtime_run.v1"), source: v.literal("acp"), userId: v.string(), workspaceId: v.optional(v.union(v.string(), v.null())), documentId: v.string(), sessionId: v.string(), runId: v.string(), title: v.optional(v.union(v.string(), v.null())), profile: v.string(), acpRuntime: v.string(), traceId: v.string(), status: v.string(), runtime: v.any(), payload: v.any(), retention: v.any(), createdAt: v.number(), updatedAt: v.number(), }; const runtimeEventArgs = { schema: v.literal("mnote.acp_runtime_event.v1"), source: v.literal("acp"), eventId: v.string(), userId: v.string(), workspaceId: v.optional(v.union(v.string(), v.null())), documentId: v.string(), sessionId: v.string(), runId: v.string(), profile: v.string(), acpRuntime: v.string(), eventType: v.string(), payload: v.any(), createdAt: v.number(), }; function isoFromMillis(value: number) { return new Date(value).toISOString(); } async function requireScopedUser(ctx: any, expectedUserId: string) { const identity = await ctx.auth.getUserIdentity(); const subject = String(identity?.subject ?? ""); const identityUserId = subject.split("|")[0] || ""; if (identityUserId && identityUserId !== expectedUserId) { throw new Error("无权限"); } if (!identityUserId && !expectedUserId.trim()) { throw new Error("未登录"); } return expectedUserId; } function rowToClient(row: any) { return { id: row.id, userId: row.user_id, workspaceId: row.workspace_id, documentId: row.document_id, sessionId: row.session_id, runId: row.run_id, profile: row.profile, title: row.title, acpRuntime: row.acp_runtime, source: row.source, status: row.status, traceId: row.trace_id, runtime: row.runtime, usage: row.usage ?? null, payload: row.payload, retention: row.retention, createdAt: row.created_at, updatedAt: row.updated_at, }; } export const upsertRuntimeRun = mutation({ args: runtimeRunArgs, handler: async (ctx, args) => { const userId = await requireScopedUser(ctx, args.userId); const workspaceId = args.workspaceId ?? null; const nowIso = isoFromMillis(args.updatedAt); const existing = await ctx.db .query("acp_runtime_runs") .withIndex("by_run_id", (q: any) => q.eq("run_id", args.runId)) .first(); const patch = { id: `acp_run:${args.runId}`, user_id: userId, workspace_id: workspaceId, document_id: args.documentId, session_id: args.sessionId, run_id: args.runId, title: args.title ?? null, profile: args.profile, acp_runtime: args.acpRuntime, source: args.source, status: args.status, trace_id: args.traceId, runtime: args.runtime, usage: null, payload: args.payload, retention: args.retention, updated_at: nowIso, deleted_at: null, }; if (existing) { if (String(existing.user_id) !== userId) { throw new Error("无权限"); } await ctx.db.patch(existing._id, patch); return { ok: true, id: existing.id, runId: args.runId, updated: true }; } await ctx.db.insert("acp_runtime_runs", { ...patch, created_at: isoFromMillis(args.createdAt), }); return { ok: true, id: patch.id, runId: args.runId, created: true }; }, }); export const getRuntimeRun = query({ args: { userId: v.string(), runId: v.string(), }, handler: async (ctx, args) => { const userId = await requireScopedUser(ctx, args.userId); const row = await ctx.db .query("acp_runtime_runs") .withIndex("by_run_id", (q: any) => q.eq("run_id", args.runId)) .first(); if (!row || row.deleted_at !== null || String(row.user_id) !== userId) { return null; } return rowToClient(row); }, }); export const listRuntimeRuns = query({ args: { userId: v.string(), workspaceId: v.optional(v.union(v.string(), v.null())), documentId: v.optional(v.union(v.string(), v.null())), sessionId: v.optional(v.union(v.string(), v.null())), limit: v.optional(v.number()), }, handler: async (ctx, args) => { const userId = await requireScopedUser(ctx, args.userId); const workspaceId = args.workspaceId ?? null; const limit = Math.min(Math.max(args.limit ?? 50, 1), 100); let rows; if (args.sessionId) { rows = await ctx.db .query("acp_runtime_runs") .withIndex("by_user_workspace_session", (q: any) => q.eq("user_id", userId).eq("workspace_id", workspaceId).eq("session_id", args.sessionId), ) .order("desc") .take(limit); } else if (args.documentId) { rows = await ctx.db .query("acp_runtime_runs") .withIndex("by_user_workspace_document", (q: any) => q.eq("user_id", userId).eq("workspace_id", workspaceId).eq("document_id", args.documentId), ) .order("desc") .take(limit); } else { rows = await ctx.db .query("acp_runtime_runs") .withIndex("by_user_created_at", (q: any) => q.eq("user_id", userId)) .order("desc") .take(limit); } return rows .filter((row: any) => row.deleted_at === null) .map(rowToClient); }, }); export const appendRuntimeEvent = mutation({ args: runtimeEventArgs, handler: async (ctx, args) => { const userId = await requireScopedUser(ctx, args.userId); const workspaceId = args.workspaceId ?? null; const existingRun = await ctx.db .query("acp_runtime_runs") .withIndex("by_run_id", (q: any) => q.eq("run_id", args.runId)) .first(); if (existingRun && String(existingRun.user_id) !== userId) { throw new Error("无权限"); } await ctx.db.insert("acp_runtime_events", { id: args.eventId, user_id: userId, workspace_id: workspaceId, document_id: args.documentId, session_id: args.sessionId, run_id: args.runId, profile: args.profile, acp_runtime: args.acpRuntime, source: args.source, event_type: args.eventType, payload: args.payload, created_at: isoFromMillis(args.createdAt), }); const usage = usageFromRuntimeEvent(args.eventType, args.payload); if (existingRun && usage) { await ctx.db.patch(existingRun._id, { usage, updated_at: isoFromMillis(args.createdAt), }); } return { ok: true, id: args.eventId, runId: args.runId }; }, }); function usageFromRuntimeEvent(eventType: string, payload: any) { if (eventType === "usage.updated") { return { source: "usage_update", contextUsed: Number(payload?.used ?? payload?.contextUsed ?? 0), contextSize: Number(payload?.size ?? payload?.contextSize ?? 0), }; } if (eventType === "run.completed" && payload?.usage && typeof payload.usage === "object") { return { source: "run_completed", ...payload.usage, }; } return null; } export const listRuntimeEvents = query({ args: { userId: v.string(), runId: v.string(), limit: v.optional(v.number()), }, handler: async (ctx, args) => { const userId = await requireScopedUser(ctx, args.userId); const limit = Math.min(Math.max(args.limit ?? 100, 1), 500); const rows = await ctx.db .query("acp_runtime_events") .withIndex("by_run_created_at", (q: any) => q.eq("run_id", args.runId)) .order("asc") .take(limit); return rows .filter((row: any) => String(row.user_id) === userId) .map((row: any) => ({ id: row.id, userId: row.user_id, workspaceId: row.workspace_id, documentId: row.document_id, sessionId: row.session_id, runId: row.run_id, profile: row.profile, acpRuntime: row.acp_runtime, source: row.source, eventType: row.event_type, payload: row.payload, createdAt: row.created_at, })); }, }); function titleFromRun(row: any) { const payload = row?.payload ?? {}; const raw = String(payload.message ?? payload.input ?? payload.title ?? "").trim(); if (!raw) return "当前页问答"; return raw.length > 32 ? `${raw.slice(0, 32)}...` : raw; } async function sessionRows(ctx: any, userId: string, workspaceId: string | null, sessionId: string) { return await ctx.db .query("acp_runtime_runs") .withIndex("by_user_workspace_session", (q: any) => q.eq("user_id", userId).eq("workspace_id", workspaceId).eq("session_id", sessionId), ) .collect(); } export const renameRuntimeSession = mutation({ args: { userId: v.string(), workspaceId: v.optional(v.union(v.string(), v.null())), sessionId: v.string(), title: v.string(), }, handler: async (ctx, args) => { const userId = await requireScopedUser(ctx, args.userId); const workspaceId = args.workspaceId ?? null; const title = args.title.trim(); if (!title) throw new Error("标题不能为空"); const rows = await sessionRows(ctx, userId, workspaceId, args.sessionId); for (const row of rows) { if (row.deleted_at === null) { await ctx.db.patch(row._id, { title, updated_at: new Date().toISOString(), }); } } return { ok: true, sessionId: args.sessionId, title, updated: rows.length }; }, }); export const autoTitleRuntimeSession = mutation({ args: { userId: v.string(), workspaceId: v.optional(v.union(v.string(), v.null())), sessionId: v.string(), }, handler: async (ctx, args) => { const userId = await requireScopedUser(ctx, args.userId); const workspaceId = args.workspaceId ?? null; const rows = await sessionRows(ctx, userId, workspaceId, args.sessionId); const title = titleFromRun(rows.find((row: any) => row.deleted_at === null)); for (const row of rows) { if (row.deleted_at === null) { await ctx.db.patch(row._id, { title, updated_at: new Date().toISOString(), }); } } return { ok: true, sessionId: args.sessionId, title, updated: rows.length }; }, }); export const deleteRuntimeSession = mutation({ args: { userId: v.string(), workspaceId: v.optional(v.union(v.string(), v.null())), sessionId: v.string(), }, handler: async (ctx, args) => { const userId = await requireScopedUser(ctx, args.userId); const workspaceId = args.workspaceId ?? null; const rows = await sessionRows(ctx, userId, workspaceId, args.sessionId); const deletedAt = new Date().toISOString(); for (const row of rows) { if (row.deleted_at === null) { await ctx.db.patch(row._id, { deleted_at: deletedAt, updated_at: deletedAt, }); } } return { ok: true, sessionId: args.sessionId, deleted: rows.length }; }, }); function searchText(row: any) { const payload = row?.payload ?? {}; return [ row.title, row.session_id, row.run_id, row.profile, row.status, payload.message, payload.input, payload.title, ] .map((value) => String(value ?? "")) .join(" ") .toLowerCase(); } function snippetFor(row: any, q: string) { const payload = row?.payload ?? {}; const raw = String(row.title ?? payload.message ?? payload.input ?? row.session_id ?? "").trim(); if (!raw) return ""; const index = raw.toLowerCase().indexOf(q.toLowerCase()); if (index < 0) return raw.length > 80 ? `${raw.slice(0, 80)}...` : raw; const start = Math.max(0, index - 24); const end = Math.min(raw.length, index + q.length + 56); return `${start > 0 ? "..." : ""}${raw.slice(start, end)}${end < raw.length ? "..." : ""}`; } export const searchRuntimeSessions = query({ args: { userId: v.string(), workspaceId: v.optional(v.union(v.string(), v.null())), q: v.string(), limit: v.optional(v.number()), }, handler: async (ctx, args) => { const userId = await requireScopedUser(ctx, args.userId); const workspaceId = args.workspaceId ?? null; const q = args.q.trim().toLowerCase(); if (!q) return []; const limit = Math.min(Math.max(args.limit ?? 20, 1), 50); const rows = await ctx.db .query("acp_runtime_runs") .withIndex("by_user_created_at", (query: any) => query.eq("user_id", userId)) .order("desc") .take(500); const seen = new Set(); const results = []; for (const row of rows) { if (row.deleted_at !== null) continue; if ((row.workspace_id ?? null) !== workspaceId) continue; if (!searchText(row).includes(q)) continue; if (seen.has(row.session_id)) continue; seen.add(row.session_id); results.push({ sessionId: row.session_id, runId: row.run_id, workspaceId: row.workspace_id, documentId: row.document_id, title: row.title ?? titleFromRun(row), profile: row.profile, status: row.status, snippet: snippetFor(row, args.q), score: 1, }); if (results.length >= limit) break; } return results; }, });