fix: continue indefinitely after each 429
This commit is contained in:
@@ -2,6 +2,11 @@
|
||||
|
||||
All notable changes to this project are documented in this file.
|
||||
|
||||
## 0.1.1 - 2026-09-12
|
||||
|
||||
- Continue indefinitely while each new structured HTTP 429 failure is emitted.
|
||||
- Keep the delay fixed at 3 seconds for every attempt.
|
||||
|
||||
## 0.1.0 - 2026-09-11
|
||||
|
||||
- Detect structured Codex HTTP 429 task failures.
|
||||
|
||||
+1
-1
@@ -7,7 +7,7 @@
|
||||
## 功能
|
||||
|
||||
- 只识别结构化的 `response_too_many_failed_attempts` HTTP 429 失败记录。
|
||||
- 每次续跑统一等待 3 秒;同一失败链最多由插件发起四次续跑。
|
||||
- 每次续跑统一等待 3 秒;每条结构化 429 都会继续触发下一次续跑,不设总次数上限。
|
||||
- Codex 提供原生重试或继续工具时优先使用;否则只向原任务发送“继续”。
|
||||
- 通过 `SessionStart` 支持新建和重新打开的任务,通过 `UserPromptSubmit` 为安装前已经加载的任务补启动监听。
|
||||
- 如果 Codex 或用户已经开始新一轮任务,取消当前等待中的续跑。
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "codex-429-auto-continue",
|
||||
"version": "0.1.0",
|
||||
"version": "0.1.1",
|
||||
"private": true,
|
||||
"description": "Automatically continue Codex Desktop tasks after exhausted HTTP 429 retries.",
|
||||
"type": "module",
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "codex-429-auto-continue",
|
||||
"version": "0.1.0",
|
||||
"version": "0.1.1",
|
||||
"description": "Automatically continues a Codex desktop task after an exhausted HTTP 429 retry cycle.",
|
||||
"author": {
|
||||
"name": "im"
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
When a Codex task ends with the structured `response_too_many_failed_attempts` HTTP 429 error, this plugin continues the same task through the Codex desktop app. It prefers a native retry/continue tool when Codex exposes one and otherwise sends `继续`.
|
||||
|
||||
The plugin starts one watcher per loaded session. `SessionStart` covers new and reopened sessions, while `UserPromptSubmit` backfills watchers for sessions that were already loaded when the plugin was installed. It gives Codex's built-in retry a chance to start first; when no new task has started, every continuation attempt waits exactly 3 seconds. A successful completion or 15 minutes without a plugin-initiated retry resets the chain. Prompt text, assistant text, and tool output containing `429` do not trigger it.
|
||||
The plugin starts one watcher per loaded session. `SessionStart` covers new and reopened sessions, while `UserPromptSubmit` backfills watchers for sessions that were already loaded when the plugin was installed. It gives Codex's built-in retry a chance to start first; when no new task has started, every continuation attempt waits exactly 3 seconds. Every structured 429 can trigger another continuation with no total attempt limit. Prompt text, assistant text, and tool output containing `429` do not trigger it.
|
||||
|
||||
The watcher requires `node` on `PATH` and the Codex desktop app to remain open. Continuations appear in the original task. The watcher exits on `SessionEnd` or after 24 hours without transcript activity. Detection and delivery events are recorded under the plugin data directory as per-session JSONL logs.
|
||||
|
||||
|
||||
@@ -8,8 +8,7 @@ import { homedir } from "node:os";
|
||||
import { dirname, join } from "node:path";
|
||||
import { fileURLToPath } from "node:url";
|
||||
|
||||
export const RETRY_DELAYS_MS = [3_000, 3_000, 3_000, 3_000];
|
||||
const RETRY_RESET_MS = 15 * 60_000;
|
||||
export const RETRY_DELAY_MS = 3_000;
|
||||
const WATCH_IDLE_MS = 24 * 60 * 60_000;
|
||||
const POLL_MS = 1_500;
|
||||
const TRANSCRIPT_TAIL_BYTES = 2 * 1024 * 1024;
|
||||
@@ -373,6 +372,26 @@ export async function shouldResumeAfterDelay(failedRecord, transcriptPath, delay
|
||||
return !hasTaskStartedAfter(currentRecords, failedRecord);
|
||||
}
|
||||
|
||||
export async function continueUntilStartedOrSent(failedRecord, input, dependencies = {}) {
|
||||
const shouldResume = dependencies.shouldResumeAfterDelay ?? shouldResumeAfterDelay;
|
||||
const send = dependencies.sendDesktopContinuation ?? sendDesktopContinuation;
|
||||
const onWaiting = dependencies.onWaiting ?? (() => {});
|
||||
const onDeliveryFailure = dependencies.onDeliveryFailure ?? (() => {});
|
||||
|
||||
while (true) {
|
||||
await onWaiting(RETRY_DELAY_MS);
|
||||
if (!(await shouldResume(failedRecord, input.transcriptPath, RETRY_DELAY_MS))) return null;
|
||||
try {
|
||||
return await send({
|
||||
...input,
|
||||
failedTurnId: failedRecord.payload?.turn_id,
|
||||
});
|
||||
} catch (error) {
|
||||
await onDeliveryFailure(error);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async function watch(input) {
|
||||
const paths = sessionPaths(input.sessionId);
|
||||
if (!(await acquireLock(paths.lock))) return;
|
||||
@@ -387,8 +406,6 @@ async function watch(input) {
|
||||
})}\n`, "utf8").catch(() => {});
|
||||
};
|
||||
|
||||
let attempts = 0;
|
||||
let lastResumeAt = 0;
|
||||
let lastActivityAt = Date.now();
|
||||
const cursor = { offset: 0, carry: "" };
|
||||
try {
|
||||
@@ -404,42 +421,27 @@ async function watch(input) {
|
||||
lastActivityAt = Date.now();
|
||||
|
||||
for (const record of records) {
|
||||
if (isSuccessfulCompletion(record)) {
|
||||
attempts = 0;
|
||||
lastResumeAt = 0;
|
||||
continue;
|
||||
}
|
||||
if (isSuccessfulCompletion(record)) continue;
|
||||
if (!is429Record(record)) continue;
|
||||
await log("429_detected", { turnId: record.payload?.turn_id });
|
||||
if (lastResumeAt && Date.now() - lastResumeAt >= RETRY_RESET_MS) attempts = 0;
|
||||
if (attempts >= RETRY_DELAYS_MS.length) continue;
|
||||
|
||||
for (let deliveryAttempt = attempts; deliveryAttempt < RETRY_DELAYS_MS.length; deliveryAttempt += 1) {
|
||||
const delay = RETRY_DELAYS_MS[deliveryAttempt];
|
||||
await log("waiting", { delayMs: delay, turnId: record.payload?.turn_id });
|
||||
if (!(await shouldResumeAfterDelay(record, input.transcriptPath, delay))) {
|
||||
await log("cancelled_new_task_started", { turnId: record.payload?.turn_id });
|
||||
break;
|
||||
}
|
||||
try {
|
||||
const sent = await sendDesktopContinuation({
|
||||
...input,
|
||||
failedTurnId: record.payload?.turn_id,
|
||||
const sent = await continueUntilStartedOrSent(record, input, {
|
||||
onWaiting: (delay) => log("waiting", {
|
||||
delayMs: delay,
|
||||
turnId: record.payload?.turn_id,
|
||||
}),
|
||||
onDeliveryFailure: (error) => log("desktop_continue_failed", {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
turnId: record.payload?.turn_id,
|
||||
}),
|
||||
});
|
||||
attempts += 1;
|
||||
lastResumeAt = Date.now();
|
||||
if (sent) {
|
||||
await log("desktop_continue_sent", {
|
||||
delivery: sent.delivery,
|
||||
pipePath: sent.pipePath,
|
||||
turnId: record.payload?.turn_id,
|
||||
});
|
||||
break;
|
||||
} catch (error) {
|
||||
await log("desktop_continue_failed", {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
turnId: record.payload?.turn_id,
|
||||
});
|
||||
}
|
||||
} else {
|
||||
await log("cancelled_new_task_started", { turnId: record.payload?.turn_id });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2,11 +2,12 @@ import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
|
||||
import {
|
||||
continueUntilStartedOrSent,
|
||||
hasTaskStartedAfter,
|
||||
is429Record,
|
||||
isSuccessfulCompletion,
|
||||
loadedThreadIds,
|
||||
RETRY_DELAYS_MS,
|
||||
RETRY_DELAY_MS,
|
||||
sendDesktopContinuation,
|
||||
shouldResumeAfterDelay,
|
||||
threadListContains,
|
||||
@@ -55,9 +56,9 @@ test("detects Codex or user retry activity after a failure", () => {
|
||||
});
|
||||
|
||||
test("starts the retry chain after three seconds", async () => {
|
||||
assert.deepEqual(RETRY_DELAYS_MS, [3_000, 3_000, 3_000, 3_000]);
|
||||
assert.equal(RETRY_DELAY_MS, 3_000);
|
||||
let waited;
|
||||
const shouldResume = await shouldResumeAfterDelay(failed, "transcript.jsonl", RETRY_DELAYS_MS[0], {
|
||||
const shouldResume = await shouldResumeAfterDelay(failed, "transcript.jsonl", RETRY_DELAY_MS, {
|
||||
pause: async (delay) => { waited = delay; },
|
||||
readTranscriptTail: async () => [failed],
|
||||
});
|
||||
@@ -75,6 +76,29 @@ test("starts the retry chain after three seconds", async () => {
|
||||
assert.equal(builtInRetryWon, false);
|
||||
});
|
||||
|
||||
test("keeps retrying delivery every three seconds without a fixed attempt limit", async () => {
|
||||
const delays = [];
|
||||
let sendAttempts = 0;
|
||||
const result = await continueUntilStartedOrSent(failed, {
|
||||
sessionId: "session-1",
|
||||
transcriptPath: "transcript.jsonl",
|
||||
}, {
|
||||
shouldResumeAfterDelay: async (_record, _path, delay) => {
|
||||
delays.push(delay);
|
||||
return true;
|
||||
},
|
||||
sendDesktopContinuation: async () => {
|
||||
sendAttempts += 1;
|
||||
if (sendAttempts < 7) throw new Error("temporary delivery failure");
|
||||
return { delivery: "message", pipePath: "app-pipe" };
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(sendAttempts, 7);
|
||||
assert.deepEqual(delays, Array(7).fill(3_000));
|
||||
assert.equal(result.delivery, "message");
|
||||
});
|
||||
|
||||
test("finds a target task in a Codex desktop tool result", () => {
|
||||
const result = {
|
||||
contentItems: [{
|
||||
|
||||
Reference in New Issue
Block a user