small-scripts/scripts/idle-draft

#!/usr/bin/env python3
"""Idle-subscription dispatcher for the O'Reilly Consulting writing pipeline.

Spec: specs/idle-draft.spec.md
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

# === Constants replicated from scripts/agent-subscriptions (keep in sync) ===
# idle-draft must resolve the same credential paths agent-subscriptions probes
# with, to run the credential-parity check (IDLE-DRAFT-PLAN.md, finding C2).

ANTHROPIC_TOKEN_PATH = Path("~/dev/claude/secrets/anthropic/api_key").expanduser()
MINIMAX_SOPS_PATH = Path(
    "~/dev/claude/projects/agent-runtime-secrets/providers/minimax/v1/provider.sops.env"
).expanduser()
MINIMAX_SOPS_KEY = Path("~/dev/claude/secrets/sops/provider-age-key.txt").expanduser()
MINIMAX_DOTENV_KEY = "ANTHROPIC_AUTH_TOKEN"

# === Work-item / pipeline constants ===

WORK_TYPES = ("research", "draft", "review", "topic_ideas")
STAGE_RANK = {"review": 3, "draft": 2, "research": 1, "topic_ideas": 0}
SUFFIX = {"research": "research", "draft": "draft", "review": "review"}
PROMPT_TEMPLATE_FILES = {
    "research": "research.md",
    "draft": "draft.md",
    "review": "review-suggest.md",
    "topic_ideas": "topic-ideas.md",
}
# Expected first-line heading prefix per work type, used by output validation.
EXPECTED_HEADING_PREFIX = {
    "research": "# ",
    "draft": "# ",
    "review": "# ",
    "topic_ideas": "## ",
}
ALLOWED_ITEM_KEYS = {
    "human_edit_done",
    "research_sampled",
    "approved",
    "blocked",
    "attempts",
}
ALLOWED_ATTEMPT_WORK_TYPES = {"research", "draft", "review"}

# Effort levels accepted by `claude --effort <level>`. Derived empirically on
# 2026-08-02 by running `claude --effort obviously-bogus-level -p ""` (no API
# call -- an invalid flag value is rejected before any network activity) and
# reading the CLI's own error message, which enumerated the valid set
# verbatim: "Warning: Unknown --effort value 'obviously-bogus-level' —
# ignoring it and using the default effort. Valid values: low, medium, high,
# xhigh, max." Cross-checked against `claude --help`'s `--effort <level>`
# description, which lists the same five values. If a future CLI version
# changes this set, re-run the same probe and update this constant.
CLAUDE_EFFORT_LEVELS = frozenset({"low", "medium", "high", "xhigh", "max"})

DEFAULT_MAX_TURNS = 25
DEFAULT_TASK_TIMEOUT_SECONDS = 1800
DEFAULT_MAX_ATTEMPTS = 2
DEFAULT_PARALLEL = 2

REQUIRED_CONFIG_KEYS = (
    "parallel",
    "providers",
    "work_types",
    "dossiers",
    "review_score_threshold",
    "max_unreviewed_research_per_dossier",
    "max_open_topic_proposals",
    "evidence_dirs",
)

RETRYABLE_STDERR_MARKERS = (
    "429",
    "rate limit",
    "rate_limit",
    "500",
    "502",
    "503",
    "504",
    "overloaded",
    "temporarily unavailable",
    "connection reset",
    "econnreset",
    "timed out",
    "timeout",
)

ITEM_FILENAME_RE = re.compile(r"^(\d+)-(.+)\.overview\.md$")
ABS_PATH_RE = re.compile(r"/home/[^\s`)\]\"'>,;]+")
LINE_RANGE_SUFFIX_RE = re.compile(r"^(.*/[^/:]+):\d+(?:-\d+)?$")

SCRIPT_PATH = Path(__file__).resolve()
REPO_SELF = SCRIPT_PATH.parent.parent  # small-scripts repo (holds data/idle-draft)
PROMPTS_DIR = REPO_SELF / "data" / "idle-draft" / "prompts"


# === Errors ===


class ConfigError(Exception):
    pass


class StateValidationError(Exception):
    pass


# === Small utilities ===


def expand(p: str | Path) -> Path:
    return Path(str(p)).expanduser()


def iso_now() -> str:
    return datetime.now(timezone.utc).isoformat()


def sha256_bytes(data: bytes) -> str:
    return hashlib.sha256(data).hexdigest()


def sha256_text(text: str) -> str:
    return sha256_bytes(text.encode("utf-8"))


def atomic_write_json(path: Path, data: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_name = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.")
    try:
        with os.fdopen(fd, "w") as f:
            json.dump(data, f, indent=2, sort_keys=True)
            f.write("\n")
        os.replace(tmp_name, path)
    finally:
        if os.path.exists(tmp_name):
            os.unlink(tmp_name)


def atomic_write_text(path: Path, text: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    fd, tmp_name = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.")
    try:
        with os.fdopen(fd, "w") as f:
            f.write(text)
        os.replace(tmp_name, path)
    finally:
        if os.path.exists(tmp_name):
            os.unlink(tmp_name)


# === Logging ===


def rotate_log_if_needed(repo: Path) -> None:
    log_path = repo / "idle-draft.log"
    try:
        if log_path.exists() and log_path.stat().st_size > 5 * 1024 * 1024:
            rotated = repo / "idle-draft.log.1"
            os.replace(log_path, rotated)
    except OSError:
        pass


def log_event(repo: Path, message: str) -> None:
    line = f"{iso_now()} {message}"
    print(line, file=sys.stderr)
    try:
        log_path = repo / "idle-draft.log"
        with open(log_path, "a") as f:
            f.write(line + "\n")
    except OSError:
        pass


# === Locking ===


def acquire_lock(repo: Path):
    """Returns an open file object holding the lock, or None if already locked."""
    lock_path = repo / ".idle-draft.lock"
    f = open(lock_path, "w")
    try:
        fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
    except BlockingIOError:
        f.close()
        return None
    return f


# === Config ===


def load_config(path: Path) -> dict:
    if not path.exists():
        raise ConfigError(f"config file not found: {path}")
    try:
        data = json.loads(path.read_text())
    except json.JSONDecodeError as e:
        raise ConfigError(f"config file is not valid JSON: {e}") from e
    if not isinstance(data, dict):
        raise ConfigError("config file must be a JSON object")
    missing = [k for k in REQUIRED_CONFIG_KEYS if k not in data]
    if missing:
        raise ConfigError(f"config missing required key(s): {', '.join(missing)}")
    if not isinstance(data["providers"], dict) or not data["providers"]:
        raise ConfigError("config 'providers' must be a non-empty object")
    if not isinstance(data["work_types"], dict) or not data["work_types"]:
        raise ConfigError("config 'work_types' must be a non-empty object")
    if not isinstance(data["dossiers"], list) or not data["dossiers"]:
        raise ConfigError("config 'dossiers' must be a non-empty array")
    if not isinstance(data["evidence_dirs"], list):
        raise ConfigError("config 'evidence_dirs' must be an array")
    for pname, pcfg in data["providers"].items():
        for key in ("profile", "threshold_pct", "five_hour_ceiling", "min_idle"):
            if key not in pcfg:
                raise ConfigError(f"provider '{pname}' missing key '{key}'")
    known_providers = set(data["providers"].keys())
    for wtype, wcfg in data["work_types"].items():
        if "providers" not in wcfg or not isinstance(wcfg["providers"], list):
            raise ConfigError(f"work_type '{wtype}' missing 'providers' list")
        for entry in wcfg["providers"]:
            # A providers-list entry is either the legacy bare provider-name
            # string, or an object carrying a per-provider model/effort
            # override: {"provider": "<name>", "model": "...", "effort": "..."}.
            # Both forms may appear in the same list (list order still encodes
            # preference for select_provider()).
            if isinstance(entry, str):
                pname_ref = entry
                if pname_ref not in known_providers:
                    raise ConfigError(
                        f"work_type '{wtype}' providers entry names unknown provider '{pname_ref}'"
                    )
            elif isinstance(entry, dict):
                unknown_keys = set(entry.keys()) - {"provider", "model", "effort"}
                if unknown_keys:
                    raise ConfigError(
                        f"work_type '{wtype}' providers entry has unknown key(s): {sorted(unknown_keys)}"
                    )
                pname_ref = entry.get("provider")
                if not isinstance(pname_ref, str) or not pname_ref:
                    raise ConfigError(
                        f"work_type '{wtype}' providers entry (object form) missing non-empty 'provider' key"
                    )
                if pname_ref not in known_providers:
                    raise ConfigError(
                        f"work_type '{wtype}' providers entry names unknown provider '{pname_ref}'"
                    )
                if "model" in entry:
                    entry_model = entry["model"]
                    if not isinstance(entry_model, str) or not entry_model:
                        raise ConfigError(
                            f"work_type '{wtype}' providers entry for '{pname_ref}' 'model' must be "
                            f"a non-empty string"
                        )
                if "effort" in entry:
                    entry_effort = entry["effort"]
                    if not isinstance(entry_effort, str) or not entry_effort:
                        raise ConfigError(
                            f"work_type '{wtype}' providers entry for '{pname_ref}' 'effort' must be "
                            f"a non-empty string"
                        )
                    if entry_effort not in CLAUDE_EFFORT_LEVELS:
                        raise ConfigError(
                            f"work_type '{wtype}' providers entry for '{pname_ref}' 'effort' must be one "
                            f"of {sorted(CLAUDE_EFFORT_LEVELS)}, got {entry_effort!r}"
                        )
            else:
                raise ConfigError(
                    f"work_type '{wtype}' providers entry must be a string or object, "
                    f"got {type(entry).__name__}"
                )
        if "allowed_tools" in wcfg:
            at = wcfg["allowed_tools"]
            if not isinstance(at, list) or not all(isinstance(x, str) for x in at):
                raise ConfigError(f"work_type '{wtype}' 'allowed_tools' must be a list of strings")
        if "effort" in wcfg:
            effort = wcfg["effort"]
            if not isinstance(effort, str) or not effort:
                raise ConfigError(f"work_type '{wtype}' 'effort' must be a non-empty string")
            if effort not in CLAUDE_EFFORT_LEVELS:
                raise ConfigError(
                    f"work_type '{wtype}' 'effort' must be one of {sorted(CLAUDE_EFFORT_LEVELS)}, got {effort!r}"
                )
        if "model" in wcfg:
            model = wcfg["model"]
            if not isinstance(model, str) or not model:
                raise ConfigError(f"work_type '{wtype}' 'model' must be a non-empty string")
    return data


# === State ===


def default_state() -> dict:
    return {"items": {}}


def load_state(path: Path, repo: Path) -> dict:
    if not path.exists():
        return default_state()
    try:
        data = json.loads(path.read_text())
    except json.JSONDecodeError as e:
        raise StateValidationError(f"state file is not valid JSON: {e}") from e
    validate_state(data, repo)
    return data


def validate_state(data: dict, repo: Path) -> None:
    if not isinstance(data, dict):
        raise StateValidationError("state file must be a JSON object")
    extra_top = set(data.keys()) - {"items"}
    if extra_top:
        raise StateValidationError(f"state file has unknown top-level key(s): {sorted(extra_top)}")
    items = data.get("items", {})
    if not isinstance(items, dict):
        raise StateValidationError("state 'items' must be an object")
    for key, entry in items.items():
        if "/" not in key:
            raise StateValidationError(f"state item key '{key}' is not of the form dossier/NN-slug")
        dossier, slug = key.split("/", 1)
        overview = repo / dossier / f"{slug}.overview.md"
        if not overview.exists():
            raise StateValidationError(
                f"state item key '{key}' does not resolve to an existing overview file: {overview}"
            )
        if not isinstance(entry, dict):
            raise StateValidationError(f"state item '{key}' value must be an object")
        extra_keys = set(entry.keys()) - ALLOWED_ITEM_KEYS
        if extra_keys:
            raise StateValidationError(f"state item '{key}' has unknown key(s): {sorted(extra_keys)}")
        for bkey in ("human_edit_done", "research_sampled", "approved"):
            if bkey in entry and not isinstance(entry[bkey], bool):
                raise StateValidationError(f"state item '{key}'.{bkey} must be a bool")
        if "blocked" in entry and entry["blocked"] is not None and not isinstance(entry["blocked"], str):
            raise StateValidationError(f"state item '{key}'.blocked must be a string or null")
        if "attempts" in entry:
            attempts = entry["attempts"]
            if not isinstance(attempts, dict):
                raise StateValidationError(f"state item '{key}'.attempts must be an object")
            for wtype, count in attempts.items():
                if wtype not in ALLOWED_ATTEMPT_WORK_TYPES:
                    raise StateValidationError(
                        f"state item '{key}'.attempts has unknown work type '{wtype}'"
                    )
                if not isinstance(count, int) or isinstance(count, bool) or count < 0:
                    raise StateValidationError(
                        f"state item '{key}'.attempts.{wtype} must be a non-negative int"
                    )


def get_item_state(state: dict, key: str) -> dict:
    entry = state.get("items", {}).get(key, {})
    return {
        "human_edit_done": entry.get("human_edit_done", False),
        "research_sampled": entry.get("research_sampled", False),
        "approved": entry.get("approved", False),
        "blocked": entry.get("blocked"),
        "attempts": dict(entry.get("attempts", {})),
    }


def save_state_atomic(path: Path, state: dict) -> None:
    atomic_write_json(path, state)


# === provider.env parsing (replicates scripts/claude-profile ~L320-440) ===


def parse_provider_env(path: Path) -> dict:
    """Parse a provider.env file the way claude-profile does.

    Returns {"model_id": str|None, "base_url": str|None, "api_key_file": str|None,
    "extra": {KEY: value, ...}}. Lines are KEY=value; blank lines and lines starting
    with '#' are skipped. Unrecognised keys fall into 'extra' verbatim, mirroring
    claude-profile's PROVIDER_EXTRA_ENV accumulation.
    """
    result: dict = {"model_id": None, "base_url": None, "api_key_file": None, "extra": {}}
    if not path.exists():
        return result
    for raw_line in path.read_text().splitlines():
        line = raw_line.strip()
        if not line or line.startswith("#"):
            continue
        if "=" not in line:
            continue
        key, _, value = line.partition("=")
        key = key.strip()
        value = value.strip()
        if key == "MODEL_ID":
            result["model_id"] = value
        elif key == "ANTHROPIC_BASE_URL":
            result["base_url"] = value
        elif key == "ANTHROPIC_API_KEY_FILE":
            result["api_key_file"] = value
        else:
            result["extra"][key] = value
    return result


def build_child_env(profile_dir: Path, base_env: dict | None = None) -> dict:
    """Build the environment for a headless `claude -p` child, replicating
    claude-profile's export logic (L423-435) without invoking claude-profile."""
    env = dict(base_env if base_env is not None else os.environ)
    env["CLAUDE_CONFIG_DIR"] = str(profile_dir)
    provider_env_path = profile_dir / "provider.env"
    parsed = parse_provider_env(provider_env_path)
    if parsed["base_url"]:
        env["ANTHROPIC_BASE_URL"] = parsed["base_url"]
    if parsed["api_key_file"]:
        key_file = expand(parsed["api_key_file"])
        if key_file.exists():
            env["ANTHROPIC_API_KEY"] = key_file.read_text().strip()
    for k, v in parsed["extra"].items():
        env[k] = v
    return env


