From 1f811f3c07d14a27fbfe8b23d997525a1dce9d63 Mon Sep 17 00:00:00 2001 From: Codex Date: Wed, 13 May 2026 09:47:47 +0000 Subject: [PATCH] fix: load code queue references from database - resolve referenced tasks through the read path instead of hot in-memory state\n- make reference injection async so it can query persisted tasks and build transitive graphs\n- keep reference injection self-test coverage aligned with async lookup --- .../microservices/code-queue/src/index.ts | 16 ++++++------ .../code-queue/src/references.ts | 26 ++++++++++++++----- .../code-queue/src/self-tests.ts | 10 +++---- 3 files changed, 33 insertions(+), 19 deletions(-) diff --git a/src/components/microservices/code-queue/src/index.ts b/src/components/microservices/code-queue/src/index.ts index c94bba89..ae30d160 100644 --- a/src/components/microservices/code-queue/src/index.ts +++ b/src/components/microservices/code-queue/src/index.ts @@ -1648,8 +1648,8 @@ function normalizePromptEditRequest(task: QueueTask, value: unknown): QueueTaskR }; } -function buildQueuedPromptUpdate(task: QueueTask, body: unknown): QueueTaskRequest { - return injectCodeQueueEnvironmentHint(injectReferencedTaskContext(normalizePromptEditRequest(task, body))); +async function buildQueuedPromptUpdate(task: QueueTask, body: unknown): Promise { + return injectCodeQueueEnvironmentHint(await injectReferencedTaskContext(normalizePromptEditRequest(task, body))); } function rewriteEnqueueOutput(task: QueueTask): void { @@ -1872,7 +1872,7 @@ configureQueueApi({ configureReferences({ addUniqueTaskId, - findTask, + findTask: findTaskForRead, nowIso, referenceInjectionMaxRounds, referenceTaskIdsFromPrompt, @@ -2730,11 +2730,11 @@ async function createTasks(req: Request): Promise { const records = typeof body === "object" && body !== null && !Array.isArray(body) && Array.isArray((body as Record).tasks) ? (body as Record).tasks as unknown[] : [body]; - const tasks = records.map((record) => { + const tasks = await Promise.all(records.map(async (record) => { const normalized = normalizeRequest(record); if (normalized.queueId === undefined && batchQueueId !== undefined) normalized.queueId = batchQueueId; - return createTask(injectCodeQueueEnvironmentHint(injectReferencedTaskContext(normalized))); - }); + return createTask(injectCodeQueueEnvironmentHint(await injectReferencedTaskContext(normalized))); + })); for (const task of tasks) appendOutput(task, "user", `${task.prompt}\n`, "enqueue"); state.tasks.push(...tasks); if (tasks.length > 0) armIdleNotification(); @@ -2771,7 +2771,7 @@ async function editQueuedTaskPrompt(task: QueueTask, req: Request): Promise { if (url.pathname === "/api/judge/probe" && (req.method === "GET" || req.method === "POST")) return await runJudgeProbe(); if (url.pathname === "/api/judge/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runJudgeInfraSelfTest()); if (url.pathname === "/api/queue-order/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runQueueOrderingSelfTest()); - if (url.pathname === "/api/reference-injection/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runReferenceInjectionSelfTest()); + if (url.pathname === "/api/reference-injection/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(await runReferenceInjectionSelfTest()); if (url.pathname === "/api/trace-port/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runTracePortSelfTest()); if (url.pathname === "/api/notifications/claudeqq" && req.method === "GET") { await loadClaudeQqNotificationOutboxFromDatabase(); diff --git a/src/components/microservices/code-queue/src/references.ts b/src/components/microservices/code-queue/src/references.ts index e68aa4bd..dd2229ca 100644 --- a/src/components/microservices/code-queue/src/references.ts +++ b/src/components/microservices/code-queue/src/references.ts @@ -4,6 +4,8 @@ import { lastAssistantMessage } from "./task-view"; import { resolvedReferenceContextTitle, userPromptForDisplay } from "./prompts"; import type { QueueTask, QueueTaskRequest, ReferenceInjectionRecord, ReferenceInjectionSummaryItem } from "./types"; +type TaskFinder = (id: string) => QueueTask | null | Promise; + export class ReferenceTaskLookupError extends Error { missingIds: string[]; @@ -16,7 +18,7 @@ export class ReferenceTaskLookupError extends Error { export interface ReferencesContext { addUniqueTaskId: (ids: string[], value: string) => void; - findTask: (id: string) => QueueTask | null; + findTask: TaskFinder; nowIso: () => string; referenceInjectionMaxRounds: number | null; referenceTaskIdsFromPrompt: (prompt: string) => string[]; @@ -67,7 +69,18 @@ function referenceSummaryItem(task: QueueTask, round: number, roundIndex: number }; } -function collectReferenceGraph(rootIds: string[], maxRounds: number | null, finder: (id: string) => QueueTask | null = ctx().findTask): { items: ReferenceInjectionSummaryItem[]; tasks: QueueTask[]; truncated: boolean } { +function cachedTaskFinder(finder: TaskFinder): (id: string) => Promise { + const cache = new Map>(); + return (id: string) => { + const existing = cache.get(id); + if (existing !== undefined) return existing; + const next = Promise.resolve(finder(id)); + cache.set(id, next); + return next; + }; +} + +async function collectReferenceGraph(rootIds: string[], maxRounds: number | null, finder: (id: string) => Promise): Promise<{ items: ReferenceInjectionSummaryItem[]; tasks: QueueTask[]; truncated: boolean }> { const seen = new Set(); let frontier = rootIds.map((id) => ({ id, viaTaskId: null as string | null })); const rawItems: Array<{ task: QueueTask; depth: number; viaTaskId: string | null; discoveryIndex: number }> = []; @@ -81,7 +94,7 @@ function collectReferenceGraph(rootIds: string[], maxRounds: number | null, find const next: Array<{ id: string; viaTaskId: string | null }> = []; for (const entry of frontier) { if (seen.has(entry.id)) continue; - const task = finder(entry.id); + const task = await finder(entry.id); if (task === null) continue; seen.add(entry.id); discoveryIndex += 1; @@ -146,15 +159,16 @@ function referencedTaskContext(task: QueueTask, summary: ReferenceInjectionSumma ].join("\n"); } -function injectReferencedTaskContext(request: QueueTaskRequest, finder: (id: string) => QueueTask | null = ctx().findTask, injectedAt = ctx().nowIso()): QueueTaskRequest { +async function injectReferencedTaskContext(request: QueueTaskRequest, finder: TaskFinder = ctx().findTask, injectedAt = ctx().nowIso()): Promise { const ids = request.referenceTaskIds ?? []; if (ids.length === 0 || request.prompt.includes(resolvedReferenceContextTitle)) return request; if (ids.length > 5) throw new Error(`referenceTaskIds supports at most 5 task ids, got ${ids.length}`); - const referencedTasks = ids.map((id) => finder(id)); + const cachedFinder = cachedTaskFinder(finder); + const referencedTasks = await Promise.all(ids.map((id) => cachedFinder(id))); const missing = ids.filter((_id, index) => referencedTasks[index] === null); if (missing.length > 0) throw new ReferenceTaskLookupError(missing); const userPrompt = request.basePrompt ?? userPromptForDisplay(request.prompt); - const graph = collectReferenceGraph(ids, ctx().referenceInjectionMaxRounds, finder); + const graph = await collectReferenceGraph(ids, ctx().referenceInjectionMaxRounds, cachedFinder); const injection: ReferenceInjectionRecord = { version: 2, injectedAt, diff --git a/src/components/microservices/code-queue/src/self-tests.ts b/src/components/microservices/code-queue/src/self-tests.ts index 0251e77c..94fe8207 100644 --- a/src/components/microservices/code-queue/src/self-tests.ts +++ b/src/components/microservices/code-queue/src/self-tests.ts @@ -15,7 +15,7 @@ export interface SelfTestsContext { availableQueueStartSlotsFor: (activeSlotCount: number, maxActiveQueues?: number) => number; defaultQueueId: string; enqueueActiveRunSlotWaiter: (task: QueueTask) => ActiveRunSlotWaiter; - injectReferencedTaskContext: (request: QueueTaskRequest, finder?: (id: string) => QueueTask | null, injectedAt?: string) => QueueTaskRequest; + injectReferencedTaskContext: (request: QueueTaskRequest, finder?: (id: string) => QueueTask | null | Promise, injectedAt?: string) => Promise; nextRunnableTaskFrom: (queueId: string, tasks?: QueueTask[]) => QueueTask | null; normalizeTask: (task: QueueTask) => QueueTask; nowIso: () => string; @@ -84,17 +84,17 @@ function assertReferenceTest(condition: boolean, message: string): void { if (!condition) throw new Error(message); } -function runReferenceInjectionSelfTest(): JsonValue { +async function runReferenceInjectionSelfTest(): Promise { const at = "2026-05-08T00:00:00.000Z"; const taskA = testTask("codex_1000_aaaaaa", "A base prompt", "A final", [], at); - const injectedB = ctx().injectReferencedTaskContext({ + const injectedB = await ctx().injectReferencedTaskContext({ prompt: "引用 Code Queue 任务 codex_1000_aaaaaa。\n\n本次任务：\nB user prompt", referenceTaskIds: [taskA.id], }, (id) => id === taskA.id ? taskA : null, "2026-05-08T00:01:00.000Z"); const taskB = testTask("codex_1001_bbbbbb", injectedB.prompt, "B final", injectedB.referenceTaskIds, "2026-05-08T00:02:00.000Z"); taskB.basePrompt = injectedB.basePrompt ?? ""; taskB.referenceInjection = injectedB.referenceInjection ?? null; - const injectedC = ctx().injectReferencedTaskContext({ + const injectedC = await ctx().injectReferencedTaskContext({ prompt: "C user prompt", referenceTaskIds: [taskB.id], }, (id) => id === taskA.id ? taskA : id === taskB.id ? taskB : null, "2026-05-08T00:03:00.000Z"); @@ -126,7 +126,7 @@ function runReferenceInjectionSelfTest(): JsonValue { deepTasks.push(testTask(id, `deep prompt ${index}`, `deep final ${index}`, parent === undefined ? [] : [parent.id], `2026-05-08T00:${String(10 + index).padStart(2, "0")}:00.000Z`)); } const deepById = new Map(deepTasks.map((task) => [task.id, task])); - const injectedDeep = ctx().injectReferencedTaskContext({ + const injectedDeep = await ctx().injectReferencedTaskContext({ prompt: "Deep user prompt", referenceTaskIds: [deepTasks.at(-1)?.id ?? ""], }, (id) => deepById.get(id) ?? null, "2026-05-08T00:30:00.000Z");