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

fix: guard code queue manager moves

Browse Source
This commit is contained in:
Codex
2026-05-17 11:48:07 +00:00
parent a24f8b9968
commit 00e5766794
3 changed files with 405 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue-mgr/src/index.ts
+351 -35
View File
@@ -205,6 +205,21 @@ interface TaskRow {
task_json: unknown;
}
type TaskStatusRow = Pick<TaskRow, "id" | "queue_id" | "status" | "started_at" | "current_attempt" | "codex_thread_id" | "active_turn_id">;
type TaskJsonRow = Pick<TaskRow, "task_json"> & Partial<Pick<TaskRow,
"id"
| "queue_id"
| "status"
| "current_attempt"
| "current_mode"
| "codex_thread_id"
| "active_turn_id"
| "updated_at"
| "started_at"
| "finished_at"
| "read_at"
>>;
interface QueueRow {
id: string;
name: string;
@@ -988,8 +1003,58 @@ function queueIdOf(task: QueueTask): string {
return safeQueueId(task.queueId);
}
function rowToTask(row: Pick<TaskRow, "task_json">): QueueTask {
return normalizeTask(row.task_json);
function rowToTask(row: TaskJsonRow): QueueTask {
const task = normalizeTask(row.task_json);
if (typeof row.id === "string" && row.id.length > 0) task.id = row.id;
if (typeof row.queue_id === "string" && row.queue_id.length > 0) task.queueId = safeQueueId(row.queue_id);
if (typeof row.status === "string" && row.status.length > 0) task.status = row.status as TaskStatus;
if (row.current_attempt !== undefined) task.currentAttempt = Number(row.current_attempt ?? 0);
if (row.current_mode !== undefined) task.currentMode = row.current_mode ?? null;
if (row.codex_thread_id !== undefined) task.codexThreadId = row.codex_thread_id ?? null;
if (row.active_turn_id !== undefined) task.activeTurnId = row.active_turn_id ?? null;
if (row.updated_at !== undefined) task.updatedAt = timestampToIso(row.updated_at) ?? task.updatedAt;
if (row.started_at !== undefined) task.startedAt = timestampToIso(row.started_at);
if (row.finished_at !== undefined) task.finishedAt = timestampToIso(row.finished_at);
if (row.read_at !== undefined) task.readAt = timestampToIso(row.read_at);
return task;
}
function taskStatusRowJson(row: TaskStatusRow | null): JsonValue {
if (row === null) return null;
return {
id: row.id,
queueId: safeQueueId(row.queue_id),
status: row.status,
startedAt: timestampToIso(row.started_at),
currentAttempt: Number(row.current_attempt ?? 0),
codexThreadId: row.codex_thread_id,
activeTurnId: row.active_turn_id,
};
}
function taskIsUnclaimedMovable(task: QueueTask): boolean {
return (task.status === "queued" || task.status === "retry_wait")
&& task.startedAt === null
&& task.currentAttempt === 0
&& task.codexThreadId === null
&& task.activeTurnId === null;
}
function taskStatusMoveBlocker(row: TaskStatusRow | null): string {
if (row === null) return "task not found";
if (row.status !== "queued" && row.status !== "retry_wait") return `status=${row.status}`;
if (row.started_at !== null) return "task already has started_at";
const currentAttempt = Number(row.current_attempt ?? 0);
if (currentAttempt !== 0) return `task already has current_attempt=${currentAttempt}`;
if (row.codex_thread_id !== null) return "task already has codex_thread_id";
if (row.active_turn_id !== null) return "task already has active_turn_id";
return "";
}
function queueMergeBlocker(row: TaskStatusRow): string {
if (row.status === "running" || row.status === "judging") return `task ${row.id} is ${row.status}`;
if ((row.status === "queued" || row.status === "retry_wait") && taskStatusMoveBlocker(row).length > 0) return `task ${row.id} has already been claimed`;
return "";
}
function queueRowToRecord(row: QueueRow): QueueRecord {
@@ -1086,6 +1151,7 @@ async function upsertQueue(queue: QueueRecord): Promise<void> {
}
async function upsertTask(client: SqlExecutor, task: QueueTask): Promise<void> {
const taskHasClaim = hasClaimMarkers(task);
await client`
INSERT INTO unidesk_code_queue_tasks (
id, queue_id, status, provider_id, execution_mode, model, cwd, prompt, base_prompt,
@@ -1132,6 +1198,26 @@ async function upsertTask(client: SqlExecutor, task: QueueTask): Promise<void> {
last_output_seq = EXCLUDED.last_output_seq,
task_json = EXCLUDED.task_json
WHERE unidesk_code_queue_tasks.status NOT IN ('running', 'judging')
AND (
${taskHasClaim}
OR unidesk_code_queue_tasks.started_at IS NULL
OR unidesk_code_queue_tasks.started_at = EXCLUDED.started_at
)
AND (
${taskHasClaim}
OR unidesk_code_queue_tasks.current_attempt = 0
OR unidesk_code_queue_tasks.current_attempt = EXCLUDED.current_attempt
)
AND (
${taskHasClaim}
OR unidesk_code_queue_tasks.codex_thread_id IS NULL
OR unidesk_code_queue_tasks.codex_thread_id = EXCLUDED.codex_thread_id
)
AND (
${taskHasClaim}
OR unidesk_code_queue_tasks.active_turn_id IS NULL
OR unidesk_code_queue_tasks.active_turn_id = EXCLUDED.active_turn_id
)
`;
}
@@ -1150,8 +1236,20 @@ async function loadQueues(): Promise<QueueRecord[]> {
}
async function loadTask(taskId: string): Promise<QueueTask | null> {
const rows = await traceSql<Array<Pick<TaskRow, "task_json">>>`
SELECT task_json
const rows = await traceSql<TaskJsonRow[]>`
SELECT
id,
queue_id,
status,
current_attempt,
current_mode,
codex_thread_id,
active_turn_id,
updated_at,
started_at,
finished_at,
read_at,
task_json
FROM unidesk_code_queue_tasks
WHERE id = ${taskId}
LIMIT 1
@@ -1166,8 +1264,21 @@ async function loadTasksForList(url: URL): Promise<{ tasks: QueueTask[]; total:
const queueIdRaw = url.searchParams.get("queueId");
const queueId = queueIdRaw === null || queueIdRaw.length === 0 ? null : safeQueueId(queueIdRaw);
const search = String(url.searchParams.get("q") ?? url.searchParams.get("search") ?? "").trim();
const rows = await traceSql<Array<Pick<TaskRow, "task_json"> & { total_count: string | number }>>`
SELECT task_json, COUNT(*) OVER() AS total_count
const rows = await traceSql<Array<TaskJsonRow & { total_count: string | number }>>`
SELECT
id,
queue_id,
status,
current_attempt,
current_mode,
codex_thread_id,
active_turn_id,
updated_at,
started_at,
finished_at,
read_at,
task_json,
COUNT(*) OVER() AS total_count
FROM unidesk_code_queue_tasks
WHERE (${status === null} OR status = ${status})
AND (${queueId === null} OR queue_id = ${queueId})
@@ -1185,8 +1296,20 @@ async function loadTasksForList(url: URL): Promise<{ tasks: QueueTask[]; total:
async function loadTasksByIds(ids: string[]): Promise<QueueTask[]> {
const unique = Array.from(new Set(ids.map((id) => id.trim()).filter(Boolean)));
if (unique.length === 0) return [];
const rows = await traceSql<Array<Pick<TaskRow, "task_json">>>`
SELECT task_json
const rows = await traceSql<TaskJsonRow[]>`
SELECT
id,
queue_id,
status,
current_attempt,
current_mode,
codex_thread_id,
active_turn_id,
updated_at,
started_at,
finished_at,
read_at,
task_json
FROM unidesk_code_queue_tasks
WHERE id IN ${traceSql(unique)}
`;
@@ -1307,8 +1430,20 @@ async function queueSummary(tasks?: QueueTask[], queueRecords?: QueueRecord[]):
}
async function loadAllTasksLite(): Promise<QueueTask[]> {
const rows = await traceSql<Array<Pick<TaskRow, "task_json">>>`
SELECT task_json
const rows = await traceSql<TaskJsonRow[]>`
SELECT
id,
queue_id,
status,
current_attempt,
current_mode,
codex_thread_id,
active_turn_id,
updated_at,
started_at,
finished_at,
read_at,
task_json - 'output' - 'events' AS task_json
FROM unidesk_code_queue_tasks
ORDER BY created_at ASC, id ASC
LIMIT 2000
@@ -1838,24 +1973,54 @@ async function mergeQueues(targetQueueIdValue: string | null, req: Request): Pro
.filter((id) => id !== targetQueueId);
if (sourceQueueIds.length === 0) return jsonResponse({ ok: false, error: "sourceQueueId is required" }, 400);
const updatedAt = nowIso();
await mgrSql.begin(async (client) => {
const result = await mgrSql.begin(async (client): Promise<{ movedTaskIds: string[]; blocker: TaskStatusRow | null }> => {
const mergeQueueIds = Array.from(new Set([targetQueueId, ...sourceQueueIds]));
const lockedRows = await client<TaskStatusRow[]>`
SELECT id, queue_id, status, started_at, current_attempt, codex_thread_id, active_turn_id
FROM unidesk_code_queue_tasks
WHERE queue_id IN ${client(mergeQueueIds)}
ORDER BY updated_at DESC, id DESC
FOR UPDATE
`;
const blocker = lockedRows.find((row) => queueMergeBlocker(row).length > 0) ?? null;
if (blocker !== null) return { movedTaskIds: [], blocker };
await client`
INSERT INTO unidesk_code_queue_queues (id, name, created_at, updated_at)
VALUES (${targetQueueId}, ${targetQueueId}, ${updatedAt}, ${updatedAt})
ON CONFLICT (id) DO UPDATE SET updated_at = EXCLUDED.updated_at
`;
await client`
const movedRows = await client<Array<{ id: string }>>`
UPDATE unidesk_code_queue_tasks
SET
queue_id = ${targetQueueId},
updated_at = ${updatedAt},
task_json = jsonb_set(jsonb_set(task_json, '{queueId}', to_jsonb(${targetQueueId}::text), true), '{updatedAt}', to_jsonb(${updatedAt}::text), true)
WHERE queue_id IN ${client(sourceQueueIds)}
AND status NOT IN ('running', 'judging')
AND (
status IN ('succeeded', 'failed', 'canceled')
OR (
status IN ('queued', 'retry_wait')
AND started_at IS NULL
AND current_attempt = 0
AND codex_thread_id IS NULL
AND active_turn_id IS NULL
)
)
RETURNING id
`;
await client`DELETE FROM unidesk_code_queue_queues WHERE id IN ${client(sourceQueueIds)}`;
return { movedTaskIds: movedRows.map((row) => row.id), blocker: null };
});
return jsonResponse({ ok: true, targetQueueId, sourceQueueIds, queue: await queueSummary() }, 202);
if (result.blocker !== null) {
const blocker = result.blocker;
const queueId = safeQueueId(blocker.queue_id);
return jsonResponse({
ok: false,
error: `cannot merge queue ${queueId}: ${queueMergeBlocker(blocker)}`,
blocker: taskStatusRowJson(blocker),
}, 409);
}
return jsonResponse({ ok: true, targetQueueId, sourceQueueIds, movedTaskIds: result.movedTaskIds, queue: await queueSummary() }, 202);
}
async function markTaskRead(taskId: string): Promise<Response> {
@@ -1900,32 +2065,155 @@ async function moveTask(taskId: string, req: Request): Promise<Response> {
const body = asRecord(await readJson(req)) ?? {};
const queueId = normalizeQueueId(body.queueId ?? body.id);
const movedAt = nowIso();
const task = await mgrSql.begin(async (client): Promise<QueueTask | null> => {
const result = await mgrSql.begin(async (client): Promise<{ task: QueueTask | null; blocker: string; row: TaskStatusRow | null }> => {
const rows = await client<Array<TaskJsonRow & TaskStatusRow>>`
SELECT
id,
queue_id,
status,
current_attempt,
current_mode,
codex_thread_id,
active_turn_id,
updated_at,
started_at,
finished_at,
read_at,
task_json
FROM unidesk_code_queue_tasks
WHERE id = ${taskId}
FOR UPDATE
`;
const row = rows[0] ?? null;
const blocker = taskStatusMoveBlocker(row);
if (row === null || blocker.length > 0) return { task: row === null ? null : rowToTask(row), blocker, row };
await client`
INSERT INTO unidesk_code_queue_queues (id, name, created_at, updated_at)
VALUES (${queueId}, ${queueId}, ${movedAt}, ${movedAt})
ON CONFLICT (id) DO UPDATE SET updated_at = EXCLUDED.updated_at
`;
const rows = await client<Array<Pick<TaskRow, "task_json">>>`
SELECT task_json
FROM unidesk_code_queue_tasks
WHERE id = ${taskId}
FOR UPDATE
`;
if (rows[0] === undefined) return null;
const nextTask = rowToTask(rows[0]);
if (nextTask.status === "running" || nextTask.status === "judging") return nextTask;
const previousQueueId = safeQueueId(row.queue_id);
const nextTask = rowToTask(row);
nextTask.queueId = queueId;
nextTask.queueEnteredAt = movedAt;
nextTask.updatedAt = movedAt;
nextTask.output.push({ seq: outputMaxSeq(nextTask) + 1, at: movedAt, channel: "system", text: `moved to queue=${queueId}\n`, method: "queue/move" });
nextTask.output.push({ seq: outputMaxSeq(nextTask) + 1, at: movedAt, channel: "system", text: `moved from queue=${previousQueueId} to queue=${queueId}\n`, method: "queue/move" });
nextTask.outputMaxSeq = outputMaxSeq(nextTask);
await upsertTask(client, nextTask);
return nextTask;
const updated = await client<TaskStatusRow[]>`
UPDATE unidesk_code_queue_tasks
SET
queue_id = ${queueId},
updated_at = ${movedAt},
output_count = ${nextTask.output.length},
event_count = ${nextTask.events.length},
attempt_count = ${nextTask.attempts.length},
last_output_seq = ${outputMaxSeq(nextTask)},
task_json = ${client.json(taskJson(nextTask) as unknown as postgres.JSONValue)}
WHERE id = ${taskId}
AND status IN ('queued', 'retry_wait')
AND started_at IS NULL
AND current_attempt = 0
AND codex_thread_id IS NULL
AND active_turn_id IS NULL
RETURNING id, queue_id, status, started_at, current_attempt, codex_thread_id, active_turn_id
`;
if (updated[0] === undefined) return { task: rowToTask(row), blocker: "conditional update matched no rows", row };
return { task: nextTask, blocker: "", row: updated[0] };
});
if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
if (task.status === "running" || task.status === "judging") return jsonResponse({ ok: false, error: `cannot move active task while status=${task.status}`, task: taskListResponse(task) }, 409);
return jsonResponse({ ok: true, task: taskListResponse(task, false), queue: await queueSummary() }, 202);
if (result.task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
if (result.blocker.length > 0) {
return jsonResponse({
ok: false,
error: `cannot move task ${taskId}: ${result.blocker}`,
blocker: taskStatusRowJson(result.row),
task: taskListResponse(result.task),
}, 409);
}
return jsonResponse({ ok: true, task: taskListResponse(result.task, false), blocker: taskStatusRowJson(result.row), queue: await queueSummary() }, 202);
}
async function runQueueClaimMoveSelfTest(): Promise<JsonRecord> {
const suffix = String(Date.now());
const taskId = `codex_mgr_claim_move_${suffix}`;
const sourceQueueId = `mgr_claim_move_source_${suffix}`;
const targetQueueId = `mgr_claim_move_target_${suffix}`;
const queuedAt = nowIso();
await mgrSql`DELETE FROM unidesk_code_queue_tasks WHERE id = ${taskId}`;
try {
const queuedTask = normalizeTask(await createTaskFromRequest({
prompt: "Code Queue manager claim/move race self-test",
queueId: sourceQueueId,
maxAttempts: 1,
}));
queuedTask.id = taskId;
queuedTask.queueId = sourceQueueId;
queuedTask.queueEnteredAt = queuedAt;
queuedTask.createdAt = queuedAt;
queuedTask.updatedAt = queuedAt;
await mgrSql.begin(async (client) => {
await client`
INSERT INTO unidesk_code_queue_queues (id, name, created_at, updated_at)
VALUES (${sourceQueueId}, ${sourceQueueId}, ${queuedAt}, ${queuedAt})
ON CONFLICT (id) DO UPDATE SET updated_at = EXCLUDED.updated_at
`;
await upsertTask(client, queuedTask);
});
const claimedAt = nowIso();
const claimedTask = normalizeTask(JSON.parse(JSON.stringify(queuedTask)) as QueueTask);
claimedTask.status = "running";
claimedTask.startedAt = claimedAt;
claimedTask.currentAttempt = 1;
claimedTask.currentMode = "initial";
claimedTask.codexThreadId = `thread_${suffix}`;
claimedTask.activeTurnId = `turn_${suffix}`;
claimedTask.updatedAt = claimedAt;
await mgrSql.begin(async (client) => {
await upsertTask(client, claimedTask);
});
const moveResponse = await moveTask(taskId, new Request(`http://code-queue-mgr.local/api/tasks/${taskId}/move`, {
method: "POST",
body: JSON.stringify({ queueId: targetQueueId }),
headers: { "content-type": "application/json" },
}));
const moveBody = await moveResponse.json() as JsonRecord;
const rows = await mgrSql<Array<TaskJsonRow & TaskStatusRow>>`
SELECT id, queue_id, status, current_attempt, current_mode, codex_thread_id, active_turn_id, updated_at, started_at, finished_at, read_at, task_json
FROM unidesk_code_queue_tasks
WHERE id = ${taskId}
LIMIT 1
`;
const after = rows[0] === undefined ? null : rowToTask(rows[0]);
const ok = moveResponse.status === 409
&& after !== null
&& after.status === "running"
&& queueIdOf(after) === sourceQueueId
&& after.currentAttempt === 1
&& after.startedAt !== null
&& after.codexThreadId !== null
&& after.activeTurnId !== null;
if (!ok) {
throw new Error(`claim/move self-test failed: moveStatus=${moveResponse.status}, after=${JSON.stringify(after === null ? null : taskListResponse(after))}`);
}
return {
ok: true,
taskId,
sourceQueueId,
targetQueueId,
moveStatus: moveResponse.status,
databaseStatus: after.status,
databaseQueueId: queueIdOf(after),
currentAttempt: after.currentAttempt,
startedAt: after.startedAt,
codexThreadId: after.codexThreadId,
activeTurnId: after.activeTurnId,
moveResponse: moveBody,
};
} finally {
await mgrSql`DELETE FROM unidesk_code_queue_tasks WHERE id = ${taskId}`;
await mgrSql`DELETE FROM unidesk_code_queue_queues WHERE id IN ${mgrSql([sourceQueueId, targetQueueId])}`;
}
}
async function retryTask(taskId: string, req: Request): Promise<Response> {
@@ -1933,8 +2221,20 @@ async function retryTask(taskId: string, req: Request): Promise<Response> {
const explicitPrompt = typeof body.prompt === "string" ? body.prompt.trim() : typeof body.continuePrompt === "string" ? body.continuePrompt.trim() : "";
const queuedAt = nowIso();
const result = await mgrSql.begin(async (client): Promise<{ task: QueueTask | null; changed: boolean }> => {
const rows = await client<Array<Pick<TaskRow, "task_json">>>`
SELECT task_json
const rows = await client<Array<TaskJsonRow & TaskStatusRow>>`
SELECT
id,
queue_id,
status,
current_attempt,
current_mode,
codex_thread_id,
active_turn_id,
updated_at,
started_at,
finished_at,
read_at,
task_json
FROM unidesk_code_queue_tasks
WHERE id = ${taskId}
FOR UPDATE
@@ -1972,8 +2272,20 @@ async function editTask(taskId: string, req: Request): Promise<Response> {
const body = asRecord(await readJson(req)) ?? {};
const editedAt = nowIso();
const result = await mgrSql.begin(async (client): Promise<{ task: QueueTask | null; changed: boolean }> => {
const rows = await client<Array<Pick<TaskRow, "task_json">>>`
SELECT task_json
const rows = await client<Array<TaskJsonRow & TaskStatusRow>>`
SELECT
id,
queue_id,
status,
current_attempt,
current_mode,
codex_thread_id,
active_turn_id,
updated_at,
started_at,
finished_at,
read_at,
task_json
FROM unidesk_code_queue_tasks
WHERE id = ${taskId}
FOR UPDATE
@@ -2092,12 +2404,16 @@ async function route(req: Request): Promise<Response> {
noDockerSocket: true,
},
endpoints: {
control: ["/api/queues", "/api/tasks", "/api/tasks/:id/(move|retry|read|edit)"],
control: ["/api/queues", "/api/tasks", "/api/tasks/:id/(move|retry|read|edit)", "/api/queue-claim-move/self-test"],
traceRead: ["/api/tasks/overview", "/api/tasks/:id/summary", "/api/tasks/:id/trace-summary", "/api/tasks/:id/trace-steps", "/api/tasks/:id/output"],
},
}, schemaReady ? 200 : 503);
}
if (url.pathname === "/logs" && req.method === "GET") return jsonResponse({ ok: true, logs: recentLogs.slice(-100) });
if (url.pathname === "/api/queue-claim-move/self-test" && (req.method === "POST" || req.method === "GET")) {
if (!schemaReady) return jsonResponse({ ok: false, error: "code-queue-mgr database schema is not ready", schemaLastError }, 503);
return jsonResponse(await runQueueClaimMoveSelfTest(), 200);
}
const activeControl = routeActiveControl(url.pathname, req.method);
if (activeControl !== null) return jsonResponse(activeControl.body, activeControl.status);
if (url.pathname === "/api/queues" && req.method === "GET") {