def resolve_model_id(profile_dir: Path) -> str | None:
    parsed = parse_provider_env(profile_dir / "provider.env")
    return parsed["model_id"]


# === Credential parity (IDLE-DRAFT-PLAN.md finding C2) ===


def _credential_parity_from_values(profile_value: str, probe_value: str) -> tuple[bool, str]:
    """Pure comparison: do two credential strings hash the same? Testable without
    touching any file, secret, or subprocess."""
    if not profile_value or not probe_value:
        return False, "one or both credential values are empty"
    a, b = sha256_text(profile_value), sha256_text(probe_value)
    if a == b:
        return True, "match"
    return False, f"hash mismatch (profile={a[:12]}... probe={b[:12]}...)"


def resolve_profile_credential_value(provider_name: str, profile_dir: Path) -> str:
    """Read the credential the given profile would export. Raises on failure."""
    parsed = parse_provider_env(profile_dir / "provider.env")
    if parsed["api_key_file"]:
        key_file = expand(parsed["api_key_file"])
        return key_file.read_text().strip()
    if provider_name == "anthropic":
        return ANTHROPIC_TOKEN_PATH.read_text().strip()
    raise ValueError(
        f"profile '{profile_dir}' has no provider.env ANTHROPIC_API_KEY_FILE and "
        f"provider '{provider_name}' has no fallback credential source"
    )


