From 6661bf26f138e0486afb17591d2cdc7d0343c601 Mon Sep 17 00:00:00 2001 From: cneicy Date: Fri, 11 Sep 2026 01:29:34 +0800 Subject: [PATCH] feat: publish Codex 429 auto-continue plugin --- .agents/plugins/marketplace.json | 20 + .gitignore | 3 + CHANGELOG.md | 11 + CONTRIBUTING.md | 17 + LICENSE | 21 + README.md | 60 ++ README.zh-CN.md | 60 ++ SECURITY.md | 11 + package.json | 14 + .../.codex-plugin/plugin.json | 19 + plugins/codex-429-auto-continue/README.md | 9 + .../hooks/codex-429-auto-continue.mjs | 523 ++++++++++++++++++ .../codex-429-auto-continue/hooks/hooks.json | 40 ++ .../tests/hook.test.mjs | 182 ++++++ 14 files changed, 990 insertions(+) create mode 100644 .agents/plugins/marketplace.json create mode 100644 .gitignore create mode 100644 CHANGELOG.md create mode 100644 CONTRIBUTING.md create mode 100644 LICENSE create mode 100644 README.md create mode 100644 README.zh-CN.md create mode 100644 SECURITY.md create mode 100644 package.json create mode 100644 plugins/codex-429-auto-continue/.codex-plugin/plugin.json create mode 100644 plugins/codex-429-auto-continue/README.md create mode 100644 plugins/codex-429-auto-continue/hooks/codex-429-auto-continue.mjs create mode 100644 plugins/codex-429-auto-continue/hooks/hooks.json create mode 100644 plugins/codex-429-auto-continue/tests/hook.test.mjs diff --git a/.agents/plugins/marketplace.json b/.agents/plugins/marketplace.json new file mode 100644 index 0000000..5a73b6a --- /dev/null +++ b/.agents/plugins/marketplace.json @@ -0,0 +1,20 @@ +{ + "name": "codex-429-auto-continue", + "interface": { + "displayName": "Codex 429 Auto Continue" + }, + "plugins": [ + { + "name": "codex-429-auto-continue", + "source": { + "source": "local", + "path": "./plugins/codex-429-auto-continue" + }, + "policy": { + "installation": "AVAILABLE", + "authentication": "ON_INSTALL" + }, + "category": "Productivity" + } + ] +} diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3c45938 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +node_modules/ +*.log +.DS_Store diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..17425b7 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,11 @@ +# Changelog + +All notable changes to this project are documented in this file. + +## 0.1.0 - 2026-09-11 + +- Detect structured Codex HTTP 429 task failures. +- Continue the original task after a fixed 3-second wait. +- Prefer native continuation and fall back to sending `继续`. +- Support new, reopened, and currently loaded tasks. +- Prevent duplicate per-task watchers and false positives from ordinary text. diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md new file mode 100644 index 0000000..34020ce --- /dev/null +++ b/CONTRIBUTING.md @@ -0,0 +1,17 @@ +# Contributing / 参与贡献 + +Use Node.js 20 or newer. Keep changes focused on reliable Codex 429 recovery and add a regression test for behavior changes. + +使用 Node.js 20 或更高版本。修改应聚焦于 Codex 429 续跑，并为行为变更补充回归测试。 + +Run the test suite before submitting a change: + +提交修改前请运行： + +```powershell +npm test +``` + +Do not commit task transcripts, plugin data, tokens, pipe names, or other machine-specific information. + +请勿提交任务会话记录、插件数据、令牌、管道名称或其他机器相关信息。 diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..afb5ee3 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2026 cneicy + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/README.md b/README.md new file mode 100644 index 0000000..d77f62e --- /dev/null +++ b/README.md @@ -0,0 +1,60 @@ +# Codex 429 Auto Continue + +[简体中文](README.zh-CN.md) + +A Codex Desktop plugin that continues a task after Codex exhausts its built-in retries with HTTP 429. + +## Features + +- Detects only structured `response_too_many_failed_attempts` failures with HTTP status 429. +- Waits exactly 3 seconds for every continuation attempt, with at most four plugin-initiated attempts in one failure chain. +- Uses a native retry or continue tool when Codex exposes one; otherwise sends only `继续` to the original task. +- Covers new and reopened tasks through `SessionStart` and backfills all currently loaded tasks through `UserPromptSubmit`. +- Cancels its pending continuation when Codex or the user has already started another turn. +- Uses one watcher per task to avoid duplicate continuations. + +## Requirements + +- Codex Desktop on Windows. +- Node.js available as `node` on `PATH`. + +The plugin currently depends on Codex Desktop's internal local task-tool pipe. That interface is not a stable public API and may require compatibility updates after a Codex release. + +## Install + +```powershell +codex plugin marketplace add https://git.crash.work/cneicy/codex-429-auto-continue.git +codex plugin add codex-429-auto-continue@codex-429-auto-continue +``` + +Start or reopen a task after installation. If Codex asks you to review the hook, approve it from `/hooks`. + +## Behavior + +The watcher reads new records from the task transcript. After a structured 429 task failure, it waits 3 seconds and checks whether a new turn has already started. If not, it attempts native continuation and falls back to sending `继续`. + +A successful task completion resets the failure chain. The chain also resets after 15 minutes without a plugin-initiated continuation. A watcher exits when its task ends or after 24 hours without transcript activity. + +Prompt text, assistant text, and tool output that merely contain `429` do not trigger the plugin. + +## Local Data + +Per-task JSONL event logs and lock files are stored under: + +```text +%USERPROFILE%\.codex\plugin-data\codex-429-auto-continue +``` + +The plugin does not make its own network requests. It reads local Codex transcripts and communicates with the local Codex Desktop task pipe. + +## Development + +```powershell +npm test +``` + +The installable plugin is under `plugins/codex-429-auto-continue`. The marketplace manifest is `.agents/plugins/marketplace.json`. + +## License + +[MIT](LICENSE) diff --git a/README.zh-CN.md b/README.zh-CN.md new file mode 100644 index 0000000..e5dc276 --- /dev/null +++ b/README.zh-CN.md @@ -0,0 +1,60 @@ +# Codex 429 自动继续 + +[English](README.md) + +这是一个 Codex Desktop 插件。当 Codex 内置重试耗尽并以 HTTP 429 中断任务时，插件会自动继续原任务。 + +## 功能 + +- 只识别结构化的 `response_too_many_failed_attempts` HTTP 429 失败记录。 +- 每次续跑统一等待 3 秒；同一失败链最多由插件发起四次续跑。 +- Codex 提供原生重试或继续工具时优先使用；否则只向原任务发送“继续”。 +- 通过 `SessionStart` 支持新建和重新打开的任务，通过 `UserPromptSubmit` 为安装前已经加载的任务补启动监听。 +- 如果 Codex 或用户已经开始新一轮任务，取消当前等待中的续跑。 +- 每个任务只运行一个监听进程，避免重复续跑。 + +## 运行要求 + +- Windows 版 Codex Desktop。 +- `PATH` 中可以直接调用 `node`。 + +插件目前依赖 Codex Desktop 的本地内部任务工具管道。这个接口不是稳定的公开 API，Codex 更新后可能需要同步适配。 + +## 安装 + +```powershell +codex plugin marketplace add https://git.crash.work/cneicy/codex-429-auto-continue.git +codex plugin add codex-429-auto-continue@codex-429-auto-continue +``` + +安装后新建或重新打开一个任务。如果 Codex 要求审查 hook，请在 `/hooks` 中批准。 + +## 工作方式 + +监听进程持续读取任务会话记录。发现结构化 429 失败后，先等待 3 秒，再检查是否已经开始新一轮任务；如果没有，就优先调用原生继续能力，不可用时发送“继续”。 + +任务成功完成后会重置失败链；连续 15 分钟没有由插件发起续跑也会重置。任务结束或会话记录连续 24 小时没有变化时，监听进程退出。 + +用户提示、助手回复或工具输出中仅仅出现 `429` 字样，不会触发插件。 + +## 本地数据 + +每个任务的 JSONL 事件日志和锁文件保存在： + +```text +%USERPROFILE%\.codex\plugin-data\codex-429-auto-continue +``` + +插件不会自行发起网络请求。它只读取本地 Codex 会话记录，并通过本地 Codex Desktop 任务工具管道继续任务。 + +## 开发 + +```powershell +npm test +``` + +可安装插件位于 `plugins/codex-429-auto-continue`，marketplace 清单位于 `.agents/plugins/marketplace.json`。 + +## 许可证 + +[MIT](LICENSE) diff --git a/SECURITY.md b/SECURITY.md new file mode 100644 index 0000000..1b92cee --- /dev/null +++ b/SECURITY.md @@ -0,0 +1,11 @@ +# Security Policy / 安全策略 + +## Reporting / 报告方式 + +Do not publish tokens, Codex transcripts, session identifiers, or local pipe details in a public issue. Report security concerns privately to `im@crash.work` with reproduction steps and the affected version. + +请勿在公开 Issue 中发布令牌、Codex 会话记录、会话 ID 或本地管道信息。请将复现步骤和受影响版本通过 `im@crash.work` 私下报告。 + +Only the latest released version receives security fixes. + +仅当前最新发布版本提供安全修复。 diff --git a/package.json b/package.json new file mode 100644 index 0000000..b0d8bf7 --- /dev/null +++ b/package.json @@ -0,0 +1,14 @@ +{ + "name": "codex-429-auto-continue", + "version": "0.1.0", + "private": true, + "description": "Automatically continue Codex Desktop tasks after exhausted HTTP 429 retries.", + "type": "module", + "scripts": { + "test": "node --test plugins/codex-429-auto-continue/tests/hook.test.mjs" + }, + "engines": { + "node": ">=20" + }, + "license": "MIT" +} diff --git a/plugins/codex-429-auto-continue/.codex-plugin/plugin.json b/plugins/codex-429-auto-continue/.codex-plugin/plugin.json new file mode 100644 index 0000000..e29a2a0 --- /dev/null +++ b/plugins/codex-429-auto-continue/.codex-plugin/plugin.json @@ -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." + } +} diff --git a/plugins/codex-429-auto-continue/README.md b/plugins/codex-429-auto-continue/README.md new file mode 100644 index 0000000..676c167 --- /dev/null +++ b/plugins/codex-429-auto-continue/README.md @@ -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. diff --git a/plugins/codex-429-auto-continue/hooks/codex-429-auto-continue.mjs b/plugins/codex-429-auto-continue/hooks/codex-429-auto-continue.mjs new file mode 100644 index 0000000..3dcd175 --- /dev/null +++ b/plugins/codex-429-auto-continue/hooks/codex-429-auto-continue.mjs @@ -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(); +} diff --git a/plugins/codex-429-auto-continue/hooks/hooks.json b/plugins/codex-429-auto-continue/hooks/hooks.json new file mode 100644 index 0000000..1324972 --- /dev/null +++ b/plugins/codex-429-auto-continue/hooks/hooks.json @@ -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 + } + ] + } + ] + } +} diff --git a/plugins/codex-429-auto-continue/tests/hook.test.mjs b/plugins/codex-429-auto-continue/tests/hook.test.mjs new file mode 100644 index 0000000..b21a111 --- /dev/null +++ b/plugins/codex-429-auto-continue/tests/hook.test.mjs @@ -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); +});