|
#!/usr/bin/env python3 |
|
"""scheduled_tasks_doctor.py -- find, diagnose and restore Claude Desktop scheduled tasks. |
|
|
|
WHY: the Desktop app keeps its routine registry in a PER-ACCOUNT, PER-WORKSPACE file: |
|
|
|
macOS ~/Library/Application Support/Claude/claude-code-sessions/<accountUuid>/<workspaceUuid>/scheduled-tasks.json |
|
Windows %APPDATA%\\Claude\\claude-code-sessions\\<accountUuid>\\<workspaceUuid>\\scheduled-tasks.json |
|
Linux ~/.config/Claude/claude-code-sessions/<accountUuid>/<workspaceUuid>/scheduled-tasks.json |
|
|
|
The task PROMPTS live somewhere else entirely -- ~/.claude/scheduled-tasks/<taskId>/SKILL.md -- |
|
and they survive everything. So when the registry is wiped by an app update, or you sign in |
|
under a different account, your routines vanish from the Routines list while every prompt is |
|
still on disk. They look deleted. They are not. |
|
|
|
This script: |
|
* finds EVERY registry on the machine and prints what each one holds (which account is which); |
|
* lists ORPHANS -- task folders with a SKILL.md that no registry references (the lost routines); |
|
* can put an orphan back into a chosen registry (--restore), with a timestamped backup. |
|
|
|
MEASURED GOTCHA (2026-08-24, refined 2026-08-26 -- the refinement is the important half): |
|
the running app holds the registry in MEMORY, and it does not merely IGNORE an edit made behind |
|
its back. It OVERWRITES it. Timeline from one machine, taken from file mtimes and the app's own |
|
lastRunAt fields: |
|
|
|
08:23:45 registry on disk holds 31 tasks (9 restored by hand) |
|
08:30:38 a scheduled task runs; the app flushes its in-memory registry to record lastRunAt |
|
08:30:38 registry on disk holds 23 tasks -- all 9 restored entries are gone, silently |
|
|
|
So the window is not "until the next restart", it is "until the next time any routine fires", |
|
which on a busy machine is minutes. Restoring while the app runs looks like it worked and then |
|
quietly loses the work. Quit the app first; this script refuses to write while it is running |
|
unless you pass --force. |
|
|
|
Usage: |
|
python3 scheduled_tasks_doctor.py # report: registries, tasks, orphans |
|
python3 scheduled_tasks_doctor.py --json # same, machine-readable |
|
python3 scheduled_tasks_doctor.py --restore ID --into <registry-path> [--cron "0 10 * * *"] [--cwd DIR] |
|
# refuses while the Claude app is running; --force overrides |
|
python3 scheduled_tasks_doctor.py --self-test # runs on a synthetic tree, touches nothing real |
|
|
|
Read-only unless --restore is given. No dependencies beyond the stdlib. |
|
""" |
|
from __future__ import annotations |
|
|
|
import argparse |
|
import json |
|
import os |
|
import platform |
|
import shutil |
|
import sys |
|
import time |
|
from pathlib import Path |
|
|
|
TASKS_DIR = Path.home() / ".claude" / "scheduled-tasks" |
|
|
|
|
|
def app_roots() -> list[Path]: |
|
"""Every plausible Claude app-data root for this OS (existing ones only).""" |
|
system = platform.system() |
|
if system == "Darwin": |
|
cands = [Path.home() / "Library" / "Application Support" / "Claude"] |
|
elif system == "Windows": |
|
appdata = os.environ.get("APPDATA") |
|
cands = [Path(appdata) / "Claude"] if appdata else [] |
|
else: |
|
cands = [Path.home() / ".config" / "Claude"] |
|
return [c for c in cands if c.is_dir()] |
|
|
|
|
|
def find_registries(roots: list[Path]) -> list[Path]: |
|
out: list[Path] = [] |
|
for root in roots: |
|
for sub in ("claude-code-sessions", "local-agent-mode-sessions"): |
|
base = root / sub |
|
if base.is_dir(): |
|
out.extend(sorted(base.glob("*/*/scheduled-tasks.json"))) |
|
return out |
|
|
|
|
|
def read_registry(path: Path) -> list[dict]: |
|
try: |
|
data = json.loads(path.read_text(encoding="utf-8")) |
|
except (OSError, json.JSONDecodeError): |
|
return [] |
|
return data.get("scheduledTasks") or [] |
|
|
|
|
|
def registry_label(path: Path) -> str: |
|
"""<accountUuid>/<workspaceUuid> -- the two path segments that decide visibility.""" |
|
return f"{path.parent.parent.name}/{path.parent.name}" |
|
|
|
|
|
def local_task_ids() -> list[str]: |
|
if not TASKS_DIR.is_dir(): |
|
return [] |
|
return sorted(d.name for d in TASKS_DIR.iterdir() if (d / "SKILL.md").is_file()) |
|
|
|
|
|
def collect() -> dict: |
|
registries = [] |
|
registered: set[str] = set() |
|
for path in find_registries(app_roots()): |
|
tasks = read_registry(path) |
|
registries.append({ |
|
"path": str(path), |
|
"label": registry_label(path), |
|
"count": len(tasks), |
|
"tasks": [ |
|
{ |
|
"id": t.get("id"), |
|
"enabled": bool(t.get("enabled")), |
|
"cron": t.get("cronExpression") or "", |
|
"fireAt": t.get("fireAt") or "", |
|
"lastRunAt": t.get("lastRunAt") or "", |
|
} |
|
for t in tasks |
|
], |
|
}) |
|
registered.update(t.get("id") for t in tasks if t.get("id")) |
|
on_disk = local_task_ids() |
|
return { |
|
"registries": registries, |
|
"tasks_dir": str(TASKS_DIR), |
|
"on_disk": on_disk, |
|
"orphans": [t for t in on_disk if t not in registered], |
|
} |
|
|
|
|
|
def describe(task_id: str) -> str: |
|
"""First non-empty line of the SKILL.md frontmatter description, if any.""" |
|
skill = TASKS_DIR / task_id / "SKILL.md" |
|
try: |
|
for line in skill.read_text(encoding="utf-8", errors="replace").splitlines()[:12]: |
|
if line.startswith("description:"): |
|
return line.split(":", 1)[1].strip()[:100] |
|
except OSError: |
|
pass |
|
return "" |
|
|
|
|
|
def report(data: dict) -> int: |
|
if not data["registries"]: |
|
print("no registry found -- is the Claude desktop app installed for this user?") |
|
return 2 |
|
print(f"registries found: {len(data['registries'])}\n") |
|
for reg in data["registries"]: |
|
print(f" [{reg['label']}] {reg['count']} task(s)") |
|
print(f" {reg['path']}") |
|
for t in reg["tasks"]: |
|
when = t["cron"] or (f"once {t['fireAt']}" if t["fireAt"] else "ad-hoc") |
|
flag = "on " if t["enabled"] else "off" |
|
ran = t["lastRunAt"][:19] if t["lastRunAt"] else "never" |
|
print(f" {flag} {t['id']:<44} {when:<20} last={ran}") |
|
print() |
|
print(f"task folders on disk ({data['tasks_dir']}): {len(data['on_disk'])}") |
|
orphans = data["orphans"] |
|
if not orphans: |
|
print("orphans: none -- every prompt on disk is registered somewhere.") |
|
return 0 |
|
print(f"\nORPHANS ({len(orphans)}) -- prompt exists, no registry references it:") |
|
for task_id in orphans: |
|
desc = describe(task_id) |
|
print(f" {task_id}" + (f" -- {desc}" if desc else "")) |
|
print("\nThese are your lost routines. Restore one with:") |
|
print(f" python3 {Path(sys.argv[0]).name} --restore <id> --into <registry-path> --cron \"0 10 * * *\"") |
|
print("Then RESTART the Claude app -- a running app will not re-read this file.") |
|
return 1 |
|
|
|
|
|
def _parse_win_count(text: str) -> int: |
|
"""Count from a PowerShell Measure-Object. Its own function so it can be tested against |
|
real captured output instead of being trusted.""" |
|
for line in (text or "").splitlines(): |
|
line = line.strip() |
|
if line.isdigit(): |
|
return int(line) |
|
return -1 # could not read a count -- caller treats this as "assume running" |
|
|
|
|
|
def _mac_has_desktop(ps_comm_output: str) -> bool: |
|
"""True if `ps -Ao comm=` output contains the DESKTOP app, not the CLI. |
|
|
|
Case matters and is the whole point: the desktop app is `Claude`, the Claude Code CLI is |
|
`claude`. Lower-casing here would make every CLI session look like a running desktop app |
|
and refuse every restore. |
|
""" |
|
for line in (ps_comm_output or "").splitlines(): |
|
base = line.strip().rsplit("/", 1)[-1] |
|
if base == "Claude" or base.startswith("Claude Helper"): |
|
return True |
|
return False |
|
|
|
|
|
def app_is_running() -> bool: |
|
"""Is a Claude Desktop process alive right now? |
|
|
|
This is a SAFETY question, not a cosmetic one: a running app rewrites the registry from |
|
memory the next time any routine fires, so a hand-written entry is not just inert, it is |
|
discarded (see the module docstring for the measured timeline). |
|
|
|
Windows detection deliberately does NOT match on the process name. On Windows the Claude |
|
Code CLI is also `claude.exe`, so a name match reports "desktop app alive" during any CLI |
|
session -- including the one you are probably running this from. Only the MSIX install |
|
under WindowsApps\\Claude_ is the desktop app, so the filter is on ExecutablePath. |
|
(Measured 2026-08-26: a name match on "Claude.exe" also missed 16 live processes outright, |
|
because tasklist prints "claude.exe" and the comparison was case-sensitive. Two bugs, one |
|
line -- hence the ExecutablePath query and the unit-tested parsers.) |
|
|
|
Conservative on purpose: when the check itself fails we answer True (assume running). |
|
A false "yes" costs the user one --force flag; a false "no" costs them the restore. |
|
""" |
|
import subprocess |
|
try: |
|
if platform.system() == "Windows": |
|
cmd = ("(Get-CimInstance Win32_Process -Filter \"Name='claude.exe'\" | " |
|
"Where-Object { $_.ExecutablePath -like '*\\WindowsApps\\Claude_*' } | " |
|
"Measure-Object).Count") |
|
out = subprocess.run(["powershell", "-NoProfile", "-Command", cmd], |
|
capture_output=True, text=True, timeout=60).stdout |
|
n = _parse_win_count(out) |
|
return True if n < 0 else n > 0 |
|
out = subprocess.run(["ps", "-Ao", "comm="], capture_output=True, text=True, |
|
timeout=20).stdout |
|
return _mac_has_desktop(out) |
|
except Exception: |
|
return True |
|
|
|
|
|
def restore(task_id: str, into: Path, cron: str | None, fire_at: str | None, cwd: str | None, |
|
force: bool = False) -> int: |
|
skill = TASKS_DIR / task_id / "SKILL.md" |
|
if not skill.is_file(): |
|
print(f"refusing: no prompt at {skill}") |
|
return 2 |
|
if not into.is_file(): |
|
print(f"refusing: no registry at {into}") |
|
return 2 |
|
if cron and fire_at: |
|
print("refusing: --cron and --fire-at are mutually exclusive") |
|
return 2 |
|
if not force and app_is_running(): |
|
print("refusing: the Claude desktop app is running.") |
|
print(" A running app holds the registry in memory and rewrites the file the next time") |
|
print(" ANY routine fires -- measured: 31 tasks on disk at 08:23, 23 at 08:30, the 9") |
|
print(" restored entries gone with no error. The write would look like it worked.") |
|
print(" Quit the app, run this again, then start the app and confirm a next-run time.") |
|
print(" --force writes anyway (only useful if you are about to quit the app yourself).") |
|
return 3 |
|
|
|
data = json.loads(into.read_text(encoding="utf-8")) |
|
tasks = data.get("scheduledTasks") or [] |
|
if any(t.get("id") == task_id for t in tasks): |
|
print(f"already present in {registry_label(into)} -- nothing to do") |
|
return 0 |
|
|
|
entry = { |
|
"id": task_id, |
|
"enabled": True, |
|
"filePath": str(skill), |
|
"createdAt": int(time.time() * 1000), |
|
"cwd": cwd or str(Path.home()), |
|
} |
|
if cron: |
|
entry["cronExpression"] = cron |
|
elif fire_at: |
|
entry["fireAt"] = fire_at |
|
|
|
backup = into.with_suffix(f".json.bak-{time.strftime('%Y%m%d-%H%M%S')}") |
|
shutil.copy2(into, backup) |
|
tasks.append(entry) |
|
data["scheduledTasks"] = tasks |
|
into.write_text(json.dumps(data, indent=2), encoding="utf-8") |
|
|
|
print(f"restored {task_id} into {registry_label(into)}") |
|
print(f"backup: {backup}") |
|
print("NEXT: start the Claude desktop app and confirm the task shows a next-run time.") |
|
print(" Verify it -- do not assume. If the app was running during this write, the entry") |
|
print(" may already have been overwritten from memory.") |
|
return 0 |
|
|
|
|
|
def self_test() -> int: |
|
"""Synthetic tree, real functions, no side effects on the user's data.""" |
|
import tempfile |
|
|
|
failures = [] |
|
|
|
# Detector parsers, against output captured from real machines. These exist because the |
|
# first version of this detector had two bugs in one line and the stubbed tests could not |
|
# see either: it matched "Claude.exe" case-sensitively against tasklist output that says |
|
# "claude.exe" (missed 16 live processes), and matching the name at all would have counted |
|
# the Claude Code CLI, which is also claude.exe on Windows. |
|
WIN_REAL = "\n16\n" # Measure-Object output, 16 desktop processes |
|
WIN_ZERO = "\n0\n" # app down |
|
WIN_JUNK = "Get-CimInstance : Access is denied." |
|
if _parse_win_count(WIN_REAL) != 16: |
|
failures.append("_parse_win_count misread a real count") |
|
if _parse_win_count(WIN_ZERO) != 0: |
|
failures.append("_parse_win_count misread zero") |
|
if _parse_win_count(WIN_JUNK) != -1: |
|
failures.append("_parse_win_count should report -1 (unknown) on junk, not 0") |
|
MAC_APP = "/Applications/Claude.app/Contents/MacOS/Claude\nClaude Helper (Renderer)\nssh\n" |
|
MAC_CLI = "claude\nnode\nssh\n" # the CLI only -- must NOT count as the app |
|
if not _mac_has_desktop(MAC_APP): |
|
failures.append("_mac_has_desktop missed the desktop app") |
|
if _mac_has_desktop(MAC_CLI): |
|
failures.append("_mac_has_desktop counted the CLI as the desktop app") |
|
|
|
with tempfile.TemporaryDirectory() as tmp: |
|
root = Path(tmp) |
|
reg = root / "claude-code-sessions" / "acct-A" / "ws-1" / "scheduled-tasks.json" |
|
reg.parent.mkdir(parents=True) |
|
reg.write_text(json.dumps({"scheduledTasks": [ |
|
{"id": "alive", "enabled": True, "cronExpression": "0 9 * * *"}, |
|
]}), encoding="utf-8") |
|
|
|
found = find_registries([root]) |
|
if [p.name for p in found] != ["scheduled-tasks.json"]: |
|
failures.append(f"find_registries returned {found}") |
|
if registry_label(reg) != "acct-A/ws-1": |
|
failures.append(f"registry_label -> {registry_label(reg)}") |
|
if [t["id"] for t in read_registry(reg)] != ["alive"]: |
|
failures.append("read_registry lost the task") |
|
|
|
bad = root / "broken.json" |
|
bad.write_text("{not json", encoding="utf-8") |
|
if read_registry(bad) != []: |
|
failures.append("read_registry should return [] on malformed JSON") |
|
if read_registry(root / "nope.json") != []: |
|
failures.append("read_registry should return [] on a missing file") |
|
|
|
# restore path: refuses when the prompt is absent |
|
global TASKS_DIR, app_is_running |
|
original, original_detector = TASKS_DIR, app_is_running |
|
try: |
|
TASKS_DIR = root / "tasks" |
|
# The app-running guard is the reason this tool is safe to hand to a stranger, so |
|
# it is tested from both sides with the detector stubbed -- the real one would make |
|
# the result depend on whether the tester happens to have the app open. |
|
app_is_running = lambda: False # noqa: E731 |
|
if restore("ghost", reg, "0 10 * * *", None, None) != 2: |
|
failures.append("restore should refuse a task with no SKILL.md") |
|
(TASKS_DIR / "revived").mkdir(parents=True) |
|
(TASKS_DIR / "revived" / "SKILL.md").write_text("---\ndescription: x\n---\nbody\n", encoding="utf-8") |
|
if restore("revived", reg, "0 10 * * *", None, None) != 0: |
|
failures.append("restore failed on a valid task") |
|
ids = [t["id"] for t in read_registry(reg)] |
|
if ids != ["alive", "revived"]: |
|
failures.append(f"after restore registry holds {ids}") |
|
if restore("revived", reg, "0 10 * * *", None, None) != 0: |
|
failures.append("second restore should be a no-op") |
|
if len(read_registry(reg)) != 2: |
|
failures.append("second restore duplicated the entry") |
|
if not list(reg.parent.glob("*.bak-*")): |
|
failures.append("no backup was written") |
|
if restore("revived", reg, "0 10 * * *", "2030-01-01T00:00:00+00:00", None) != 2: |
|
failures.append("restore should refuse cron+fireAt together") |
|
|
|
# guard: with the app up, a NEW task must not be written at all |
|
(TASKS_DIR / "guarded").mkdir(parents=True) |
|
(TASKS_DIR / "guarded" / "SKILL.md").write_text("---\nd: x\n---\nb\n", encoding="utf-8") |
|
app_is_running = lambda: True # noqa: E731 |
|
if restore("guarded", reg, "0 10 * * *", None, None) != 3: |
|
failures.append("restore should refuse (exit 3) while the app is running") |
|
if "guarded" in [t["id"] for t in read_registry(reg)]: |
|
failures.append("refused restore still wrote the entry") |
|
if restore("guarded", reg, "0 10 * * *", None, None, force=True) != 0: |
|
failures.append("--force should write despite the running app") |
|
if "guarded" not in [t["id"] for t in read_registry(reg)]: |
|
failures.append("--force did not write the entry") |
|
finally: |
|
TASKS_DIR, app_is_running = original, original_detector |
|
|
|
if failures: |
|
print("SELF-TEST FAILED") |
|
for f in failures: |
|
print(" -", f) |
|
return 1 |
|
print("SELF-TEST OK (18 checks)") |
|
return 0 |
|
|
|
|
|
def main() -> int: |
|
ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) |
|
ap.add_argument("--json", action="store_true", help="machine-readable report") |
|
ap.add_argument("--restore", metavar="TASK_ID", help="put an orphan task back into a registry") |
|
ap.add_argument("--into", metavar="PATH", help="registry file to restore into") |
|
ap.add_argument("--cron", metavar="EXPR", help="5-field cron expression, app-local time") |
|
ap.add_argument("--fire-at", metavar="ISO", help="one-time ISO timestamp instead of --cron") |
|
ap.add_argument("--cwd", metavar="DIR", help="working directory for the task (default: home)") |
|
ap.add_argument("--force", action="store_true", |
|
help="write even though the Claude app is running (it will likely be overwritten)") |
|
ap.add_argument("--self-test", action="store_true", help="run built-in tests and exit") |
|
args = ap.parse_args() |
|
|
|
if args.self_test: |
|
return self_test() |
|
if args.restore: |
|
if not args.into: |
|
print("--restore needs --into <registry-path> (run without arguments to list them)") |
|
return 2 |
|
return restore(args.restore, Path(args.into).expanduser(), args.cron, args.fire_at, |
|
args.cwd, force=args.force) |
|
data = collect() |
|
if args.json: |
|
print(json.dumps(data, indent=2)) |
|
return 0 |
|
return report(data) |
|
|
|
|
|
if __name__ == "__main__": |
|
sys.exit(main()) |