#!/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 concurrent.futures
import fcntl
import hashlib
import json
import os
import re
import subprocess
import sys
import tempfile
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"}

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`)\]\"'>,;]+")

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}'")
    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")
    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"]
    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:
                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 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 pname in providers:
        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 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(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]) -> list[str]:
    argv = ["claude", "-p", "--max-turns", str(max_turns)]
    if model_id:
        argv += ["--model", model_id]
    for d in add_dirs:
        argv += ["--add-dir", d]
    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 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_model_id(profile_dir)
    add_dirs = resolve_add_dirs(config, work_type)
    argv = build_claude_argv(model_id, max_turns, add_dirs)
    canonical = canonical_output_path(repo, candidate)
    tmp_path = canonical.with_name(canonical.name + ".tmp")
    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),
    }


# === 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 = canonical.with_name(canonical.name + ".tmp")

    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_model_id(profile_dir)
    add_dirs = resolve_add_dirs(config, work_type)
    argv = build_claude_argv(model_id, max_turns, add_dirs)
    env = build_child_env(profile_dir)

    timed_out = False
    returncode = -1
    stderr = ""
    try:
        proc = subprocess.run(
            argv, cwd=str(repo), env=env, input=prompt_text,
            capture_output=True, text=True, timeout=task_timeout,
        )
        returncode = proc.returncode
        stderr = proc.stderr
    except subprocess.TimeoutExpired:
        timed_out = True

    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:
    # Successful dispatch doesn't itself set a human gate; nothing to flip here.
    # (human_edit_done / research_sampled / approved are human-only via `mark`.)
    # Present for symmetry with apply_failure_state and as the extension point
    # if a future work type needs to record something on success.
    return


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)
    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

    try:
        state = load_state(state_path, repo)
    except StateValidationError as e:
        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:
        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
            if args.once and dispatched_count >= 1:
                return
            nonlocal queue
            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
                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, "idle-draft.state.json"], 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()