def read_anthropic_probe_credential() -> str:
    return ANTHROPIC_TOKEN_PATH.read_text().strip()


def _parse_dotenv_key(content: str, key: str) -> str:
    for line in content.splitlines():
        line = line.strip()
        if not line or line.startswith("#"):
            continue
        if "=" in line:
            k, _, v = line.partition("=")
        elif ": " in line:
            k, _, v = line.partition(": ")
        else:
            continue
        if k.strip() == key:
            v = v.strip()
            if len(v) >= 2 and v[0] == v[-1] and v[0] in ('"', "'"):
                v = v[1:-1]
            return v
    raise KeyError(f"Key '{key}' not found in decrypted content")


def read_minimax_probe_credential() -> str:
    env = {**os.environ, "SOPS_AGE_KEY_FILE": str(MINIMAX_SOPS_KEY)}
    result = subprocess.run(
        ["sops", "--decrypt", "--output-type", "dotenv", str(MINIMAX_SOPS_PATH)],
        capture_output=True, text=True, check=True, env=env,
    )
    return _parse_dotenv_key(result.stdout, MINIMAX_DOTENV_KEY)


PROBE_CREDENTIAL_READERS = {
    "anthropic": read_anthropic_probe_credential,
    "minimax": read_minimax_probe_credential,
}


def credential_parity(provider_name: str, profile_dir: Path, probe_credential_fn=None) -> tuple[bool, str]:
    """Full parity check for one provider: resolves both sides and compares.

    `probe_credential_fn` is injectable (tests pass a fixture callable instead of
    hitting sops/secrets); defaults to the real reader for `provider_name`.
    """
    reader = probe_credential_fn or PROBE_CREDENTIAL_READERS.get(provider_name)
    if reader is None:
        return False, f"no probe credential reader for provider '{provider_name}'"
    try:
        profile_value = resolve_profile_credential_value(provider_name, profile_dir)
    except Exception as e:
        return False, f"profile credential unreadable: {e}"
    try:
        probe_value = reader()
    except Exception as e:
        return False, f"probe credential unreadable: {e}"
    return _credential_parity_from_values(profile_value, probe_value)


# === Probe ===


def run_probe(probe_json_path: Path | None) -> dict:
    if probe_json_path is not None:
        return json.loads(probe_json_path.read_text())
    result = subprocess.run(
        ["agent-subscriptions", "--output", "json"],
        capture_output=True, text=True, timeout=30,
    )
    if result.returncode != 0:
        raise RuntimeError(f"agent-subscriptions exited {result.returncode}: {result.stderr.strip()}")
    return json.loads(result.stdout)


def provider_probe_lookup(report: dict, provider_key: str) -> dict | None:
    name_map = {"anthropic": "anthropic", "minimax": "minimax"}
    wanted = name_map.get(provider_key, provider_key).lower()
    for entry in report.get("providers", []):
        if str(entry.get("provider", "")).lower() == wanted:
            return entry
    return None


# === Gates ===


