feat: publish Codex 429 auto-continue plugin
This commit is contained in:
@@ -0,0 +1,19 @@
|
||||
{
|
||||
"name": "codex-429-auto-continue",
|
||||
"version": "0.1.0",
|
||||
"description": "Automatically continues a Codex desktop task after an exhausted HTTP 429 retry cycle.",
|
||||
"author": {
|
||||
"name": "im"
|
||||
},
|
||||
"interface": {
|
||||
"displayName": "Codex 429 Auto Continue",
|
||||
"shortDescription": "Resume interrupted tasks after Codex HTTP 429 errors.",
|
||||
"longDescription": "Detects a structured HTTP 429 task failure, waits 3 seconds, and continues the original Codex desktop task.",
|
||||
"developerName": "im",
|
||||
"category": "Productivity",
|
||||
"capabilities": [
|
||||
"Automation"
|
||||
],
|
||||
"defaultPrompt": "Show whether Codex 429 Auto Continue is installed."
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,9 @@
|
||||
# Codex 429 Auto Continue
|
||||
|
||||
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 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.
|
||||
|
||||
After installation, start a new task and review the plugin hook once in `/hooks` when Codex asks you to trust it.
|
||||
@@ -0,0 +1,523 @@
|
||||
#!/usr/bin/env node
|
||||
|
||||
import { createHash } from "node:crypto";
|
||||
import { spawn } from "node:child_process";
|
||||
import { appendFile, mkdir, open, readFile, readdir, rename, stat, unlink, writeFile } from "node:fs/promises";
|
||||
import net from "node:net";
|
||||
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;
|
||||
const WATCH_IDLE_MS = 24 * 60 * 60_000;
|
||||
const POLL_MS = 1_500;
|
||||
const TRANSCRIPT_TAIL_BYTES = 2 * 1024 * 1024;
|
||||
const PIPE_TIMEOUT_MS = 5_000;
|
||||
const THREAD_LIST_LIMIT = 50;
|
||||
const CONTINUATION_PROMPT = "继续";
|
||||
const NATIVE_CONTINUATION_TOOL_NAMES = [
|
||||
"retry_turn",
|
||||
"continue_thread",
|
||||
"retry_thread",
|
||||
"resume_thread",
|
||||
];
|
||||
|
||||
function pluginDataDirectory() {
|
||||
return process.env.PLUGIN_DATA
|
||||
|| join(homedir(), ".codex", "plugin-data", "codex-429-auto-continue");
|
||||
}
|
||||
|
||||
function sessionKey(sessionId) {
|
||||
return createHash("sha256").update(sessionId).digest("hex").slice(0, 24);
|
||||
}
|
||||
|
||||
function sessionPaths(sessionId) {
|
||||
const root = pluginDataDirectory();
|
||||
const key = sessionKey(sessionId);
|
||||
return {
|
||||
lock: join(root, `${key}.lock.json`),
|
||||
log: join(root, `${key}.log.jsonl`),
|
||||
stop: join(root, `${key}.stop`),
|
||||
};
|
||||
}
|
||||
|
||||
async function readHookInput() {
|
||||
let input = "";
|
||||
for await (const chunk of process.stdin) input += chunk;
|
||||
try {
|
||||
const payload = JSON.parse(input);
|
||||
return payload && typeof payload === "object" ? payload : {};
|
||||
} catch {
|
||||
return {};
|
||||
}
|
||||
}
|
||||
|
||||
export function is429Record(record) {
|
||||
const payload = record?.type === "event_msg" ? record.payload : null;
|
||||
if (payload?.type !== "task_complete" || typeof payload.error !== "object") {
|
||||
return false;
|
||||
}
|
||||
|
||||
const status = payload.error?.codex_error_info
|
||||
?.response_too_many_failed_attempts?.http_status_code;
|
||||
if (status === 429) return true;
|
||||
|
||||
const message = payload.error?.message;
|
||||
return typeof message === "string"
|
||||
&& /exceeded retry limit/i.test(message)
|
||||
&& /\b429\b/.test(message);
|
||||
}
|
||||
|
||||
export function isSuccessfulCompletion(record) {
|
||||
const payload = record?.type === "event_msg" ? record.payload : null;
|
||||
return payload?.type === "task_complete" && !payload.error;
|
||||
}
|
||||
|
||||
export function hasTaskStartedAfter(records, failedRecord) {
|
||||
return records.some((record) => {
|
||||
if (record?.type !== "event_msg" || record.payload?.type !== "task_started") {
|
||||
return false;
|
||||
}
|
||||
if (Number.isFinite(record.ordinal) && Number.isFinite(failedRecord?.ordinal)) {
|
||||
return record.ordinal > failedRecord.ordinal;
|
||||
}
|
||||
return record.payload.turn_id !== failedRecord?.payload?.turn_id;
|
||||
});
|
||||
}
|
||||
|
||||
async function readTranscriptTail(path) {
|
||||
const metadata = await stat(path);
|
||||
const start = Math.max(0, metadata.size - TRANSCRIPT_TAIL_BYTES);
|
||||
const handle = await open(path, "r");
|
||||
try {
|
||||
const buffer = Buffer.alloc(metadata.size - start);
|
||||
await handle.read(buffer, 0, buffer.length, start);
|
||||
let text = buffer.toString("utf8");
|
||||
if (start > 0) {
|
||||
const firstNewline = text.indexOf("\n");
|
||||
text = firstNewline >= 0 ? text.slice(firstNewline + 1) : "";
|
||||
}
|
||||
return text.split(/\r?\n/).filter(Boolean).flatMap((line) => {
|
||||
try {
|
||||
return [JSON.parse(line)];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
});
|
||||
} finally {
|
||||
await handle.close();
|
||||
}
|
||||
}
|
||||
|
||||
async function readNewRecords(path, cursor) {
|
||||
const metadata = await stat(path);
|
||||
if (metadata.size < cursor.offset) {
|
||||
cursor.offset = 0;
|
||||
cursor.carry = "";
|
||||
}
|
||||
if (metadata.size === cursor.offset) return [];
|
||||
|
||||
const handle = await open(path, "r");
|
||||
try {
|
||||
const buffer = Buffer.alloc(metadata.size - cursor.offset);
|
||||
await handle.read(buffer, 0, buffer.length, cursor.offset);
|
||||
cursor.offset = metadata.size;
|
||||
const parts = `${cursor.carry}${buffer.toString("utf8")}`.split(/\r?\n/);
|
||||
cursor.carry = parts.pop() ?? "";
|
||||
return parts.flatMap((line) => {
|
||||
if (!line) return [];
|
||||
try {
|
||||
return [JSON.parse(line)];
|
||||
} catch {
|
||||
return [];
|
||||
}
|
||||
});
|
||||
} finally {
|
||||
await handle.close();
|
||||
}
|
||||
}
|
||||
|
||||
function pause(milliseconds) {
|
||||
return new Promise((resolve) => setTimeout(resolve, milliseconds));
|
||||
}
|
||||
|
||||
async function processIsAlive(pid) {
|
||||
if (!Number.isInteger(pid) || pid <= 0) return false;
|
||||
try {
|
||||
process.kill(pid, 0);
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
async function acquireLock(path) {
|
||||
await mkdir(dirname(path), { recursive: true });
|
||||
try {
|
||||
const existing = JSON.parse(await readFile(path, "utf8"));
|
||||
if (await processIsAlive(existing.pid)) return false;
|
||||
await unlink(path).catch(() => {});
|
||||
} catch {
|
||||
await unlink(path).catch(() => {});
|
||||
}
|
||||
|
||||
try {
|
||||
const handle = await open(path, "wx");
|
||||
await handle.writeFile(`${JSON.stringify({ pid: process.pid, startedAt: Date.now() })}\n`);
|
||||
await handle.close();
|
||||
return true;
|
||||
} catch {
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
function encodeFrame(message) {
|
||||
const payload = Buffer.from(JSON.stringify(message), "utf8");
|
||||
const frame = Buffer.alloc(payload.length + 4);
|
||||
frame.writeUInt32LE(payload.length, 0);
|
||||
payload.copy(frame, 4);
|
||||
return frame;
|
||||
}
|
||||
|
||||
function pipeRequest(pipePath, method, params, timeoutMs = PIPE_TIMEOUT_MS) {
|
||||
return new Promise((resolve, reject) => {
|
||||
const socket = net.createConnection(pipePath);
|
||||
let settled = false;
|
||||
let pending = Buffer.alloc(0);
|
||||
const finish = (error, result) => {
|
||||
if (settled) return;
|
||||
settled = true;
|
||||
clearTimeout(timer);
|
||||
socket.destroy();
|
||||
if (error) reject(error);
|
||||
else resolve(result);
|
||||
};
|
||||
const timer = setTimeout(() => finish(new Error("Codex desktop pipe timed out")), timeoutMs);
|
||||
socket.once("connect", () => {
|
||||
socket.write(encodeFrame({ id: 1, jsonrpc: "2.0", method, params }));
|
||||
});
|
||||
socket.on("data", (chunk) => {
|
||||
pending = Buffer.concat([pending, chunk]);
|
||||
if (pending.length < 4) return;
|
||||
const length = pending.readUInt32LE(0);
|
||||
if (pending.length < length + 4) return;
|
||||
try {
|
||||
const response = JSON.parse(pending.subarray(4, length + 4).toString("utf8"));
|
||||
if (response.error) finish(new Error(response.error.message));
|
||||
else finish(null, response.result);
|
||||
} catch (error) {
|
||||
finish(error);
|
||||
}
|
||||
});
|
||||
socket.once("error", (error) => finish(error));
|
||||
socket.once("close", () => {
|
||||
if (!settled) finish(new Error("Codex desktop pipe closed"));
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
function parseToolText(result) {
|
||||
const text = result?.contentItems?.find((item) => item?.type === "inputText")?.text;
|
||||
if (typeof text !== "string") return null;
|
||||
try {
|
||||
return JSON.parse(text);
|
||||
} catch {
|
||||
return text;
|
||||
}
|
||||
}
|
||||
|
||||
export function threadListContains(result, sessionId) {
|
||||
const parsed = parseToolText(result);
|
||||
if (!parsed || typeof parsed !== "object") return false;
|
||||
return [...(parsed.pinnedThreads ?? []), ...(parsed.threads ?? [])]
|
||||
.some((thread) => thread?.id === sessionId);
|
||||
}
|
||||
|
||||
export function loadedThreadIds(result) {
|
||||
const parsed = parseToolText(result);
|
||||
if (!parsed || typeof parsed !== "object") return [];
|
||||
return [...(parsed.pinnedThreads ?? []), ...(parsed.threads ?? [])]
|
||||
.filter((thread) => {
|
||||
const status = typeof thread?.status === "string"
|
||||
? thread.status
|
||||
: thread?.status?.type;
|
||||
return thread?.id && status !== "notLoaded";
|
||||
})
|
||||
.map((thread) => thread.id);
|
||||
}
|
||||
|
||||
async function pipeCandidates() {
|
||||
const configured = process.env.CODEX_APP_TOOLS_PIPE_PATH?.trim();
|
||||
const candidates = configured ? [configured] : [];
|
||||
if (process.platform !== "win32") return candidates;
|
||||
const names = await readdir("\\\\.\\pipe\\").catch(() => []);
|
||||
for (const name of names) {
|
||||
if (/^codex-browser-use-/i.test(name)) {
|
||||
candidates.push(`\\\\.\\pipe\\${name}`);
|
||||
}
|
||||
}
|
||||
return [...new Set(candidates)];
|
||||
}
|
||||
|
||||
function toolCallParams(input, tool, args) {
|
||||
const callId = `codex-429-${Date.now()}-${process.pid}`;
|
||||
return {
|
||||
arguments: args,
|
||||
callId,
|
||||
namespace: "codex_app",
|
||||
threadId: input.sessionId,
|
||||
tool,
|
||||
turnId: input.failedTurnId || callId,
|
||||
};
|
||||
}
|
||||
|
||||
async function loadedDesktopSessionIds(input, dependencies = {}) {
|
||||
const listCandidates = dependencies.pipeCandidates ?? pipeCandidates;
|
||||
const request = dependencies.pipeRequest ?? pipeRequest;
|
||||
for (const pipePath of await listCandidates()) {
|
||||
try {
|
||||
await request(pipePath, "tools/list", { threadStartKind: "all" });
|
||||
const listed = await request(pipePath, "tools/call", toolCallParams(input, "list_threads", {
|
||||
limit: THREAD_LIST_LIMIT,
|
||||
}));
|
||||
if (listed?.success) return loadedThreadIds(listed);
|
||||
} catch {
|
||||
// Other Codex pipes do not expose app task tools.
|
||||
}
|
||||
}
|
||||
return [];
|
||||
}
|
||||
|
||||
async function findSessionTranscripts(sessionIds, root = join(homedir(), ".codex", "sessions")) {
|
||||
const wanted = new Set(sessionIds);
|
||||
const found = new Map();
|
||||
async function visit(directory) {
|
||||
const entries = await readdir(directory, { withFileTypes: true }).catch(() => []);
|
||||
await Promise.all(entries.map(async (entry) => {
|
||||
const path = join(directory, entry.name);
|
||||
if (entry.isDirectory()) {
|
||||
await visit(path);
|
||||
return;
|
||||
}
|
||||
if (!entry.isFile() || !entry.name.endsWith(".jsonl")) return;
|
||||
const sessionId = [...wanted].find((id) => entry.name.includes(id));
|
||||
if (!sessionId) return;
|
||||
const modifiedAt = (await stat(path)).mtimeMs;
|
||||
if (!found.has(sessionId) || found.get(sessionId).modifiedAt < modifiedAt) {
|
||||
found.set(sessionId, { modifiedAt, path });
|
||||
}
|
||||
}));
|
||||
}
|
||||
await visit(root);
|
||||
return new Map([...found].map(([sessionId, value]) => [sessionId, value.path]));
|
||||
}
|
||||
|
||||
function nativeContinuationArguments(tool, input) {
|
||||
const schema = tool?.inputSchema;
|
||||
const properties = schema?.properties ?? {};
|
||||
const args = {};
|
||||
if (properties.threadId) args.threadId = input.sessionId;
|
||||
if (properties.thread_id) args.thread_id = input.sessionId;
|
||||
if (properties.turnId && input.failedTurnId) args.turnId = input.failedTurnId;
|
||||
if (properties.turn_id && input.failedTurnId) args.turn_id = input.failedTurnId;
|
||||
const required = Array.isArray(schema?.required) ? schema.required : [];
|
||||
return required.every((name) => Object.hasOwn(args, name)) ? args : null;
|
||||
}
|
||||
|
||||
export async function sendDesktopContinuation(input, dependencies = {}) {
|
||||
const listCandidates = dependencies.pipeCandidates ?? pipeCandidates;
|
||||
const request = dependencies.pipeRequest ?? pipeRequest;
|
||||
const candidates = await listCandidates();
|
||||
let lastError = new Error("No Codex desktop task pipe was found");
|
||||
|
||||
for (const pipePath of candidates) {
|
||||
try {
|
||||
const catalog = await request(pipePath, "tools/list", { threadStartKind: "all" });
|
||||
const tools = Array.isArray(catalog?.tools) ? catalog.tools : [];
|
||||
const listed = await request(pipePath, "tools/call", toolCallParams(input, "list_threads", { limit: THREAD_LIST_LIMIT }));
|
||||
if (!listed?.success || !threadListContains(listed, input.sessionId)) continue;
|
||||
|
||||
for (const toolName of NATIVE_CONTINUATION_TOOL_NAMES) {
|
||||
const tool = tools.find((candidate) => candidate?.name === toolName);
|
||||
const args = nativeContinuationArguments(tool, input);
|
||||
if (!tool || args == null) continue;
|
||||
try {
|
||||
const resumed = await request(pipePath, "tools/call", toolCallParams(input, tool.name, args));
|
||||
if (resumed?.success) {
|
||||
return { delivery: "native", pipePath, result: parseToolText(resumed) };
|
||||
}
|
||||
} catch {
|
||||
// Fall back to the normal Codex message path below.
|
||||
}
|
||||
}
|
||||
|
||||
const sent = await request(pipePath, "tools/call", toolCallParams(input, "send_message_to_thread", {
|
||||
prompt: CONTINUATION_PROMPT,
|
||||
threadId: input.sessionId,
|
||||
}));
|
||||
if (!sent?.success) throw new Error("Codex desktop rejected the continuation message");
|
||||
return { delivery: "message", pipePath, result: parseToolText(sent) };
|
||||
} catch (error) {
|
||||
lastError = error;
|
||||
}
|
||||
}
|
||||
throw lastError;
|
||||
}
|
||||
|
||||
export async function shouldResumeAfterDelay(failedRecord, transcriptPath, delay, dependencies = {}) {
|
||||
const sleep = dependencies.pause ?? pause;
|
||||
const readTail = dependencies.readTranscriptTail ?? readTranscriptTail;
|
||||
await sleep(delay);
|
||||
const currentRecords = await readTail(transcriptPath);
|
||||
return !hasTaskStartedAfter(currentRecords, failedRecord);
|
||||
}
|
||||
|
||||
async function watch(input) {
|
||||
const paths = sessionPaths(input.sessionId);
|
||||
if (!(await acquireLock(paths.lock))) return;
|
||||
await unlink(paths.stop).catch(() => {});
|
||||
|
||||
const log = async (event, details = {}) => {
|
||||
await appendFile(paths.log, `${JSON.stringify({
|
||||
timestamp: new Date().toISOString(),
|
||||
event,
|
||||
sessionId: input.sessionId,
|
||||
...details,
|
||||
})}\n`, "utf8").catch(() => {});
|
||||
};
|
||||
|
||||
let attempts = 0;
|
||||
let lastResumeAt = 0;
|
||||
let lastActivityAt = Date.now();
|
||||
const cursor = { offset: 0, carry: "" };
|
||||
try {
|
||||
cursor.offset = (await stat(input.transcriptPath)).size;
|
||||
await log("watcher_started", { transcriptPath: input.transcriptPath });
|
||||
while (Date.now() - lastActivityAt < WATCH_IDLE_MS) {
|
||||
if (await stat(paths.stop).then(() => true).catch(() => false)) break;
|
||||
const records = await readNewRecords(input.transcriptPath, cursor).catch(() => []);
|
||||
if (records.length === 0) {
|
||||
await pause(POLL_MS);
|
||||
continue;
|
||||
}
|
||||
lastActivityAt = Date.now();
|
||||
|
||||
for (const record of records) {
|
||||
if (isSuccessfulCompletion(record)) {
|
||||
attempts = 0;
|
||||
lastResumeAt = 0;
|
||||
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,
|
||||
});
|
||||
attempts += 1;
|
||||
lastResumeAt = Date.now();
|
||||
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,
|
||||
});
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
} finally {
|
||||
await log("watcher_stopped");
|
||||
await unlink(paths.lock).catch(() => {});
|
||||
await unlink(paths.stop).catch(() => {});
|
||||
}
|
||||
}
|
||||
|
||||
async function signalStop(input) {
|
||||
if (!input.session_id) return;
|
||||
const path = sessionPaths(input.session_id).stop;
|
||||
await mkdir(dirname(path), { recursive: true });
|
||||
const temporary = `${path}.${process.pid}.tmp`;
|
||||
await writeFile(temporary, "stop\n", "utf8");
|
||||
await rename(temporary, path);
|
||||
}
|
||||
|
||||
function launchWatcher(input) {
|
||||
const environment = {
|
||||
...process.env,
|
||||
CODEX_429_SESSION_ID: input.session_id,
|
||||
CODEX_429_TRANSCRIPT_PATH: input.transcript_path,
|
||||
CODEX_429_CWD: input.cwd || "",
|
||||
CODEX_429_MODEL: input.model || "",
|
||||
};
|
||||
const child = spawn(process.execPath, [fileURLToPath(import.meta.url), "watch"], {
|
||||
detached: true,
|
||||
env: environment,
|
||||
stdio: "ignore",
|
||||
windowsHide: true,
|
||||
});
|
||||
child.unref();
|
||||
}
|
||||
|
||||
async function launchLoadedSessionWatchers(input) {
|
||||
const sessionIds = new Set([input.session_id]);
|
||||
const loaded = await loadedDesktopSessionIds({ sessionId: input.session_id });
|
||||
for (const sessionId of loaded) sessionIds.add(sessionId);
|
||||
const transcripts = await findSessionTranscripts(sessionIds);
|
||||
if (input.transcript_path) transcripts.set(input.session_id, input.transcript_path);
|
||||
|
||||
for (const [sessionId, transcriptPath] of transcripts) {
|
||||
launchWatcher({
|
||||
session_id: sessionId,
|
||||
transcript_path: transcriptPath,
|
||||
cwd: sessionId === input.session_id ? input.cwd : "",
|
||||
model: sessionId === input.session_id ? input.model : "",
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async function main() {
|
||||
const mode = process.argv[2] || "start";
|
||||
if (mode === "watch") {
|
||||
await watch({
|
||||
sessionId: process.env.CODEX_429_SESSION_ID,
|
||||
transcriptPath: process.env.CODEX_429_TRANSCRIPT_PATH,
|
||||
cwd: process.env.CODEX_429_CWD,
|
||||
model: process.env.CODEX_429_MODEL,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const input = await readHookInput();
|
||||
if (process.env.CODEX_429_AUTO_CONTINUE_CHILD === "1") return;
|
||||
if (mode === "stop") {
|
||||
await signalStop(input);
|
||||
return;
|
||||
}
|
||||
if ((input.hook_event_name === "SessionStart" || input.hook_event_name === "UserPromptSubmit")
|
||||
&& input.session_id
|
||||
&& input.transcript_path) {
|
||||
await launchLoadedSessionWatchers(input);
|
||||
}
|
||||
}
|
||||
|
||||
if (process.argv[1] && fileURLToPath(import.meta.url) === process.argv[1]) {
|
||||
await main();
|
||||
}
|
||||
@@ -0,0 +1,40 @@
|
||||
{
|
||||
"description": "Watch a Codex task and resume it after a structured HTTP 429 failure.",
|
||||
"hooks": {
|
||||
"SessionStart": [
|
||||
{
|
||||
"matcher": "startup|resume",
|
||||
"hooks": [
|
||||
{
|
||||
"type": "command",
|
||||
"command": "node \"${PLUGIN_ROOT}/hooks/codex-429-auto-continue.mjs\"",
|
||||
"timeout": 5,
|
||||
"statusMessage": "Starting the Codex 429 watcher"
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"UserPromptSubmit": [
|
||||
{
|
||||
"hooks": [
|
||||
{
|
||||
"type": "command",
|
||||
"command": "node \"${PLUGIN_ROOT}/hooks/codex-429-auto-continue.mjs\"",
|
||||
"timeout": 5
|
||||
}
|
||||
]
|
||||
}
|
||||
],
|
||||
"SessionEnd": [
|
||||
{
|
||||
"hooks": [
|
||||
{
|
||||
"type": "command",
|
||||
"command": "node \"${PLUGIN_ROOT}/hooks/codex-429-auto-continue.mjs\" stop",
|
||||
"timeout": 3
|
||||
}
|
||||
]
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,182 @@
|
||||
import assert from "node:assert/strict";
|
||||
import test from "node:test";
|
||||
|
||||
import {
|
||||
hasTaskStartedAfter,
|
||||
is429Record,
|
||||
isSuccessfulCompletion,
|
||||
loadedThreadIds,
|
||||
RETRY_DELAYS_MS,
|
||||
sendDesktopContinuation,
|
||||
shouldResumeAfterDelay,
|
||||
threadListContains,
|
||||
} from "../hooks/codex-429-auto-continue.mjs";
|
||||
|
||||
const failed = {
|
||||
ordinal: 10,
|
||||
type: "event_msg",
|
||||
payload: {
|
||||
type: "task_complete",
|
||||
turn_id: "turn-1",
|
||||
error: {
|
||||
message: "exceeded retry limit, last status: 429 Too Many Requests",
|
||||
codex_error_info: {
|
||||
response_too_many_failed_attempts: { http_status_code: 429 },
|
||||
},
|
||||
},
|
||||
},
|
||||
};
|
||||
|
||||
test("recognizes only a structured 429 task failure", () => {
|
||||
assert.equal(is429Record(failed), true);
|
||||
assert.equal(is429Record({ type: "message", message: failed.payload.error.message }), false);
|
||||
assert.equal(is429Record({ type: "event_msg", payload: { type: "task_complete" } }), false);
|
||||
});
|
||||
|
||||
test("recognizes a successful task completion", () => {
|
||||
assert.equal(isSuccessfulCompletion({
|
||||
type: "event_msg",
|
||||
payload: { type: "task_complete", last_agent_message: "done" },
|
||||
}), true);
|
||||
assert.equal(isSuccessfulCompletion(failed), false);
|
||||
});
|
||||
|
||||
test("detects Codex or user retry activity after a failure", () => {
|
||||
assert.equal(hasTaskStartedAfter([{
|
||||
ordinal: 11,
|
||||
type: "event_msg",
|
||||
payload: { type: "task_started", turn_id: "turn-2" },
|
||||
}], failed), true);
|
||||
assert.equal(hasTaskStartedAfter([{
|
||||
ordinal: 9,
|
||||
type: "event_msg",
|
||||
payload: { type: "task_started", turn_id: "turn-0" },
|
||||
}], failed), false);
|
||||
});
|
||||
|
||||
test("starts the retry chain after three seconds", async () => {
|
||||
assert.deepEqual(RETRY_DELAYS_MS, [3_000, 3_000, 3_000, 3_000]);
|
||||
let waited;
|
||||
const shouldResume = await shouldResumeAfterDelay(failed, "transcript.jsonl", RETRY_DELAYS_MS[0], {
|
||||
pause: async (delay) => { waited = delay; },
|
||||
readTranscriptTail: async () => [failed],
|
||||
});
|
||||
assert.equal(waited, 3_000);
|
||||
assert.equal(shouldResume, true);
|
||||
|
||||
const builtInRetryWon = await shouldResumeAfterDelay(failed, "transcript.jsonl", 3_000, {
|
||||
pause: async () => {},
|
||||
readTranscriptTail: async () => [failed, {
|
||||
ordinal: 11,
|
||||
type: "event_msg",
|
||||
payload: { type: "task_started", turn_id: "turn-2" },
|
||||
}],
|
||||
});
|
||||
assert.equal(builtInRetryWon, false);
|
||||
});
|
||||
|
||||
test("finds a target task in a Codex desktop tool result", () => {
|
||||
const result = {
|
||||
contentItems: [{
|
||||
type: "inputText",
|
||||
text: JSON.stringify({ pinnedThreads: [], threads: [{ id: "session-1" }] }),
|
||||
}],
|
||||
success: true,
|
||||
};
|
||||
assert.equal(threadListContains(result, "session-1"), true);
|
||||
assert.equal(threadListContains(result, "session-2"), false);
|
||||
});
|
||||
|
||||
test("selects existing loaded tasks for watcher backfill", () => {
|
||||
const result = {
|
||||
contentItems: [{
|
||||
type: "inputText",
|
||||
text: JSON.stringify({
|
||||
pinnedThreads: [{ id: "active-1", status: "active" }],
|
||||
threads: [
|
||||
{ id: "idle-1", status: "idle" },
|
||||
{ id: "failed-1", status: "systemError" },
|
||||
{ id: "closed-1", status: "notLoaded" },
|
||||
],
|
||||
}),
|
||||
}],
|
||||
success: true,
|
||||
};
|
||||
assert.deepEqual(loadedThreadIds(result), ["active-1", "idle-1", "failed-1"]);
|
||||
});
|
||||
|
||||
test("sends continuation through the desktop pipe that owns the task", async () => {
|
||||
const calls = [];
|
||||
const result = await sendDesktopContinuation({ sessionId: "session-1", failedTurnId: "turn-1" }, {
|
||||
pipeCandidates: async () => ["wrong-pipe", "right-pipe"],
|
||||
pipeRequest: async (pipePath, method, params) => {
|
||||
calls.push({ pipePath, method, params });
|
||||
if (method === "tools/list") {
|
||||
if (pipePath === "wrong-pipe") throw new Error("wrong pipe");
|
||||
return { tools: [{ name: "list_threads" }, { name: "send_message_to_thread" }] };
|
||||
}
|
||||
if (params.tool === "list_threads") {
|
||||
return {
|
||||
success: true,
|
||||
contentItems: [{
|
||||
type: "inputText",
|
||||
text: JSON.stringify({
|
||||
pinnedThreads: [],
|
||||
threads: pipePath === "right-pipe" ? [{ id: "session-1" }] : [],
|
||||
}),
|
||||
}],
|
||||
};
|
||||
}
|
||||
return { success: true, contentItems: [{ type: "inputText", text: "{}" }] };
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(result.pipePath, "right-pipe");
|
||||
const listed = calls.find((call) => call.params.tool === "list_threads");
|
||||
assert.equal(listed.params.arguments.limit, 50);
|
||||
const sent = calls.find((call) => call.params.tool === "send_message_to_thread");
|
||||
assert.equal(sent.pipePath, "right-pipe");
|
||||
assert.equal(sent.params.arguments.threadId, "session-1");
|
||||
assert.equal(sent.params.arguments.prompt, "继续");
|
||||
assert.equal(result.delivery, "message");
|
||||
});
|
||||
|
||||
test("prefers a native continuation tool when Codex exposes one", async () => {
|
||||
const calls = [];
|
||||
const result = await sendDesktopContinuation({ sessionId: "session-1", failedTurnId: "turn-1" }, {
|
||||
pipeCandidates: async () => ["app-pipe"],
|
||||
pipeRequest: async (pipePath, method, params) => {
|
||||
calls.push({ pipePath, method, params });
|
||||
if (method === "tools/list") {
|
||||
return {
|
||||
tools: [
|
||||
{ name: "list_threads" },
|
||||
{
|
||||
name: "retry_turn",
|
||||
inputSchema: {
|
||||
properties: { threadId: {}, turnId: {} },
|
||||
required: ["threadId", "turnId"],
|
||||
},
|
||||
},
|
||||
{ name: "send_message_to_thread" },
|
||||
],
|
||||
};
|
||||
}
|
||||
if (params.tool === "list_threads") {
|
||||
return {
|
||||
success: true,
|
||||
contentItems: [{
|
||||
type: "inputText",
|
||||
text: JSON.stringify({ pinnedThreads: [], threads: [{ id: "session-1" }] }),
|
||||
}],
|
||||
};
|
||||
}
|
||||
return { success: true, contentItems: [{ type: "inputText", text: "{}" }] };
|
||||
},
|
||||
});
|
||||
|
||||
assert.equal(result.delivery, "native");
|
||||
const nativeCall = calls.find((call) => call.params?.tool === "retry_turn");
|
||||
assert.deepEqual(nativeCall.params.arguments, { threadId: "session-1", turnId: "turn-1" });
|
||||
assert.equal(calls.some((call) => call.params?.tool === "send_message_to_thread"), false);
|
||||
});
|
||||
Reference in New Issue
Block a user