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

fix: persist code queue retry claims

Browse Source
This commit is contained in:
Codex
2026-05-17 15:20:22 +00:00
parent 1cf0b13aa8
commit 50ba23fae1
1 changed files with 7 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/components/microservices/code-queue/src/index.ts
+7 -3
View File
@@ -3398,10 +3398,12 @@ function queueActiveTasksForRestartRetry(reason: string, method: string, options
return recovered;
}
async function recoverOrphanedActiveTasks(reason: string, method: string): Promise<number> {
async function recoverOrphanedActiveTasks(reason: string, method: string, options: { taskIds?: Set<string>; staleMs?: number } = {}): Promise<number> {
const staleMs = options.staleMs ?? orphanedActiveTaskRecoveryStaleMs;
const staleOrphanTaskIds = new Set(
state.tasks
.filter((task) => taskIsRecoverableOrphanedActive(task))
.filter((task) => options.taskIds === undefined || options.taskIds.has(task.id))
.filter((task) => taskIsRecoverableOrphanedActive(task, staleMs))
.map((task) => task.id),
);
if (staleOrphanTaskIds.size === 0) return 0;
@@ -3492,7 +3494,7 @@ async function runTask(task: QueueTask): Promise<void> {
}
if (!claimed) {
releaseRunSlot();
await recoverOrphanedActiveTasks("Database claim found stale active task without local run", "database/claim-recovery");
await recoverOrphanedActiveTasks("Database claim found stale active task without local run", "database/claim-recovery", { taskIds: new Set([task.id]), staleMs: 0 });
return;
}
publishTaskOaEvent(task, "claim");
@@ -3550,6 +3552,7 @@ async function runTask(task: QueueTask): Promise<void> {
task.updatedAt = nowIso();
appendOutput(task, "system", `judge=fail treated as retry (${task.judgeFailCount}/${judgeFailRetryLimit}); appending continuation prompt to existing session\n`, "queue");
persistTaskState(task);
await flushDirtyTasksToDatabase(true);
logger("warn", "task_judge_fail_treated_as_retry", {
taskId: task.id,
attempt: task.currentAttempt,
@@ -3587,6 +3590,7 @@ async function runTask(task: QueueTask): Promise<void> {
setAttemptFeedbackPrompt(latestAttempt, nextPrompt, judge.continuePrompt?.trim() ? "judge-continue-prompt" : "judge-retry-generated", task.attempts.length + 1);
task.updatedAt = nowIso();
persistTaskState(task);
await flushDirtyTasksToDatabase(true);
const delayMs = retryBackoffMs(task.attempts.length);
appendOutput(task, "system", `retry backoff ${Math.round(delayMs / 1000)}s before appending continuation to existing session\n`, "queue");
await sleepForRetryBackoff(task, delayMs);