def compute_gate(provider_cfg: dict, provider_probe: dict | None) -> dict:
    result = {
        "eligible": False,
        "idle_points": None,
        "five_hour_pct": None,
        "seven_day_pct": None,
        "elapsed_pct": None,
        "reason": "",
    }
    if provider_probe is None:
        result["reason"] = "no probe data for this provider"
        return result
    if not provider_probe.get("available"):
        result["reason"] = f"provider unavailable: {provider_probe.get('error', 'unknown')}"
        return result
    windows = provider_probe.get("windows", {})
    sd = windows.get("seven_day", {}) or {}
    fh = windows.get("five_hour", {}) or {}
    elapsed = sd.get("elapsed_pct")
    usage_sd = sd.get("utilization_pct")
    usage_fh = fh.get("utilization_pct")
    result["elapsed_pct"] = elapsed
    result["seven_day_pct"] = usage_sd
    result["five_hour_pct"] = usage_fh
    if elapsed is None:
        result["reason"] = "seven_day.elapsed_pct is null (cannot pace)"
        return result
    if usage_sd is None:
        result["reason"] = "seven_day.utilization_pct missing"
        return result
    if usage_fh is None:
        result["reason"] = "five_hour.utilization_pct missing"
        return result
    threshold_pct = provider_cfg["threshold_pct"]
    min_idle = provider_cfg["min_idle"]
    five_hour_ceiling = provider_cfg["five_hour_ceiling"]
    idle_points = threshold_pct * elapsed - usage_sd
    result["idle_points"] = idle_points
    if idle_points <= min_idle:
        result["reason"] = f"idle_points {idle_points:.2f} <= min_idle {min_idle}"
        return result
    if not (usage_fh < five_hour_ceiling):
        result["reason"] = f"five_hour.utilization_pct {usage_fh} >= ceiling {five_hour_ceiling}"
        return result
    result["eligible"] = True
    result["reason"] = "ok"
    return result


def compute_all_gates(config: dict, probe_report: dict) -> dict:
    gates = {}
    for pname, pcfg in config["providers"].items():
        probe = provider_probe_lookup(probe_report, pname)
        gates[pname] = compute_gate(pcfg, probe)
    return gates


# === Work-item discovery / stage derivation ===


def list_items(repo: Path, dossier: str) -> list[tuple[str, str]]:
    """Returns [(nn_str, slug), ...] sorted by numeric prefix ascending."""
    dossier_dir = repo / dossier
    if not dossier_dir.is_dir():
        return []
    found = []
    for f in dossier_dir.glob("*.overview.md"):
        m = ITEM_FILENAME_RE.match(f.name)
        if m:
            found.append((m.group(1), m.group(2)))
    found.sort(key=lambda t: int(t[0]))
    return found


def item_files(repo: Path, dossier: str, nn_slug: str) -> dict:
    """nn_slug is the full on-disk identifier, e.g. '03-shadow-agents...' —
    NOT the bare slug returned as the second element of list_items()'s tuples."""
    base = repo / dossier
    return {
        "overview": (base / f"{nn_slug}.overview.md").exists(),
        "agent": (base / f"{nn_slug}.agent.md").exists(),
        "research": (base / f"{nn_slug}.research.md").exists(),
        "draft": (base / f"{nn_slug}.draft.md").exists(),
        "review": (base / f"{nn_slug}.review.md").exists(),
    }


def next_work_type_for_item(files: dict, state_entry: dict) -> tuple[str | None, str]:
    if state_entry.get("blocked"):
        return None, f"blocked: {state_entry['blocked']}"
    if state_entry.get("approved"):
        return None, "approved (terminal)"
    if not files["agent"]:
        return None, "missing commissioning brief (.agent.md)"
    if not files["research"]:
        return "research", "research not yet produced"
    if not files["draft"]:
        return "draft", "research complete, draft not yet produced"
    if files["review"]:
        return None, "review complete, awaiting human revise/approve"
    if not state_entry.get("human_edit_done"):
        return None, "waiting on human edit of draft"
    return "review", "draft human-edited, ready for review"


def count_unreviewed_research(repo: Path, dossier: str, state: dict) -> int:
    count = 0
    for nn, slug in list_items(repo, dossier):
        files = item_files(repo, dossier, f"{nn}-{slug}")
        if not files["research"]:
            continue
        entry = get_item_state(state, f"{dossier}/{nn}-{slug}")
        if not entry["research_sampled"]:
            count += 1
    return count


def count_open_topic_proposals(repo: Path, dossier: str) -> int:
    proposals_path = repo / dossier / "TOPIC-PROPOSALS.md"
    if not proposals_path.exists():
        return 0
    text = proposals_path.read_text()
    return len(re.findall(r"^##\s+", text, flags=re.MULTILINE))


def build_ready_queue(repo: Path, config: dict, state: dict, in_flight: set) -> list[dict]:
    candidates: list[dict] = []
    dossiers = config["dossiers"]
    sample_override = bool(config.get("sample_override", False))
    for idx, dossier in enumerate(dossiers):
        unreviewed = count_unreviewed_research(repo, dossier, state)
        max_unreviewed = config["max_unreviewed_research_per_dossier"]
        for nn, slug in list_items(repo, dossier):
            key = f"{dossier}/{nn}-{slug}"
            if key in in_flight:
                continue
            files = item_files(repo, dossier, f"{nn}-{slug}")
            entry = get_item_state(state, key)
            wtype, reason = next_work_type_for_item(files, entry)
            if wtype == "research" and unreviewed >= max_unreviewed and not sample_override:
                continue
            if wtype is None:
                continue
            candidates.append({
                "kind": "item",
                "dossier": dossier,
                "slug": f"{nn}-{slug}",
                "item_key": key,
                "work_type": wtype,
                "rank": STAGE_RANK[wtype],
                "nn": int(nn),
                "dossier_idx": idx,
                "reason": reason,
            })

    if not candidates:
        for idx, dossier in enumerate(dossiers):
            if f"{dossier}/topic_ideas" in in_flight:
                continue
            open_count = count_open_topic_proposals(repo, dossier)
            if open_count < config["max_open_topic_proposals"]:
                candidates.append({
                    "kind": "dossier",
                    "dossier": dossier,
                    "slug": None,
                    "item_key": f"{dossier}/topic_ideas",
                    "work_type": "topic_ideas",
                    "rank": STAGE_RANK["topic_ideas"],
                    "nn": 0,
                    "dossier_idx": idx,
                    "reason": f"no other work eligible; {open_count} open proposals",
                })

    candidates.sort(key=lambda c: (-c["rank"], c["nn"], c["dossier_idx"]))
    return candidates


def provider_entry_name(entry) -> str:
    """A providers-list entry is either a plain provider-name string (the
    legacy form) or an object {"provider": name, "model": ..., "effort": ...}.
    This extracts just the name -- the only field select_provider() itself
    needs; the model/effort overrides are resolved later, once a provider has
    actually been selected, by resolve_work_type_model/_effort below."""
    return entry if isinstance(entry, str) else entry["provider"]


