idle-draft: process-group child termination, unique tmp paths, stale sweep

Lock-contention test added; children die with the dispatcher (SIGTERM/
SIGINT handlers + atexit); tmp files are per-invocation (.tmp.<pid>) so
a SIGKILL-orphaned child can never collide with a new instance.

Claude-Session: https://claude.ai/code/session_01YQDoWNM7XPPii28khFWoMc

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
+#       (`<canonical>.tmp.<dispatcher-pid>`), 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.<pid>` 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: