#!/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=" 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 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 ", 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()