def select_provider(work_type: str, config: dict, gates: dict, credential_ok: dict) -> tuple[str | None, str]:
    providers = config["work_types"].get(work_type, {}).get("providers", [])
    for entry in providers:
        pname = provider_entry_name(entry)
        gate = gates.get(pname, {})
        if gate.get("eligible") and credential_ok.get(pname, (False, ""))[0]:
            return pname, "eligible"
    return None, "no eligible provider"


# === Prompt rendering ===


def load_prompt_template(work_type: str) -> str:
    fname = PROMPT_TEMPLATE_FILES[work_type]
    path = PROMPTS_DIR / fname
    return path.read_text()


def render_prompt(work_type: str, mapping: dict) -> str:
    template = Template(load_prompt_template(work_type))
    return template.substitute(mapping)


def build_prompt_mapping(repo: Path, dossier: str, slug: str | None, work_type: str,
                          config: dict, output_path: Path) -> dict:
    base = repo / dossier
    evidence_dirs_text = "\n".join(f"- {d}" for d in config["evidence_dirs"])
    mapping = {
        "dossier": dossier,
        "slug": slug or "",
        "overview_path": str(base / f"{slug}.overview.md") if slug else "",
        "agent_path": str(base / f"{slug}.agent.md") if slug else "",
        "research_path": str(base / f"{slug}.research.md") if slug else "",
        "draft_path": str(base / f"{slug}.draft.md") if slug else "",
        "output_path": str(output_path),
        "style_dir": str(repo / "style"),
        "source_register_path": str(base / "SOURCE-REGISTER.md"),
        "agents_root_path": str(repo / "AGENTS.md"),
        "dossier_agents_path": str(base / "AGENTS.md"),
        "review_prompt_path": str(repo / "style" / "review-prompt.md"),
        "review_score_threshold": str(config["review_score_threshold"]),
        "evidence_dirs": evidence_dirs_text,
        "existing_titles": "",
        "max_new": str(config["max_open_topic_proposals"]),
    }
    if work_type == "topic_ideas":
        mapping["existing_titles"] = "\n".join(f"- {t}" for t in collect_existing_titles(repo, dossier))
    return mapping


def collect_existing_titles(repo: Path, dossier: str) -> list[str]:
    titles = []
    for nn, slug in list_items(repo, dossier):
        overview = repo / dossier / f"{nn}-{slug}.overview.md"
        try:
            first_line = overview.read_text().splitlines()[0]
            titles.append(first_line.lstrip("#").strip())
        except (IndexError, OSError):
            titles.append(f"{nn}-{slug}")
    proposals_path = repo / dossier / "TOPIC-PROPOSALS.md"
    if proposals_path.exists():
        for line in proposals_path.read_text().splitlines():
            if line.startswith("## "):
                titles.append(line[3:].strip())
    return titles


# === Output validation ===


def extract_absolute_paths(text: str) -> list[str]:
    return ABS_PATH_RE.findall(text)


def strip_line_range(p: str) -> str:
    """Strip a trailing ':<digits>' or ':<digits>-<digits>' citation line-range suffix.

    Only strips when the remainder still looks like a path (has a filename
    segment before the colon); a plain path with no suffix is unchanged.
    """
    m = LINE_RANGE_SUFFIX_RE.match(p)
    return m.group(1) if m else p


def validate_output(work_type: str, text: str) -> tuple[bool, str]:
    if not text or not text.strip():
        return False, "output is empty"
    first_line = next((l for l in text.splitlines() if l.strip()), "")
    prefix = EXPECTED_HEADING_PREFIX[work_type]
    if not first_line.startswith(prefix):
        return False, f"first non-blank line does not start with '{prefix}': {first_line!r}"
    if work_type == "research":
        dead = [p for p in extract_absolute_paths(text) if not Path(strip_line_range(p.rstrip(".,;:"))).exists()]
        if dead:
            return False, f"dead cited path(s): {dead}"
    return True, "ok"


# === Claude argv / execution ===


def build_claude_argv(
    model_id: str | None,
    max_turns: int,
    add_dirs: list[str],
    allowed_tools: list[str] | None = None,
    effort: str | None = None,
) -> list[str]:
    """`model_id` is the already-resolved effective model (see
    resolve_effective_model: work-type "model" beats the profile's MODEL_ID) --
    this function emits at most one --model flag, never two."""
    argv = ["claude", "-p", "--max-turns", str(max_turns)]
    if model_id:
        argv += ["--model", model_id]
    if effort:
        argv += ["--effort", effort]
    for d in add_dirs:
        argv += ["--add-dir", d]
    if allowed_tools:
        argv += ["--allowedTools", ",".join(allowed_tools)]
    return argv


def classify_failure(returncode: int, stderr: str, timed_out: bool) -> str | None:
    if timed_out:
        return "transient"
    if returncode == 0:
        return None
    lowered = (stderr or "").lower()
    if any(marker in lowered for marker in RETRYABLE_STDERR_MARKERS):
        return "transient"
    return "content"


def canonical_output_path(repo: Path, candidate: dict) -> Path:
    dossier = candidate["dossier"]
    work_type = candidate["work_type"]
    if work_type == "topic_ideas":
        return repo / dossier / "TOPIC-PROPOSALS.md"
    slug = candidate["slug"]
    return repo / dossier / f"{slug}.{SUFFIX[work_type]}.md"


def rejected_output_path(canonical: Path) -> Path:
    return canonical.with_name(canonical.name + ".rejected")


def git_commit(repo: Path, paths: list[str], message: str) -> tuple[bool, str]:
    add = subprocess.run(["git", "-C", str(repo), "add", "--"] + paths, capture_output=True, text=True)
    if add.returncode != 0:
        return False, f"git add failed: {add.stderr.strip()}"
    commit = subprocess.run(
        ["git", "-C", str(repo), "commit", "-m", message, "--"] + paths,
        capture_output=True, text=True,
    )
    if commit.returncode != 0:
        return False, f"git commit failed: {commit.stderr.strip()}"
    return True, "ok"


def resolve_add_dirs(config: dict, work_type: str) -> list[str]:
    if work_type in ("research", "topic_ideas"):
        return [str(expand(d)) for d in config["evidence_dirs"]]
    return []


def resolve_allowed_tools(config: dict, work_type: str) -> list[str]:
    """Tool names granted to the headless child via --allowedTools, from
    config["work_types"][work_type]["allowed_tools"]. Empty/absent -> []
    (caller omits the flag; headless runs cannot answer permission prompts,
    so only tools explicitly listed here are usable by that work type)."""
    return list(config["work_types"].get(work_type, {}).get("allowed_tools") or [])


