mirrors/pikasTech-unidesk
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: load code queue references from database

Browse Source 
- 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
This commit is contained in:
Codex
2026-05-13 09:47:47 +00:00
parent b8417ee18a
commit 1f811f3c07
3 changed files with 33 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue/src/index.ts
+8 -8
View File
@@ -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<QueueTaskRequest> {
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<Response> {
const records = typeof body === "object" && body !== null && !Array.isArray(body) && Array.isArray((body as Record<string, unknown>).tasks)
? (body as Record<string, unknown>).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<Resp
}
let update: QueueTaskRequest;
try {
update = buildQueuedPromptUpdate(task, await readJson(req));
update = await buildQueuedPromptUpdate(task, await readJson(req));
} catch (error) {
return jsonResponse({ ok: false, error: error instanceof Error ? error.message : String(error) }, 400);
}
@@ -3067,7 +3067,7 @@ async function route(req: Request): Promise<Response> {
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();
src/components/microservices/code-queue/src/references.ts
+20 -6
View File
@@ -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<QueueTask | null>;
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<QueueTask | null> {
const cache = new Map<string, Promise<QueueTask | null>>();
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<QueueTask | null>): Promise<{ items: ReferenceInjectionSummaryItem[]; tasks: QueueTask[]; truncated: boolean }> {
const seen = new Set<string>();
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<QueueTaskRequest> {
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,
src/components/microservices/code-queue/src/self-tests.ts
+5 -5
View File
@@ -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<QueueTask | null>, injectedAt?: string) => Promise<QueueTaskRequest>;
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<JsonValue> {
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");