diff --git a/.github/release-notes/v0.4.0.md b/.github/release-notes/v0.4.0.md new file mode 100644 index 0000000..bc76995 --- /dev/null +++ b/.github/release-notes/v0.4.0.md @@ -0,0 +1,30 @@ +# WAKE v0.4.0 + +WAKE now resumes long-running work by persisting the **task**, not by reopening the old Codex chat. + +## Highlights + +- Replaces automatic `codex exec resume ` with checkpoint-first job continuation. +- Stores durable jobs under `~/.codex/wake/jobs/`. +- Starts a fresh lightweight Codex thread after quota recovery and reads `checkpoint.md` first. +- Keeps `ordinaryUsageAllowed` as the authoritative recovery signal; reset timestamps remain hints only. +- Adds bounded retry handling for temporary model-capacity and network failures. +- Adds explicit job states for `retry_waiting`, `stopped`, and `needs_attention`. +- Supports optional per-job model and reasoning-effort overrides. +- Preserves job/checkpoint history across Skill reinstall. + +## Validation + +Validated on Windows with: + +- Python syntax checks. +- Unit tests for checkpoint lifecycle, quota transitions, retry behavior, and handoff launch. +- Installer/start/stop/status lifecycle tests. +- A real Codex handoff through the recovered-quota path that created a new thread, read the checkpoint, produced and byte-verified the requested artifact, updated the checkpoint, and marked the job `completed`. + +The real handoff also exercised transient timeout/capacity retries before succeeding. + +## Compatibility note + +v0.3.2 exact-thread resume could hit a Desktop active-writer conflict. v0.4.0 deliberately avoids taking ownership of the original Desktop thread. + diff --git a/.github/workflows/validate.yml b/.github/workflows/validate.yml index f3f536e..c58ccca 100644 --- a/.github/workflows/validate.yml +++ b/.github/workflows/validate.yml @@ -14,7 +14,7 @@ jobs: - name: Check out repository uses: actions/checkout@v4 - - name: Validate WAKE metadata + - name: Validate metadata and checkpoint-first contract shell: python run: | from pathlib import Path @@ -23,40 +23,41 @@ jobs: skill = Path("SKILL.md") agent = Path("agents/openai.yaml") watcher = Path("watcher/wake_watcher.py") + docs = Path("docs/quota-watcher.md") - assert skill.is_file() - assert agent.is_file() - assert watcher.is_file() + for path in (skill, agent, watcher, docs): + assert path.is_file(), f"Missing {path}" text = skill.read_text(encoding="utf-8") - assert text.startswith("---\n") - assert re.search(r"(?m)^name:\s*wake\s*$", text) - - required = [ - "$wake", - "$wake start", - "$wake stop", - "$wake status", - "CODEX_THREAD_ID", - "ordinaryUsageAllowed", - "QUOTA_BLOCKED", - "WAITING_USER", - "BYMINUTE=0,15,30,45", - ] - for item in required: - assert item in text, f"Missing required content: {item}" - - version = Path("VERSION").read_text(encoding="utf-8").strip() - assert version in {"0.2.0", "0.3.2"} - + agent_text = agent.read_text(encoding="utf-8") watcher_text = watcher.read_text(encoding="utf-8") - assert 'VERSION = "0.3.2"' in watcher_text - - name: Compile watcher - run: python -m py_compile watcher/wake_watcher.py + assert text.startswith("---\n") + assert re.search(r"(?m)^name:\s*wake\s*$", text) + for required in ( + "$wake", "$wake start", "$wake stop", "$wake status", + "checkpoint", "quota_waiting", "needs_attention", + "ordinaryUsageAllowed", "job-create-current", "job-complete", + ): + assert required in text, f"Missing SKILL contract: {required}" + + assert "codex exec resume" not in watcher_text + assert '"exec"' in watcher_text + assert "checkpoint.md" in watcher_text + assert re.search( + r"(?m)^\s*allow_implicit_invocation:\s*true\s*$", + agent_text, + ) + + - name: Run Python tests + run: | + python -m py_compile watcher/wake_watcher.py + python -m unittest discover -s tests -v - name: Check shell scripts run: | - for file in scripts/*.sh; do - bash -n "$file" - done + bash -n scripts/install.sh + bash -n scripts/uninstall.sh + bash -n scripts/start-watcher.sh + bash -n scripts/status-watcher.sh + bash -n scripts/stop-watcher.sh diff --git a/.gitignore b/.gitignore index f1755e6..7fabadf 100644 --- a/.gitignore +++ b/.gitignore @@ -1,9 +1,3 @@ -# Runtime state -.state/ -*.log -__pycache__/ -*.py[cod] - # OS .DS_Store Thumbs.db @@ -14,6 +8,10 @@ Thumbs.db *.swp *.swo +# Python +__pycache__/ +*.py[cod] + # Temporary files *.tmp *.bak diff --git a/CHANGELOG.md b/CHANGELOG.md index a953aa6..b215cf1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,31 @@ # Changelog +All notable changes to this project will be documented here. + +## [0.4.0] - 2026-09-26 + +### Changed + +- Replaced thread-resume recovery with checkpoint-first job continuation. +- WAKE now persists durable jobs under `~/.codex/wake/jobs/`. +- Quota recovery launches a new lightweight `codex exec` thread instead of reopening the old chat. +- The watcher uses one hidden persistent `codex app-server` for quota reads. +- Skill reinstall no longer owns or deletes durable job history. + +### Added + +- Compact `checkpoint.md` handoff contract. +- Job states: draft, monitoring, quota_waiting, running, retry_waiting, waiting_user, needs_attention, stopped, completed. +- Commands: `job-create-current`, `job-arm`, `job-status`, `job-list`, `job-pause`, `job-rearm`, `job-stop`, `job-complete`. +- Optional per-job model and reasoning-effort overrides. +- Bounded retries for temporary capacity/network failures. +- Unit tests for checkpoint lifecycle, quota transitions, handoff launch, and transient retries. + +### Removed + +- Automatic `codex exec resume `. +- Dependency on Desktop Goal owner/control-socket recovery for the primary path. + ## [0.3.2] - 2026-09-24 ### Added @@ -26,8 +52,7 @@ ### Known limitation -- Exact quota recovery currently resumes through detached `codex exec resume `; - Codex Desktop UI synchronization may vary by version. +- Exact quota recovery used detached `codex exec resume `; v0.4.0 replaces that path with checkpoint-first job continuation. ## [0.2.0] - 2026-09-19 @@ -38,11 +63,14 @@ - Automatic pause when progress requires user input, approval, credentials, confirmation, restart, or another manual action. - Re-arm flow after the user resolves a WAITING_USER blocker. -- State-machine documentation. +- Optional implicit invocation for re-arming an already-paused WAKE conversation. +- State-machine documentation and central-supervisor design notes. -### Fixed +### Changed -- Prevent repeated quarter-hour wakeups while Codex is waiting for the user. +- WAKE no longer treats every unfinished conversation as immediately runnable. +- Scheduled runs classify state before doing project work. +- `$wake status` now reports paused state and pause reason when available. ## [0.1.0] - 2026-09-18 @@ -51,6 +79,6 @@ - Initial public WAKE skill. - `$wake`, `$wake start`, `$wake stop`, and `$wake status`. - Shared quarter-hour synchronization at `:00 / :15 / :30 / :45`. -- Windows/macOS/Linux install scripts. -- English and Simplified Chinese documentation. -- GitHub Actions validation. +- Duplicate-schedule prevention guidance. +- Windows PowerShell install/uninstall scripts. +- macOS/Linux install/uninstall scripts. diff --git a/README.md b/README.md index 9b59db9..ff8f443 100644 --- a/README.md +++ b/README.md @@ -1,34 +1,39 @@ # wake -[![Validate skill](https://github.com/Gordon-Jang/codex-wake/actions/workflows/validate.yml/badge.svg)](https://github.com/Gordon-Jang/codex-wake/actions/workflows/validate.yml) -[![GitHub release](https://img.shields.io/github/v/release/Gordon-Jang/codex-wake)](https://github.com/Gordon-Jang/codex-wake/releases) -[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) +**wake** is a Codex Skill for long-running tasks that may span multiple quota windows. -[English](README.md) · [简体中文](README.zh-CN.md) +## v0.4.0: persist the task, not the chat -**wake** is a state-aware Codex Skill for long-running work. v0.3.2 adds a local -quota watcher so quota recovery no longer depends on blind 15-minute model retries. +Earlier WAKE versions tried to resume the original Codex thread. On Codex Desktop that +can conflict with the active writer already owned by Desktop. v0.4 changes the recovery +unit from a thread to a durable **job**. -## What v0.3.2 does +A job stores a compact checkpoint under: -- registers the exact current Codex thread from `CODEX_THREAD_ID` -- reads quota locally through `codex app-server` -- treats `ordinaryUsageAllowed` as the recovery authority -- uses `resetsAt` only to decide when to check again -- tracks `monitoring -> quota_waiting -> monitoring` -- never wakes `waiting_user` threads -- never uses `--last` or guesses a thread id -- keeps synchronized Scheduled Tasks as a conservative fallback +`~/.codex/wake/jobs//checkpoint.md` -> [!IMPORTANT] -> WAKE does not bypass OpenAI usage limits, quotas, rate limits, credits, or safety controls. +When included usage is unavailable, WAKE waits locally without making model turns. When +`ordinaryUsageAllowed` becomes true again, WAKE starts a **new** lightweight +`codex exec` thread that reads the checkpoint first and continues only unfinished work. + +## Why this design + +- avoids `thread already has an active writer` conflicts; +- does not depend on Codex Desktop Goal UI automation; +- does not require the experimental Windows app-server daemon; +- avoids blindly resuming a huge old chat; +- keeps quota checks local through one hidden persistent `codex app-server`; +- never infers recovery from reset time alone. + +WAKE does not bypass OpenAI usage limits, credits, rate limits, or safety controls. ## Requirements - Codex CLI available as `codex` - Python 3 -- Codex authentication already configured -- a Codex host that exposes `CODEX_THREAD_ID` to shell/tool execution for exact-thread registration +- existing Codex authentication +- a Codex host that exposes `CODEX_THREAD_ID` to shell/tool execution when `$wake` + creates the initial job ## Install @@ -37,81 +42,65 @@ Windows: ```powershell git clone https://github.com/Gordon-Jang/codex-wake.git cd codex-wake -.scriptsinstall.cmd +.\scripts\install.cmd ``` Default install path: -```text -%USERPROFILE%.codexskillswake -``` +`%USERPROFILE%\.codex\skills\wake` -macOS/Linux: +Durable job/checkpoint data lives separately under: -```bash -git clone https://github.com/Gordon-Jang/codex-wake.git -cd codex-wake -./scripts/install.sh -``` +`%USERPROFILE%\.codex\wake` -## Verify quota integration +Reinstalling the Skill does not erase that job history. -```powershell -cd $HOME.codexskillswake -python .watcherwake_watcher.py doctor -``` +## Use -Working integration reports: +In the Codex conversation that owns the task: ```text -"quotaReadOk": true +$wake ``` -## Start watcher +The Skill creates a job, writes a compact checkpoint from the current task context, arms +the job, starts the local watcher, and continues working. -Windows: +Useful diagnostics: ```powershell -.scriptsstart-watcher.cmd -.scriptsstatus-watcher.cmd +python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" doctor +python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" status +python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" job-list ``` -Then, in the Codex conversation to manage: +For advanced use, `job-create` / `job-create-current` also accept `--model` and `--reasoning-effort`; if omitted, Codex uses the normal configuration. -```text -$wake -``` +## Job states -The Skill should register the exact thread from inside that conversation. A normal -external terminal does not have `CODEX_THREAD_ID`, so `register-current` is -intentionally rejected there. +- `draft` — checkpoint exists but is not armed; +- `monitoring` — current work is active and quota is watched; +- `quota_waiting` — ordinary included usage is unavailable; +- `running` — a checkpoint handoff thread is executing; +- `retry_waiting` — transient capacity/network failure; retries with bounded backoff; +- `waiting_user` — explicit user/manual dependency; never auto-launched; +- `needs_attention` — continuation exited unexpectedly; never loops blindly; +- `stopped` — user disabled automatic continuation but kept the checkpoint; +- `completed` — verified done; ignored by watcher. -Check registrations: +## Recovery contract -```powershell -python $HOME.codexskillswakewatcherwake_watcher.py list -``` +A handoff thread reads the checkpoint first, then normally checks only +`git status --short` and `git diff --stat`. It is explicitly told not to reconstruct, +resume, or reread the previous chat. -## State behavior +See [docs/quota-watcher.md](docs/quota-watcher.md) for the state machine. -| State | Behavior | -|---|---| -| ACTIVE | Do not duplicate work | -| WAITING_USER | Pause; quota refresh must not wake it | -| QUOTA_BLOCKED | Watch locally until ordinary usage is actually allowed | -| RETRYABLE_BLOCKED | Use conservative Scheduled Task fallback | -| RUNNABLE | Continue unfinished work | -| DONE | Unregister and stop | -| UNKNOWN | Ask once, then pause | +## Development -See [docs/quota-watcher.md](docs/quota-watcher.md). - -## Known limitation - -Quota recovery currently launches exact `codex exec resume `. It is -safer than `--last`, but a detached CLI continuation may not always render in -Codex Desktop exactly like a native Desktop turn. - -## License +```powershell +python -m py_compile .\watcher\wake_watcher.py +python -m unittest discover -s tests -v +``` -MIT. +v0.4.0 has been validated with unit tests plus a real Windows handoff through the recovered-quota path: a fresh Codex thread read the checkpoint, produced and byte-verified the requested artifact, updated the checkpoint, and marked the job completed. diff --git a/README.zh-CN.md b/README.zh-CN.md index 290e431..77a3ce8 100644 --- a/README.zh-CN.md +++ b/README.zh-CN.md @@ -1,99 +1,111 @@ # wake -[![Validate skill](https://github.com/Gordon-Jang/codex-wake/actions/workflows/validate.yml/badge.svg)](https://github.com/Gordon-Jang/codex-wake/actions/workflows/validate.yml) -[![GitHub release](https://img.shields.io/github/v/release/Gordon-Jang/codex-wake)](https://github.com/Gordon-Jang/codex-wake/releases) -[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](LICENSE) +**wake** 是一个用于跨多个额度窗口持续执行长任务的 Codex Skill。 -[English](README.md) · [简体中文](README.zh-CN.md) +## v0.4.0:恢复任务,不恢复旧聊天 -**wake** 是一个用于长时间 Codex 任务的状态感知 Skill。v0.3.2 新增本地 -quota watcher,额度耗尽后不再依赖每 15 分钟盲目发送模型任务。 +旧版 WAKE 尝试恢复原来的 Codex thread。在 Codex Desktop 中,这可能和 Desktop +已经持有的 active writer 冲突。v0.4 把恢复单位从 thread 改成了持久化的 **job**。 -## v0.3.2 +每个 job 都保存一个紧凑 checkpoint: -- 从 Codex 对话内部的 `CODEX_THREAD_ID` 登记精确线程 -- 通过本机 `codex app-server` 读取额度 -- 以 `ordinaryUsageAllowed` 作为额度真正恢复的依据 -- `resetsAt` 只用于决定下一次检查时间 -- 自动执行 `monitoring -> quota_waiting -> monitoring` -- `waiting_user` 状态即使额度恢复也不会被唤醒 -- 永远不使用 `--last`,也不猜线程 ID -- 原来的整刻钟 Scheduled Task 只作为保守后备方案 +`~/.codex/wake/jobs//checkpoint.md` -> [!IMPORTANT] -> WAKE 不会绕过 OpenAI 的额度、rate limit、credits 或安全限制。 +额度耗尽时,WAKE 只在本地等待,不发送模型 turn。只有当 +`ordinaryUsageAllowed == true` 时,才启动一个新的轻量 `codex exec` thread, +先读 checkpoint,再只继续未完成的工作。 + +## 这版解决什么 + +- 避免 `thread already has an active writer` 冲突; +- 不依赖 Codex Desktop 的 Goal“继续”按钮; +- 不依赖 Windows 上当前仍较实验性的 app-server daemon; +- 不再为了续跑重新灌入超长旧聊天; +- 额度查询使用一个隐藏、持久的 `codex app-server`; +- 绝不把 reset time 当作额度已经恢复的证明。 + +WAKE 不绕过 OpenAI 的额度、credits、rate limits 或安全机制。 + +## 环境要求 + +- 可用的 Codex CLI:`codex` +- Python 3 +- 已完成 Codex 登录 +- 首次 `$wake` 创建 job 时,Codex shell/tool 环境能提供 `CODEX_THREAD_ID` ## Windows 安装 ```powershell git clone https://github.com/Gordon-Jang/codex-wake.git cd codex-wake -.scriptsinstall.cmd +.\scripts\install.cmd ``` -默认安装到: +默认 Skill 安装目录: -```text -C:Users<用户名>.codexskillswake -``` +`%USERPROFILE%\.codex\skills\wake` -`install.cmd` 会为本次调用使用 ExecutionPolicy Bypass,不需要修改全局策略。 +持久 job/checkpoint 单独保存在: -## 检查额度接口 +`%USERPROFILE%\.codex\wake` -```powershell -cd $HOME.codexskillswake -python .watcherwake_watcher.py doctor -``` +因此重新安装 Skill 不会删除任务 checkpoint。 -正常时应看到: +## 使用 + +在需要持续完成的 Codex 对话中输入: ```text -"quotaReadOk": true +$wake ``` -## 启动 watcher +Skill 会: -```powershell -.scriptsstart-watcher.cmd -.scriptsstatus-watcher.cmd -``` +1. 定义当前任务 Goal; +2. 创建 job; +3. 根据当前上下文生成精简 checkpoint; +4. arm job; +5. 启动本地 watcher; +6. 继续当前任务。 -然后在需要自动续作的 Codex 对话中输入: +常用诊断: -```text -$wake +```powershell +python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" doctor +python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" status +python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" job-list ``` -普通 PowerShell 没有 `CODEX_THREAD_ID`,所以不能在外部终端用 -`register-current` 猜测当前线程。 +高级用法中,`job-create` / `job-create-current` 还支持 `--model` 与 `--reasoning-effort`;不指定时沿用 Codex 的正常配置。 -查看登记: +## Job 状态 -```powershell -python $HOME.codexskillswakewatcherwake_watcher.py list -``` +- `draft`:checkpoint 尚未完成,不能自动续跑; +- `monitoring`:当前工作正常,watcher 只监控额度; +- `quota_waiting`:普通 included usage 不可用; +- `running`:新的 checkpoint handoff thread 正在执行; +- `retry_waiting`:遇到临时容量/网络故障,按有限退避重试; +- `waiting_user`:需要用户输入/人工操作,不自动启动; +- `needs_attention`:续跑进程异常结束,不盲目循环; +- `stopped`:用户停止自动续跑,但保留 checkpoint; +- `completed`:任务已验证完成,watcher 忽略。 + +## 续跑约定 -## 状态逻辑 +新 thread 必须先读 checkpoint,通常只额外检查: -| 状态 | 行为 | -|---|---| -| ACTIVE | 已经在工作,不重复启动 | -| WAITING_USER | 等用户,暂停;额度恢复也不唤醒 | -| QUOTA_BLOCKED | 本地等待额度真正恢复 | -| RETRYABLE_BLOCKED | 使用保守的 Scheduled Task 后备方案 | -| RUNNABLE | 继续未完成工作 | -| DONE | 注销并停止 | -| UNKNOWN | 只问一次,然后暂停 | +- `git status --short` +- `git diff --stat` -详细设计见 [docs/quota-watcher.md](docs/quota-watcher.md)。 +并明确禁止重新构建、resume 或重读上一条超长聊天。 -## 已知限制 +详细状态机见 [docs/quota-watcher.md](docs/quota-watcher.md)。 -恢复动作目前使用精确的 `codex exec resume `。它比 `--last` -安全,但 detached CLI continuation 在某些 Codex Desktop 版本里不一定完全像 -原生 Desktop turn 一样显示。 +## 自测 -## License +```powershell +python -m py_compile .\watcher\wake_watcher.py +python -m unittest discover -s tests -v +``` -MIT。 +v0.4.0 已在 Windows 上通过单元测试与恢复额度路径的真实接班验证:新 Codex thread 从 checkpoint 接班、完成并按字节校验测试产物、更新 checkpoint,并最终将 job 标记为 completed。 diff --git a/SKILL.md b/SKILL.md index e126b09..a2e5f96 100644 --- a/SKILL.md +++ b/SKILL.md @@ -1,199 +1,165 @@ --- name: wake description: > - Manage state-aware continuation for the current Codex conversation. Prefer the - local quota watcher when installed: register the exact current Codex thread via - CODEX_THREAD_ID, monitor account quota locally, and wake the thread when ordinary - usage becomes available again. Fall back to synchronized Scheduled Tasks when - exact-thread registration or the watcher is unavailable. Explicitly invoke with - $wake, $wake start, $wake stop, or $wake status. Re-arm implicitly only when the - user resolves a blocker in a conversation already using WAKE. + Checkpoint-first continuation for long Codex work across quota windows. Persist the + current task as a durable WAKE job with a compact checkpoint. When ordinary included + usage returns, continue the job in a new lightweight Codex exec thread instead of + resuming the old chat. Explicit commands: $wake, $wake start, $wake stop, $wake status. --- -# WAKE +# WAKE v0.4.0 -WAKE keeps long Codex work moving without blindly retrying quota every 15 minutes. +WAKE persists the TASK, not the chat thread. -## Backends +## Core rule -### Local quota watcher +Never use `codex exec resume`, `--last`, recency, title, or another heuristic to +reopen the old conversation. The old conversation is an archive. The durable object is +the WAKE job and its checkpoint. -Preferred for quota exhaustion: +## $wake / $wake start -- register the exact current thread from `CODEX_THREAD_ID` -- read limits locally through `codex app-server` -- use `ordinaryUsageAllowed` as the recovery authority -- use `resetsAt` only as a hint for when to check again -- resume only exact registered threads that observed quota blocking -- never use `--last` and never guess a thread id +For the CURRENT Codex conversation only: -### Scheduled Task fallback - -Use the synchronized quarter-hour grid only when the watcher cannot safely handle -the thread, or for non-quota transient failures: - -`RRULE:FREQ=HOURLY;BYMINUTE=0,15,30,45;BYSECOND=0` - -## Commands - -- `$wake` — same as `$wake start` -- `$wake start` — enable or re-arm WAKE -- `$wake stop` — unregister/stop WAKE for this conversation -- `$wake status` — report backend and state without modifying it - -## START - -When `$wake` or `$wake start` is invoked: - -1. Operate on the CURRENT conversation only. -2. Do not create or infer another thread. -3. Start the local watcher if installed and not already running. -4. From a shell/tool execution inside THIS conversation, register the exact current thread. +1. Start the local watcher if it is not running. Windows: - -`"$HOME\.codex\skills\wake\scripts\register-current.cmd"` +`"$HOME\.codex\skills\wake\scripts\start-watcher.cmd"` macOS/Linux: +`"$HOME/.codex/skills/wake/scripts/start-watcher.sh"` -`"$HOME/.codex/skills/wake/scripts/register-current.sh"` - -The helper must obtain the id from `CODEX_THREAD_ID`. Never substitute recency, -title, project path, ordering, or `--last`. - -5. If exact registration succeeds, prefer the watcher for quota recovery. -6. Use at most one synchronized Scheduled Task as fallback when useful. - -## STATE MACHINE +2. Define one concise task Goal from the user's actual request. +3. From a shell/tool inside THIS conversation, create the job: -### ACTIVE +Windows example: +`python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" job-create-current --cwd "" --goal ""` -Work is already progressing. +The helper MUST obtain the source thread from `CODEX_THREAD_ID`; never guess it. +If the thread already has a non-completed WAKE job, the helper returns that existing job +with `alreadyExists: true` instead of creating a duplicate. +4. Read the returned `checkpointPath` and refresh it into a compact, +authoritative checkpoint containing: -- Do not duplicate work. -- Do not restart running commands, tests, builds, agents, or processes. -- Leave watcher registration in `monitoring`. +- Goal +- Constraints +- Completed +- Decisions +- Files changed +- Verification +- Current state +- Next actions +- Do not repeat +- Blockers +- Handoff -### WAITING_USER +Keep it concise. Prefer a few KB; do not dump chat history. +5. Arm the job only after Goal and Next actions are real: -Progress requires the user to answer, approve, provide information/credentials, -restart something, perform a manual step, or confirm an external action. +`python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" job-arm --job-id ` -Before ending the turn, pause the exact registration: +6. Continue the task normally. -Windows: - -`"$HOME\.codex\skills\wake\scripts\pause-current.cmd" --reason ""` - -macOS/Linux: +## Checkpoint discipline -`"$HOME/.codex/skills/wake/scripts/pause-current.sh" --reason ""` +While a WAKE job is active: -Then: +- update the checkpoint after every meaningful milestone; +- update it before long-running, risky, or quota-heavy operations; +- preserve verified work and explicit user constraints; +- record abandoned approaches under Decisions / Do not repeat; +- do not copy large logs, diffs, or chat transcripts into the checkpoint; +- use paths, short results, commit ids, and test summaries instead. -- pause/disable fallback Scheduled Task when possible -- do not create retry tasks -- do not send periodic waiting messages -- preserve all existing work -- emit at most one concise message explaining what is awaited +The checkpoint is the recovery authority after a quota handoff. -A `waiting_user` entry must never be woken by quota recovery. +## Quota behavior -### QUOTA_BLOCKED +The watcher keeps one hidden persistent local `codex app-server` only for +`account/rateLimits/read`. -The local watcher handles quota without needing a model turn at the moment of exhaustion. +- `ordinaryUsageAllowed == false`: active jobs become `quota_waiting`. +- `resetsAt` is only a check-time hint. +- only `ordinaryUsageAllowed == true` permits a handoff. +- recovery starts a NEW `codex exec` thread with the checkpoint path. +- transient capacity/network failures use bounded exponential backoff (`retry_waiting`) instead of immediately giving up or spinning. +- the recovery prompt explicitly forbids reconstructing or resuming the old chat. -1. A thread registered by `$wake` starts in `monitoring`. -2. The watcher independently reads `account/rateLimits/read`. -3. When `ordinaryUsageAllowed == false`, `monitoring` becomes `quota_waiting`. -4. `resetsAt` may schedule the next check, but is never proof of recovery. -5. Only `ordinaryUsageAllowed == true` permits resume. -6. Resume only exact `quota_waiting` thread ids. -7. After a launch, return the registration to `monitoring`. -8. The resumed turn must classify state again before doing work. +This avoids Desktop active-writer conflicts and reduces repeated context processing. -If the watcher or exact registration is unavailable, use conservative Scheduled -Task fallback on the quarter-hour grid. +## New-thread handoff -WAKE must never bypass or evade platform limits. +The continuation thread must: -### RETRYABLE_BLOCKED +1. read the checkpoint first; +2. inspect only minimal live workspace state, normally `git status --short` and + `git diff --stat`; +3. continue from Next actions; +4. not redo items listed under Completed or Do not repeat unless stale; +5. keep updating the checkpoint; +6. mark WAITING_USER or DONE with the helper commands below. -For transient non-quota network/service/model failures: +WAKE does not bypass, evade, or increase platform quota. -- preserve valid work -- use synchronized Scheduled Task fallback -- do not create extra retry schedules +## WAITING_USER -### RUNNABLE +If progress requires an answer, approval, credential, restart, or manual action: -If unfinished work can continue without user input: +1. update the checkpoint with the exact blocker; +2. run: -- inspect current conversation, files, worktree, runtime state, and tests -- preserve completed work -- continue only unfinished work -- do not restart from scratch -- do not create duplicate schedules or registrations +`python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" job-pause --job-id --reason ""` -### DONE +A `waiting_user` job is never auto-launched. -After verifying the original request is complete: +After the user resolves the blocker: -- report the result -- stop/remove fallback Scheduled Task -- unregister the exact thread +`python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" job-rearm --job-id ` -Windows: +## DONE -`"$HOME\.codex\skills\wake\scripts\unregister-current.cmd"` +After the original Goal is fully verified: -macOS/Linux: +1. update the checkpoint with final verification; +2. run: -`"$HOME/.codex/skills/wake/scripts/unregister-current.sh"` +`python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" job-complete --job-id ` -### UNKNOWN +Completed jobs remain as local history and are ignored by the watcher. -If user dependency is unclear: +## $wake status -- ask one concise question -- treat as WAITING_USER -- pause watcher registration and fallback task +Report both watcher status and the current thread's active WAKE job without changing state: -## RE-ARM AFTER USER INPUT +`python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" status` -If the user resolves a previous WAITING_USER blocker, re-enable exact monitoring: - -Windows: - -`"$HOME\.codex\skills\wake\scripts\resume-current.cmd"` - -macOS/Linux: +`python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" job-current` -`"$HOME/.codex/skills/wake/scripts/resume-current.sh"` +If no active job is associated with the current thread, say so plainly. -If implicit invocation does not happen, `$wake` is the manual re-arm. +## $wake stop -## WATCHER COMMANDS +Stop automatic continuation for the current task without deleting its checkpoint: -Installed watcher: +`python "$HOME\.codex\skills\wake\watcher\wake_watcher.py" job-stop-current` -`~/.codex/skills/wake/watcher/wake_watcher.py` +A `stopped` job is retained locally and is never auto-launched. A later `$wake start` +may refresh its checkpoint and arm/rearm it again. -Useful commands: +## Diagnostics -- `doctor` — verify Codex CLI/App Server quota integration -- `quota` — read quota once -- `status` — watcher process and registration counts -- `list` — list exact-thread registrations -- `register-current` — register current `CODEX_THREAD_ID` -- `pause-current` — mark current thread WAITING_USER -- `resume-current` — return current thread to monitoring -- `unregister-current` — remove current registration +- `python ...\wake_watcher.py doctor` +- `python ...\wake_watcher.py status` +- `python ...\wake_watcher.py quota` +- `python ...\wake_watcher.py job-list` +- `python ...\wake_watcher.py job-status --job-id ` -The watcher never stores OpenAI credentials, never redeems credits, never infers -recovery from `resetsAt` alone, never uses `--last`, and never wakes -`waiting_user` entries. +## Safety -Watcher-driven resume currently uses exact detached `codex exec resume `. -Codex Desktop UI synchronization may vary by version, so Scheduled Task fallback -remains available. +- Never resume the original thread automatically. +- Never launch a job without a completed checkpoint. +- Never auto-launch `waiting_user`, `needs_attention`, `stopped`, `completed`, or `draft`; only `quota_waiting` and due `retry_waiting` may launch a handoff. +- Never infer quota recovery from time alone. +- Never delete user work to recover a job. +- If a continuation process exits while quota is available without marking DONE or + WAITING_USER, set `needs_attention` instead of looping. diff --git a/VERSION b/VERSION index d15723f..1d0ba9e 100644 --- a/VERSION +++ b/VERSION @@ -1 +1 @@ -0.3.2 +0.4.0 diff --git a/agents/openai.yaml b/agents/openai.yaml index eaf0d46..91ed753 100644 --- a/agents/openai.yaml +++ b/agents/openai.yaml @@ -1,7 +1,7 @@ interface: display_name: "Wake" - short_description: "Quota-aware continuation for long Codex tasks with a local watcher and safe Scheduled Task fallback." - default_prompt: "Use $wake to keep this Codex conversation progressing. Prefer the local quota watcher only when an exact thread id is available; pause while waiting for the user; never guess a thread id or bypass usage limits." + short_description: "Checkpoint-first continuation for long Codex jobs across quota windows." + default_prompt: "Use $wake to persist the current task as a compact checkpoint-backed job. On quota recovery, continue in a new lightweight thread from the checkpoint instead of resuming the old chat." policy: allow_implicit_invocation: true diff --git a/docs/quota-watcher.md b/docs/quota-watcher.md index 80dcdaa..b7ae224 100644 --- a/docs/quota-watcher.md +++ b/docs/quota-watcher.md @@ -1,48 +1,130 @@ -# Quota-aware watcher design — v0.3.2 +# Checkpoint-first quota watcher — v0.4 -The watcher is registered before quota is exhausted. This matters because a model -turn may not be available at the exact moment the account hits its limit. +## Goal -## Exact thread identity +Continue a long Codex task across included-usage windows without reopening the entire +old conversation and without competing with Codex Desktop for the same thread writer. -Codex shell/tool executions receive `CODEX_THREAD_ID`. `$wake start` invokes a -local helper from the target conversation so the watcher records that exact id. +The durable object is a **WAKE job**, not a Codex thread. -A normal external terminal does not have this variable and therefore cannot use -`register-current`. WAKE never substitutes `--last`, recency, title, or working -directory for exact identity. +## Persistent layout -## State transition +```text +~/.codex/wake/ +├─ watcher.pid +├─ watcher.log +└─ jobs/ + └─ / + ├─ state.json + ├─ checkpoint.md + └─ runs/ + └─ run-0001-.jsonl +``` + +The Skill install lives separately under `~/.codex/skills/wake`. Reinstalling the Skill +therefore does not erase job/checkpoint history. + +## Initial registration + +`$wake` runs inside the source Codex conversation and requires `CODEX_THREAD_ID`. +The thread id is recorded only as provenance. It is never used for automatic resume. + +The current model writes a compact checkpoint and then arms the job. ```text -$wake start - ↓ -monitoring - ↓ ordinaryUsageAllowed == false -quota_waiting - ↓ ordinaryUsageAllowed == true -resume exact thread once - ↓ +draft + | + | checkpoint filled + job-arm + v monitoring ``` -A `waiting_user` entry is outside that transition and is not woken by quota recovery. +A draft job is never launched. ## Quota authority -The watcher calls `account/rateLimits/read` through local `codex app-server`. +WAKE reads `account/rateLimits/read` from one hidden persistent local +`codex app-server`. + +- `ordinaryUsageAllowed == false` means wait. +- `ordinaryUsageAllowed == true` permits a checkpoint handoff. +- null/unknown means do not infer recovery. +- `resetsAt` is only a scheduling hint. + +Multiple windows can exist. WAKE always re-reads the authoritative boolean before launch. + +## State machine + +```text +monitoring + | + | ordinaryUsageAllowed == false + v +quota_waiting + | + | ordinaryUsageAllowed == true + v +running -- starts NEW codex exec thread + | + +--> completed + | + +--> waiting_user + | + +--> quota_waiting (run ended while quota blocked) + | + +--> retry_waiting (temporary capacity/network failure) + | | + | +--> running (bounded exponential backoff) + | + +--> needs_attention (non-retryable failure / retry limit exceeded) +``` + +`waiting_user`, `needs_attention`, `stopped`, `draft`, and `completed` are not auto-launched. + +## Handoff contract + +Recovery uses a fresh: + +```text +codex exec --json -C --add-dir ... +``` + +It does **not** use `codex exec resume`. + +The continuation prompt requires the new thread to: + +1. read `checkpoint.md` first; +2. normally inspect only `git status --short` and `git diff --stat`; +3. continue from `Next actions`; +4. skip work already listed under `Completed` or `Do not repeat`; +5. update the checkpoint after meaningful milestones; +6. call `job-pause` when user input is required; +7. call `job-complete` after the Goal is fully verified. + +## Why not resume the old Desktop thread? + +In testing, detached `codex exec resume ` failed with: + +```text +thread-store conflict: thread ... already has an active writer +``` + +Codex Desktop already owned that thread. v0.4 avoids that control-plane conflict entirely +by making a new thread for each quota handoff. + +## Failure handling + +WAKE distinguishes transient failures from terminal ones: -- `ordinaryUsageAllowed == true` → recovery is allowed -- `ordinaryUsageAllowed == false` → keep waiting -- `ordinaryUsageAllowed == null` → unknown; never assume recovery -- `resetsAt` → check-time hint only +- selected-model capacity and common temporary network/HTTP 502/503/504 failures enter `retry_waiting`; +- retry delay grows exponentially from 60 seconds, is capped at 15 minutes, and stops after 8 retries; +- if a continuation process exits while quota is available for a non-retryable reason, WAKE sets `needs_attention`; +- if it exits while quota is unavailable, WAKE returns to `quota_waiting`; +- missing workspace paths or launch errors also become `needs_attention`. -Multiple reset windows may exist. The earliest future reset is only a useful next -check time; the watcher always re-reads `ordinaryUsageAllowed` before waking anything. +## Context-efficiency rule -## Desktop limitation +The checkpoint should stay small and operational. Record conclusions, file paths, short +verification results, and next actions. Do not paste old chat history, full logs, or large +diffs. Old chats remain available as archives if a human later needs them. -Recovery currently uses exact `codex exec resume `. Exact identity is -safer than `--last`, but a detached CLI continuation may not always render in -Codex Desktop exactly like a native Desktop turn. Keep Scheduled Task fallback -available where detached resume is not satisfactory. diff --git a/docs/state-machine.md b/docs/state-machine.md index 8fb57d8..7638f12 100644 --- a/docs/state-machine.md +++ b/docs/state-machine.md @@ -1,95 +1,35 @@ -# WAKE state machine - -WAKE v0.2 replaces blind periodic continuation with a small state machine. - -```text - ┌───────────────┐ - │ RUNNABLE │ - └───────┬───────┘ - │ continue - ▼ - ┌───────────────┐ - │ work │ - └───────┬───────┘ - │ - ┌─────────────────┼──────────────────┐ - ▼ ▼ ▼ - ┌─────────┐ ┌──────────────┐ ┌─────────┐ - │ ACTIVE │ │ WAITING_USER │ │ DONE │ - └────┬────┘ └──────┬───────┘ └────┬────┘ - │ │ │ - no duplicate pause WAKE remove WAKE - │ │ - │ user replies - │ │ - └──────────────┬───┘ - ▼ - re-arm WAKE - - RETRYABLE_BLOCKED → keep schedule → retry next quarter-hour -``` - -## Why WAITING_USER pauses - -A scheduled message is already visible by the time a scheduled run begins. -Therefore a prompt-only guard cannot prevent the *first* wake that discovers -the conversation is waiting for the user. - -The useful goal is to prevent every later quarter-hour wake. - -When a run determines that progress requires user input or a manual action, -it pauses its own WAKE schedule immediately. The conversation should then stay -quiet until the user responds. - -## Re-arm - -When the user provides the requested answer, approval, confirmation, or manual -completion, WAKE can re-arm the existing schedule. - -WAKE enables implicit invocation only for this re-arm path. It must not start -WAKE implicitly in unrelated conversations. - -If implicit invocation is not selected by the host/model, the fallback is: - -```text -$wake -``` - -or: +# WAKE v0.4 job state machine ```text -$wake start +draft + | + | checkpoint filled + job-arm + v +monitoring + | + | ordinaryUsageAllowed == false + v +quota_waiting + | + | ordinaryUsageAllowed == true + v +running + |\ + | +--> completed + | +--> waiting_user + | +--> quota_waiting + +----> needs_attention ``` -## State definitions - -| State | Meaning | Scheduled-task action | -|---|---|---| -| ACTIVE | Work is already progressing without user action | Do not duplicate work; leave schedule enabled | -| WAITING_USER | A user answer, approval, credential, restart, or manual step is required | Pause WAKE | -| RETRYABLE_BLOCKED | Quota, rate limit, temporary service/network failure | Keep WAKE enabled | -| RUNNABLE | Unfinished work can proceed now | Continue from current state | -| DONE | Original request has been verified complete | Remove WAKE | -| UNKNOWN | It is unclear whether user action is required | Ask once, then pause as WAITING_USER | - -## Central supervisor status - -A future WAKE backend may use Codex App Server as a central supervisor. The -official App Server protocol exposes non-resuming thread reads, runtime status, -and user-input/approval events. - -WAKE does not enable that backend by default yet. An external process does not -currently have a documented, generally safe way to attach to the exact active -Codex Desktop runtime instance for every user-visible local thread. Starting a -separate App Server can read persisted threads, but that is not equivalent to -observing the Desktop process's live runtime state. - -Until that boundary is supported reliably, v0.2 prefers the safer -conversation-local state machine over cross-thread automation that might resume -the wrong or already-active thread. +## Rules -References: +- draft: never auto-launch. +- monitoring: current source/handoff thread is working; watcher only observes quota. +- quota_waiting: no model turns are generated until quota is actually available. +- running: a fresh checkpoint-based `codex exec` is active. +- waiting_user: explicit human dependency; never auto-launch. +- needs_attention: abnormal exit while quota is available; stop instead of looping. +- stopped: user-disabled automatic continuation; checkpoint is retained. +- completed: terminal state; retained as local history. -- https://developers.openai.com/docs/app-server -- https://developers.openai.com/docs/automations -- https://github.com/openai/codex/issues/25914 +The source thread id is provenance only. WAKE v0.4 never automatically resumes it. diff --git a/scripts/install.ps1 b/scripts/install.ps1 index 23dfb6c..1e237c2 100644 --- a/scripts/install.ps1 +++ b/scripts/install.ps1 @@ -4,63 +4,44 @@ param( ) $ErrorActionPreference = "Stop" - $RepoRoot = (Resolve-Path (Split-Path -Parent $PSScriptRoot)).Path -$TargetRoot = [System.IO.Path]::GetFullPath($TargetRoot) -$Target = Join-Path $TargetRoot "wake" - -if (-not (Test-Path (Join-Path $RepoRoot "SKILL.md"))) { - throw "SKILL.md not found at $RepoRoot" -} - -New-Item -ItemType Directory -Force -Path $TargetRoot | Out-Null - +$Target = Join-Path ([System.IO.Path]::GetFullPath($TargetRoot)) "wake" $RepoFull = [System.IO.Path]::GetFullPath($RepoRoot).TrimEnd('\') $TargetFull = [System.IO.Path]::GetFullPath($Target).TrimEnd('\') +New-Item -ItemType Directory -Force -Path $Target | Out-Null + if ($RepoFull -ieq $TargetFull) { - Write-Host "" - Write-Host "WAKE is already located at the target path:" - Write-Host " $Target" - Write-Host "No copy is needed." + Write-Host "WAKE is already at $Target" } else { - $SavedState = $null - $ExistingState = Join-Path $Target ".state" - if (Test-Path $ExistingState) { - $SavedState = Join-Path $env:TEMP ("wake-state-" + [guid]::NewGuid().ToString()) - Copy-Item -Recurse -Force $ExistingState $SavedState + foreach ($File in @("SKILL.md","VERSION","README.md","README.zh-CN.md","CHANGELOG.md")) { + $Source = Join-Path $RepoRoot $File + if (Test-Path $Source) { Copy-Item -Force $Source $Target } } - - if (Test-Path $Target) { - Remove-Item -Recurse -Force $Target + foreach ($Dir in @("agents","watcher","docs","scripts")) { + $SourceDir = Join-Path $RepoRoot $Dir + $DestDir = Join-Path $Target $Dir + New-Item -ItemType Directory -Force -Path $DestDir | Out-Null + Copy-Item -Recurse -Force (Join-Path $SourceDir "*") $DestDir } - - New-Item -ItemType Directory -Force -Path $Target | Out-Null - Copy-Item -Force (Join-Path $RepoRoot "SKILL.md") $Target - Copy-Item -Force (Join-Path $RepoRoot "VERSION") $Target - Copy-Item -Recurse -Force (Join-Path $RepoRoot "agents") $Target - Copy-Item -Recurse -Force (Join-Path $RepoRoot "watcher") $Target - Copy-Item -Recurse -Force (Join-Path $RepoRoot "docs") $Target - Copy-Item -Recurse -Force (Join-Path $RepoRoot "scripts") $Target - - if ($SavedState -and (Test-Path $SavedState)) { - Copy-Item -Recurse -Force $SavedState (Join-Path $Target ".state") - Remove-Item -Recurse -Force $SavedState + foreach ($Old in @( + "scripts\register-current.cmd","scripts\register-current.sh", + "scripts\pause-current.cmd","scripts\pause-current.sh", + "scripts\resume-current.cmd","scripts\resume-current.sh", + "scripts\unregister-current.cmd","scripts\unregister-current.sh", + "docs\v0.3.3-local.md","docs\v0.3.3-local2.md","README.local.md" + )) { + Remove-Item -Force (Join-Path $Target $Old) -ErrorAction SilentlyContinue } - - Write-Host "" - Write-Host "WAKE v0.3.2 installed:" - Write-Host " $Target" + Remove-Item -Recurse -Force (Join-Path $Target ".state") -ErrorAction SilentlyContinue + Remove-Item -Recurse -Force (Join-Path $Target "watcher\__pycache__") -ErrorAction SilentlyContinue + Write-Host "WAKE v0.4.0 installed to $Target" } -Write-Host "" -Write-Host "Check quota integration:" -Write-Host " python `"$Target\watcher\wake_watcher.py`" doctor" - +$WakeHome = Join-Path $HOME ".codex\wake" +New-Item -ItemType Directory -Force -Path $WakeHome | Out-Null +Write-Host "Durable jobs: $WakeHome" +Write-Host "Doctor: python $Target\watcher\wake_watcher.py doctor" if ($StartWatcher) { powershell.exe -NoProfile -ExecutionPolicy Bypass -File (Join-Path $Target "scripts\start-watcher.ps1") -} else { - Write-Host "" - Write-Host "Start watcher with:" - Write-Host " `"$Target\scripts\start-watcher.cmd`"" } diff --git a/scripts/install.sh b/scripts/install.sh index 13f8785..30f6055 100644 --- a/scripts/install.sh +++ b/scripts/install.sh @@ -1,38 +1,21 @@ #!/usr/bin/env bash set -euo pipefail - SCRIPT_DIR="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)" REPO_ROOT="$(cd -- "$SCRIPT_DIR/.." && pwd)" TARGET_ROOT="$HOME/.codex/skills" TARGET="$TARGET_ROOT/wake" mkdir -p "$TARGET_ROOT" - -REPO_FULL="$(cd -- "$REPO_ROOT" && pwd)" -TARGET_FULL="$(cd -- "$(dirname -- "$TARGET")" && pwd)/$(basename -- "$TARGET")" - -if [[ "$REPO_FULL" == "$TARGET_FULL" ]]; then - echo "WAKE is already located at the target path: $TARGET" - echo "No copy is needed." -else - SAVED_STATE="" - if [[ -d "$TARGET/.state" ]]; then - TMPROOT="$(mktemp -d)" - SAVED_STATE="$TMPROOT/state" - cp -R "$TARGET/.state" "$SAVED_STATE" - fi - +if [[ "$REPO_ROOT" != "$TARGET" ]]; then rm -rf "$TARGET" mkdir -p "$TARGET" cp "$REPO_ROOT/SKILL.md" "$REPO_ROOT/VERSION" "$TARGET/" + for f in README.md README.zh-CN.md CHANGELOG.md; do + [[ -f "$REPO_ROOT/$f" ]] && cp "$REPO_ROOT/$f" "$TARGET/" + done cp -R "$REPO_ROOT/agents" "$REPO_ROOT/watcher" "$REPO_ROOT/docs" "$REPO_ROOT/scripts" "$TARGET/" - - if [[ -n "$SAVED_STATE" && -d "$SAVED_STATE" ]]; then - cp -R "$SAVED_STATE" "$TARGET/.state" - fi - - echo "WAKE v0.3.2 installed: $TARGET" fi - +mkdir -p "$HOME/.codex/wake" +echo "WAKE v0.4.0 installed: $TARGET" +echo "Durable jobs: $HOME/.codex/wake" echo "Run: python3 '$TARGET/watcher/wake_watcher.py' doctor" -echo "Watcher is not auto-started." diff --git a/scripts/start-watcher.ps1 b/scripts/start-watcher.ps1 index 76dcfbd..746916c 100644 --- a/scripts/start-watcher.ps1 +++ b/scripts/start-watcher.ps1 @@ -1,14 +1,12 @@ $ErrorActionPreference = "Stop" - $SkillRoot = Split-Path -Parent $PSScriptRoot $Watcher = Join-Path $SkillRoot "watcher\wake_watcher.py" -$StateDir = Join-Path $SkillRoot ".state" -$PidFile = Join-Path $StateDir "watcher.pid" - -New-Item -ItemType Directory -Force -Path $StateDir | Out-Null +$WakeHome = Join-Path $HOME ".codex\wake" +$PidFile = Join-Path $WakeHome "watcher.pid" +New-Item -ItemType Directory -Force -Path $WakeHome | Out-Null if (Test-Path $PidFile) { - $OldPid = Get-Content $PidFile -ErrorAction SilentlyContinue + $OldPid = Get-Content $PidFile -ErrorAction SilentlyContinue | Select-Object -First 1 if ($OldPid -and (Get-Process -Id $OldPid -ErrorAction SilentlyContinue)) { Write-Host "WAKE watcher already running (PID $OldPid)." exit 0 @@ -18,16 +16,12 @@ if (Test-Path $PidFile) { $Python = $null foreach ($Name in @("pythonw.exe", "python.exe", "pythonw", "python")) { $Cmd = Get-Command $Name -ErrorAction SilentlyContinue - if ($Cmd) { - $Python = $Cmd.Source - break - } -} -if (-not $Python) { - throw "Python 3 was not found in PATH." + if ($Cmd) { $Python = $Cmd.Source; break } } +if (-not $Python) { throw "Python 3 was not found in PATH." } +$env:PYTHONUTF8 = "1" $Proc = Start-Process -FilePath $Python -ArgumentList @($Watcher, "run") -WindowStyle Hidden -PassThru -Start-Sleep -Milliseconds 500 +Start-Sleep -Milliseconds 800 Write-Host "WAKE watcher start requested (PID $($Proc.Id))." -Write-Host "Status: python `"$Watcher`" status" +python $Watcher status diff --git a/scripts/start-watcher.sh b/scripts/start-watcher.sh index 0bd9ec5..ed14ba7 100644 --- a/scripts/start-watcher.sh +++ b/scripts/start-watcher.sh @@ -1,10 +1,12 @@ #!/usr/bin/env bash set -euo pipefail SKILL_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)" -mkdir -p "$SKILL_ROOT/.state" -if [[ -f "$SKILL_ROOT/.state/watcher.pid" ]] && kill -0 "$(cat "$SKILL_ROOT/.state/watcher.pid")" 2>/dev/null; then - echo "WAKE watcher already running." +WAKE_HOME="$HOME/.codex/wake" +PIDFILE="$WAKE_HOME/watcher.pid" +mkdir -p "$WAKE_HOME" +if [[ -f "$PIDFILE" ]] && kill -0 "$(cat "$PIDFILE")" 2>/dev/null; then + echo "WAKE watcher already running (PID $(cat "$PIDFILE"))." exit 0 fi -nohup python3 "$SKILL_ROOT/watcher/wake_watcher.py" run >> "$SKILL_ROOT/.state/nohup.log" 2>&1 & +nohup python3 "$SKILL_ROOT/watcher/wake_watcher.py" run >> "$WAKE_HOME/nohup.log" 2>&1 & echo "WAKE watcher start requested (PID $!)." diff --git a/scripts/status-watcher.ps1 b/scripts/status-watcher.ps1 index 27266b5..6d978c7 100644 --- a/scripts/status-watcher.ps1 +++ b/scripts/status-watcher.ps1 @@ -1,4 +1,5 @@ $ErrorActionPreference = "Stop" $SkillRoot = Split-Path -Parent $PSScriptRoot $Watcher = Join-Path $SkillRoot "watcher\wake_watcher.py" +$env:PYTHONUTF8 = "1" python $Watcher status diff --git a/scripts/status-watcher.sh b/scripts/status-watcher.sh new file mode 100644 index 0000000..25b8f86 --- /dev/null +++ b/scripts/status-watcher.sh @@ -0,0 +1,4 @@ +#!/usr/bin/env bash +set -euo pipefail +SKILL_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)" +python3 "$SKILL_ROOT/watcher/wake_watcher.py" status diff --git a/scripts/stop-watcher.ps1 b/scripts/stop-watcher.ps1 index ed37395..a8f941f 100644 --- a/scripts/stop-watcher.ps1 +++ b/scripts/stop-watcher.ps1 @@ -1,19 +1,14 @@ $ErrorActionPreference = "Stop" - -$SkillRoot = Split-Path -Parent $PSScriptRoot -$PidFile = Join-Path $SkillRoot ".state\watcher.pid" - +$PidFile = Join-Path $HOME ".codex\wake\watcher.pid" if (-not (Test-Path $PidFile)) { Write-Host "WAKE watcher is not running (no PID file)." exit 0 } - -$PidValue = (Get-Content $PidFile -ErrorAction SilentlyContinue | Select-Object -First 1) +$PidValue = Get-Content $PidFile -ErrorAction SilentlyContinue | Select-Object -First 1 if ($PidValue -and (Get-Process -Id $PidValue -ErrorAction SilentlyContinue)) { Stop-Process -Id $PidValue -Force Write-Host "Stopped WAKE watcher PID $PidValue." } else { Write-Host "WAKE watcher PID file was stale." } - Remove-Item -Force $PidFile -ErrorAction SilentlyContinue diff --git a/scripts/stop-watcher.sh b/scripts/stop-watcher.sh index c1b5636..83e466b 100644 --- a/scripts/stop-watcher.sh +++ b/scripts/stop-watcher.sh @@ -1,7 +1,6 @@ #!/usr/bin/env bash set -euo pipefail -SKILL_ROOT="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")/.." && pwd)" -PIDFILE="$SKILL_ROOT/.state/watcher.pid" +PIDFILE="$HOME/.codex/wake/watcher.pid" if [[ -f "$PIDFILE" ]]; then PID="$(cat "$PIDFILE")" kill "$PID" 2>/dev/null || true diff --git a/scripts/uninstall.ps1 b/scripts/uninstall.ps1 index f449ec4..0307154 100644 --- a/scripts/uninstall.ps1 +++ b/scripts/uninstall.ps1 @@ -1,19 +1,22 @@ param( + [switch]$PurgeJobs, [string]$TargetRoot = "$HOME\.codex\skills" ) $ErrorActionPreference = "Stop" $Target = Join-Path $TargetRoot "wake" - +$Stop = Join-Path $Target "scripts\stop-watcher.ps1" +if (Test-Path $Stop) { + powershell.exe -NoProfile -ExecutionPolicy Bypass -File $Stop | Out-Null +} if (Test-Path $Target) { - $Stop = Join-Path $Target "scripts\stop-watcher.ps1" - if (Test-Path $Stop) { - powershell.exe -NoProfile -ExecutionPolicy Bypass -File $Stop | Out-Null - } Remove-Item -Recurse -Force $Target - Write-Host "WAKE removed from:" - Write-Host " $Target" + Write-Host "WAKE removed: $Target" +} +if ($PurgeJobs) { + $WakeHome = Join-Path $HOME ".codex\wake" + if (Test-Path $WakeHome) { Remove-Item -Recurse -Force $WakeHome } + Write-Host "WAKE durable jobs purged." } else { - Write-Host "WAKE is not installed at:" - Write-Host " $Target" + Write-Host "Durable job/checkpoint history preserved under $HOME\.codex\wake" } diff --git a/scripts/uninstall.sh b/scripts/uninstall.sh index 7b0af13..3efd9ef 100644 --- a/scripts/uninstall.sh +++ b/scripts/uninstall.sh @@ -6,3 +6,4 @@ if [[ -x "$TARGET/scripts/stop-watcher.sh" ]]; then fi rm -rf "$TARGET" echo "WAKE removed: $TARGET" +echo "Durable jobs preserved: $HOME/.codex/wake" diff --git a/tests/test_wake.py b/tests/test_wake.py new file mode 100644 index 0000000..68fa910 --- /dev/null +++ b/tests/test_wake.py @@ -0,0 +1,207 @@ +import importlib.util +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +MODULE_PATH = Path(__file__).parents[1] / "watcher" / "wake_watcher.py" +spec = importlib.util.spec_from_file_location("wake_watcher", MODULE_PATH) +wake = importlib.util.module_from_spec(spec) +assert spec.loader is not None +spec.loader.exec_module(wake) + + +class FakeApp: + def __init__(self, ordinary): + self.ordinary = ordinary + + def request(self, method): + assert method == "account/rateLimits/read" + return { + "ordinaryUsageAllowed": self.ordinary, + "accountId": "x", + "rateLimits": { + "primary": {"usedPercent": 10, "resetsAt": wake.now() + 60}, + "secondary": {"usedPercent": 20, "resetsAt": wake.now() + 3600}, + }, + } + + def close(self): + pass + + +class WakeTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + root = Path(self.tmp.name) + self.old = (wake.WAKE_HOME, wake.JOBS_DIR, wake.PID_FILE, wake.LOG_FILE) + wake.WAKE_HOME = root / "wake" + wake.JOBS_DIR = wake.WAKE_HOME / "jobs" + wake.PID_FILE = wake.WAKE_HOME / "watcher.pid" + wake.LOG_FILE = wake.WAKE_HOME / "watcher.log" + self.cwd = root / "repo" + self.cwd.mkdir() + + def tearDown(self): + wake.WAKE_HOME, wake.JOBS_DIR, wake.PID_FILE, wake.LOG_FILE = self.old + self.tmp.cleanup() + + def create_ready_job(self): + job = wake.create_job("thread-12345678", str(self.cwd), "Finish the project") + cp = wake.checkpoint_path(job["jobId"]) + text = cp.read_text(encoding="utf-8").replace("1. REPLACE_ME", "1. Run tests") + cp.write_text(text, encoding="utf-8") + return wake.arm_job(job["jobId"]) + + def test_create_job_is_draft_and_durable(self): + job = wake.create_job("thread-12345678", str(self.cwd), "Ship it") + self.assertEqual(job["state"], "draft") + self.assertFalse(job["checkpointReady"]) + self.assertTrue(wake.checkpoint_path(job["jobId"]).exists()) + + def test_arm_rejects_placeholder(self): + job = wake.create_job("thread-12345678", str(self.cwd), None) + with self.assertRaises(wake.WakeError): + wake.arm_job(job["jobId"]) + + def test_arm_pause_rearm_complete(self): + job = self.create_ready_job() + jid = job["jobId"] + self.assertEqual(wake.pause_job(jid, "need approval")["state"], "waiting_user") + self.assertEqual(wake.rearm_job(jid)["state"], "monitoring") + self.assertEqual(wake.complete_job(jid)["state"], "completed") + with self.assertRaises(wake.WakeError): + wake.rearm_job(jid) + + def test_duplicate_thread_reuses_existing_job(self): + first = wake.create_job("thread-12345678", str(self.cwd), "One") + second = wake.create_job("thread-12345678", str(self.cwd), "Two") + self.assertEqual(first["jobId"], second["jobId"]) + self.assertTrue(second["alreadyExists"]) + + def test_stop_prevents_auto_launch(self): + job = self.create_ready_job() + wake.stop_job(job["jobId"]) + with mock.patch.object(wake, "launch_handoff") as launch: + wake.process_once(FakeApp(True)) + launch.assert_not_called() + self.assertEqual(wake.load_job(job["jobId"])["state"], "stopped") + + def test_quota_false_moves_monitoring_to_waiting(self): + job = self.create_ready_job() + delay = wake.process_once(FakeApp(False)) + self.assertGreaterEqual(delay, 5) + self.assertEqual(wake.load_job(job["jobId"])["state"], "quota_waiting") + + def test_quota_recovery_launches_new_handoff(self): + job = self.create_ready_job() + job["state"] = "quota_waiting" + wake.save_job(job) + with mock.patch.object(wake, "launch_handoff") as launch: + launch.side_effect = lambda j: j + wake.process_once(FakeApp(True)) + launch.assert_called_once() + self.assertEqual(launch.call_args.args[0]["jobId"], job["jobId"]) + + def test_dead_run_needs_attention_when_quota_available(self): + job = self.create_ready_job() + job["state"] = "running" + job["lastRunPid"] = 99999999 + wake.save_job(job) + with mock.patch.object(wake, "is_process_alive", return_value=False): + wake.process_once(FakeApp(True)) + self.assertEqual(wake.load_job(job["jobId"])["state"], "needs_attention") + + def test_dead_run_returns_to_waiting_when_blocked(self): + job = self.create_ready_job() + job["state"] = "running" + job["lastRunPid"] = 99999999 + wake.save_job(job) + with mock.patch.object(wake, "is_process_alive", return_value=False): + wake.process_once(FakeApp(False)) + self.assertEqual(wake.load_job(job["jobId"])["state"], "quota_waiting") + + def test_prompt_is_checkpoint_first(self): + job = self.create_ready_job() + prompt = wake.continuation_prompt(job) + self.assertIn(job["checkpointPath"], prompt) + self.assertIn("Do NOT reconstruct, resume, or reread", prompt) + self.assertIn("job-complete", prompt) + self.assertIn("job-pause", prompt) + + def test_launch_handoff_creates_new_exec_not_resume(self): + job = self.create_ready_job() + fake_proc = mock.Mock(pid=4242) + with mock.patch.object(wake, "find_codex", return_value="codex"), \ + mock.patch.object(wake.subprocess, "Popen", return_value=fake_proc) as popen: + result = wake.launch_handoff(job) + cmd = popen.call_args.args[0] + self.assertEqual(cmd[1], "exec") + self.assertNotIn("resume", cmd) + self.assertIn("--add-dir", cmd) + self.assertIn("--sandbox", cmd) + self.assertIn("workspace-write", cmd) + self.assertEqual(result["state"], "running") + self.assertEqual(result["lastRunPid"], 4242) + + def test_transient_capacity_failure_enters_retry_waiting(self): + job = self.create_ready_job() + runlog = wake.run_dir(job["jobId"]) / "run.jsonl" + runlog.write_text( + '{"type":"turn.failed","error":{"message":"Selected model is at capacity. Please try a different model."}}\n', + encoding="utf-8", + ) + job["state"] = "running" + job["lastRunPid"] = 99999999 + job["lastRunLog"] = str(runlog) + wake.save_job(job) + with mock.patch.object(wake, "is_process_alive", return_value=False): + wake.process_once(FakeApp(True)) + result = wake.load_job(job["jobId"]) + self.assertEqual(result["state"], "retry_waiting") + self.assertEqual(result["retryCount"], 1) + self.assertGreater(result["nextRetryAt"], wake.now()) + + def test_retry_waiting_launches_when_due(self): + job = self.create_ready_job() + job["state"] = "retry_waiting" + job["retryCount"] = 1 + job["nextRetryAt"] = wake.now() - 1 + wake.save_job(job) + with mock.patch.object(wake, "launch_handoff") as launch: + launch.side_effect = lambda j: j + wake.process_once(FakeApp(True)) + launch.assert_called_once() + + def test_launch_honors_optional_model_and_effort(self): + job = wake.create_job( + "thread-model-12345678", str(self.cwd), "Finish", + model="gpt-5.6-luna", reasoning_effort="low", + ) + cp = wake.checkpoint_path(job["jobId"]) + cp.write_text(cp.read_text(encoding="utf-8").replace("1. REPLACE_ME", "1. Test"), encoding="utf-8") + job = wake.arm_job(job["jobId"]) + fake_proc = mock.Mock(pid=4243) + with mock.patch.object(wake, "find_codex", return_value="codex"), \ + mock.patch.object(wake.subprocess, "Popen", return_value=fake_proc) as popen: + wake.launch_handoff(job) + cmd = popen.call_args.args[0] + self.assertIn("gpt-5.6-luna", cmd) + self.assertIn('model_reasoning_effort="low"', cmd) + + def test_summarize_quota_hides_account_id(self): + future = wake.now() + 120 + result = wake.summarize_quota({ + "ordinaryUsageAllowed": False, + "accountId": "secret", + "rateLimits": {"primary": {"resetsAt": future}, "secondary": {}}, + }) + self.assertFalse(result["ordinaryUsageAllowed"]) + self.assertTrue(result["accountPresent"]) + self.assertEqual(result["nextResetAt"], future) + self.assertNotIn("accountId", result) + + +if __name__ == "__main__": + unittest.main() + diff --git a/watcher/wake_watcher.py b/watcher/wake_watcher.py index cb323d0..e2221cc 100644 --- a/watcher/wake_watcher.py +++ b/watcher/wake_watcher.py @@ -1,13 +1,9 @@ #!/usr/bin/env python3 -"""Local quota watcher for WAKE v0.3. +"""WAKE v0.4.0: checkpoint-first, job-first Codex continuation. -This helper does not store OpenAI credentials and does not make model turns when -checking quota. It asks the locally installed Codex App Server for -account/rateLimits/read and only launches an exact registered thread after -ordinaryUsageAllowed becomes true. - -Thread resume is intentionally explicit and experimental for Codex Desktop -sessions because host/UI synchronization can vary across Codex versions. +WAKE does not resume or take ownership of an existing Desktop thread. +It tracks a durable job, waits for ordinary included usage to return, then +starts a NEW lightweight Codex exec thread that reads only the checkpoint. """ from __future__ import annotations @@ -15,52 +11,61 @@ import argparse import json import os +import queue import re import shutil import signal import subprocess import sys +import threading import time +import uuid from datetime import datetime, timezone from pathlib import Path from typing import Any -VERSION = "0.3.2" +VERSION = "0.4.0" SKILL_ROOT = Path.home() / ".codex" / "skills" / "wake" -STATE_DIR = SKILL_ROOT / ".state" -REGISTRY_FILE = STATE_DIR / "registrations.json" -PID_FILE = STATE_DIR / "watcher.pid" -LOG_FILE = STATE_DIR / "watcher.log" - -DEFAULT_RECOVERY_PROMPT = ( - "[WAKE quota recovery] Included Codex usage is available again. " - "Return to this exact conversation and classify state before acting: " - "if WAITING_USER, pause WAKE; if ACTIVE, do not duplicate work; " - "if DONE, stop WAKE; otherwise continue only the unfinished task." -) - +WAKE_HOME = Path.home() / ".codex" / "wake" +JOBS_DIR = WAKE_HOME / "jobs" +PID_FILE = WAKE_HOME / "watcher.pid" +LOG_FILE = WAKE_HOME / "watcher.log" THREAD_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$") +JOB_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{5,95}$") _STOP = False +MAX_TRANSIENT_RETRIES = 8 +RETRY_BASE_SECONDS = 60 +RETRY_MAX_SECONDS = 900 +REASONING_EFFORTS = {"low", "medium", "high", "xhigh", "ultra", "persistent", "max"} +MODEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$") + +class WakeError(RuntimeError): + pass -def utc_now() -> int: + +def now() -> int: return int(time.time()) def iso(ts: int | None) -> str | None: - if ts is None: - return None - return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat() + return datetime.fromtimestamp(ts, tz=timezone.utc).isoformat() if ts else None + +def ensure_dirs() -> None: + JOBS_DIR.mkdir(parents=True, exist_ok=True) -def ensure_state_dir() -> None: - STATE_DIR.mkdir(parents=True, exist_ok=True) + +def atomic_json(path: Path, obj: dict[str, Any]) -> None: + path.parent.mkdir(parents=True, exist_ok=True) + tmp = path.with_suffix(path.suffix + ".tmp") + tmp.write_text(json.dumps(obj, ensure_ascii=False, indent=2), encoding="utf-8") + tmp.replace(path) def log(message: str) -> None: - ensure_state_dir() - stamp = datetime.now().astimezone().isoformat(timespec="seconds") - line = f"[{stamp}] {message}" + ensure_dirs() + line = f"[{datetime.now().astimezone().isoformat(timespec='seconds')}] {message}" print(line, flush=True) try: with LOG_FILE.open("a", encoding="utf-8") as f: @@ -69,366 +74,264 @@ def log(message: str) -> None: pass -def load_registry() -> dict[str, dict[str, Any]]: - ensure_state_dir() - if not REGISTRY_FILE.exists(): - return {} - try: - data = json.loads(REGISTRY_FILE.read_text(encoding="utf-8")) - except (OSError, json.JSONDecodeError): - return {} - if not isinstance(data, dict): - return {} - entries = data.get("registrations", {}) - return entries if isinstance(entries, dict) else {} - - -def save_registry(entries: dict[str, dict[str, Any]]) -> None: - ensure_state_dir() - tmp = REGISTRY_FILE.with_suffix(".tmp") - payload = { - "version": VERSION, - "updatedAt": utc_now(), - "registrations": entries, - } - tmp.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") - tmp.replace(REGISTRY_FILE) +def hidden_creationflags() -> int: + if os.name != "nt": + return 0 + return int(getattr(subprocess, "CREATE_NO_WINDOW", 0)) def find_codex() -> str: - codex = shutil.which("codex") - if not codex: - raise RuntimeError("codex executable was not found in PATH") - return codex - - -def app_server_commands(codex: str) -> list[list[str]]: - return [ - [codex, "app-server"], - [codex, "app-server", "--stdio"], - ] + value = shutil.which("codex") + if not value: + raise WakeError("codex executable was not found in PATH") + return value -def _stderr_reader(stream: Any, sink: list[str]) -> None: - try: - for line in stream: - sink.append(line.rstrip()) - if len(sink) > 200: - del sink[:100] - except Exception: - pass +def validate_thread_id(value: str) -> str: + if not THREAD_ID_RE.match(value): + raise WakeError("unsafe thread id") + return value -def _stdout_reader(stream: Any, out_queue: Any) -> None: - try: - for line in stream: - out_queue.put(line) - except Exception as exc: - out_queue.put(exc) - finally: - out_queue.put(None) +def validate_job_id(value: str) -> str: + if not JOB_ID_RE.match(value): + raise WakeError("unsafe job id") + return value -def _send_json(proc: subprocess.Popen[str], message: dict[str, Any]) -> None: - if proc.stdin is None: - raise RuntimeError("app-server stdin is unavailable") - proc.stdin.write(json.dumps(message, separators=(",", ":")) + "\n") - proc.stdin.flush() +def current_thread_id(required: bool = True) -> str | None: + value = os.environ.get("CODEX_THREAD_ID", "").strip() + if not value: + if required: + raise WakeError("CODEX_THREAD_ID is unavailable; run inside the target Codex conversation") + return None + return validate_thread_id(value) -def _wait_for_response(out_queue: Any, request_id: int, timeout: float) -> dict[str, Any]: - import queue - - deadline = time.time() + timeout - while True: - remaining = deadline - time.time() - if remaining <= 0: - raise RuntimeError(f"timed out waiting for app-server response id={request_id}") - try: - item = out_queue.get(timeout=remaining) - except queue.Empty: - raise RuntimeError(f"timed out waiting for app-server response id={request_id}") - if item is None: - raise RuntimeError("app-server stdout closed before the requested response") - if isinstance(item, Exception): - raise RuntimeError(f"app-server stdout reader failed: {item}") - line = str(item).strip() - if not line: - continue - try: - obj = json.loads(line) - except json.JSONDecodeError: - continue - if obj.get("id") != request_id: - continue - if "error" in obj: - raise RuntimeError(f"app-server error: {obj['error']}") - result = obj.get("result") - if not isinstance(result, dict): - raise RuntimeError("app-server returned a non-object result") - return result +def new_job_id() -> str: + return f"wake-{time.strftime('%Y%m%d-%H%M%S')}-{uuid.uuid4().hex[:8]}" -def read_rate_limits(timeout: int = 30) -> dict[str, Any]: - import queue - import threading +def job_dir(job_id: str) -> Path: + return JOBS_DIR / validate_job_id(job_id) - codex = find_codex() - last_error: str | None = None - for cmd in app_server_commands(codex): - proc: subprocess.Popen[str] | None = None - stderr_lines: list[str] = [] - try: - proc = subprocess.Popen( - cmd, - stdin=subprocess.PIPE, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - encoding="utf-8", - errors="replace", - bufsize=1, - ) - assert proc.stdout is not None - assert proc.stderr is not None - - out_queue: queue.Queue[Any] = queue.Queue() - threading.Thread(target=_stdout_reader, args=(proc.stdout, out_queue), daemon=True).start() - threading.Thread(target=_stderr_reader, args=(proc.stderr, stderr_lines), daemon=True).start() - - init_id = 1 - quota_id = 7 - _send_json( - proc, - { - "method": "initialize", - "id": init_id, - "params": { - "clientInfo": { - "name": "codex-wake-watcher", - "title": "Codex WAKE Watcher", - "version": VERSION, - } - }, - }, - ) - _wait_for_response(out_queue, init_id, timeout=min(timeout, 15)) - _send_json(proc, {"method": "initialized", "params": {}}) - _send_json(proc, {"method": "account/rateLimits/read", "id": quota_id}) - return _wait_for_response(out_queue, quota_id, timeout=timeout) - except Exception as exc: - detail = "\n".join(stderr_lines[-20:]).strip() - last_error = f"{exc}; stderr={detail}" if detail else str(exc) - finally: - if proc is not None: - try: - if proc.stdin: - proc.stdin.close() - except Exception: - pass - try: - proc.terminate() - proc.wait(timeout=2) - except Exception: - try: - proc.kill() - except Exception: - pass - - raise RuntimeError(last_error or "unable to query Codex rate limits") - - -def collect_reset_times(value: Any, out: set[int]) -> None: - if isinstance(value, dict): - for key, item in value.items(): - if key == "resetsAt" and isinstance(item, int): - out.add(item) - else: - collect_reset_times(item, out) - elif isinstance(value, list): - for item in value: - collect_reset_times(item, out) - - -def summarize_quota(result: dict[str, Any]) -> dict[str, Any]: - ordinary = result.get("ordinaryUsageAllowed") - if ordinary not in (True, False, None): - ordinary = None - - resets: set[int] = set() - collect_reset_times(result.get("rateLimits"), resets) - collect_reset_times(result.get("rateLimitsByLimitId"), resets) - - now = utc_now() - future = sorted(ts for ts in resets if ts > now - 2) - return { - "ordinaryUsageAllowed": ordinary, - "accountPresent": bool(result.get("accountId")), - "nextResetAt": future[0] if future else None, - "nextResetAtUtc": iso(future[0]) if future else None, - "allResetTimes": future, - "rateLimits": result.get("rateLimits"), - "rateLimitsByLimitId": result.get("rateLimitsByLimitId"), - } +def state_path(job_id: str) -> Path: + return job_dir(job_id) / "state.json" -def validate_thread_id(thread_id: str) -> None: - if not THREAD_ID_RE.match(thread_id): - raise SystemExit( - "Refusing unsafe thread id. Supply the exact Codex thread/session id; " - "WAKE never guesses or uses --last." - ) +def checkpoint_path(job_id: str) -> Path: + return job_dir(job_id) / "checkpoint.md" -def current_thread_id(required: bool = True) -> str | None: - value = os.environ.get("CODEX_THREAD_ID", "").strip() - if not value: - if required: - raise SystemExit( - "CODEX_THREAD_ID is not present. Run this command from a shell/tool " - "invoked by the target Codex conversation, not from a normal terminal." - ) - return None - validate_thread_id(value) - return value +def run_dir(job_id: str) -> Path: + return job_dir(job_id) / "runs" -def normalize_cwd(cwd: str | None) -> Path: - path = Path(cwd or os.getcwd()).expanduser().resolve() - if not path.exists() or not path.is_dir(): - raise SystemExit(f"cwd does not exist or is not a directory: {path}") - return path +def load_job(job_id: str) -> dict[str, Any]: + path = state_path(job_id) + if not path.exists(): + raise WakeError(f"job not found: {job_id}") + data = json.loads(path.read_text(encoding="utf-8")) + if not isinstance(data, dict): + raise WakeError(f"invalid job state: {job_id}") + return data -def register_thread(thread_id: str, cwd: str | None, prompt: str) -> dict[str, Any]: - validate_thread_id(thread_id) - path = normalize_cwd(cwd) - entries = load_registry() - now = utc_now() - old = entries.get(thread_id, {}) - entry = { - "threadId": thread_id, - "cwd": str(path), - "prompt": prompt, - "state": "monitoring", - "registeredAt": old.get("registeredAt") or now, - "updatedAt": now, - "quotaBlockedObservedAt": old.get("quotaBlockedObservedAt"), - "lastLaunchAt": old.get("lastLaunchAt"), - "launchLog": old.get("launchLog"), - "lastError": None, - "pauseReason": None, - } - entries[thread_id] = entry - save_registry(entries) - return entry +def save_job(job: dict[str, Any]) -> None: + job["updatedAt"] = now() + atomic_json(state_path(str(job["jobId"])), job) -def register_current(cwd: str | None, prompt: str) -> None: - thread_id = current_thread_id(required=True) - assert thread_id is not None - print(json.dumps(register_thread(thread_id, cwd, prompt), ensure_ascii=False, indent=2)) +def all_jobs() -> list[dict[str, Any]]: + ensure_dirs() + jobs: list[dict[str, Any]] = [] + for path in sorted(JOBS_DIR.glob("*/state.json")): + try: + obj = json.loads(path.read_text(encoding="utf-8")) + if isinstance(obj, dict): + jobs.append(obj) + except Exception: + continue + return jobs -def pause_thread(thread_id: str, reason: str) -> None: +def find_job_for_thread(thread_id: str) -> dict[str, Any] | None: validate_thread_id(thread_id) - entries = load_registry() - entry = entries.get(thread_id) - if not entry: - raise SystemExit("thread is not registered with WAKE") - entry["state"] = "waiting_user" - entry["pauseReason"] = reason or "waiting for user" - entry["updatedAt"] = utc_now() - entries[thread_id] = entry - save_registry(entries) - print(json.dumps(entry, ensure_ascii=False, indent=2)) - + matches = [ + job for job in all_jobs() + if job.get("state") != "completed" + and thread_id in {job.get("originalThreadId"), job.get("currentThreadId")} + ] + if not matches: + return None + matches.sort(key=lambda x: int(x.get("updatedAt") or 0), reverse=True) + return matches[0] -def pause_current(reason: str) -> None: - thread_id = current_thread_id(required=True) - assert thread_id is not None - pause_thread(thread_id, reason) +CHECKPOINT_TEMPLATE = """# WAKE Task Checkpoint -def resume_thread(thread_id: str, cwd: str | None = None) -> None: - validate_thread_id(thread_id) - entries = load_registry() - entry = entries.get(thread_id) - if not entry: - entry = register_thread(thread_id, cwd, DEFAULT_RECOVERY_PROMPT) - else: - if cwd: - entry["cwd"] = str(normalize_cwd(cwd)) - entry["state"] = "monitoring" - entry["pauseReason"] = None - entry["updatedAt"] = utc_now() - entry["lastError"] = None - entries[thread_id] = entry - save_registry(entries) - print(json.dumps(entry, ensure_ascii=False, indent=2)) - - -def resume_current(cwd: str | None = None) -> None: - thread_id = current_thread_id(required=True) - assert thread_id is not None - resume_thread(thread_id, cwd) - - -def unregister(thread_id: str) -> None: - validate_thread_id(thread_id) - entries = load_registry() - existed = entries.pop(thread_id, None) - save_registry(entries) - print("unregistered" if existed else "not registered") +## Goal +REPLACE_ME +## Constraints +- Preserve completed work. +- Do not reconstruct or reread the old chat unless this checkpoint explicitly requires it. -def unregister_current() -> None: - thread_id = current_thread_id(required=True) - assert thread_id is not None - unregister(thread_id) +## Completed +- None recorded yet. +## Decisions +- None recorded yet. -def show_current() -> None: - thread_id = current_thread_id(required=True) - assert thread_id is not None - entry = load_registry().get(thread_id) - print(json.dumps({"threadId": thread_id, "registration": entry}, ensure_ascii=False, indent=2)) +## Files changed +- None recorded yet. +## Verification +- None recorded yet. -def arm_quota(thread_id: str, cwd: str, prompt: str) -> None: - entry = register_thread(thread_id, cwd, prompt) - entries = load_registry() - entry["state"] = "quota_waiting" - entry["quotaBlockedObservedAt"] = utc_now() - entry["updatedAt"] = utc_now() - entries[thread_id] = entry - save_registry(entries) - print(json.dumps(entry, ensure_ascii=False, indent=2)) +## Current state +Checkpoint created; fill this before arming WAKE. +## Next actions +1. REPLACE_ME -def disarm(thread_id: str) -> None: - unregister(thread_id) +## Do not repeat +- Do not redo completed investigation without evidence it is stale. +## Blockers +- None. -def list_regs() -> None: - print(json.dumps(load_registry(), ensure_ascii=False, indent=2)) +## Handoff +Update this file after every meaningful milestone and before long-running work. +""" -def is_process_alive(pid: int) -> bool: - if pid <= 0: +def create_job( + thread_id: str, + cwd: str | None, + goal: str | None, + model: str | None = None, + reasoning_effort: str | None = None, +) -> dict[str, Any]: + validate_thread_id(thread_id) + if model is not None: + model = model.strip() + if not MODEL_RE.match(model): + raise WakeError("unsafe model name") + if reasoning_effort is not None: + reasoning_effort = reasoning_effort.strip().lower() + if reasoning_effort not in REASONING_EFFORTS: + raise WakeError(f"unsupported reasoning effort: {reasoning_effort}") + existing = find_job_for_thread(thread_id) + if existing is not None: + result = dict(existing) + result["alreadyExists"] = True + return result + root = Path(cwd or os.getcwd()).expanduser().resolve() + if not root.exists() or not root.is_dir(): + raise WakeError(f"cwd does not exist: {root}") + jid = new_job_id() + jdir = job_dir(jid) + jdir.mkdir(parents=True, exist_ok=False) + run_dir(jid).mkdir() + cp = CHECKPOINT_TEMPLATE + if goal: + cp = cp.replace("## Goal\nREPLACE_ME", f"## Goal\n{goal.strip()}", 1) + checkpoint_path(jid).write_text(cp, encoding="utf-8") + job = { + "version": VERSION, + "jobId": jid, + "originalThreadId": thread_id, + "currentThreadId": thread_id, + "cwd": str(root), + "checkpointPath": str(checkpoint_path(jid)), + "state": "draft", + "checkpointReady": False, + "generation": 0, + "model": model, + "reasoningEffort": reasoning_effort, + "retryCount": 0, + "nextRetryAt": None, + "createdAt": now(), + "updatedAt": now(), + "lastError": None, + "pauseReason": None, + "lastRunPid": None, + "lastRunLog": None, + } + save_job(job) + return job + + +def arm_job(job_id: str) -> dict[str, Any]: + job = load_job(job_id) + cp_path = checkpoint_path(job_id) + text = cp_path.read_text(encoding="utf-8") + if "## Goal\nREPLACE_ME" in text or "1. REPLACE_ME" in text: + raise WakeError("checkpoint still contains REPLACE_ME; fill Goal and Next actions first") + if cp_path.stat().st_size > 65536: + raise WakeError("checkpoint exceeds 64 KiB; compact it before arming") + job["checkpointReady"] = True + job["state"] = "monitoring" + job["lastError"] = None + save_job(job) + return job + + +def pause_job(job_id: str, reason: str) -> dict[str, Any]: + job = load_job(job_id) + job["state"] = "waiting_user" + job["pauseReason"] = reason.strip() or "waiting for user" + save_job(job) + return job + + +def rearm_job(job_id: str) -> dict[str, Any]: + job = load_job(job_id) + if job.get("state") == "completed": + raise WakeError("completed jobs are immutable; create a new job instead") + if not job.get("checkpointReady"): + raise WakeError("job checkpoint is not armed") + job["state"] = "monitoring" + job["pauseReason"] = None + job["lastError"] = None + save_job(job) + return job + + +def stop_job(job_id: str) -> dict[str, Any]: + job = load_job(job_id) + if job.get("state") == "completed": + return job + job["state"] = "stopped" + job["pauseReason"] = "stopped by user" + job["lastRunPid"] = None + save_job(job) + return job + + +def complete_job(job_id: str) -> dict[str, Any]: + job = load_job(job_id) + job["state"] = "completed" + job["completedAt"] = now() + job["lastRunPid"] = None + save_job(job) + return job + + +def is_process_alive(pid: int | None) -> bool: + if not pid or pid <= 0: return False if os.name == "nt": try: - completed = subprocess.run( + cp = subprocess.run( ["tasklist", "/FI", f"PID eq {pid}", "/FO", "CSV", "/NH"], - stdout=subprocess.PIPE, - stderr=subprocess.DEVNULL, - text=True, - timeout=5, - check=False, + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + text=True, timeout=5, check=False, + creationflags=hidden_creationflags(), ) - return str(pid) in completed.stdout + return str(pid) in cp.stdout except Exception: return False try: @@ -438,150 +341,445 @@ def is_process_alive(pid: int) -> bool: return False -def watcher_status() -> dict[str, Any]: - pid: int | None = None - if PID_FILE.exists(): +class StdioAppServer: + """Persistent hidden app-server used only for account/rateLimits/read.""" + + def __init__(self) -> None: + self.proc: subprocess.Popen[str] | None = None + self.out: queue.Queue[Any] = queue.Queue() + self.stderr: list[str] = [] + self.next_id = 1 + self.lock = threading.RLock() + + def start(self) -> None: + if self.proc is not None and self.proc.poll() is None: + return + self.close() + self.out = queue.Queue() + self.stderr = [] + kwargs: dict[str, Any] = { + "stdin": subprocess.PIPE, "stdout": subprocess.PIPE, "stderr": subprocess.PIPE, + "text": True, "encoding": "utf-8", "errors": "replace", "bufsize": 1, + } + flags = hidden_creationflags() + if flags: + kwargs["creationflags"] = flags + self.proc = subprocess.Popen([find_codex(), "app-server"], **kwargs) + assert self.proc.stdout is not None and self.proc.stderr is not None + threading.Thread(target=self._read_stdout, daemon=True).start() + threading.Thread(target=self._read_stderr, daemon=True).start() + self.request("initialize", {"clientInfo": { + "name": "codex-wake-watcher", "title": "Codex WAKE Watcher", "version": VERSION, + }}, timeout=15, skip_start=True) + self.notify("initialized", {}, skip_start=True) + + def _read_stdout(self) -> None: + assert self.proc is not None and self.proc.stdout is not None try: - pid = int(PID_FILE.read_text(encoding="utf-8").strip()) - except (OSError, ValueError): - pid = None - entries = load_registry() - counts: dict[str, int] = {} - for entry in entries.values(): - state = str(entry.get("state") or "unknown") - counts[state] = counts.get(state, 0) + 1 + for line in self.proc.stdout: + self.out.put(line) + except Exception as exc: + self.out.put(exc) + finally: + self.out.put(None) + + def _read_stderr(self) -> None: + assert self.proc is not None and self.proc.stderr is not None + try: + for line in self.proc.stderr: + self.stderr.append(line.rstrip()) + self.stderr[:] = self.stderr[-200:] + except Exception: + pass + + def _send(self, obj: dict[str, Any]) -> None: + if self.proc is None or self.proc.stdin is None or self.proc.poll() is not None: + raise WakeError("app-server is not running") + self.proc.stdin.write(json.dumps(obj, separators=(",", ":")) + "\n") + self.proc.stdin.flush() + + def notify(self, method: str, params: dict[str, Any] | None = None, *, skip_start: bool = False) -> None: + if not skip_start: + self.start() + obj: dict[str, Any] = {"method": method} + if params is not None: + obj["params"] = params + self._send(obj) + + def request(self, method: str, params: dict[str, Any] | None = None, *, + timeout: float = 30, skip_start: bool = False) -> dict[str, Any]: + with self.lock: + if not skip_start: + self.start() + req_id = self.next_id + self.next_id += 1 + obj: dict[str, Any] = {"method": method, "id": req_id} + if params is not None: + obj["params"] = params + self._send(obj) + deadline = time.time() + timeout + while True: + remaining = deadline - time.time() + if remaining <= 0: + raise WakeError(f"timed out waiting for {method}") + try: + item = self.out.get(timeout=remaining) + except queue.Empty as exc: + raise WakeError(f"timed out waiting for {method}") from exc + if item is None: + raise WakeError("app-server stdout closed; " + " | ".join(self.stderr[-10:])) + if isinstance(item, Exception): + raise WakeError(str(item)) + try: + msg = json.loads(str(item).strip()) + except json.JSONDecodeError: + continue + if msg.get("id") != req_id: + continue + if "error" in msg: + raise WakeError(f"{method} failed: {msg['error']}") + result = msg.get("result") + if not isinstance(result, dict): + raise WakeError(f"{method} returned non-object result") + return result + + def close(self) -> None: + proc, self.proc = self.proc, None + if proc is None: + return + try: + if proc.stdin: + proc.stdin.close() + except Exception: + pass + try: + proc.terminate() + proc.wait(timeout=2) + except Exception: + try: + proc.kill() + except Exception: + pass + + +def summarize_quota(raw: dict[str, Any]) -> dict[str, Any]: + ordinary = raw.get("ordinaryUsageAllowed") + limits = raw.get("rateLimits") + if not isinstance(limits, dict): + limits = {} + times: list[int] = [] + for key in ("primary", "secondary"): + item = limits.get(key) + if isinstance(item, dict) and isinstance(item.get("resetsAt"), int): + times.append(int(item["resetsAt"])) + future = sorted(x for x in times if x > now()) + next_reset = future[0] if future else (min(times) if times else None) return { - "pid": pid, - "running": bool(pid and is_process_alive(pid)), - "registry": str(REGISTRY_FILE), - "log": str(LOG_FILE), - "registeredCount": len(entries), - "stateCounts": counts, - "currentThreadEnvPresent": bool(os.environ.get("CODEX_THREAD_ID")), + "ordinaryUsageAllowed": ordinary if isinstance(ordinary, bool) else None, + "accountPresent": bool(raw.get("accountId")), + "nextResetAt": next_reset, + "nextResetAtUtc": iso(next_reset), + "allResetTimes": sorted(set(times)), + "rateLimits": limits, } -def safe_name(thread_id: str) -> str: - return re.sub(r"[^A-Za-z0-9_.-]", "_", thread_id)[:80] +def next_sleep(summary: dict[str, Any]) -> int: + if summary.get("ordinaryUsageAllowed") is False: + reset = summary.get("nextResetAt") + if isinstance(reset, int): + return min(max(5, reset - now() + 2), 300) + return 300 + return 60 -def launch_resume(entry: dict[str, Any]) -> tuple[bool, str]: - thread_id = str(entry["threadId"]) - validate_thread_id(thread_id) - cwd = Path(str(entry["cwd"])) - prompt = str(entry.get("prompt") or DEFAULT_RECOVERY_PROMPT) - if not cwd.exists(): - return False, f"cwd no longer exists: {cwd}" +def continuation_prompt(job: dict[str, Any]) -> str: + jid = str(job["jobId"]) + cp = str(job["checkpointPath"]) + cwd = str(job["cwd"]) + helper = str(SKILL_ROOT / "watcher" / "wake_watcher.py") + return f"""Continue WAKE job {jid}. This is a checkpoint handoff, not a chat resume. + +Read FIRST and treat as authoritative: +{cp} + +Then inspect only the minimum live workspace state needed in: +{cwd} + +If that directory is inside a Git worktree, normally inspect: +- git status --short +- git diff --stat +If it is not a Git worktree, skip Git checks without treating that as an error. + +Do NOT reconstruct, resume, or reread the previous chat/thread. Do NOT repeat work listed +under Completed or Do not repeat unless verification is demonstrably stale. + +Continue from Next actions and work toward the Goal. Update the checkpoint after every +meaningful milestone and before long-running or risky operations. Keep it concise. - codex = find_codex() - ensure_state_dir() - log_path = STATE_DIR / f"resume-{safe_name(thread_id)}-{utc_now()}.log" +If user input/manual action is required, update the checkpoint, then run: +python "{helper}" job-pause --job-id {jid} --reason "" + +When the Goal is fully verified complete, update the checkpoint, then run: +python "{helper}" job-complete --job-id {jid} + +If usage quota stops execution, exit normally; WAKE will create the next checkpoint handoff. +""" + + +def launch_handoff(job: dict[str, Any]) -> dict[str, Any]: + jid = str(job["jobId"]) + source_state = str(job.get("state") or "") + cwd = Path(str(job["cwd"])) + if not cwd.exists(): + raise WakeError(f"cwd no longer exists: {cwd}") + jdir = job_dir(jid) + run_dir(jid).mkdir(parents=True, exist_ok=True) + generation = int(job.get("generation") or 0) + 1 + log_path = run_dir(jid) / f"run-{generation:04d}-{now()}.jsonl" stream = log_path.open("a", encoding="utf-8") cmd = [ - codex, - "exec", - "resume", - thread_id, - "--json", - "--skip-git-repo-check", - prompt, + find_codex(), "exec", "--json", "--color", "never", + "--sandbox", "workspace-write", "--skip-git-repo-check", + "-C", str(cwd), "--add-dir", str(jdir), ] - + model = job.get("model") + if isinstance(model, str) and model: + cmd.extend(["-m", model]) + effort = job.get("reasoningEffort") + if isinstance(effort, str) and effort: + cmd.extend(["-c", f'model_reasoning_effort="{effort}"']) + cmd.append(continuation_prompt(job)) kwargs: dict[str, Any] = { + "stdin": subprocess.DEVNULL, "stdout": stream, "stderr": subprocess.STDOUT, "cwd": str(cwd), - "stdin": subprocess.DEVNULL, - "stdout": stream, - "stderr": subprocess.STDOUT, - "close_fds": os.name != "nt", } - if os.name == "nt": - creationflags = 0 - creationflags |= getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) - creationflags |= getattr(subprocess, "DETACHED_PROCESS", 0) - kwargs["creationflags"] = creationflags - - try: - subprocess.Popen(cmd, **kwargs) - except OSError as exc: - stream.close() - return False, str(exc) + flags = hidden_creationflags() + if flags: + kwargs["creationflags"] = flags + proc = subprocess.Popen(cmd, **kwargs) stream.close() - return True, str(log_path) - + job["generation"] = generation + job["state"] = "running" + if source_state == "quota_waiting": + job["retryCount"] = 0 + job["nextRetryAt"] = None + job["currentThreadId"] = None + job["lastRunPid"] = proc.pid + job["lastRunLog"] = str(log_path) + job["lastRunStartedAt"] = now() + job["lastError"] = None + save_job(job) + log(f"job {jid} launched checkpoint handoff generation={generation} pid={proc.pid}") + return job + + +def discover_run_thread(job: dict[str, Any]) -> dict[str, Any]: + path_value = job.get("lastRunLog") + if not path_value: + return job + path = Path(str(path_value)) + if not path.exists(): + return job + try: + with path.open("r", encoding="utf-8", errors="replace") as f: + for index, line in enumerate(f): + if index >= 80: + break + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + if event.get("type") == "thread.started": + tid = event.get("thread_id") or event.get("threadId") + if isinstance(tid, str) and THREAD_ID_RE.match(tid): + if job.get("currentThreadId") != tid: + job["currentThreadId"] = tid + save_job(job) + break + except OSError: + pass + return job -def next_sleep_seconds(summary: dict[str, Any]) -> int: - ordinary = summary.get("ordinaryUsageAllowed") - if ordinary is None: - return 60 - next_reset = summary.get("nextResetAt") - if ordinary is False and isinstance(next_reset, int): - until = max(5, next_reset - utc_now() + 2) - return min(until, 300) - return 60 +def classify_run_failure(job: dict[str, Any]) -> tuple[bool, str]: + path_value = job.get("lastRunLog") + if not path_value: + return False, "continuation exited without a run log" + path = Path(str(path_value)) + if not path.exists(): + return False, "continuation run log is missing" -def process_once() -> int: - entries = load_registry() - tracked = { - k: v for k, v in entries.items() - if v.get("state") in {"monitoring", "quota_waiting"} - } - if not tracked: - return 30 + try: + size = path.stat().st_size + with path.open("rb") as f: + if size > 131072: + f.seek(size - 131072) + text = f.read().decode("utf-8", errors="replace") + except OSError as exc: + return False, f"could not read run log: {exc}" + messages: list[str] = [] + for line in text.splitlines(): + try: + event = json.loads(line) + except json.JSONDecodeError: + continue + if event.get("type") == "turn.failed": + err = event.get("error") + if isinstance(err, dict) and isinstance(err.get("message"), str): + messages.append(err["message"]) + elif event.get("type") == "error" and isinstance(event.get("message"), str): + messages.append(event["message"]) + elif event.get("type") == "item.completed": + item = event.get("item") + if isinstance(item, dict) and item.get("type") == "error" and isinstance(item.get("message"), str): + messages.append(item["message"]) + + reason = messages[-1] if messages else "continuation exited without marking job complete or waiting_user" + haystack = ("\n".join(messages) + "\n" + text[-32768:]).lower() + retryable_patterns = ( + "selected model is at capacity", + "request timed out", + "timed out", + "temporarily unavailable", + "service unavailable", + "connection reset", + "connection error", + "connection closed", + "network error", + "reconnecting", + "bad gateway", + "gateway timeout", + "http 502", + "http 503", + "http 504", + ) + return any(pattern in haystack for pattern in retryable_patterns), reason + + +def reconcile_running(job: dict[str, Any], ordinary: bool | None) -> None: + jid = str(job["jobId"]) + fresh = load_job(jid) + if fresh.get("state") != "running": + return + job = discover_run_thread(fresh) + pid = job.get("lastRunPid") + if is_process_alive(int(pid) if isinstance(pid, int) else None): + return + if ordinary is False: + job["state"] = "quota_waiting" + job["quotaBlockedObservedAt"] = now() + job["retryCount"] = 0 + job["nextRetryAt"] = None + job["lastRunPid"] = None + save_job(job) + log(f"job {jid} run ended while quota blocked; waiting for next window") + return + + retryable, reason = classify_run_failure(job) + if ordinary is True and retryable: + retry_count = int(job.get("retryCount") or 0) + 1 + if retry_count <= MAX_TRANSIENT_RETRIES: + delay = min(RETRY_BASE_SECONDS * (2 ** (retry_count - 1)), RETRY_MAX_SECONDS) + job["state"] = "retry_waiting" + job["retryCount"] = retry_count + job["nextRetryAt"] = now() + delay + job["lastRunPid"] = None + job["lastError"] = reason + save_job(job) + log(f"job {jid} transient failure; retry {retry_count}/{MAX_TRANSIENT_RETRIES} in {delay}s: {reason}") + return + reason = f"transient retry limit exceeded ({MAX_TRANSIENT_RETRIES}): {reason}" + + job["state"] = "needs_attention" + job["lastRunPid"] = None + job["nextRetryAt"] = None + job["lastError"] = reason + save_job(job) + log(f"job {jid} needs attention: {reason}") + + +def process_once(app: StdioAppServer) -> int: + jobs = [j for j in all_jobs() if j.get("checkpointReady") and j.get("state") not in {"completed", "draft", "stopped"}] + if not jobs: + return 60 try: - summary = summarize_quota(read_rate_limits()) + summary = summarize_quota(app.request("account/rateLimits/read")) except Exception as exc: log(f"quota read failed: {exc}") + app.close() return 60 - ordinary = summary.get("ordinaryUsageAllowed") - waiting = sum(1 for v in tracked.values() if v.get("state") == "quota_waiting") - log( - "quota: ordinaryUsageAllowed=" - f"{ordinary} nextResetAt={summary.get('nextResetAtUtc')} " - f"tracked={len(tracked)} quota_waiting={waiting}" - ) + log(f"quota ordinaryUsageAllowed={ordinary} nextResetAt={summary.get('nextResetAtUtc')} jobs={len(jobs)}") - changed = False - if ordinary is False: - now = utc_now() - for thread_id, entry in tracked.items(): - if entry.get("state") == "monitoring": - entry["state"] = "quota_waiting" - entry["quotaBlockedObservedAt"] = now - entry["updatedAt"] = now - entries[thread_id] = entry - changed = True - log(f"thread {thread_id} entered quota_waiting") - if changed: - save_registry(entries) - return next_sleep_seconds(summary) - - if ordinary is not True: - return next_sleep_seconds(summary) - - for thread_id, entry in list(entries.items()): - if entry.get("state") != "quota_waiting": + for job in jobs: + jid = str(job["jobId"]) + try: + job = load_job(jid) + except Exception: + continue + state = str(job.get("state")) + if state in {"completed", "draft", "stopped", "waiting_user", "needs_attention"}: + continue + if state == "running": + reconcile_running(job, ordinary) continue - ok, detail = launch_resume(entry) - if ok: - entry["state"] = "monitoring" - entry["lastLaunchAt"] = utc_now() - entry["launchLog"] = detail - entry["lastError"] = None - entry["updatedAt"] = utc_now() - log(f"quota recovered; launched exact thread {thread_id}; log={detail}") - else: - entry["lastError"] = detail - entry["updatedAt"] = utc_now() - log(f"failed to launch exact thread {thread_id}: {detail}") - entries[thread_id] = entry - changed = True - - if changed: - save_registry(entries) - return 30 + if ordinary is False: + if state != "quota_waiting": + job["state"] = "quota_waiting" + job["quotaBlockedObservedAt"] = now() + job["retryCount"] = 0 + job["nextRetryAt"] = None + job["lastError"] = None + save_job(job) + log(f"job {jid} entered quota_waiting") + continue + if ordinary is True and state in {"quota_waiting", "retry_waiting"}: + if state == "retry_waiting": + retry_at = job.get("nextRetryAt") + if isinstance(retry_at, int) and now() < retry_at: + continue + try: + latest = load_job(jid) + if latest.get("state") not in {"quota_waiting", "retry_waiting"}: + continue + retry_at = latest.get("nextRetryAt") + if latest.get("state") == "retry_waiting" and isinstance(retry_at, int) and now() < retry_at: + continue + launch_handoff(latest) + except Exception as exc: + latest = load_job(jid) + if latest.get("state") in {"quota_waiting", "retry_waiting"}: + latest["state"] = "needs_attention" + latest["lastError"] = str(exc) + latest["nextRetryAt"] = None + save_job(latest) + log(f"job {jid} handoff launch failed: {exc}") + return next_sleep(summary) + +def watcher_status() -> dict[str, Any]: + pid: int | None = None + if PID_FILE.exists(): + try: + pid = int(PID_FILE.read_text(encoding="utf-8").strip()) + except Exception: + pid = None + jobs = all_jobs() + counts: dict[str, int] = {} + for job in jobs: + key = str(job.get("state") or "unknown") + counts[key] = counts.get(key, 0) + 1 + return { + "version": VERSION, + "pid": pid, + "running": is_process_alive(pid), + "wakeHome": str(WAKE_HOME), + "jobsDir": str(JOBS_DIR), + "jobCount": len(jobs), + "stateCounts": counts, + "currentThreadEnvPresent": bool(os.environ.get("CODEX_THREAD_ID")), + } def handle_stop(_signum: int, _frame: Any) -> None: @@ -590,19 +788,21 @@ def handle_stop(_signum: int, _frame: Any) -> None: def run_daemon() -> None: - ensure_state_dir() + ensure_dirs() PID_FILE.write_text(str(os.getpid()), encoding="utf-8") signal.signal(signal.SIGINT, handle_stop) if hasattr(signal, "SIGTERM"): signal.signal(signal.SIGTERM, handle_stop) + app = StdioAppServer() log(f"WAKE watcher {VERSION} started pid={os.getpid()}") try: while not _STOP: - delay = process_once() - end = time.time() + max(1, delay) + delay = max(1, process_once(app)) + end = time.time() + delay while not _STOP and time.time() < end: time.sleep(min(1, end - time.time())) finally: + app.close() try: if PID_FILE.exists() and PID_FILE.read_text(encoding="utf-8").strip() == str(os.getpid()): PID_FILE.unlink() @@ -616,129 +816,116 @@ def doctor() -> int: "version": VERSION, "python": sys.executable, "codex": shutil.which("codex"), + "windowsNoWindow": bool(hidden_creationflags()) if os.name == "nt" else None, "watcher": watcher_status(), } + app = StdioAppServer() try: - result["quota"] = summarize_quota(read_rate_limits()) + result["quota"] = summarize_quota(app.request("account/rateLimits/read")) result["quotaReadOk"] = True except Exception as exc: result["quotaReadOk"] = False result["quotaError"] = str(exc) + finally: + app.close() print(json.dumps(result, ensure_ascii=False, indent=2)) - return 0 if result["quotaReadOk"] else 1 - - -def build_parser() -> argparse.ArgumentParser: - p = argparse.ArgumentParser(description="WAKE local quota watcher") - p.add_argument("--version", action="version", version=VERSION) - sub = p.add_subparsers(dest="command", required=True) - - sub.add_parser("quota", help="Read Codex rate limits once") - sub.add_parser("run", help="Run the watcher loop in the foreground") - sub.add_parser("once", help="Process registrations once") - sub.add_parser("list", help="List exact-thread registrations") - sub.add_parser("status", help="Show watcher process/registry status") - sub.add_parser("doctor", help="Check Codex and quota-read integration") - - register = sub.add_parser("register", help="Register one exact thread for quota monitoring") - register.add_argument("--thread-id", required=True) - register.add_argument("--cwd") - register.add_argument("--prompt", default=DEFAULT_RECOVERY_PROMPT) - - register_current_p = sub.add_parser( - "register-current", - help="Register CODEX_THREAD_ID from the current Codex shell/tool environment", - ) - register_current_p.add_argument("--cwd") - register_current_p.add_argument("--prompt", default=DEFAULT_RECOVERY_PROMPT) - - pause = sub.add_parser("pause", help="Mark one exact thread WAITING_USER") - pause.add_argument("--thread-id", required=True) - pause.add_argument("--reason", default="waiting for user") - - pause_current_p = sub.add_parser("pause-current", help="Mark current CODEX_THREAD_ID WAITING_USER") - pause_current_p.add_argument("--reason", default="waiting for user") + return 0 if result.get("quotaReadOk") else 1 - resume = sub.add_parser("resume-monitoring", help="Re-enable quota monitoring for one thread") - resume.add_argument("--thread-id", required=True) - resume.add_argument("--cwd") - resume_current_p = sub.add_parser( - "resume-current", - help="Re-enable monitoring for current CODEX_THREAD_ID", - ) - resume_current_p.add_argument("--cwd") - - unregister_p = sub.add_parser("unregister", help="Remove one exact thread registration") - unregister_p.add_argument("--thread-id", required=True) - sub.add_parser("unregister-current", help="Remove current CODEX_THREAD_ID registration") - sub.add_parser("current", help="Show registration for current CODEX_THREAD_ID") +def current_job(required: bool = True) -> dict[str, Any] | None: + tid = current_thread_id(required) + if tid is None: + return None + job = find_job_for_thread(tid) + if job is None and required: + raise WakeError(f"no active WAKE job is associated with thread {tid}") + return job - arm = sub.add_parser("arm-quota", help="Immediately mark one exact thread quota-waiting") - arm.add_argument("--thread-id", required=True) - arm.add_argument("--cwd", required=True) - arm.add_argument("--prompt", default=DEFAULT_RECOVERY_PROMPT) - dis = sub.add_parser("disarm", help="Alias for unregister") - dis.add_argument("--thread-id", required=True) - return p +def print_json(obj: Any) -> None: + print(json.dumps(obj, ensure_ascii=False, indent=2)) def main() -> int: - args = build_parser().parse_args() - if args.command == "quota": - print(json.dumps(summarize_quota(read_rate_limits()), ensure_ascii=False, indent=2)) - return 0 - if args.command == "run": - run_daemon() - return 0 - if args.command == "once": - delay = process_once() - print(json.dumps({"nextCheckSeconds": delay}, indent=2)) - return 0 - if args.command == "list": - list_regs() - return 0 - if args.command == "status": - print(json.dumps(watcher_status(), ensure_ascii=False, indent=2)) - return 0 - if args.command == "doctor": + p = argparse.ArgumentParser(description="WAKE checkpoint-first job continuation") + p.add_argument("--version", action="version", version=VERSION) + sub = p.add_subparsers(dest="cmd", required=True) + for name in ( + "run", "once", "quota", "status", "doctor", "job-list", + "job-current", "job-stop-current", "job-rearm-current", "job-complete-current", + ): + sub.add_parser(name) + + jc = sub.add_parser("job-create") + jc.add_argument("--thread-id", required=True) + jc.add_argument("--cwd") + jc.add_argument("--goal") + jc.add_argument("--model") + jc.add_argument("--reasoning-effort") + jcc = sub.add_parser("job-create-current") + jcc.add_argument("--cwd") + jcc.add_argument("--goal") + jcc.add_argument("--model") + jcc.add_argument("--reasoning-effort") + for name in ("job-arm", "job-status", "job-rearm", "job-stop", "job-complete"): + sp = sub.add_parser(name) + sp.add_argument("--job-id", required=True) + jp = sub.add_parser("job-pause") + jp.add_argument("--job-id", required=True) + jp.add_argument("--reason", default="waiting for user") + jpc = sub.add_parser("job-pause-current") + jpc.add_argument("--reason", default="waiting for user") + args = p.parse_args() + + if args.cmd == "run": + run_daemon(); return 0 + if args.cmd == "status": + print_json(watcher_status()); return 0 + if args.cmd == "doctor": return doctor() - if args.command == "register": - print(json.dumps(register_thread(args.thread_id, args.cwd, args.prompt), ensure_ascii=False, indent=2)) - return 0 - if args.command == "register-current": - register_current(args.cwd, args.prompt) - return 0 - if args.command == "pause": - pause_thread(args.thread_id, args.reason) - return 0 - if args.command == "pause-current": - pause_current(args.reason) - return 0 - if args.command == "resume-monitoring": - resume_thread(args.thread_id, args.cwd) - return 0 - if args.command == "resume-current": - resume_current(args.cwd) - return 0 - if args.command == "unregister": - unregister(args.thread_id) - return 0 - if args.command == "unregister-current": - unregister_current() - return 0 - if args.command == "current": - show_current() - return 0 - if args.command == "arm-quota": - arm_quota(args.thread_id, args.cwd, args.prompt) - return 0 - if args.command == "disarm": - disarm(args.thread_id) - return 0 + if args.cmd == "job-list": + print_json(all_jobs()); return 0 + if args.cmd == "job-current": + print_json(current_job(True)); return 0 + if args.cmd == "job-create": + print_json(create_job( + args.thread_id, args.cwd, args.goal, args.model, args.reasoning_effort + )); return 0 + if args.cmd == "job-create-current": + print_json(create_job( + current_thread_id(True), args.cwd, args.goal, args.model, args.reasoning_effort + )); return 0 + if args.cmd == "job-arm": + print_json(arm_job(args.job_id)); return 0 + if args.cmd == "job-status": + print_json(load_job(args.job_id)); return 0 + if args.cmd == "job-rearm": + print_json(rearm_job(args.job_id)); return 0 + if args.cmd == "job-stop": + print_json(stop_job(args.job_id)); return 0 + if args.cmd == "job-pause": + print_json(pause_job(args.job_id, args.reason)); return 0 + if args.cmd == "job-complete": + print_json(complete_job(args.job_id)); return 0 + if args.cmd == "job-rearm-current": + print_json(rearm_job(str(current_job(True)["jobId"]))); return 0 + if args.cmd == "job-stop-current": + print_json(stop_job(str(current_job(True)["jobId"]))); return 0 + if args.cmd == "job-pause-current": + print_json(pause_job(str(current_job(True)["jobId"]), args.reason)); return 0 + if args.cmd == "job-complete-current": + print_json(complete_job(str(current_job(True)["jobId"]))); return 0 + app = StdioAppServer() + try: + if args.cmd == "quota": + print_json(summarize_quota(app.request("account/rateLimits/read"))); return 0 + if args.cmd == "once": + print_json({"nextCheckSeconds": process_once(app)}); return 0 + finally: + app.close() return 2 if __name__ == "__main__": raise SystemExit(main()) +