def find_provider_entry(config: dict, work_type: str, provider_name: str):
    """The raw providers-list entry (str or dict) for provider_name within
    work_type's providers list, or None if not present. Defensive lookup --
    callers only ever pass a provider name that select_provider() just chose
    from this same list, so a miss shouldn't happen, but resolve_work_type_*
    below treat it as "no per-provider override" rather than raising."""
    for entry in config["work_types"].get(work_type, {}).get("providers", []):
        if provider_entry_name(entry) == provider_name:
            return entry
    return None


def resolve_work_type_effort(config: dict, work_type: str, provider_name: str | None = None) -> str | None:
    """Effort level for (work_type, provider_name). Precedence:
    per-provider entry "effort" > work-type-level "effort" > None (CLI
    default). `provider_name` is optional -- omit it to get just the
    work-type-level value (e.g. before a provider has been selected). Both
    layers are already validated (non-empty string, member of
    CLAUDE_EFFORT_LEVELS) at config-load time -- this is a plain lookup."""
    wcfg = config["work_types"].get(work_type, {})
    if provider_name is not None:
        entry = find_provider_entry(config, work_type, provider_name)
        per_provider = entry.get("effort") if isinstance(entry, dict) else None
        if per_provider:
            return per_provider
    return wcfg.get("effort")


def resolve_work_type_model(config: dict, work_type: str, provider_name: str | None = None) -> str | None:
    """Model override for (work_type, provider_name). Precedence:
    per-provider entry "model" > work-type-level "model" > None (falls back
    to the profile's MODEL_ID via resolve_effective_model). `provider_name`
    is optional -- omit it to get just the work-type-level value. Both
    layers are already type-checked (non-empty string) at config-load time --
    this is a plain lookup."""
    wcfg = config["work_types"].get(work_type, {})
    if provider_name is not None:
        entry = find_provider_entry(config, work_type, provider_name)
        per_provider = entry.get("model") if isinstance(entry, dict) else None
        if per_provider:
            return per_provider
    return wcfg.get("model")


def resolve_effective_model(profile_model_id: str | None, work_type_model: str | None) -> str | None:
    """Precedence: a work-type "model" override always wins over the
    profile's provider.env MODEL_ID. Pure function so the precedence rule is
    independently testable without touching a profile dir."""
    return work_type_model or profile_model_id


def dispatch_preview(repo: Path, config: dict, candidate: dict, max_turns: int) -> dict:
    """Compute everything a dryrun needs to print, without executing anything."""
    dossier = candidate["dossier"]
    work_type = candidate["work_type"]
    provider = candidate["provider"]
    profile_dir = expand(config["providers"][provider]["profile"])
    model_id = resolve_effective_model(
        resolve_model_id(profile_dir), resolve_work_type_model(config, work_type, provider)
    )
    effort = resolve_work_type_effort(config, work_type, provider)
    add_dirs = resolve_add_dirs(config, work_type)
    allowed_tools = resolve_allowed_tools(config, work_type)
    argv = build_claude_argv(model_id, max_turns, add_dirs, allowed_tools, effort)
    canonical = canonical_output_path(repo, candidate)
    tmp_path = tmp_output_path(canonical)
    parsed_env = parse_provider_env(profile_dir / "provider.env")
    return {
        "dossier": dossier,
        "slug": candidate["slug"],
        "work_type": work_type,
        "provider": provider,
        "prompt_template": str(PROMPTS_DIR / PROMPT_TEMPLATE_FILES[work_type]),
        "argv": argv,
        "cwd": str(repo),
        "env_config_dir": str(profile_dir),
        "env_base_url": parsed_env["base_url"],
        "env_has_api_key": bool(parsed_env["api_key_file"]),
        "output_tmp_path": str(tmp_path),
        "output_canonical_path": str(canonical),
    }


# === 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 ===


def run_task(repo: Path, config: dict, candidate: dict, max_turns: int, task_timeout: int) -> dict:
    dossier = candidate["dossier"]
    work_type = candidate["work_type"]
    provider = candidate["provider"]
    slug = candidate["slug"]

    canonical = canonical_output_path(repo, candidate)
    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)

    profile_dir = expand(config["providers"][provider]["profile"])
    model_id = resolve_effective_model(
        resolve_model_id(profile_dir), resolve_work_type_model(config, work_type, provider)
    )
    effort = resolve_work_type_effort(config, work_type, provider)
    add_dirs = resolve_add_dirs(config, work_type)
    allowed_tools = resolve_allowed_tools(config, work_type)
    argv = build_claude_argv(model_id, max_turns, add_dirs, allowed_tools, effort)
    env = build_child_env(profile_dir)

    timed_out = False
    returncode = -1
    stderr = ""
    try:
        proc = subprocess.Popen(
            argv, cwd=str(repo), env=env,
            stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
            text=True, start_new_session=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)

    if failure == "transient":
        if tmp_path.exists():
            tmp_path.unlink()
        return {"outcome": "transient", "candidate": candidate, "reason": "timeout" if timed_out else stderr}

    if failure == "content":
        rejected = rejected_output_path(canonical)
        if tmp_path.exists():
            os.replace(tmp_path, rejected)
        return {"outcome": "content_failure", "candidate": candidate, "reason": stderr or "non-zero exit"}

    # returncode == 0: validate
    text = tmp_path.read_text() if tmp_path.exists() else ""
    ok, reason = validate_output(work_type, text)
    if not ok:
        rejected = rejected_output_path(canonical)
        if tmp_path.exists():
            os.replace(tmp_path, rejected)
        return {"outcome": "content_failure", "candidate": candidate, "reason": reason}

    if work_type == "topic_ideas":
        existing = canonical.read_text() if canonical.exists() else ""
        atomic_write_text(canonical, existing + ("\n" if existing and not existing.endswith("\n") else "") + text)
        if tmp_path.exists():
            tmp_path.unlink()
    else:
        os.replace(tmp_path, canonical)

    return {"outcome": "success", "candidate": candidate, "canonical": canonical}


# === Dispatch loop ===


def get_config_defaults(config: dict) -> dict:
    return {
        "max_turns": config.get("max_turns", DEFAULT_MAX_TURNS),
        "task_timeout": config.get("task_timeout_seconds", DEFAULT_TASK_TIMEOUT_SECONDS),
        "max_attempts": config.get("max_attempts", DEFAULT_MAX_ATTEMPTS),
    }


def evaluate_credentials(config: dict) -> dict:
    result = {}
    for pname in config["providers"]:
        profile_dir = expand(config["providers"][pname]["profile"])
        result[pname] = credential_parity(pname, profile_dir)
    return result


