lix/mnote
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(kernel): complete tree-first graph tasks 074-080

Browse Source
This commit is contained in:
lix-2026
2026-04-16 22:01:51 +08:00
parent 2ff10fa86c
commit b1d5d97142
65 changed files with 11579 additions and 4606 deletions
Download Patch File Download Diff File Expand all files Collapse all files
wolai-frontend/convex/crons.ts
+8 -1
View File
@@ -11,5 +11,12 @@ crons.weekly(
(internal as any).maintenance.cleanupWeekly,
);
export default crons;
// 每日触发一次 kernel-aware refresh 过渡链。
// 说明：当前仍复用 LightRAG 入库，但任务结果与语料口径已带上 kernel-aware 刷新语义。
crons.daily(
"kernel_aware_refresh_daily_transition",
{ hourUTC: 4, minuteUTC: 15 },
(internal as any).jobs.enqueueKernelAwareRefreshSweep,
);
export default crons;
wolai-frontend/convex/jobs.ts
+424
View File
@@ -7,6 +7,211 @@ import { extractTextFromDocumentContent, extractTextFromMindmapData } from "./_u
import { enqueueIngestDocumentJob, enqueueIngestMediaAssetJob, enqueueIngestMindmapJob } from "./_utils/ingestJobs";
import { extractTextFromAttachment } from "./_utils/attachmentExtract";
type KernelAwareRefreshTarget = {
documentIds: string[];
mindmapRefs: Array<{ docId: string; mindmapId: string }>;
assetIds: string[];
};
function uniqueNonEmptyStrings(values: Iterable<string | null | undefined>, limit: number): string[] {
const seen = new Set<string>();
const output: string[] = [];
for (const value of values) {
const normalized = String(value ?? "").trim();
if (!normalized || seen.has(normalized)) {
continue;
}
seen.add(normalized);
output.push(normalized);
if (output.length >= limit) {
break;
}
}
return output;
}
function truncateKernelText(value: string, limit = 800): string {
const normalized = String(value ?? "").replace(/\s+/g, " ").trim();
if (!normalized) {
return "";
}
return normalized.length > limit ? `${normalized.slice(0, Math.max(0, limit - 1))}…` : normalized;
}
function makeKernelAwareRefreshText(input: {
workspaceId: string;
documentRows: Array<{ id: string; title?: string | null; raw_text?: string | null; updated_at?: string | null }>;
mindmapRows: Array<{ document_id?: string | null; mindmap_id?: string | null; data?: unknown }>;
assetRows: Array<{ id: string; document_id?: string | null; file_name?: string | null; ocr_text?: string | null }>;
}): string {
const lines: string[] = [];
lines.push(`# kernel-aware refresh workspace ${input.workspaceId}`);
lines.push("");
for (const doc of input.documentRows) {
const title = String(doc.title ?? "").trim() || "无标题";
const updatedAt = String(doc.updated_at ?? "").trim() || "unknown";
const rawText = truncateKernelText(String(doc.raw_text ?? ""), 600);
lines.push(`## node:${doc.id}`);
lines.push(`title=${title}`);
lines.push(`subtreeRoot=${doc.id}`);
lines.push(`updatedAt=${updatedAt}`);
if (rawText) {
lines.push(`evidence=${rawText}`);
}
lines.push("");
}
for (const mindmap of input.mindmapRows) {
const docId = String(mindmap.document_id ?? "").trim();
const mindmapId = String(mindmap.mindmap_id ?? "").trim();
if (!docId || !mindmapId) {
continue;
}
const text = truncateKernelText(extractTextFromMindmapData(mindmap.data ?? null), 400);
if (!text) {
continue;
}
lines.push(`## subtree:${docId}:${mindmapId}`);
lines.push(`node=${docId}`);
lines.push(`subtreeRoot=${docId}`);
lines.push(`evidence=${text}`);
lines.push("");
}
for (const asset of input.assetRows) {
const assetId = String(asset.id ?? "").trim();
const docId = String(asset.document_id ?? "").trim();
if (!assetId || !docId) {
continue;
}
const title = String(asset.file_name ?? "").trim() || assetId;
const ocrText = truncateKernelText(String(asset.ocr_text ?? ""), 400);
if (!ocrText) {
continue;
}
lines.push(`## evidence:${assetId}`);
lines.push(`node=${docId}`);
lines.push(`subtreeRoot=${docId}`);
lines.push(`title=${title}`);
lines.push(`evidence=${ocrText}`);
lines.push("");
}
return lines.join("\n").trim();
}
async function selectKernelAwareRefreshTargets(ctx: any, args: { workspaceId: string; userId: string }): Promise<KernelAwareRefreshTarget> {
const [documents, mindmaps, assets] = await Promise.all([
ctx.db
.query("documents")
.withIndex("by_workspace", (q: any) => q.eq("workspace_id", args.workspaceId))
.collect(),
ctx.db
.query("mindmaps")
.withIndex("by_workspace", (q: any) => q.eq("workspace_id", args.workspaceId))
.collect(),
ctx.db
.query("media_assets")
.withIndex("by_workspace", (q: any) => q.eq("workspace_id", args.workspaceId))
.collect(),
]);
const aliveDocuments = documents
.filter((row: any) => row.deleted_at == null)
.filter((row: any) => String(row.user_id ?? "") === args.userId)
.sort((left: any, right: any) => String(right.updated_at ?? "").localeCompare(String(left.updated_at ?? "")));
const ownedMindmaps = mindmaps
.filter((row: any) => row.deleted_at == null)
.filter((row: any) => String(row.user_id ?? "") === args.userId)
.sort((left: any, right: any) => String(right.updated_at ?? "").localeCompare(String(left.updated_at ?? "")));
const aliveAssets = assets
.filter((row: any) => row.deleted_at == null && row.purged_at == null)
.sort((left: any, right: any) => String(right.updated_at ?? "").localeCompare(String(left.updated_at ?? "")));
const documentIds = uniqueNonEmptyStrings(aliveDocuments.map((row: any) => row.id), 12);
const selectedDocumentIds = new Set(documentIds);
const mindmapRefs = ownedMindmaps
.filter((row: any) => selectedDocumentIds.has(String(row.document_id ?? "").trim()))
.slice(0, 8)
.map((row: any) => ({
docId: String(row.document_id ?? "").trim(),
mindmapId: String(row.mindmap_id ?? "").trim(),
}))
.filter((row: { docId: string; mindmapId: string }) => row.docId && row.mindmapId);
const assetIds = uniqueNonEmptyStrings(
aliveAssets
.filter((row: any) => selectedDocumentIds.has(String(row.document_id ?? "").trim()))
.map((row: any) => row.id),
10,
);
return { documentIds, mindmapRefs, assetIds };
}
async function buildKernelAwareRefreshPayload(ctx: any, args: {
workspaceId: string;
userId: string;
target: KernelAwareRefreshTarget;
}) {
const documentRows = await Promise.all(
args.target.documentIds.map(async (documentId) => {
const [meta, contentRes] = await Promise.all([
ctx.runQuery(internal.documents.getMetaForIngest, { userId: args.userId, id: documentId }),
ctx.runQuery(internal.documents.getContentForIngest, { userId: args.userId, id: documentId }),
]);
if (!meta) {
return null;
}
return {
id: documentId,
title: meta.title ?? "无标题",
raw_text: extractTextFromDocumentContent(contentRes?.content ?? null),
updated_at: meta.updated_at ?? null,
};
}),
);
const mindmapRows = await Promise.all(
args.target.mindmapRefs.map(async (ref) => {
const result = await ctx.runQuery(internal.mindmaps.getForIngest, {
userId: args.userId,
docId: ref.docId,
mindmapId: ref.mindmapId,
});
if (!result?.ok || !result.meta?.exists) {
return null;
}
return {
document_id: ref.docId,
mindmap_id: ref.mindmapId,
data: result.data ?? null,
};
}),
);
const assetRows = await Promise.all(
args.target.assetIds.map(async (assetId) => {
const asset = await ctx.runQuery(api.mediaAssets.getById, { userId: args.userId, id: assetId });
if (!asset) {
return null;
}
return {
id: assetId,
document_id: asset.document_id ?? null,
file_name: asset.file_name ?? null,
ocr_text: asset.ocr_text ?? null,
};
}),
);
return {
documentRows: documentRows.filter((item): item is NonNullable<typeof item> => Boolean(item)),
mindmapRows: mindmapRows.filter((item): item is NonNullable<typeof item> => Boolean(item)),
assetRows: assetRows.filter((item): item is NonNullable<typeof item> => Boolean(item)),
};
}
export const get = query({
args: { userId: v.string(), id: v.string() },
handler: async (ctx, args) => {
@@ -85,6 +290,133 @@ export const enqueueRagIndexMediaAsset = mutation({
},
});
export const enqueueKernelAwareRefresh = mutation({
args: { userId: v.string(), workspaceId: v.string() },
handler: async (ctx, args) => {
const workspaceId = String(args.workspaceId ?? "").trim();
if (!workspaceId) {
throw new Error("缺少 workspaceId");
}
const membership = await ctx.db
.query("workspace_members")
.withIndex("by_workspace_user", (q) => q.eq("workspace_id", workspaceId).eq("user_id", args.userId))
.first();
if (!membership) {
throw new Error("无权访问该工作空间");
}
const ts = nowIso();
const id = `refresh:kernel-aware:${workspaceId}`;
const payload = {
workspaceId,
enqueuedAt: ts,
trigger: "manual",
};
const existing = await ctx.db
.query("jobs")
.withIndex("by_job_id", (q) => q.eq("id", id))
.first();
if (existing) {
await ctx.db.patch(existing._id, {
user_id: args.userId,
type: "refresh.kernel_aware_transition",
status: "queued",
payload,
result: null,
error: null,
updated_at: ts,
started_at: null,
finished_at: null,
});
} else {
await ctx.db.insert("jobs", {
id,
user_id: args.userId,
type: "refresh.kernel_aware_transition",
status: "queued",
payload,
result: null,
error: null,
created_at: ts,
updated_at: ts,
started_at: null,
finished_at: null,
});
}
await ctx.scheduler.runAfter(0, internal.jobs.start, { id });
return { ok: true, id };
},
});
export const enqueueKernelAwareRefreshSweep = internalMutation({
args: {},
handler: async (ctx) => {
const memberships = await ctx.db.query("workspace_members").collect();
const ownersByWorkspace = new Map<string, string>();
for (const membership of memberships) {
if (membership.role !== "owner") {
continue;
}
if (!ownersByWorkspace.has(membership.workspace_id)) {
ownersByWorkspace.set(membership.workspace_id, membership.user_id);
}
}
const scheduled: string[] = [];
for (const [workspaceId, userId] of ownersByWorkspace.entries()) {
const id = `refresh:kernel-aware:${workspaceId}`;
const ts = nowIso();
const payload = {
workspaceId,
enqueuedAt: ts,
trigger: "cron",
};
const existing = await ctx.db
.query("jobs")
.withIndex("by_job_id", (q) => q.eq("id", id))
.first();
if (existing && (existing.status === "queued" || existing.status === "running")) {
scheduled.push(id);
continue;
}
if (existing) {
await ctx.db.patch(existing._id, {
user_id: userId,
type: "refresh.kernel_aware_transition",
status: "queued",
payload,
result: null,
error: null,
updated_at: ts,
started_at: null,
finished_at: null,
});
} else {
await ctx.db.insert("jobs", {
id,
user_id: userId,
type: "refresh.kernel_aware_transition",
status: "queued",
payload,
result: null,
error: null,
created_at: ts,
updated_at: ts,
started_at: null,
finished_at: null,
});
}
await ctx.scheduler.runAfter(0, internal.jobs.start, { id });
scheduled.push(id);
}
return {
ok: true,
scheduledCount: scheduled.length,
jobIds: scheduled,
};
},
});
export const start = internalMutation({
args: { id: v.string() },
handler: async (ctx, args) => {
@@ -320,6 +652,81 @@ export const run = internalAction({
return;
}
if (job.type === "refresh.kernel_aware_transition") {
const workspaceId = String(job.payload?.workspaceId ?? "").trim();
if (!workspaceId) throw new Error("缺少 workspaceId");
const membership = await ctx.runQuery((internal as any).jobs._getWorkspaceMembership, {
workspaceId,
userId: job.user_id,
});
if (!membership) {
throw new Error("工作空间不存在或无权限");
}
const target = await ctx.runQuery((internal as any).jobs._selectKernelAwareRefreshTargets, {
workspaceId,
userId: job.user_id,
});
const payload = await buildKernelAwareRefreshPayload(ctx, {
workspaceId,
userId: job.user_id,
target,
});
const text = makeKernelAwareRefreshText({
workspaceId,
documentRows: payload.documentRows,
mindmapRows: payload.mindmapRows,
assetRows: payload.assetRows,
});
const ingest = await lightragIngestText({
fileSource: `kernel-aware-refresh:${workspaceId}`,
text,
});
await Promise.all([
...target.documentIds.slice(0, 6).map((documentId: string) =>
ctx.runMutation(api.jobs.enqueueRagIndexDocument, { userId: job.user_id, documentId }).catch(() => null),
),
...target.mindmapRefs.slice(0, 4).map((item: { docId: string; mindmapId: string }) =>
ctx.runMutation(api.jobs.enqueueRagIndexMindmap, {
userId: job.user_id,
docId: item.docId,
mindmapId: item.mindmapId,
}).catch(() => null),
),
...target.assetIds.slice(0, 4).map((assetId: string) =>
ctx.runMutation(api.jobs.enqueueRagIndexMediaAsset, { userId: job.user_id, assetId }).catch(() => null),
),
]);
await ctx.runMutation(internal.jobs.finishSuccess, {
id: args.id,
result: {
ok: true,
kind: "kernel_aware_refresh",
workspaceId,
refreshMode: "kernel_aware_transition",
bridge: {
backend: "lightrag",
fileSource: `kernel-aware-refresh:${workspaceId}`,
skipped: Boolean(ingest.skipped),
reason: ingest.reason ?? null,
trackId: ingest.trackId,
},
refreshedDocuments: target.documentIds.length,
refreshedMindmaps: target.mindmapRefs.length,
refreshedAssets: target.assetIds.length,
kernelPreview: {
nodeIds: payload.documentRows.map((item) => item.id).slice(0, 8),
subtreeRootIds: payload.documentRows.map((item) => item.id).slice(0, 8),
evidenceAssetIds: payload.assetRows.map((item) => item.id).slice(0, 8),
},
},
});
return;
}
throw new Error(`未知任务类型：${job.type}`);
} catch (err) {
const message = err instanceof Error ? err.message : String(err);
@@ -346,6 +753,23 @@ export const _getInternal = internalQuery({
},
});
export const _getWorkspaceMembership = internalQuery({
args: { workspaceId: v.string(), userId: v.string() },
handler: async (ctx, args) => {
return await ctx.db
.query("workspace_members")
.withIndex("by_workspace_user", (q) => q.eq("workspace_id", args.workspaceId).eq("user_id", args.userId))
.first();
},
});
export const _selectKernelAwareRefreshTargets = internalQuery({
args: { workspaceId: v.string(), userId: v.string() },
handler: async (ctx, args) => {
return await selectKernelAwareRefreshTargets(ctx, args);
},
});
export const finishSuccess = internalMutation({
args: { id: v.string(), result: v.any() },
handler: async (ctx, args) => {