diff --git a/scripts/idle-draft b/scripts/idle-draft index 9c6e1d1..ee664bf 100755 --- a/scripts/idle-draft +++ b/scripts/idle-draft @@ -7,15 +7,19 @@ Design contract: ~/dev/claude/writing/oreillyconsulting/IDLE-DRAFT-PLAN.md from __future__ import annotations +import atexit import concurrent.futures import fcntl import hashlib import json import os import re +import signal import subprocess import sys import tempfile +import threading +import time from datetime import datetime, timezone from pathlib import Path from string import Template @@ -828,7 +832,7 @@ def dispatch_preview(repo: Path, config: dict, candidate: dict, max_turns: int) allowed_tools = resolve_allowed_tools(config, work_type) argv = build_claude_argv(model_id, max_turns, add_dirs, allowed_tools) canonical = canonical_output_path(repo, candidate) - tmp_path = canonical.with_name(canonical.name + ".tmp") + tmp_path = tmp_output_path(canonical) parsed_env = parse_provider_env(profile_dir / "provider.env") return { "dossier": dossier, @@ -846,6 +850,133 @@ def dispatch_preview(repo: Path, config: dict, candidate: dict, max_turns: int) } +# === Child process tracking, unique tmp paths, stale-tmp sweep === +# +# Concurrency-protection hardening (see specs/idle-draft.spec.md "Concurrency +# protection"): the flock (§1) already keeps two dispatcher instances from +# running concurrently in the common case. The gap it does *not* close is a +# dispatcher that dies without releasing its children (SIGTERM/SIGKILL, +# machine reboot, OOM kill): the flock is released when the fd closes, but +# `claude` grandchildren spawned via subprocess survive as orphans and can +# still be mid-write to a tmp file when a fresh cron-fired instance starts. +# Two independent mechanisms close that gap: +# (a) children run in their own process group (`start_new_session=True`); +# a caught signal (SIGTERM/SIGINT) or normal interpreter exit kills +# every live child's group via os.killpg. This cannot catch SIGKILL of +# the dispatcher itself -- hence (b). +# (b) every task's tmp output path is unique per dispatcher invocation +# (`.tmp.`), so even an orphan that outlives +# its dispatcher can never collide with the tmp path a *new* dispatcher +# instance uses for the same item. At startup a new instance sweeps and +# deletes tmp files older than the task timeout -- they are inert +# litter (never promoted by anyone) but worth reclaiming. + +_live_children_lock = threading.Lock() +_live_children: set[int] = set() + + +def register_child(pid: int) -> None: + with _live_children_lock: + _live_children.add(pid) + + +def unregister_child(pid: int) -> None: + with _live_children_lock: + _live_children.discard(pid) + + +def live_children() -> set[int]: + """Snapshot of currently-tracked child pids. Test/inspection hook.""" + with _live_children_lock: + return set(_live_children) + + +def kill_all_children() -> None: + """Kill the process group of every tracked live child. Safe to call + repeatedly / with a stale or empty registry -- a dead or nonexistent pid + just raises ProcessLookupError, which is swallowed. Each child was + started with start_new_session=True, so its pid is also its pgid.""" + with _live_children_lock: + pids = list(_live_children) + _live_children.clear() + for pid in pids: + try: + os.killpg(pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + pass + + +def install_signal_handlers() -> None: + """Install SIGTERM/SIGINT handlers so a killed dispatcher takes its + in-flight `claude` children down with it rather than orphaning them. + Only meaningful in the main thread of a real (non-dryrun) dispatch run, + where children actually get spawned.""" + + def _handler(signum, frame): + kill_all_children() + sys.exit(1) + + signal.signal(signal.SIGTERM, _handler) + signal.signal(signal.SIGINT, _handler) + + +# Belt-and-braces: even a normal unhandled-exception exit (no signal +# involved) should not leave children behind. atexit fires on interpreter +# shutdown from any cause except os._exit()/SIGKILL. +atexit.register(kill_all_children) + + +TMP_SUFFIX_RE = re.compile(r"\.tmp\.\d+$") + + +def tmp_output_path(canonical: Path, pid: int | None = None) -> Path: + """Per-invocation-unique tmp path for a task's output. Defaults to this + process's pid (shared by all worker threads of one dispatcher run, so + every task in one invocation uses the same suffix); a pid can be passed + explicitly for testing or preview purposes.""" + pid = pid if pid is not None else os.getpid() + return canonical.with_name(canonical.name + f".tmp.{pid}") + + +def find_tmp_files(repo: Path, dossiers: list) -> list: + """Every leftover `*.tmp.` file directly under any configured + dossier directory (where canonical outputs and their tmp siblings live).""" + found = [] + for dossier in dossiers: + d = repo / dossier + if not d.is_dir(): + continue + for f in d.iterdir(): + if f.is_file() and TMP_SUFFIX_RE.search(f.name): + found.append(f) + return found + + +def sweep_stale_tmp_files(repo: Path, dossiers: list, task_timeout: int, now: float | None = None) -> list: + """Delete tmp files whose mtime is older than task_timeout seconds -- + orphaned litter from a dispatcher that died mid-task (GAP 2b). Tmp files + younger than the timeout are left alone: they could belong to a live + orphan still finishing up, or to another concurrent dispatch cycle's + in-flight task. Since a tmp file is only ever promoted by the exact + run_task() call that created it, an untouched one is inert at worst -- + never silently adopted by anyone else. Returns the paths actually + deleted, for logging by the caller.""" + now = now if now is not None else time.time() + deleted = [] + for f in find_tmp_files(repo, dossiers): + try: + mtime = f.stat().st_mtime + except OSError: + continue + if now - mtime > task_timeout: + try: + f.unlink() + deleted.append(f) + except OSError: + continue + return deleted + + # === Real task execution === @@ -856,7 +987,7 @@ def run_task(repo: Path, config: dict, candidate: dict, max_turns: int, task_tim slug = candidate["slug"] canonical = canonical_output_path(repo, candidate) - tmp_path = canonical.with_name(canonical.name + ".tmp") + tmp_path = tmp_output_path(canonical) mapping = build_prompt_mapping(repo, dossier, slug, work_type, config, tmp_path) prompt_text = render_prompt(work_type, mapping) @@ -872,14 +1003,34 @@ def run_task(repo: Path, config: dict, candidate: dict, max_turns: int, task_tim returncode = -1 stderr = "" try: - proc = subprocess.run( - argv, cwd=str(repo), env=env, input=prompt_text, - capture_output=True, text=True, timeout=task_timeout, + proc = subprocess.Popen( + argv, cwd=str(repo), env=env, + stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, + text=True, start_new_session=True, ) - returncode = proc.returncode - stderr = proc.stderr - except subprocess.TimeoutExpired: - timed_out = True + except OSError as e: + return {"outcome": "transient", "candidate": candidate, "reason": f"failed to start claude subprocess: {e}"} + + # GAP 2a: own process group + tracked pid, so a killed dispatcher can + # take this (and any grandchildren it spawns) down with it instead of + # orphaning it. See "Child process tracking" section above. + register_child(proc.pid) + try: + try: + _, stderr = proc.communicate(input=prompt_text, timeout=task_timeout) + returncode = proc.returncode + except subprocess.TimeoutExpired: + timed_out = True + try: + os.killpg(proc.pid, signal.SIGKILL) + except (ProcessLookupError, PermissionError, OSError): + pass + try: + proc.communicate(timeout=5) + except Exception: + pass + finally: + unregister_child(proc.pid) failure = classify_failure(returncode, stderr, timed_out) @@ -972,12 +1123,20 @@ def apply_failure_state(state: dict, candidate: dict, max_attempts: int) -> str def run_dispatch(args, config: dict, repo: Path, state_path: Path) -> int: rotate_log_if_needed(repo) + defaults = get_config_defaults(config) lock = None if not args.dryrun: lock = acquire_lock(repo) if lock is None: log_event(repo, "lock held by another invocation; exiting") return 0 + # Held for the full dispatch loop, including while children run (see + # module-level "Child process tracking" section) -- a second cron-fired + # invocation exits 0 immediately per the check above, it never races + # this instance even if this instance runs for hours. + install_signal_handlers() + for swept in sweep_stale_tmp_files(repo, config["dossiers"], defaults["task_timeout"]): + log_event(repo, f"swept stale tmp file (orphaned litter): {swept}") try: state = load_state(state_path, repo) @@ -985,7 +1144,6 @@ def run_dispatch(args, config: dict, repo: Path, state_path: Path) -> int: log_event(repo, f"FATAL: state validation failed: {e}") return 2 - defaults = get_config_defaults(config) parallel = args.parallel or config.get("parallel", DEFAULT_PARALLEL) try: diff --git a/specs/idle-draft.spec.md b/specs/idle-draft.spec.md index d195f34..f7d1f0a 100644 --- a/specs/idle-draft.spec.md +++ b/specs/idle-draft.spec.md @@ -65,6 +65,14 @@ lock), `flock` (`fcntl.flock`, `LOCK_EX | LOCK_NB`) on `/.idle-draft.lock` already locked by another invocation, log one line and exit **0** — a concurrent cron tick is not a failure. +**Guarantee:** the lock is acquired before any other work and held for the *entire* +dispatch loop, including while dispatched children are running — not released until the +process is about to exit. A second cron-fired invocation therefore always exits 0 +immediately for as long as the first is alive, however long that is (a low-usage +provider can let one instance run for hours "catching up"; the lock does not expire or +time out). This is the primary concurrency guard; §13 covers what happens when the +*lock-holding process itself* dies uncleanly rather than exiting normally. + ### 2. Log rotation At startup, if `/idle-draft.log` exceeds 5 MB, rename it to `idle-draft.log.1` @@ -216,6 +224,11 @@ For the selected `(item_or_dossier, work_type, provider)`: 2. Render the template (`string.Template`, `$placeholder` substitution) with the resolved paths for that item (overview, agent, research, draft as applicable), dossier name, slug, style directory, source register path, and a temp output path. + The temp output path is `.tmp.` — unique per dispatcher + *invocation* (all tasks within one dispatch run share the same suffix, since it's + this process's own pid), not per task. This is what makes tmp paths collision-proof + across dispatcher instances even when one instance dies leaving orphaned children + behind — see §13. 3. Resolve the profile directory from `config["providers"][provider]["profile"]` (`~` expanded). Build the child environment: `CLAUDE_CONFIG_DIR=`, plus — if `/provider.env` exists — `ANTHROPIC_BASE_URL` (if set), @@ -237,7 +250,11 @@ For the selected `(item_or_dossier, work_type, provider)`: 5. Run the subprocess: `cwd=`, `env=`, prompt piped via **stdin** (not as an argv element — avoids `ARG_MAX` on large rendered prompts, same lesson `claude-profile` already applies to its system-prompt injection), timeout = per-task - timeout (1800s unless overridden). + timeout (1800s unless overridden). The child is started in its own process group + (`start_new_session=True`) and its pid is tracked in a live-children registry for the + duration of the call — see §13. On timeout, the whole process group is sent + `SIGKILL` (not just the immediate `claude` process), so any grandchild it spawned + dies too, before the task is classified as a transient failure. 6. Classify the result: - **Timeout** → transient failure. Do not increment `attempts`. - **Non-zero exit** whose stderr matches a retryable signature (`429`, `5xx`, @@ -265,7 +282,10 @@ For the selected `(item_or_dossier, work_type, provider)`: `//..md.rejected` (kept for human inspection, never promoted, never committed to the canonical name; `topic_ideas` content failures are simply discarded — nothing is appended, nothing is blocked, since - `topic_ideas` has no per-item state entry to carry an attempt counter). + `topic_ideas` has no per-item state entry to carry an attempt counter). The + `.rejected` path is derived from the canonical path, not the tmp path, so it is + unaffected by the `.tmp.` suffix (§13) — always + `.rejected`, never `.rejected.`. 10. On transient failure: temp file is discarded; no state change; no commit; the candidate may be retried on a later cycle. 11. On success: write `idle-draft.state.json` atomically (temp file + `os.replace()` in @@ -284,6 +304,53 @@ For the selected `(item_or_dossier, work_type, provider)`: One line per event (gate decision, dispatch, completion, failure, commit) appended to `/idle-draft.log` with an ISO-8601 timestamp, mirrored to stderr. +### 13. Concurrency protection beyond the lock + +§1's flock is the primary guard and handles the common case: a second cron-fired +invocation always sees the lock held and exits 0. This section covers the gap the lock +alone cannot close — a dispatcher process that dies *without* releasing its children +cleanly (`SIGTERM`, `SIGKILL`, OOM kill, machine reboot). The flock itself is released +the instant the holding process's file descriptors close (on any death, clean or not), +but a `claude` child spawned via `subprocess` is a separate OS process and does **not** +die automatically when its parent does — left alone it becomes an orphan, reparented to +init, still writing its output. A fresh cron-fired instance starting immediately after +would then see the lock free, pick up the *same* ready item, and dispatch a *second* +`claude` process against the *same* deterministic output path — two writers racing on +the "same" canonical file. Two independent mechanisms close this: + +**a. Children die with the dispatcher (the common death path).** Every `claude` child is +spawned with `start_new_session=True`, putting it (and any process it forks) in its own +process group with the child's pid as the group id. The dispatcher tracks each live +child's pid in an in-process registry for the duration of the call (`register_child` / +`unregister_child`, guarded by a lock since tasks run in worker threads). On `SIGTERM` or +`SIGINT`, an installed handler walks the registry and `os.killpg`s each group with +`SIGKILL`, then exits — no child outlives a dispatcher that receives a catchable signal. +The same cleanup also runs via `atexit`, so a normal unhandled-exception exit (no signal +involved) still cannot leave children behind. **This cannot catch `SIGKILL` of the +dispatcher itself** — no userspace handler can — which is exactly the case mechanism (b) +exists for. + +**b. Tmp-path collisions are structurally impossible (the uncatchable-death path).** Each +task's temp output file is named `.tmp.` (§11 step 2), not +the old bare `.tmp`. Every dispatcher process has a distinct pid, so even an +orphaned child that outlives a `SIGKILL`ed dispatcher is writing to a filename no other +dispatcher instance — past, present, or future — will ever target. There is no race to +resolve: the collision that used to be structurally possible (two writers, one path) is +now structurally impossible (two writers, two paths). + +**Startup sweep.** After acquiring the lock, before touching config-driven state, a +fresh dispatch instance globs each configured dossier directory for leftover +`*.tmp.` files. Any whose mtime is older than the per-task timeout (1800s unless +overridden) is deleted and logged as swept — it is orphaned litter from a dispatcher +that died mid-task; since it was never promoted (only the exact `run_task()` call that +created it ever calls `os.replace()` on it) it can never silently become the canonical +file no matter how long it sits there. Files younger than the timeout are left alone — +they may belong to a live orphan (mechanism (a) failed to reap it, e.g. it was itself +`SIGKILL`ed independently, or belongs to another concurrent dispatch cycle that legitimately +still has it in flight) still finishing up; touching them risks nothing since, again, +nothing ever silently adopts a tmp file it didn't create — at worst a stale one sits +unswept until the *next* startup sweep, once it ages past the timeout. + ## Dryrun behaviour `--dryrun` runs the full probe (or reads `--probe-json`), computes all gates, builds @@ -346,7 +413,9 @@ a lighter-touch check); `draft` and `review` carry no `allowed_tools` key. | Scenario | Handling | |---|---| -| Lock already held | Log one line, exit 0 (not an error — another cron tick is running) | +| Lock already held | Log one line, exit 0 (not an error — another cron tick is running), regardless of how long the holder has been running | +| Dispatcher receives `SIGTERM`/`SIGINT` while children are running | Handler `os.killpg`s every live child's process group, then exits (§13a) | +| Dispatcher is `SIGKILL`ed (uncatchable) | Children orphaned, but each was writing to a `.tmp.` path unique to that dead instance — no live instance can ever collide with it; stale ones swept once they age past the task timeout (§13b) | | `idle-draft.state.json` missing | Treated as `{"items": {}}`, not an error | | `idle-draft.state.json` present but invalid | Exit 2, loud message, run never starts | | Item's `.agent.md` missing | Item excluded from all dispatch (commissioning briefs exist today for every current item; this guards future additions) | diff --git a/tests/test-idle-draft.sh b/tests/test-idle-draft.sh index 78bb84d..742b42a 100755 --- a/tests/test-idle-draft.sh +++ b/tests/test-idle-draft.sh @@ -162,6 +162,50 @@ after_hash=$(cd "$FIXTURE_REPO" && git rev-parse HEAD) [[ "$before_hash" == "$after_hash" ]] && pass "dryrun makes no git commits" || fail "dryrun makes no git commits" [[ ! -f "$FIXTURE_REPO/ai/03-topic-three.review.md" ]] && pass "dryrun writes no output file" || fail "dryrun writes no output file" +echo "-- concurrency: second invocation exits 0 when lock is held (GAP 1) --" +# Holds .idle-draft.lock in a background helper (independent process, so the +# flock is real -- fcntl locks are per-process/per-fd, a second flock from +# this same bash process wouldn't contend). The helper touches a sentinel +# file only *after* it has the lock, so we never race the flock call itself. +LOCK_HELPER="$TMPDIR/lock_helper.py" +cat > "$LOCK_HELPER" <<'PYEOF' +import fcntl, sys, time +lock_path, sentinel_path = sys.argv[1], sys.argv[2] +f = open(lock_path, "w") +fcntl.flock(f.fileno(), fcntl.LOCK_EX) +with open(sentinel_path, "w") as s: + s.write("locked\n") +time.sleep(30) +PYEOF +LOCK_SENTINEL="$TMPDIR/lock-sentinel" +rm -f "$LOCK_SENTINEL" +python3 "$LOCK_HELPER" "$FIXTURE_REPO/.idle-draft.lock" "$LOCK_SENTINEL" & +HELPER_PID=$! + +waited=0 +while [[ ! -f "$LOCK_SENTINEL" && $waited -lt 50 ]]; do + sleep 0.1 + waited=$((waited + 1)) +done + +if [[ ! -f "$LOCK_SENTINEL" ]]; then + fail "lock-contention test setup: helper never confirmed it holds the lock" +else + before_hash=$(cd "$FIXTURE_REPO" && git rev-parse HEAD) + # Real dispatch mode (no --dryrun): a fixture with plenty of idle capacity, + # so absent the lock this would dispatch item 03's review task for real. + output=$("$SCRIPT" --config "$FIXTURE_CONFIG" --repo "$FIXTURE_REPO" --probe-json "$FIXTURE_PROBE" --once 2>&1) + code=$? + assert_exit_code "$code" 0 "second invocation with lock held exits 0" + assert_contains "$output" "lock held" "log line reports lock held" + after_hash=$(cd "$FIXTURE_REPO" && git rev-parse HEAD) + [[ "$before_hash" == "$after_hash" ]] && pass "no commit occurred while lock held" || fail "no commit occurred while lock held" + [[ ! -f "$FIXTURE_REPO/ai/03-topic-three.review.md" ]] && pass "no output file written while lock held" || fail "no output file written while lock held" +fi + +kill "$HELPER_PID" 2>/dev/null +wait "$HELPER_PID" 2>/dev/null + echo "-- state validation failure exits 2 --" BADSTATE="$TMPDIR/badstate-repo" mkdir -p "$BADSTATE/ai" @@ -547,6 +591,85 @@ except m.StateValidationError: m.validate_state({"items": {"ai/01-topic-one": {"human_edit_done": True, "attempts": {"draft": 1}}}}, repo) check("state validation: well-formed state passes", True) +# --- GAP 2b: unique tmp path per invocation --- +import os as _os +import time as _time + +canonical = repo / "ai" / "03-topic-three.review.md" +tmp_a = m.tmp_output_path(canonical, pid=11111) +tmp_b = m.tmp_output_path(canonical, pid=22222) +check("tmp path uniqueness: two different pids produce two different tmp paths", + tmp_a != tmp_b and tmp_a.name == "03-topic-three.review.md.tmp.11111" + and tmp_b.name == "03-topic-three.review.md.tmp.22222", (str(tmp_a), str(tmp_b))) + +tmp_default = m.tmp_output_path(canonical) +check("tmp path uniqueness: default pid is os.getpid()", + tmp_default.name == f"03-topic-three.review.md.tmp.{_os.getpid()}", str(tmp_default)) + +rejected = m.rejected_output_path(canonical) +check("rejected-path derivation is unaffected by the new tmp naming (derived from canonical, not tmp)", + rejected.name == "03-topic-three.review.md.rejected", str(rejected)) + +# --- GAP 2b: startup sweep of stale tmp files --- +sweep_repo = tmpdir / "sweep-repo" +(sweep_repo / "ai").mkdir(parents=True, exist_ok=True) +old_tmp = sweep_repo / "ai" / "01-a.research.md.tmp.111" +fresh_tmp = sweep_repo / "ai" / "02-b.research.md.tmp.222" +old_tmp.write_text("stale orphan output\n") +fresh_tmp.write_text("still-in-progress output\n") +now = _time.time() +task_timeout = 1800 +old_mtime = now - task_timeout - 100 # older than the timeout -> stale +fresh_mtime = now - 10 # well within the timeout -> live orphan candidate, keep +_os.utime(old_tmp, (old_mtime, old_mtime)) +_os.utime(fresh_tmp, (fresh_mtime, fresh_mtime)) + +deleted = m.sweep_stale_tmp_files(sweep_repo, ["ai"], task_timeout, now=now) +check("stale sweep: deletes a tmp file older than the task timeout", + old_tmp in deleted and not old_tmp.exists(), deleted) +check("stale sweep: leaves a tmp file younger than the task timeout untouched", + fresh_tmp not in deleted and fresh_tmp.exists(), deleted) + +# A non-.tmp. file (e.g. the canonical output itself) must never be swept. +canonical_lookalike = sweep_repo / "ai" / "01-a.research.md" +canonical_lookalike.write_text("# Research: a\n") +_os.utime(canonical_lookalike, (old_mtime, old_mtime)) +deleted2 = m.sweep_stale_tmp_files(sweep_repo, ["ai"], task_timeout, now=now) +check("stale sweep: never touches a file without a .tmp. suffix, however old", + canonical_lookalike.exists() and canonical_lookalike not in deleted2, canonical_lookalike) + +# --- GAP 2a: child process-group tracking / signal handling (smoke tests) --- +# No real subprocess is spawned here -- just the bookkeeping primitives that +# the signal handler and run_task() rely on. +fake_pid = 2**30 # astronomically unlikely to collide with a real pid/pgid +m.register_child(fake_pid) +check("child tracking: register_child adds the pid to the live set", fake_pid in m.live_children(), m.live_children()) + +m.unregister_child(fake_pid) +check("child tracking: unregister_child removes the pid", fake_pid not in m.live_children(), m.live_children()) + +m.register_child(fake_pid) +m.register_child(fake_pid + 1) +try: + m.kill_all_children() + kill_all_ok = True +except Exception as e: + kill_all_ok = False +check("child tracking: kill_all_children on nonexistent pids raises nothing (ProcessLookupError swallowed)", + kill_all_ok, "") +check("child tracking: kill_all_children clears the registry", m.live_children() == set(), m.live_children()) + +m.install_signal_handlers() +import signal as _signal +check("signal handling: install_signal_handlers installs a non-default SIGTERM handler", + _signal.getsignal(_signal.SIGTERM) not in (_signal.SIG_DFL, _signal.SIG_IGN, None), "") +check("signal handling: install_signal_handlers installs a non-default SIGINT handler", + _signal.getsignal(_signal.SIGINT) not in (_signal.SIG_DFL, _signal.SIG_IGN, None), "") +# Restore defaults so this subprocess (which is about to exit anyway) doesn't +# leave anything surprising behind for the interpreter teardown. +_signal.signal(_signal.SIGTERM, _signal.SIG_DFL) +_signal.signal(_signal.SIGINT, _signal.SIG_DFL) + for status, name, detail in results: print(f"{status}: {name}", detail if detail else "")