def log_gates(repo: Path, gates: dict, credential_ok: dict) -> None:
    for pname, gate in gates.items():
        cred_ok, cred_reason = credential_ok.get(pname, (False, "not checked"))
        eligible = gate["eligible"] and cred_ok
        log_event(
            repo,
            f"gate {pname}: idle_points={gate['idle_points']} five_hour_pct={gate['five_hour_pct']} "
            f"eligible={eligible} reason={gate['reason']!r} credential_parity={cred_ok} ({cred_reason})",
        )


def apply_success_state(state: dict, candidate: dict) -> None:
    """On success, clear any failed-attempt history for this (item, work_type).
    Without this, a success following one or more content failures leaves
    attempts[work_type] at its prior nonzero value; if a human later deletes
    the produced file to force a redo, the item would start already one
    content failure away from `blocked` instead of fresh. Pruning (rather
    than zeroing) the key/subkey keeps state minimal and matches how
    get_item_state()/validate_state() already treat an absent attempts entry
    as equivalent to zero -- nothing else (human_edit_done / research_sampled
    / approved) is a success-path concern; those are human-only via `mark`."""
    if candidate["kind"] != "item":
        return
    key = candidate["item_key"]
    work_type = candidate["work_type"]
    if work_type not in ALLOWED_ATTEMPT_WORK_TYPES:
        return
    entry = state.get("items", {}).get(key)
    if entry is None:
        return
    attempts = entry.get("attempts")
    if not attempts or work_type not in attempts:
        return
    del attempts[work_type]
    if not attempts:
        del entry["attempts"]


def apply_failure_state(state: dict, candidate: dict, max_attempts: int) -> str | None:
    if candidate["kind"] != "item":
        return None
    key = candidate["item_key"]
    work_type = candidate["work_type"]
    if work_type not in ALLOWED_ATTEMPT_WORK_TYPES:
        return None
    items = state.setdefault("items", {})
    entry = items.setdefault(key, {})
    attempts = entry.setdefault("attempts", {})
    attempts[work_type] = attempts.get(work_type, 0) + 1
    blocked_reason = None
    if attempts[work_type] >= max_attempts:
        blocked_reason = f"{work_type} failed {attempts[work_type]} times (max_attempts={max_attempts})"
        entry["blocked"] = blocked_reason
    return blocked_reason


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)
    except StateValidationError as e:
        log_event(repo, f"FATAL: state validation failed: {e}")
        return 2

    parallel = args.parallel or config.get("parallel", DEFAULT_PARALLEL)

    try:
        probe_report = run_probe(args.probe_json)
    except Exception as e:
        log_event(repo, f"FATAL: probe failed: {e}")
        return 1

    credential_ok = evaluate_credentials(config)
    gates = compute_all_gates(config, probe_report)
    log_gates(repo, gates, credential_ok)

    in_flight: set = set()
    queue = build_ready_queue(repo, config, state, in_flight)

    if args.dryrun:
        slots = 1 if args.once else parallel
        picked = []
        for cand in queue:
            if len(picked) >= slots:
                break
            provider, why = select_provider(cand["work_type"], config, gates, credential_ok)
            if provider is None:
                continue
            cand = dict(cand)
            cand["provider"] = provider
            in_flight.add(cand["item_key"])
            picked.append(cand)
        if not picked:
            print("[dryrun] No eligible (item, provider) pair to dispatch this cycle.")
            return 0
        print(f"[dryrun] Would dispatch {len(picked)} task(s) (initial wave; re-probe not simulated):")
        for cand in picked:
            preview = dispatch_preview(repo, config, cand, defaults["max_turns"])
            print(f"\n  item:      {preview['dossier']}/{preview['slug'] or '(dossier-level)'}")
            print(f"  work_type: {preview['work_type']}")
            print(f"  provider:  {preview['provider']}")
            print(f"  prompt:    {preview['prompt_template']}")
            print(f"  cwd:       {preview['cwd']}")
            print(f"  env:       CLAUDE_CONFIG_DIR={preview['env_config_dir']}"
                  + (f" ANTHROPIC_BASE_URL={preview['env_base_url']}" if preview['env_base_url'] else "")
                  + (" ANTHROPIC_API_KEY=<redacted>" if preview['env_has_api_key'] else ""))
            print(f"  argv:      {preview['argv']}")
            print(f"  output:    {preview['output_tmp_path']} -> {preview['output_canonical_path']}")
        return 0

    dispatched_count = 0
    fatal = False

    with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, parallel)) as executor:
        futures = {}

        def try_submit():
            nonlocal dispatched_count
            nonlocal queue
            while len(futures) < max(1, parallel):
                if args.once and dispatched_count >= 1:
                    return
                queue = build_ready_queue(repo, config, state, in_flight)
                for cand in queue:
                    provider, _ = select_provider(cand["work_type"], config, gates, credential_ok)
                    if provider is None:
                        continue
                    cand = dict(cand)
                    cand["provider"] = provider
                    cand["before_pct"] = gates.get(provider, {}).get("seven_day_pct")
                    in_flight.add(cand["item_key"])
                    dispatched_count += 1
                    log_event(repo, f"dispatch: {cand['item_key']} work_type={cand['work_type']} provider={provider}")
                    fut = executor.submit(run_task, repo, config, cand, defaults["max_turns"], defaults["task_timeout"])
                    futures[fut] = cand
                    break
                else:
                    return

        try_submit()

        while futures:
            done, _ = concurrent.futures.wait(futures, return_when=concurrent.futures.FIRST_COMPLETED)
            for fut in done:
                cand = futures.pop(fut)
                in_flight.discard(cand["item_key"])
                try:
                    result = fut.result()
                except Exception as e:
                    log_event(repo, f"FATAL task exception for {cand['item_key']}: {e}")
                    fatal = True
                    continue

                outcome = result["outcome"]
                if outcome == "success":
                    apply_success_state(state, cand)
                    save_state_atomic(state_path, state)
                    canonical = result["canonical"]
                    try:
                        after_probe = run_probe(args.probe_json)
                        after_gate = compute_gate(config["providers"][cand["provider"]],
                                                    provider_probe_lookup(after_probe, cand["provider"]))
                        after_pct = after_gate["seven_day_pct"]
                        gates.update(compute_all_gates(config, after_probe))
                    except Exception as e:
                        log_event(repo, f"re-probe after completion failed: {e}")
                        after_pct = None
                    before_pct = cand.get("before_pct")
                    msg = (
                        f"idle-draft: {cand['work_type']} {cand['item_key']} via {cand['provider']} "
                        f"(7d {before_pct}%→{after_pct}%)"
                    )
                    rel_canonical = str(Path(canonical).relative_to(repo))
                    ok, reason = git_commit(repo, [rel_canonical], msg)
                    log_event(repo, f"success: {cand['item_key']} -> {rel_canonical}; commit={ok} ({reason})")
                elif outcome == "content_failure":
                    blocked_reason = apply_failure_state(state, cand, defaults["max_attempts"])
                    save_state_atomic(state_path, state)
                    log_event(
                        repo,
                        f"content failure: {cand['item_key']} work_type={cand['work_type']} "
                        f"reason={result['reason']!r} blocked={blocked_reason!r}",
                    )
                else:  # transient
                    log_event(repo, f"transient failure: {cand['item_key']} reason={result['reason']!r}")

                if not (args.once and dispatched_count >= 1):
                    try_submit()

    if lock is not None:
        lock.close()

    return 1 if fatal else 0


