1480 lines
54 KiB
Python
Executable File
1480 lines
54 KiB
Python
Executable File
#!/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"}
|
|
|
|
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}'")
|
|
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")
|
|
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")
|
|
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 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
|
|
) -> 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]
|
|
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 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)
|
|
allowed_tools = resolve_allowed_tools(config, work_type)
|
|
argv = build_claude_argv(model_id, max_turns, add_dirs, allowed_tools)
|
|
canonical = canonical_output_path(repo, candidate)
|
|
tmp_path = 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_model_id(profile_dir)
|
|
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)
|
|
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
|
|
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()
|