# === status / mark ===


def cmd_status(args, config: dict, repo: Path, state_path: Path) -> int:
    try:
        state = load_state(state_path, repo)
    except StateValidationError as e:
        print(f"FATAL: state validation failed: {e}", file=sys.stderr)
        return 2

    try:
        probe_report = run_probe(args.probe_json)
        credential_ok = evaluate_credentials(config)
        gates = compute_all_gates(config, probe_report)
    except Exception as e:
        print(f"Warning: probe failed: {e}", file=sys.stderr)
        gates, credential_ok = {}, {}

    print("Providers")
    print("=========")
    for pname, gate in gates.items():
        cred_ok, cred_reason = credential_ok.get(pname, (False, "not checked"))
        print(f"  {pname}: idle_points={gate['idle_points']} five_hour_pct={gate['five_hour_pct']} "
              f"eligible={gate['eligible'] and cred_ok} reason={gate['reason']} credential_parity={cred_ok} ({cred_reason})")

    print()
    print("Items")
    print("=====")
    for dossier in config["dossiers"]:
        for nn, slug in list_items(repo, dossier):
            key = f"{dossier}/{nn}-{slug}"
            files = item_files(repo, dossier, f"{nn}-{slug}")
            entry = get_item_state(state, key)
            wtype, reason = next_work_type_for_item(files, entry)
            flags = []
            if entry["human_edit_done"]:
                flags.append("edited")
            if entry["research_sampled"]:
                flags.append("sampled")
            if entry["approved"]:
                flags.append("approved")
            if entry["blocked"]:
                flags.append(f"blocked({entry['blocked']})")
            flags_str = ",".join(flags) if flags else "-"
            next_action = wtype or "none"
            print(f"  {key}: next={next_action} flags=[{flags_str}] reason={reason!r}")
    return 0


def cmd_mark(args, config: dict, repo: Path, state_path: Path) -> int:
    item = args.item
    action = args.action
    if "/" not in item:
        print(f"Error: item must be of the form dossier/NN-slug, got: {item}", file=sys.stderr)
        return 2
    dossier, slug = item.split("/", 1)
    overview = repo / dossier / f"{slug}.overview.md"
    if not overview.exists():
        print(f"Error: no such item (overview file not found): {overview}", file=sys.stderr)
        return 2

    try:
        state = load_state(state_path, repo)
    except StateValidationError as e:
        print(f"FATAL: state validation failed: {e}", file=sys.stderr)
        return 2

    items = state.setdefault("items", {})
    entry = items.setdefault(item, {})

    if action == "edited":
        entry["human_edit_done"] = True
    elif action == "sampled":
        entry["research_sampled"] = True
    elif action == "approved":
        entry["approved"] = True
    elif action == "unblock":
        entry["blocked"] = None
    else:
        print(f"Error: unknown mark action: {action}", file=sys.stderr)
        return 2

    validate_state(state, repo)
    save_state_atomic(state_path, state)
    print(f"Marked {item}: {action}")
    return 0


# === CLI ===


HELP = """\
idle-draft — idle-subscription dispatcher for the writing pipeline

Usage:
  idle-draft [OPTIONS]
  idle-draft mark <dossier/NN-slug> edited|sampled|approved|unblock [OPTIONS]
  idle-draft status [OPTIONS]

Options:
  --config FILE       Path to config file (default: ./idle-draft.config.json)
  --repo DIR          Writing repo root (default: config file's directory)
  --once              Dispatch at most one task, then exit
  --parallel N        Max concurrent tasks (overrides config)
  --dryrun, -n        Preview dispatch without executing or mutating anything
  --probe-json FILE   Read probe output from FILE instead of running agent-subscriptions
  --help, -h          Show this help and exit

Spec: specs/idle-draft.spec.md
"""


def parse_args(argv: list[str]):
    class Args:
        pass

    args = Args()
    args.subcommand = None
    args.item = None
    args.action = None
    args.config = Path("idle-draft.config.json")
    args.repo = None
    args.once = False
    args.parallel = None
    args.dryrun = False
    args.probe_json = None

    rest = list(argv)
    if rest and rest[0] in ("mark", "status"):
        args.subcommand = rest[0]
        rest = rest[1:]
        if args.subcommand == "mark":
            positionals = [a for a in rest if not a.startswith("-")]
            if len(positionals) < 2:
                print("Error: mark requires <dossier/NN-slug> <action>", file=sys.stderr)
                sys.exit(2)
            args.item, args.action = positionals[0], positionals[1]
            rest = [a for a in rest if a not in (args.item, args.action)]

    i = 0
    while i < len(rest):
        a = rest[i]
        if a in ("-h", "--help"):
            print(HELP)
            sys.exit(0)
        elif a == "--config":
            args.config = Path(rest[i + 1]); i += 2
        elif a == "--repo":
            args.repo = Path(rest[i + 1]); i += 2
        elif a == "--once":
            args.once = True; i += 1
        elif a == "--parallel":
            args.parallel = int(rest[i + 1]); i += 2
        elif a in ("--dryrun", "-n"):
            args.dryrun = True; i += 1
        elif a == "--probe-json":
            args.probe_json = Path(rest[i + 1]); i += 2
        else:
            print(f"Error: unknown option: {a}", file=sys.stderr)
            sys.exit(2)

    return args


def main() -> None:
    args = parse_args(sys.argv[1:])

    try:
        config = load_config(args.config)
    except ConfigError as e:
        print(f"FATAL: {e}", file=sys.stderr)
        sys.exit(2)

    repo = args.repo if args.repo is not None else args.config.resolve().parent
    state_path = repo / "idle-draft.state.json"

    if args.subcommand == "mark":
        sys.exit(cmd_mark(args, config, repo, state_path))
    elif args.subcommand == "status":
        sys.exit(cmd_status(args, config, repo, state_path))
    else:
        sys.exit(run_dispatch(args, config, repo, state_path))


if __name__ == "__main__":
    main()