alethe — Skillopedia

Orchestrate CLI workers Finish the work with the least lead context and the cheapest capable workers, or the strongest mix when the user asks. Plan, delegate bounded units, and integrate results supported by evidence. - Playbook = the phases and roles. - Profile = which models each tier uses ( , , ). - Goal = how eagerly to delegate while a mode is on ( , , , ). - Runner strategy = how one wave runs ( , , , , , ). Choose the backend Honor an explicit request for the local skill or no MCP. It always wins. Otherwise run and use its : selects this runner, selects tools when present, and prefers…

, frontmatter, re.MULTILINE)\n return match.group(1) if match else None\n\n\ndef default_policy_path() -> Path:\n configured = os.environ.get(\"ALETHE_ORCHESTRATE_POLICY\")\n if configured:\n return Path(configured).expanduser()\n if os.name == \"nt\":\n base = Path(os.environ.get(\"APPDATA\", Path.home() / \"AppData\" / \"Roaming\"))\n return base / \"Alethe\" / \"alethe-orchestrate.json\"\n base = Path(os.environ.get(\"XDG_CONFIG_HOME\", Path.home() / \".config\"))\n return base / \"alethe\" / \"orchestrate.json\"\n\n\nPLAYBOOKS_FILE = \"playbooks.yaml\"\nUSER_PLAYBOOKS_FILE = \"alethe-playbooks.yaml\"\n\n\ndef bundled_playbooks_path() -> Path:\n return Path(__file__).resolve().parent.parent / PLAYBOOKS_FILE\n\n\ndef user_playbooks_path(policy_path: Path) -> Path:\n configured = os.environ.get(\"ALETHE_ORCHESTRATE_PLAYBOOKS\")\n if configured:\n return Path(configured).expanduser()\n return policy_path.with_name(USER_PLAYBOOKS_FILE)\n\n\ndef effective_playbooks_path(policy_path: Path) -> Path:\n user = user_playbooks_path(policy_path)\n return user if user.is_file() else bundled_playbooks_path()\n\n\nPLAYBOOKS_VERSION = 1\nPLAYBOOK_TOP_FIELDS = {\n \"version\", \"goal\", \"setup\", \"principles\", \"providers\", \"models\", \"adapters\", \"routes\",\n \"tiers\", \"quota\", \"budget\", \"profile\", \"profiles\", \"pickOrder\", \"playbooks\", \"routing\", \"sources\",\n}\nPLAYBOOK_PHASE_FIELDS = {\n \"id\", \"role\", \"who\", \"tier\", \"provider\", \"model\", \"effort\", \"readOnly\", \"workers\", \"isolate\",\n \"runner\", \"does\", \"returns\", \"limit\", \"gate\", \"if\", \"with\", \"after\", \"then\", \"rules\",\n}\nPLAYBOOK_FIELDS = {\"name\", \"goal\", \"pickWhen\", \"saves\", \"avoidWhen\", \"source\", \"next\", \"phases\"}\nPLAYBOOK_PROFILE_FIELDS = {\"name\", \"summary\", \"order\", \"quota\", \"tiers\"}\nPLAYBOOK_PHASE_ENUMS = {\n \"who\": (\"lead\", \"worker\"), \"tier\": MODEL_TIERS,\n \"runner\": tuple(name for name in STRATEGIES if name != \"auto\"), \"then\": (\"next\", \"recheck\", \"done\"),\n}\nPLAYBOOK_QUOTA_FIELDS = {\n \"criticalThreshold\", \"stopAt\", \"whenLow\", \"onBothCritical\", \"whenSpare\", \"spareWithinHours\", \"spareMinLeft\",\n \"planEnds\",\n}\nPLAYBOOK_MAX_DEPTH = 100\nPLAYBOOK_CACHE_SIZE = 16\n\n\nclass _PlaybookMap(dict):\n def __init__(self, line: int) -> None:\n super().__init__()\n self.line = line\n self.key_lines: Dict[str, int] = {}\n\n\nclass _PlaybookYamlReader:\n def __init__(self, text: str) -> None:\n self.lines: List[Tuple[int, int, str]] = []\n for number, raw in enumerate(text.splitlines(), 1):\n indent = len(raw) - len(raw.lstrip(\" \\t\"))\n if \"\\t\" in raw[:indent]:\n self.fail(number, \"tabs are not supported for indentation\")\n content = self.uncomment(raw[indent:]).rstrip()\n if content:\n self.lines.append((number, indent, content))\n self.index = 0\n\n def fail(self, line: int, message: str) -> None:\n raise PlanError(f\"line {line}: {message}\")\n\n def uncomment(self, text: str) -> str:\n quote = None\n index = 0\n previous = \"\"\n while index \u003c len(text):\n char = text[index]\n if quote:\n if char == \"\\\\\" and quote == '\"':\n index += 2\n continue\n if char == quote:\n if quote == \"'\" and text[index:index + 2] == \"''\":\n index += 2\n continue\n quote = None\n elif char == \"#\" and (index == 0 or text[index - 1].isspace()):\n return text[:index]\n elif char in \"\\\"'\" and (not previous or previous in \":,[{-\"):\n quote = char\n if not char.isspace():\n previous = char\n index += 1\n return text\n\n def entry(self, text: str, line: int) -> Tuple[str, str]:\n quote = None\n index = 0\n while index \u003c len(text):\n char = text[index]\n if quote:\n if char == \"\\\\\" and quote == '\"':\n index += 2\n continue\n if char == quote:\n if quote == \"'\" and text[index:index + 2] == \"''\":\n index += 2\n continue\n quote = None\n elif index == 0 and char in \"\\\"'\":\n quote = char\n elif char == \":\" and (index + 1 == len(text) or text[index + 1].isspace()):\n key = self.value(text[:index].strip(), line)\n if not isinstance(key, str) or not key or key == \"\u003c\u003c\":\n self.fail(line, \"mapping keys must be non-empty strings; merges are unsupported\")\n return key, text[index + 1:].strip()\n index += 1\n self.fail(line, \"expected a block mapping entry\")\n\n def value(self, text: str, line: int, depth: int = 0) -> Any:\n value, end = self.flow_value(text, 0, line, depth, False)\n if text[end:].strip():\n self.fail(line, \"unexpected text after value\")\n return value\n\n def flow_value(self, text: str, start: int, line: int, depth: int, flow: bool) -> Tuple[Any, int]:\n if depth > PLAYBOOK_MAX_DEPTH:\n self.fail(line, \"YAML nesting is too deep\")\n index = start\n while index \u003c len(text) and text[index].isspace():\n index += 1\n if index == len(text):\n self.fail(line, \"missing scalar value\")\n char = text[index]\n if char in \"\\\"'\":\n return self.quoted(text, index, line)\n if char in \"[{\":\n mapping = char == \"{\"\n closing = \"}\" if mapping else \"]\"\n result: Any = _PlaybookMap(line) if mapping else []\n index += 1\n while index \u003c len(text) and text[index].isspace():\n index += 1\n if index \u003c len(text) and text[index] == closing:\n return result, index + 1\n while index \u003c len(text):\n if mapping:\n if text[index] in \"\\\"'\":\n key, index = self.quoted(text, index, line)\n else:\n end = index\n while end \u003c len(text) and text[end] not in \":,{}[]\":\n end += 1\n key = self.scalar(text[index:end].strip(), line)\n index = end\n while index \u003c len(text) and text[index].isspace():\n index += 1\n if not isinstance(key, str) or not key or key == \"\u003c\u003c\":\n self.fail(line, \"flow mapping keys must be non-empty strings\")\n if index == len(text) or text[index] != \":\":\n self.fail(line, \"expected ':' in flow mapping\")\n if key in result:\n self.fail(line, f\"duplicate mapping key {key!r}\")\n value, index = self.flow_value(text, index + 1, line, depth + 1, True)\n result[key] = value\n result.key_lines[key] = line\n else:\n value, index = self.flow_value(text, index, line, depth + 1, True)\n result.append(value)\n while index \u003c len(text) and text[index].isspace():\n index += 1\n if index \u003c len(text) and text[index] == closing:\n return result, index + 1\n if index == len(text) or text[index] != \",\":\n self.fail(line, f\"expected ',' or '{closing}'\")\n index += 1\n while index \u003c len(text) and text[index].isspace():\n index += 1\n if index == len(text) or text[index] == closing:\n self.fail(line, \"missing value after ','\")\n self.fail(line, f\"unclosed flow collection, expected '{closing}'\")\n end = index\n while end \u003c len(text) and (not flow or text[end] not in \",]}{\"):\n end += 1\n return self.scalar(text[index:end].strip(), line), end\n\n def quoted(self, text: str, start: int, line: int) -> Tuple[str, int]:\n quote = text[start]\n result = []\n index = start + 1\n escapes = {\n \"0\": \"\\0\", \"a\": \"\\a\", \"b\": \"\\b\", \"t\": \"\\t\", \"n\": \"\\n\", \"v\": \"\\v\", \"f\": \"\\f\",\n \"r\": \"\\r\", \"e\": \"\\x1b\", \" \": \" \", '\"': '\"', \"/\": \"/\", \"\\\\\": \"\\\\\",\n \"N\": \"\\x85\", \"_\": \"\\xa0\", \"L\": \"\\u2028\", \"P\": \"\\u2029\",\n }\n while index \u003c len(text):\n char = text[index]\n if char == quote:\n if quote == \"'\" and text[index:index + 2] == \"''\":\n result.append(\"'\")\n index += 2\n continue\n return \"\".join(result), index + 1\n if char == \"\\\\\" and quote == '\"':\n index += 1\n if index == len(text):\n self.fail(line, \"unfinished quoted escape\")\n escape = text[index]\n if escape in (\"x\", \"u\", \"U\"):\n count = {\"x\": 2, \"u\": 4, \"U\": 8}[escape]\n digits = text[index + 1:index + 1 + count]\n if len(digits) != count or not re.fullmatch(r\"[0-9a-fA-F]+\", digits):\n self.fail(line, \"invalid Unicode escape\")\n codepoint = int(digits, 16)\n if codepoint > 0x10ffff or 0xd800 \u003c= codepoint \u003c= 0xdfff:\n self.fail(line, \"invalid Unicode code point\")\n result.append(chr(codepoint))\n index += count + 1\n continue\n if escape not in escapes:\n self.fail(line, f\"unsupported escape \\\\{escape}\")\n result.append(escapes[escape])\n index += 1\n continue\n result.append(char)\n index += 1\n self.fail(line, \"unterminated quoted string\")\n\n def scalar(self, text: str, line: int) -> Any:\n if not text or text[0] in \"&*!|>@%\\x60,]}{[?:\" or text in (\"---\", \"...\", \"-\"):\n self.fail(line, \"unsupported YAML syntax or missing scalar\")\n if re.search(r\":(?:\\s|$)\", text):\n self.fail(line, \"unexpected mapping in scalar\")\n if text in (\"true\", \"false\"):\n return text == \"true\"\n if text.lower() in (\"null\", \"~\", \"yes\", \"no\", \"on\", \"off\", \"true\", \"false\", \".inf\", \"+.inf\", \"-.inf\", \".nan\"):\n self.fail(line, \"quote this scalar; only true/false booleans are supported\")\n if re.fullmatch(r\"[-+]?(?:0|[1-9][0-9]*)\", text):\n return int(text)\n if re.fullmatch(r\"[-+]?(?:[0-9]+\\.[0-9]*|\\.[0-9]+|[0-9]+)(?:[eE][-+]?[0-9]+)?\", text):\n value = float(text)\n if not math.isfinite(value):\n self.fail(line, \"non-finite numbers are unsupported\")\n return value\n if re.match(r\"[-+]?[0-9]\", text) and re.fullmatch(r\"[-+]?[0-9_:xXoObBa-fA-F.]+\", text):\n self.fail(line, \"unsupported numeric syntax; use a decimal number or quote it\")\n return text\n\n def block_entry(self, text: str, line: int, indent: int, depth: int) -> Tuple[str, Any]:\n key, value = self.entry(text, line)\n if value:\n return key, self.value(value, line)\n if self.index == len(self.lines):\n self.fail(line, f\"{key} needs a value; use {{}} or [] for an empty collection\")\n _, child_indent, child_text = self.lines[self.index]\n indentless_list = child_indent == indent and child_text.startswith(\"- \")\n if child_indent \u003c= indent and not indentless_list:\n self.fail(line, f\"{key} needs a value; use {{}} or [] for an empty collection\")\n return key, self.block(child_indent, depth + 1)\n\n def block(self, indent: int, depth: int = 0) -> Any:\n if depth > PLAYBOOK_MAX_DEPTH:\n self.fail(self.lines[self.index][0], \"YAML nesting is too deep\")\n line, actual_indent, text = self.lines[self.index]\n is_list = text == \"-\" or text.startswith(\"- \")\n result: Any = [] if is_list else _PlaybookMap(line)\n while self.index \u003c len(self.lines):\n line, actual_indent, text = self.lines[self.index]\n if actual_indent \u003c indent:\n break\n if actual_indent != indent:\n self.fail(line, \"inconsistent indentation\")\n item_is_list = text == \"-\" or text.startswith(\"- \")\n if is_list and not item_is_list:\n break\n if not is_list and item_is_list:\n self.fail(line, \"cannot mix a block list and mapping\")\n self.index += 1\n if not is_list:\n key, value = self.block_entry(text, line, indent, depth)\n if key in result:\n self.fail(line, f\"duplicate mapping key {key!r}\")\n result[key] = value\n result.key_lines[key] = line\n continue\n item = text[1:].strip()\n if not item:\n self.fail(line, \"block list items need a scalar or mapping\")\n mapping_item = item[0] not in \"[{\" and bool(re.search(r\":(?:\\s|$)\", item))\n if item[0] in \"\\\"'\":\n _, end = self.quoted(item, 0, line)\n mapping_item = bool(re.match(r\":(?:\\s|$)\", item[end:].lstrip()))\n if not mapping_item:\n result.append(self.value(item, line))\n continue\n mapping_indent = indent + text.index(item)\n entry = _PlaybookMap(line)\n key, value = self.block_entry(item, line, mapping_indent, depth + 1)\n entry[key] = value\n entry.key_lines[key] = line\n if self.index \u003c len(self.lines) and self.lines[self.index][1] > indent:\n continuation = self.block(mapping_indent, depth + 1)\n if not isinstance(continuation, dict):\n self.fail(self.lines[self.index - 1][0], \"expected mapping fields after list item\")\n for key, value in continuation.items():\n if key in entry:\n self.fail(continuation.key_lines[key], f\"duplicate mapping key {key!r}\")\n entry[key] = value\n entry.key_lines[key] = continuation.key_lines[key]\n result.append(entry)\n return result\n\n\ndef parse_playbooks_yaml(text: str) -> Any:\n reader = _PlaybookYamlReader(text)\n if not reader.lines:\n raise PlanError(\"line 1: playbooks must not be empty\")\n if reader.lines[0][1] != 0:\n reader.fail(reader.lines[0][0], \"top-level indentation must be zero\")\n if len(reader.lines) == 1 and reader.lines[0][2][0] in \"[{\":\n return reader.value(reader.lines[0][2], reader.lines[0][0])\n value = reader.block(0)\n if reader.index \u003c len(reader.lines):\n reader.fail(reader.lines[reader.index][0], \"unexpected indentation\")\n return value\n\n\ndef playbook_structure_errors(data: Any) -> List[str]:\n errors: List[str] = []\n\n def problem(owner: Any, key: str, label: str, message: str) -> None:\n line = getattr(owner, \"key_lines\", {}).get(key, getattr(owner, \"line\", 1))\n errors.append(f\"line {line}: {label} {message}\")\n\n def mapping(value: Any, owner: Any, key: str, label: str, allowed: Any = None) -> bool:\n if not isinstance(value, dict):\n problem(owner, key, label, \"must be a mapping\")\n return False\n if allowed is not None:\n for field in value:\n if field not in allowed:\n problem(value, field, f\"{label}.{field}\", \"is an unknown field\")\n return True\n\n def strings(value: Any, owner: Any, key: str, label: str) -> bool:\n if not isinstance(value, list) or any(not isinstance(item, str) or not item for item in value):\n problem(owner, key, label, \"must be a list of non-empty strings\")\n return False\n return True\n\n def named(value: Any, owner: Any, key: str, label: str) -> None:\n if not isinstance(value, str) or not SAFE_NAME_PATTERN.fullmatch(value) or value.startswith(\"-\"):\n problem(owner, key, label, \"must be a safe model or effort name\")\n\n def tiers(value: Any, owner: Any, key: str, label: str) -> None:\n if not mapping(value, owner, key, label, MODEL_TIERS):\n return\n for tier, entries in value.items():\n where = f\"{label}.{tier}\"\n if not mapping(entries, value, tier, where):\n continue\n for provider, spec in entries.items():\n entry_label = f\"{where}.{provider}\"\n if provider not in providers:\n problem(entries, provider, entry_label, \"is not in providers\")\n if mapping(spec, entries, provider, entry_label, (\"model\", \"effort\")):\n for field, entry in spec.items():\n named(entry, spec, field, f\"{entry_label}.{field}\")\n\n def quota(value: Any, owner: Any, key: str, label: str) -> None:\n if not mapping(value, owner, key, label, PLAYBOOK_QUOTA_FIELDS):\n return\n enums = {\n \"whenLow\": (\"switch-provider\", \"keep\"), \"onBothCritical\": (\"ask\", \"run-cheapest\", \"block\"),\n \"whenSpare\": (\"spend-strong\", \"prefer\", \"ignore\"),\n }\n for field, choices in enums.items():\n if field in value and value[field] not in choices:\n problem(value, field, f\"{label}.{field}\", f\"must be one of {', '.join(choices)}\")\n for field, upper in ((\"criticalThreshold\", 100), (\"stopAt\", 100), (\"spareWithinHours\", 168), (\"spareMinLeft\", 100)):\n if field not in value:\n continue\n number = value[field]\n if type(number) is not int or not 1 \u003c= number \u003c= upper:\n problem(value, field, f\"{label}.{field}\", f\"must be an integer between 1 and {upper}\")\n if \"planEnds\" in value:\n mapping(value[\"planEnds\"], value, \"planEnds\", f\"{label}.planEnds\")\n\n if not mapping(data, data, \"\", \"playbooks\", PLAYBOOK_TOP_FIELDS):\n return errors\n if type(data.get(\"version\")) is not int or data[\"version\"] != PLAYBOOKS_VERSION:\n problem(data, \"version\", \"version\", f\"must be {PLAYBOOKS_VERSION}\")\n providers = data.get(\"providers\")\n if not strings(providers, data, \"providers\", \"providers\"):\n providers = []\n elif not providers or len(set(providers)) != len(providers):\n problem(data, \"providers\", \"providers\", \"must list each provider once and cannot be empty\")\n for provider in providers:\n if not re.fullmatch(r\"[a-z][a-z0-9-]{0,31}\", provider):\n problem(data, \"providers\", \"providers\", f\"contains invalid provider id {provider!r}\")\n tiers(data.get(\"tiers\"), data, \"tiers\", \"tiers\")\n for field in (\"principles\", \"sources\"):\n if field in data:\n strings(data[field], data, field, field)\n for field in (\"setup\", \"models\", \"adapters\", \"routes\", \"routing\"):\n if field in data:\n mapping(data[field], data, field, field)\n if \"quota\" in data:\n quota(data[\"quota\"], data, \"quota\", \"quota\")\n if \"budget\" in data and mapping(data[\"budget\"], data, \"budget\", \"budget\", (\"warnAt\", \"whenLow\")):\n budget = data[\"budget\"]\n if \"warnAt\" in budget and (type(budget[\"warnAt\"]) is not int or not 1 \u003c= budget[\"warnAt\"] \u003c= 100):\n problem(budget, \"warnAt\", \"budget.warnAt\", \"must be an integer between 1 and 100\")\n if \"whenLow\" in budget and budget[\"whenLow\"] not in (\"finish-and-report\", \"downgrade-tier\", \"stop\"):\n problem(budget, \"whenLow\", \"budget.whenLow\", \"must be finish-and-report, downgrade-tier or stop\")\n profiles = data.get(\"profiles\", {})\n if mapping(profiles, data, \"profiles\", \"profiles\"):\n active = data.get(\"profile\")\n if not isinstance(active, str) or active not in profiles:\n problem(data, \"profile\", \"profile\", \"must name an existing profile\")\n for profile, value in profiles.items():\n label = f\"profiles.{profile}\"\n if not mapping(value, profiles, profile, label, PLAYBOOK_PROFILE_FIELDS):\n continue\n if \"tiers\" in value:\n tiers(value[\"tiers\"], value, \"tiers\", f\"{label}.tiers\")\n if \"quota\" in value:\n quota(value[\"quota\"], value, \"quota\", f\"{label}.quota\")\n if \"order\" in value and strings(value[\"order\"], value, \"order\", f\"{label}.order\"):\n order = value[\"order\"]\n if len(set(order)) != len(order) or any(provider not in providers for provider in order):\n problem(value, \"order\", f\"{label}.order\", \"must contain unique ids from providers\")\n books = data.get(\"playbooks\")\n if mapping(books, data, \"playbooks\", \"playbooks\"):\n for book_id, book in books.items():\n label = f\"playbooks.{book_id}\"\n if not mapping(book, books, book_id, label, PLAYBOOK_FIELDS):\n continue\n for field in PLAYBOOK_FIELDS - {\"phases\"}:\n if field in book and (not isinstance(book[field], str) or not book[field]):\n problem(book, field, f\"{label}.{field}\", \"must be a non-empty string\")\n phases = book.get(\"phases\")\n if not isinstance(phases, list):\n problem(book, \"phases\", f\"{label}.phases\", \"must be a list\")\n continue\n seen = set()\n for phase in phases:\n where = f\"{label}.phases\"\n if not mapping(phase, book, \"phases\", where, PLAYBOOK_PHASE_FIELDS):\n continue\n phase_id = phase.get(\"id\")\n if not isinstance(phase_id, str) or not ID_PATTERN.fullmatch(phase_id):\n problem(phase, \"id\", f\"{where}.id\", \"must be a safe phase id\")\n elif phase_id in seen:\n problem(phase, \"id\", f\"{where}.id\", f\"duplicates {phase_id}\")\n else:\n seen.add(phase_id)\n for field, choices in PLAYBOOK_PHASE_ENUMS.items():\n if field in phase and phase[field] not in choices:\n problem(phase, field, f\"{where}.{field}\", f\"must be one of {', '.join(choices)}\")\n for field in (\"readOnly\", \"isolate\"):\n if field in phase and not isinstance(phase[field], bool):\n problem(phase, field, f\"{where}.{field}\", \"must be true or false\")\n if \"workers\" in phase and (type(phase[\"workers\"]) is not int or not 1 \u003c= phase[\"workers\"] \u003c= MAX_WORKERS):\n problem(phase, \"workers\", f\"{where}.workers\", f\"must be an integer between 1 and {MAX_WORKERS}\")\n if \"provider\" in phase and phase[\"provider\"] not in [*providers, \"auto\"]:\n problem(phase, \"provider\", f\"{where}.provider\", \"must be auto or an id from providers\")\n for field in (\"model\", \"effort\"):\n if field in phase:\n named(phase[field], phase, field, f\"{where}.{field}\")\n if \"rules\" not in phase:\n continue\n if not isinstance(phase[\"rules\"], list):\n problem(phase, \"rules\", f\"{where}.rules\", \"must be a list\")\n continue\n for rule in phase[\"rules\"]:\n if not mapping(rule, phase, \"rules\", f\"{where}.rules\", (\"if\", \"use\")):\n continue\n use = rule.get(\"use\")\n if not mapping(use, rule, \"use\", f\"{where}.rules.use\", (\"tier\", \"provider\", \"model\", \"effort\")):\n continue\n if \"tier\" in use and use[\"tier\"] not in MODEL_TIERS:\n problem(use, \"tier\", f\"{where}.rules.use.tier\", f\"must be one of {', '.join(MODEL_TIERS)}\")\n for field in (\"model\", \"effort\"):\n if field in use:\n named(use[field], use, field, f\"{where}.rules.use.{field}\")\n picks = data.get(\"pickOrder\", [])\n if not isinstance(picks, list):\n problem(data, \"pickOrder\", \"pickOrder\", \"must be a list\")\n return errors\n for pick in picks:\n if not mapping(pick, data, \"pickOrder\", \"pickOrder\", (\"when\", \"playbook\")):\n continue\n target = pick.get(\"playbook\")\n if not isinstance(target, str) or not isinstance(books, dict) or target not in books:\n problem(pick, \"playbook\", \"pickOrder.playbook\", \"must name an existing playbook\")\n if not isinstance(pick.get(\"when\"), str) or not pick[\"when\"]:\n problem(pick, \"when\", \"pickOrder.when\", \"must be a non-empty string\")\n return errors\n\n\[email protected]_cache(maxsize=PLAYBOOK_CACHE_SIZE)\ndef _cached_playbooks(path: str, modified: int, size: int) -> Tuple[str, Dict[str, Any]]:\n text = read_playbooks(Path(path))\n try:\n data = parse_playbooks_yaml(text)\n errors = playbook_structure_errors(data)\n except PlanError as error:\n raise PlanError(f\"invalid playbooks at {path}: {error}\") from error\n if errors:\n raise PlanError(f\"invalid playbooks at {path}: \" + \"; \".join(errors))\n return text, data\n\n\ndef loaded_playbooks(policy_path: Path) -> Tuple[str, Dict[str, Any]]:\n path = effective_playbooks_path(policy_path).resolve()\n try:\n status = path.stat()\n except OSError as error:\n raise PlanError(f\"cannot read playbooks at {path}: {error}\") from error\n return _cached_playbooks(str(path), status.st_mtime_ns, status.st_size)\n\n\ndef playbook_settings(policy_path: Optional[Path] = None) -> Dict[str, Any]:\n _, parsed = loaded_playbooks(policy_path or default_policy_path())\n data = copy.deepcopy(parsed)\n profile = data[\"profiles\"][data[\"profile\"]]\n data[\"tiers\"] = {\n tier: {\n provider: {**data[\"tiers\"].get(tier, {}).get(provider, {}), **profile.get(\"tiers\", {}).get(tier, {}).get(provider, {})}\n for provider in data[\"providers\"]\n if provider in data[\"tiers\"].get(tier, {}) or provider in profile.get(\"tiers\", {}).get(tier, {})\n }\n for tier in MODEL_TIERS\n }\n order = profile.get(\"order\", data[\"providers\"])\n data[\"providers\"] = [*order, *(provider for provider in data[\"providers\"] if provider not in order)]\n data[\"quota\"] = {**data.get(\"quota\", {}), **profile.get(\"quota\", {})}\n return data\n\n\ndef read_playbooks(path: Path) -> str:\n try:\n text = path.read_text(encoding=\"utf-8\")\n except OSError as error:\n raise PlanError(f\"cannot read playbooks at {path}: {error}\") from error\n return text\n\n\ndef playbook_doctor(policy_path: Path) -> Dict[str, Any]:\n path = effective_playbooks_path(policy_path)\n try:\n loaded_playbooks(policy_path)\n except PlanError as error:\n return {\"path\": str(path), \"valid\": False, \"error\": str(error)}\n return {\"path\": str(path), \"valid\": True, \"error\": None}\n\n\ndef selected_playbook_block(text: str, parsed: Dict[str, Any], book_id: str) -> str:\n books = parsed[\"playbooks\"]\n lines = text.splitlines()\n start = books.key_lines[book_id] - 1\n first = lines[start]\n if not first.strip().startswith(book_id + \":\"):\n return f\"{json.dumps(book_id)}: {json.dumps(books[book_id], ensure_ascii=False)}\"\n indent = len(first) - len(first.lstrip())\n end = start + 1\n while end \u003c len(lines):\n line = lines[end]\n if line.strip() and not line.lstrip().startswith(\"#\") and len(line) - len(line.lstrip()) \u003c= indent:\n break\n end += 1\n return \"\\n\".join(lines[start:end]).rstrip()\n\n\ndef playbook_output(policy_path: Path, book_id: Optional[str], pick: bool, as_json: bool) -> str:\n text, parsed = loaded_playbooks(policy_path)\n settings = playbook_settings(policy_path)\n if book_id and pick:\n raise PlanError(\"pass a playbook id or --pick, not both\")\n if book_id and book_id not in parsed[\"playbooks\"]:\n raise PlanError(f\"unknown playbook {book_id}; choose from {', '.join(parsed['playbooks'])}\")\n if pick:\n payload = {\n \"profile\": settings[\"profile\"], \"providers\": settings[\"providers\"],\n \"pickOrder\": settings.get(\"pickOrder\", []),\n \"playbooks\": [\n {\"id\": key, \"name\": book.get(\"name\", key), \"pickWhen\": book.get(\"pickWhen\", \"\")}\n for key, book in settings[\"playbooks\"].items()\n ],\n }\n if as_json:\n return json.dumps(payload, indent=2, ensure_ascii=False)\n lines = [f\"profile: {payload['profile']}\", f\"providers: {', '.join(payload['providers'])}\"]\n lines.extend(f\"pick {index}: {item['when']} -> {item['playbook']}\" for index, item in enumerate(payload[\"pickOrder\"], 1))\n lines.extend(f\"{item['id']} | {item['name']} | {item['pickWhen']}\" for item in payload[\"playbooks\"])\n return \"\\n\".join(lines)\n if not book_id:\n if as_json:\n return json.dumps(parsed, indent=2, ensure_ascii=False)\n return f\"# source: {effective_playbooks_path(policy_path)}\\n{text.rstrip()}\"\n payload = {\n \"id\": book_id, \"playbook\": parsed[\"playbooks\"][book_id], \"profile\": settings[\"profile\"],\n \"tiers\": settings[\"tiers\"], \"providers\": settings[\"providers\"],\n \"quota\": settings.get(\"quota\", {}), \"budget\": settings.get(\"budget\", {}),\n }\n if as_json:\n return json.dumps(payload, indent=2, ensure_ascii=False)\n lines = [selected_playbook_block(text, parsed, book_id), \"\", f\"profile: {settings['profile']}\", \"resolved tiers:\"]\n for provider in settings[\"providers\"]:\n for tier in MODEL_TIERS:\n spec = settings[\"tiers\"].get(tier, {}).get(provider, {})\n model = spec.get(\"model\", \"\u003cCLI default>\")\n effort = spec.get(\"effort\", \"\u003cCLI default>\")\n lines.append(f\" {provider} x {tier} -> model={model} effort={effort}\")\n lines.extend([\n f\"providers: {', '.join(settings['providers'])}\",\n f\"quota: {json.dumps(payload['quota'], ensure_ascii=False)}\",\n f\"budget: {json.dumps(payload['budget'], ensure_ascii=False)}\",\n ])\n return \"\\n\".join(lines)\n\n\ndef init_playbooks(policy_path: Path, force: bool) -> Path:\n target = user_playbooks_path(policy_path)\n if target.exists() and not force:\n raise PlanError(f\"{target} already exists; pass --force to replace it with the bundled playbooks\")\n text = read_playbooks(bundled_playbooks_path())\n target.parent.mkdir(parents=True, exist_ok=True)\n temporary = target.with_name(f\".{target.name}.{uuid.uuid4().hex}.tmp\")\n temporary.write_text(text, encoding=\"utf-8\")\n os.replace(temporary, target)\n return target\n\n\nUSAGE_FILE = \"alethe-usage.json\"\nPRICES_FILE = \"prices.json\"\nUSER_PRICES_FILE = \"alethe-prices.json\"\nUSAGE_STALE_MINUTES = 30\nPRICE_STALE_DAYS = 60\nDEFAULT_CRITICAL_THRESHOLD = 80\nDEFAULT_STOP_THRESHOLD = 95\nQUOTA_SAMPLE_LIMIT = 40\nQUOTA_FORECAST_SAMPLES = 10\nQUOTA_FORECAST_MIN_SAMPLES = 2\nQUOTA_WINDOW_SLACK_SECONDS = 600\nCLAUDE_PROBE_MODEL = \"haiku\"\nCLAUDE_PROBE_SECONDS = 120\nCLAUDE_PROBE_PROMPT = \"Reply with OK.\"\nCODEX_TAIL_BYTES = 256 * 1024\nTOKENS_PER_MILLION = 1000000\nMAIN_QUOTA_WINDOWS = (\"5h\", \"week\")\nQUOTA_PROVIDERS = (\"claude\", \"codex\")\nCLAUDE_QUOTA_WINDOWS = {\"five_hour\": \"5h\", \"seven_day\": \"week\", \"seven_day_opus\": \"opus\", \"opus\": \"opus\"}\nCODEX_QUOTA_WINDOWS = {300: \"5h\", 10080: \"week\"}\nTOKEN_PARTS = (\"inputTokens\", \"cacheReadTokens\", \"cacheWriteTokens\", \"outputTokens\")\nUSAGE_LOCK = threading.Lock()\n\n\ndef timestamp(value: Any) -> Optional[dt.datetime]:\n try:\n if isinstance(value, (int, float)) and not isinstance(value, bool):\n return dt.datetime.fromtimestamp(value, dt.timezone.utc)\n if isinstance(value, str):\n stamp = dt.datetime.fromisoformat(value.replace(\"Z\", \"+00:00\"))\n return stamp.replace(tzinfo=dt.timezone.utc) if stamp.tzinfo is None else stamp\n except (ValueError, OverflowError, OSError):\n pass\n return None\n\n\ndef quota_thresholds(policy_path: Path) -> Tuple[float, float]:\n try:\n quota = playbook_settings(policy_path).get(\"quota\", {})\n except PlanError:\n # doctor and validate report an invalid playbooks copy; usage still answers\n return DEFAULT_CRITICAL_THRESHOLD, DEFAULT_STOP_THRESHOLD\n critical = number(quota.get(\"criticalThreshold\", DEFAULT_CRITICAL_THRESHOLD))\n if critical is None or not 0 \u003c= critical \u003c= 100:\n raise PlanError(\"quota.criticalThreshold must be between 0 and 100\")\n stop = number(quota.get(\"stopAt\", max(DEFAULT_STOP_THRESHOLD, critical)))\n if stop is None or not critical \u003c= stop \u003c= 100:\n raise PlanError(\"quota.stopAt must be between quota.criticalThreshold and 100\")\n return critical, stop\n\n\ndef local_clock(value: Any) -> str:\n stamp = timestamp(value)\n return stamp.astimezone().strftime(\"%H:%M\") if stamp else \"unknown\"\n\n\ndef quota_window(percent: Any, reset: Any) -> Optional[Dict[str, Any]]:\n value, stamp = number(percent), timestamp(reset)\n if value is None or not math.isfinite(value) or not 0 \u003c= value \u003c= 100 or stamp is None:\n return None\n return {\"usedPercent\": round(value, 4), \"resetsAt\": stamp.isoformat()}\n\n\ndef claude_quota(values: Sequence[Any], observed_at: str) -> Optional[Dict[str, Any]]:\n event = next((value for value in reversed(values) if isinstance(value, dict)\n and value.get(\"type\") == \"rate_limit_event\"), None)\n if event is None:\n return None\n info = event.get(\"rate_limit_info\")\n if not isinstance(info, dict):\n return None\n unified = info.get(\"unifiedWindows\")\n unified = unified if isinstance(unified, dict) else {}\n windows = {}\n for key, name in CLAUDE_QUOTA_WINDOWS.items():\n raw = unified.get(key)\n if not isinstance(raw, dict):\n continue\n utilization = number(raw.get(\"utilization\"))\n window = quota_window(utilization * 100 if utilization is not None else None, raw.get(\"resetsAt\"))\n if window:\n windows[name] = window\n if not windows:\n return None\n return {\"provider\": \"claude\", \"windows\": windows, \"observedAt\": observed_at, \"source\": \"stream-json\"}\n\n\ndef codex_home() -> Path:\n return Path(os.environ.get(\"CODEX_HOME\", str(Path.home() / \".codex\"))).expanduser()\n\n\ndef codex_quota(thread_id: Optional[str] = None) -> Optional[Dict[str, Any]]:\n try:\n paths = list((codex_home() / \"sessions\").glob(\"*/*/*/rollout-*.jsonl\"))\n if not paths:\n return None\n preferred = [path for path in paths if thread_id and thread_id in path.name]\n path = max(preferred or paths, key=lambda item: item.stat().st_mtime)\n stat = path.stat()\n with path.open(\"rb\") as handle:\n offset = max(0, stat.st_size - CODEX_TAIL_BYTES)\n handle.seek(offset)\n if offset:\n handle.readline()\n tail = handle.read(CODEX_TAIL_BYTES).decode(\"utf-8\", errors=\"replace\")\n except OSError:\n return None\n for line in reversed(tail.splitlines()):\n try:\n event = json.loads(line)\n except json.JSONDecodeError:\n continue\n if not isinstance(event, dict):\n continue\n payload = event.get(\"payload\")\n payload = payload if isinstance(payload, dict) else event\n limits = payload.get(\"rate_limits\")\n if not isinstance(limits, dict) or limits.get(\"limit_id\", \"codex\") != \"codex\":\n continue\n windows = {}\n for key in (\"primary\", \"secondary\"):\n raw = limits.get(key)\n if not isinstance(raw, dict):\n continue\n name = CODEX_QUOTA_WINDOWS.get(raw.get(\"window_minutes\"))\n window = quota_window(raw.get(\"used_percent\"), raw.get(\"resets_at\"))\n if name and window:\n windows[name] = window\n if not windows:\n continue\n observed = timestamp(event.get(\"timestamp\")) or timestamp(stat.st_mtime)\n return {\"provider\": \"codex\", \"windows\": windows, \"observedAt\": observed.isoformat(), \"source\": str(path)}\n return None\n\n\ndef usage_file(policy_path: Path) -> Dict[str, Any]:\n path = policy_path.with_name(USAGE_FILE)\n if not path.exists():\n return {\"providers\": {}, \"samples\": []}\n raw = read_json(path)\n if not isinstance(raw, dict) or not isinstance(raw.get(\"providers\"), dict):\n raise PlanError(f\"{path} must contain a providers object\")\n for provider, reading in raw[\"providers\"].items():\n if not isinstance(reading, dict) or not isinstance(reading.get(\"windows\"), dict) or timestamp(reading.get(\"observedAt\")) is None:\n raise PlanError(f\"invalid cached quota reading for {provider} in {path}\")\n for name, window in reading[\"windows\"].items():\n if not isinstance(window, dict) or quota_window(window.get(\"usedPercent\"), window.get(\"resetsAt\")) is None:\n raise PlanError(f\"invalid cached quota window for {provider}/{name} in {path}\")\n samples = raw.get(\"samples\", [])\n if not isinstance(samples, list) or not all(\n isinstance(sample, dict) and sample.get(\"provider\") in QUOTA_PROVIDERS and isinstance(sample.get(\"perWorker\"), dict)\n and all(number(value) is not None and value >= 0 for value in sample[\"perWorker\"].values())\n for sample in samples\n ):\n raise PlanError(f\"invalid quota samples in {path}\")\n return {\"providers\": raw[\"providers\"], \"samples\": samples}\n\n\ndef saved_usage(policy_path: Path) -> Dict[str, Any]:\n return usage_file(policy_path)[\"providers\"]\n\n\ndef write_usage_file(policy_path: Path, data: Dict[str, Any]) -> None:\n path = policy_path.with_name(USAGE_FILE)\n temporary = path.with_name(f\".{path.name}.{uuid.uuid4().hex}.tmp\")\n try:\n temporary.parent.mkdir(parents=True, exist_ok=True)\n temporary.write_text(json.dumps({\"version\": 1, **data}, indent=2) + \"\\n\", encoding=\"utf-8\")\n os.replace(temporary, path)\n finally:\n if temporary.exists():\n temporary.unlink()\n\n\ndef save_usage(policy_path: Path, quota: Optional[Dict[str, Any]]) -> None:\n with USAGE_LOCK:\n data = usage_file(policy_path)\n providers = dict(data[\"providers\"])\n if quota:\n provider = quota[\"provider\"]\n previous = providers.get(provider, {})\n old = timestamp(previous.get(\"observedAt\"))\n observed = timestamp(quota.get(\"observedAt\"))\n if observed is None:\n raise PlanError(\"quota.observedAt must be an ISO timestamp\")\n if old is None or observed >= old:\n providers[provider] = quota\n write_usage_file(policy_path, {**data, \"providers\": providers})\n\n\ndef save_quota_samples(policy_path: Path, samples: Sequence[Dict[str, Any]]) -> None:\n if not samples:\n return\n with USAGE_LOCK:\n data = usage_file(policy_path)\n write_usage_file(policy_path, {**data, \"samples\": [*data[\"samples\"], *samples][-QUOTA_SAMPLE_LIMIT:]})\n\n\ndef usage_payload(policy_path: Path, refresh: bool = True, now: Optional[dt.datetime] = None) -> Dict[str, Any]:\n current = now or dt.datetime.now(dt.timezone.utc)\n if refresh:\n save_usage(policy_path, codex_quota())\n threshold, stop = quota_thresholds(policy_path)\n readings = saved_usage(policy_path)\n providers = {}\n for provider in BUILTIN_PROVIDERS:\n reading = readings.get(provider, {})\n observed = timestamp(reading.get(\"observedAt\"))\n age = max(0.0, (current - observed).total_seconds() / 60) if observed else None\n windows = {}\n for name, raw in reading.get(\"windows\", {}).items():\n window = quota_window(raw.get(\"usedPercent\"), raw.get(\"resetsAt\"))\n if window is None:\n raise PlanError(f\"invalid cached quota window for {provider}/{name}\")\n reset = timestamp(window[\"resetsAt\"]) \u003c= current\n windows[name] = {**window, \"usedPercent\": 0 if reset else window[\"usedPercent\"],\n \"observedAt\": reading.get(\"observedAt\"), \"ageMinutes\": round(age, 2) if age is not None else None,\n \"stale\": age is None or age > USAGE_STALE_MINUTES, \"reset\": reset,\n \"low\": not reset and window[\"usedPercent\"] >= threshold,\n \"blocked\": not reset and window[\"usedPercent\"] >= stop}\n main = [windows[name] for name in MAIN_QUOTA_WINDOWS if name in windows]\n providers[provider] = {\"windows\": windows, \"observedAt\": reading.get(\"observedAt\"),\n \"ageMinutes\": round(age, 2) if age is not None else None,\n \"low\": any(window[\"low\"] for window in main), \"blocked\": any(window[\"blocked\"] for window in main),\n \"stale\": age is None or age > USAGE_STALE_MINUTES, \"source\": reading.get(\"source\"),\n \"missing\": not windows, \"reset\": any(window[\"reset\"] for window in windows.values())}\n return {\"version\": 1, \"criticalThreshold\": threshold, \"stopAt\": stop,\n \"cachePath\": str(policy_path.with_name(USAGE_FILE)), \"providers\": providers}\n\n\nEDITOR_TEMPLATE = Path(__file__).resolve().parent.parent / \"assets\" / \"editor.html\"\nEDITOR_FILE = \"alethe-playbooks.html\"\nEDITOR_PLAYBOOKS_SLOT = \"\\n__PLAYBOOKS_YAML__\\n\"\nEDITOR_USAGE_SLOT = '\u003cscript type=\"application/json\" id=\"live-usage\">{}\u003c/script>'\nEDITOR_PRICES_SLOT = '\u003cscript type=\"application/json\" id=\"model-prices\">{}\u003c/script>'\nEDITOR_WINDOWS = {\"claude\": {\"5h\": \"short\", \"week\": \"week\", \"opus\": \"opus\"}, \"codex\": {\"5h\": \"short\", \"week\": \"week\"}}\n\n\ndef editor_usage(usage: Dict[str, Any], providers: Sequence[str], now: Optional[dt.datetime] = None) -> Dict[str, Any]:\n current = now or dt.datetime.now(dt.timezone.utc)\n live = {}\n for provider, keys in EDITOR_WINDOWS.items():\n reading = usage[\"providers\"].get(provider, {})\n if provider not in providers or reading.get(\"missing\", True):\n continue\n windows = reading[\"windows\"]\n entry: Dict[str, Any] = {key: round(windows[name][\"usedPercent\"]) if name in windows else 0 for name, key in keys.items()}\n week = windows.get(\"week\")\n reset = timestamp(week[\"resetsAt\"]) if week and not week[\"reset\"] else None\n entry[\"reset\"] = max(1, math.ceil((reset - current).total_seconds() / 3600)) if reset else None\n entry[\"expires\"] = None\n live[provider] = entry\n return live\n\n\ndef script_safe(text: str) -> str:\n return re.sub(r\"\u003c/(script)\", r\"\u003c\\\\/\\1\", text, flags=re.IGNORECASE)\n\n\ndef write_editor(policy_path: Path, target: Optional[Path] = None) -> Dict[str, Any]:\n text, parsed = loaded_playbooks(policy_path)\n try:\n usage = editor_usage(usage_payload(policy_path), parsed.get(\"providers\", []))\n except PlanError:\n # a broken usage cache only drops the live quota preset\n usage = {}\n try:\n models = load_prices(policy_path)[\"models\"]\n prices = {key: {\"input\": entry[\"input\"], \"output\": entry[\"output\"]} for key, entry in models.items()}\n except PlanError:\n prices = {}\n page = read_playbooks(EDITOR_TEMPLATE)\n for slot in (EDITOR_PLAYBOOKS_SLOT, EDITOR_USAGE_SLOT, EDITOR_PRICES_SLOT):\n if page.count(slot) != 1:\n raise PlanError(f\"the editor template at {EDITOR_TEMPLATE} has no {slot.strip()} slot\")\n page = page.replace(EDITOR_PLAYBOOKS_SLOT, \"\\n\" + script_safe(text.rstrip(\"\\n\")) + \"\\n\", 1)\n page = page.replace(EDITOR_USAGE_SLOT, EDITOR_USAGE_SLOT.replace(\"{}\", script_safe(json.dumps(usage))), 1)\n page = page.replace(EDITOR_PRICES_SLOT, EDITOR_PRICES_SLOT.replace(\"{}\", script_safe(json.dumps(prices))), 1)\n path = (target or policy_path.with_name(EDITOR_FILE)).expanduser().resolve()\n atomic_write_text(path, page)\n return {\"path\": str(path), \"playbooks\": str(effective_playbooks_path(policy_path)),\n \"saveTo\": str(user_playbooks_path(policy_path)), \"liveUsage\": sorted(usage)}\n\n\nREPORT_TEMPLATE = EDITOR_TEMPLATE.with_name(\"report.html\")\nREPORT_PAGE_FILE = \"report.html\"\nREPORT_SLOT = \"\\n__REPORT_JSON__\\n\"\n\n\ndef write_report_page(report: Dict[str, Any], target: Path) -> Path:\n page = read_playbooks(REPORT_TEMPLATE)\n if page.count(REPORT_SLOT) != 1:\n raise PlanError(f\"the report template at {REPORT_TEMPLATE} has no {REPORT_SLOT.strip()} slot\")\n data = json.dumps(report, ensure_ascii=False).replace(\"\u003c/\", \"\u003c\\\\/\")\n path = target.expanduser().resolve()\n atomic_write_text(path, page.replace(REPORT_SLOT, \"\\n\" + data + \"\\n\", 1))\n return path\n\n\ndef open_page(path: Path) -> bool:\n try:\n if os.name == \"nt\":\n os.startfile(str(path))\n return True\n opener = shutil.which(\"open\" if sys.platform == \"darwin\" else \"xdg-open\")\n if not opener:\n return False\n subprocess.Popen([opener, str(path)], stdin=subprocess.DEVNULL, stdout=subprocess.DEVNULL,\n stderr=subprocess.DEVNULL, start_new_session=True)\n return True\n except OSError:\n return False\n\n\ndef worker_windows(worker: Dict[str, Any]) -> Tuple[str, ...]:\n if worker[\"provider\"] == \"claude\" and \"opus\" in str(worker.get(\"model\") or \"\").lower():\n return (*MAIN_QUOTA_WINDOWS, \"opus\")\n return MAIN_QUOTA_WINDOWS\n\n\ndef quota_refusal(workers: Sequence[Dict[str, Any]], policy_path: Path) -> Optional[str]:\n usage = usage_payload(policy_path)\n for worker in workers:\n provider = worker[\"provider\"]\n if provider not in QUOTA_PROVIDERS:\n continue\n reading = usage[\"providers\"][provider][\"windows\"]\n blocked = [(name, reading[name]) for name in worker_windows(worker) if name in reading and reading[name][\"blocked\"]]\n if blocked:\n name, window = max(blocked, key=lambda item: item[1][\"usedPercent\"])\n return (f\"{provider} is at {window['usedPercent']:g}% of its {name} window, at or past the \"\n f\"{usage['stopAt']:g}% stop line (resets {local_clock(window['resetsAt'])}); ask the user to wait \"\n \"for the reset, move the work to another provider, or confirm --allow-low-quota\")\n return None\n\n\ndef probe_claude_quota(policy_path: Path) -> Optional[Dict[str, Any]]:\n worker = {\"provider\": \"claude\", \"mode\": \"read-only\", \"model\": CLAUDE_PROBE_MODEL, \"effort\": \"low\",\n \"resume\": None, \"mcp\": False}\n with tempfile.TemporaryDirectory(prefix=\"alethe-usage-\") as folder:\n cwd = Path(folder)\n args, _, _, _ = build_invocation(worker, cwd, cwd, {}, threading.Lock(), CLAUDE_PROBE_SECONDS)\n try:\n completed = subprocess.run(args, input=CLAUDE_PROBE_PROMPT.encode(\"utf-8\"), cwd=cwd,\n env=worker_environment(), stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,\n timeout=CLAUDE_PROBE_SECONDS)\n except subprocess.TimeoutExpired as error:\n raise PlanError(f\"Claude did not answer within {CLAUDE_PROBE_SECONDS} seconds\") from error\n except OSError as error:\n raise PlanError(f\"could not start Claude: {error}\") from error\n quota = claude_quota(json_values(completed.stdout.decode(\"utf-8\", errors=\"replace\"), \"jsonl\"), now_iso())\n save_usage(policy_path, quota)\n return quota\n\n\ndef quota_samples(\n workers: Sequence[Dict[str, Any]], outcomes: Sequence[Dict[str, Any]],\n before: Dict[str, Any], after: Dict[str, Any], started: dt.datetime,\n) -> List[Dict[str, Any]]:\n ran = {outcome[\"id\"] for outcome in outcomes if outcome[\"status\"] != \"skipped\"}\n samples = []\n for provider in QUOTA_PROVIDERS:\n mine = [worker for worker in workers if worker[\"provider\"] == provider and worker[\"id\"] in ran]\n start, end = before.get(provider), after.get(provider)\n if not mine or not start or not end:\n continue\n opened, closed = timestamp(start[\"observedAt\"]), timestamp(end[\"observedAt\"])\n # Claude readings come only from workers, so an old one would hide the lead's own use\n if closed \u003c= opened or (provider == \"claude\" and (started - opened).total_seconds() > USAGE_STALE_MINUTES * 60):\n continue\n windows = {}\n for name in MAIN_QUOTA_WINDOWS:\n first, last = start[\"windows\"].get(name), end[\"windows\"].get(name)\n if first and last and abs((timestamp(first[\"resetsAt\"]) - timestamp(last[\"resetsAt\"])).total_seconds()) \u003c= QUOTA_WINDOW_SLACK_SECONDS:\n windows[name] = round(max(0.0, last[\"usedPercent\"] - first[\"usedPercent\"]) / len(mine), 4)\n if windows:\n models = {worker[\"model\"] for worker in mine}\n samples.append({\"provider\": provider, \"model\": models.pop() if len(models) == 1 else None,\n \"workers\": len(mine), \"observedAt\": end[\"observedAt\"], \"perWorker\": windows})\n return samples\n\n\ndef quota_forecast(workers: Sequence[Dict[str, Any]], policy_path: Path, usage: Dict[str, Any]) -> Tuple[Dict[str, Any], List[str]]:\n samples = usage_file(policy_path)[\"samples\"]\n forecast, warnings = {}, []\n for provider in QUOTA_PROVIDERS:\n mine = [worker for worker in workers if worker[\"provider\"] == provider]\n if not mine:\n continue\n reading = usage[\"providers\"][provider]\n history = [sample for sample in samples if sample[\"provider\"] == provider]\n entry: Dict[str, Any] = {\"workers\": len(mine), \"samples\": len(history), \"windows\": {}}\n for name in MAIN_QUOTA_WINDOWS:\n if name not in reading[\"windows\"]:\n continue\n window = reading[\"windows\"][name]\n shares: Optional[List[float]] = []\n for worker in mine:\n same = [sample[\"perWorker\"][name] for sample in history if sample[\"model\"] == worker[\"model\"] and name in sample[\"perWorker\"]]\n pool = same if len(same) >= QUOTA_FORECAST_MIN_SAMPLES else [\n sample[\"perWorker\"][name] for sample in history if name in sample[\"perWorker\"]]\n if len(pool) \u003c QUOTA_FORECAST_MIN_SAMPLES:\n shares = None\n break\n recent = sorted(pool[-QUOTA_FORECAST_SAMPLES:])\n shares.append(recent[len(recent) // 2])\n projected = round(sum(shares), 1) if shares is not None else None\n after = round(window[\"usedPercent\"] + projected, 1) if projected is not None else None\n entry[\"windows\"][name] = {\"usedPercent\": window[\"usedPercent\"], \"projectedPercent\": projected,\n \"afterPercent\": after, \"resetsAt\": window[\"resetsAt\"], \"stale\": window[\"stale\"]}\n if after is not None and after >= usage[\"stopAt\"] and not window[\"blocked\"]:\n warnings.append(f\"{provider} may use about {projected:g}% of its {name} window in this wave, from \"\n f\"{len(history)} past waves: {window['usedPercent']:g}% now would reach {after:g}%, past the \"\n f\"{usage['stopAt']:g}% stop line before the {local_clock(window['resetsAt'])} reset; ask the user first\")\n forecast[provider] = entry\n return forecast, warnings\n\n\ndef usage_summary(payload: Dict[str, Any]) -> List[str]:\n lines = []\n for provider, reading in payload[\"providers\"].items():\n if provider not in QUOTA_PROVIDERS and reading[\"missing\"]:\n continue\n parts = []\n for name, window in reading[\"windows\"].items():\n reset = timestamp(window[\"resetsAt\"]).astimezone().strftime(\"%H:%M\")\n parts.append(f\"{name} {window['usedPercent']:g}% (reset passed)\" if window[\"reset\"]\n else f\"{name} {window['usedPercent']:g}% (resets {reset})\")\n detail = \"; \".join(parts) if parts else \"no reading\"\n flags = [name for name in (\"low\", \"blocked\", \"stale\") if reading[name]]\n lines.append(f\"Usage {provider}: {detail}\" + (f\" [{', '.join(flags)}]\" if flags else \"\"))\n return lines\n\n\ndef worker_quota(provider: str, values: Sequence[Any], policy_path: Path) -> Optional[Dict[str, Any]]:\n quota = claude_quota(values, now_iso()) if provider == \"claude\" else None\n if provider == \"codex\":\n quota = codex_quota(extract_session(values))\n save_usage(policy_path, quota)\n return quota\n\n\nPRICE_RANKS = (\"low\", \"mid\", \"high\")\nCLI_COST_LOW = 1\nCLI_COST_HIGH = 3\n\n\ndef load_prices(policy_path: Path) -> Dict[str, Any]:\n bundled = Path(__file__).resolve().parent.parent / PRICES_FILE\n raw = read_json(bundled)\n if not isinstance(raw, dict) or raw.get(\"version\") != 1 or raw.get(\"currency\") != \"USD\" or raw.get(\"unit\") != \"per 1M tokens\":\n raise PlanError(f\"invalid price table: {bundled}\")\n models = raw.get(\"models\")\n if not isinstance(models, dict):\n raise PlanError(f\"{bundled} must contain a models object\")\n models = dict(models)\n billing = {}\n override = policy_path.with_name(USER_PRICES_FILE)\n if override.exists():\n user = read_json(override)\n if not isinstance(user, dict) or not isinstance(user.get(\"models\", {}), dict) or not isinstance(user.get(\"billing\", {}), dict):\n raise PlanError(f\"{override} must contain models and billing objects\")\n for key, entry in user.get(\"models\", {}).items():\n if not isinstance(entry, dict):\n raise PlanError(f\"price for {key} must be an object\")\n models[key] = {**models.get(key, {}), **entry}\n billing = user.get(\"billing\", {})\n for provider, mode in billing.items():\n if mode not in (\"subscription\", \"api\"):\n raise PlanError(f\"billing for {provider} must be subscription or api\")\n for key, entry in models.items():\n if \"/\" not in key or not isinstance(entry, dict):\n raise PlanError(f\"invalid model price: {key}\")\n for name in (\"input\", \"output\", \"cacheRead\", \"cacheWrite\"):\n value = number(entry.get(name))\n if name not in entry and name not in (\"input\", \"output\"):\n continue\n if value is None or not math.isfinite(value) or value \u003c 0:\n raise PlanError(f\"price {key}.{name} must be a nonnegative number\")\n try:\n dt.date.fromisoformat(entry[\"asOf\"])\n except (KeyError, TypeError, ValueError):\n raise PlanError(f\"price {key}.asOf must be YYYY-MM-DD\")\n if not isinstance(entry.get(\"source\"), str) or not entry[\"source\"].startswith(\"https://\"):\n raise PlanError(f\"price {key}.source must be an HTTPS URL\")\n return {**raw, \"models\": models, \"billing\": billing, \"overridePath\": str(override)}\n\n\ndef playbook_model_inventory(policy_path: Path) -> Dict[str, Dict[str, Optional[float]]]:\n text = read_playbooks(effective_playbooks_path(policy_path))\n inventory: Dict[str, Dict[str, Optional[float]]] = {}\n section = provider = model = None\n provider_indent = -1\n for line in text.splitlines():\n clean = line.split(\"#\", 1)[0].rstrip()\n if not clean.strip():\n continue\n indent = len(clean) - len(clean.lstrip())\n if indent == 0:\n section = clean.split(\":\", 1)[0]\n provider = model = None\n continue\n if section not in (\"models\", \"tiers\"):\n continue\n content = clean.strip()\n head = re.match(r\"([\\w.-]+):\\s*(.*)\", content)\n if head and (head.group(1) in BUILTIN_PROVIDERS or head.group(1) in PROVIDER_ALIASES or (section == \"models\" and indent == 2)):\n provider = PROVIDER_ALIASES.get(head.group(1), head.group(1))\n provider_indent = indent\n model = None\n inventory.setdefault(provider, {})\n rest = head.group(2)\n if section == \"models\" and rest.startswith(\"[\"):\n for item in re.findall(r\"\\{[^}]*\\}|[^,\\[\\]]+\", rest[1:-1]):\n match = re.search(r\"\\bid:\\s*['\\\"]?([\\w./:-]+)\", item)\n name = match.group(1) if match else item.strip().strip(\"'\\\"\")\n cost = re.search(r\"\\b(?:cost|costMultiplier):\\s*([\\d.]+)\", item)\n if name:\n inventory[provider][name] = float(cost.group(1)) if cost else None\n if provider is None:\n continue\n match = re.search(r\"\\b(?:model|id):\\s*['\\\"]?([\\w./:-]+)\", content)\n if match:\n model = match.group(1)\n inventory[provider].setdefault(model, None)\n if section == \"models\" and indent > provider_indent and not match:\n item = re.match(r\"-\\s+['\\\"]?([\\w./:-]+)['\\\"]?\\s*$\", content)\n mapped = re.match(r\"['\\\"]?([\\w./-]+)['\\\"]?:\\s*\\{\", content)\n if item or mapped:\n model = (item or mapped).group(1)\n inventory[provider].setdefault(model, None)\n cost = re.search(r\"\\b(?:cost|costMultiplier):\\s*([\\d.]+)\", content)\n if model and cost:\n inventory[provider][model] = float(cost.group(1))\n return inventory\n\n\ndef models_payload(policy_path: Path, now: Optional[dt.datetime] = None) -> Dict[str, Any]:\n current = now or dt.datetime.now(dt.timezone.utc)\n table = load_prices(policy_path)\n inventory = playbook_model_inventory(policy_path)\n cache = read_optional_json(codex_home() / \"models_cache.json\")\n cached = cache.get(\"models\", []) if isinstance(cache, dict) else []\n for entry in cached if isinstance(cached, list) else []:\n name = entry.get(\"slug\") or entry.get(\"id\") if isinstance(entry, dict) else None\n if isinstance(name, str):\n inventory.setdefault(\"codex\", {}).setdefault(name, None)\n for key in table[\"models\"]:\n provider, model = key.split(\"/\", 1)\n inventory.setdefault(provider, {}).setdefault(model, None)\n providers = {}\n for provider in dict.fromkeys((*BUILTIN_PROVIDERS, *inventory)):\n known = inventory.get(provider, {})\n scores = sorted({entry[\"input\"] + entry[\"output\"] for key, entry in table[\"models\"].items()\n if key.split(\"/\", 1)[0] == provider})\n models = []\n for model, multiplier in sorted(known.items()):\n price = table[\"models\"].get(f\"{provider}/{model}\")\n rank = None\n if price:\n score = price[\"input\"] + price[\"output\"]\n index = scores.index(score)\n rank = PRICE_RANKS[min(len(PRICE_RANKS)-1, index * len(PRICE_RANKS) // len(scores))]\n if multiplier is not None:\n rank = \"low\" if multiplier \u003c= CLI_COST_LOW else \"high\" if multiplier >= CLI_COST_HIGH else \"mid\"\n models.append({\"model\": model, \"input\": price[\"input\"] if price else None,\n \"output\": price[\"output\"] if price else None, \"cacheRead\": price.get(\"cacheRead\") if price else None,\n \"cacheWrite\": price.get(\"cacheWrite\") if price else None, \"rank\": rank,\n \"costMultiplier\": multiplier, \"asOf\": price[\"asOf\"] if price else None,\n \"stale\": (current.date() - dt.date.fromisoformat(price[\"asOf\"])).days > PRICE_STALE_DAYS if price else False,\n \"missing\": price is None, \"source\": price[\"source\"] if price else None})\n providers[provider] = {\"billing\": table[\"billing\"].get(provider), \"models\": models}\n return {\"version\": 1, \"currency\": table[\"currency\"], \"unit\": table[\"unit\"],\n \"overridePath\": table[\"overridePath\"], \"providers\": providers}\n\n\ndef preview_usage_cost(workers: Sequence[Dict[str, Any]], policy_path: Path) -> Dict[str, Any]:\n table = load_prices(policy_path)\n usage = usage_payload(policy_path, refresh=False)\n warnings = []\n estimates = []\n for worker in workers:\n provider = worker[\"provider\"]\n price = table[\"models\"].get(f\"{provider}/{worker['model']}\")\n estimates.append({\"listPriceEstimateUsd\": round(worker[\"estimatedPromptTokens\"] * price[\"input\"] / TOKENS_PER_MILLION, 8) if price else None,\n \"listPriceAsOf\": price[\"asOf\"] if price else None})\n reading = usage[\"providers\"].get(provider, {})\n if not reading.get(\"low\"):\n continue\n name = max((name for name in MAIN_QUOTA_WINDOWS if name in reading[\"windows\"]),\n key=lambda name: reading[\"windows\"][name][\"usedPercent\"])\n window = reading[\"windows\"][name]\n reset = timestamp(window[\"resetsAt\"]).astimezone().strftime(\"%H:%M\")\n warning = f\"{worker['id']}: {provider} is at {window['usedPercent']:g}% of its {name} window (resets {reset})\"\n alternatives = [(other, max((entry[\"windows\"][key][\"usedPercent\"] for key in MAIN_QUOTA_WINDOWS if key in entry[\"windows\"]), default=0))\n for other, entry in usage[\"providers\"].items() if other != provider and not entry[\"missing\"]\n and not entry[\"low\"] and not entry[\"stale\"]]\n if window[\"blocked\"]:\n warning += f\", past the {usage['stopAt']:g}% stop line, so run refuses it; ask the user\"\n if alternatives:\n other, percent = min(alternatives, key=lambda item: item[1])\n warning += f\"; consider {other} at {percent:g}%\"\n if reading[\"stale\"]:\n warning += \"; reading is stale\"\n warnings.append(warning)\n forecast, forecast_warnings = quota_forecast(workers, policy_path, usage)\n values = [entry[\"listPriceEstimateUsd\"] for entry in estimates]\n return {\"workers\": estimates, \"warnings\": warnings + forecast_warnings, \"quotaForecast\": forecast,\n \"total\": round(sum(values), 8) if values and all(value is not None for value in values) else None,\n \"asOf\": sorted({entry[\"listPriceAsOf\"] for entry in estimates if entry[\"listPriceAsOf\"]}),\n \"label\": \"Standard short-context API list-price estimate for prompt input only; excludes output, cache and provider-side context; not CLI-reported cost\"}\n\n\ndef normalized_strategy(value: Any, label: str) -> str:\n strategy = require_string(value, label).lower()\n strategy = STRATEGY_ALIASES.get(strategy, strategy)\n if strategy not in STRATEGIES:\n raise PlanError(f\"{label} must be one of {', '.join(STRATEGIES)}\")\n return strategy\n\n\ndef validate_provider_list(value: Any, label: str, *, allow_empty: bool) -> List[str]:\n if not isinstance(value, list):\n raise PlanError(f\"{label} must be an array\")\n providers: List[str] = []\n for raw in value:\n provider = require_string(raw, label).lower()\n provider = PROVIDER_ALIASES.get(provider, provider)\n if provider not in BUILTIN_PROVIDERS:\n raise PlanError(f\"{label} contains unsupported provider {provider}\")\n if provider not in providers:\n providers.append(provider)\n if not allow_empty and not providers:\n raise PlanError(f\"{label} must contain at least one provider\")\n return providers\n\n\ndef validate_budget(raw: Any, label: str) -> Dict[str, Any]:\n if raw is None:\n return {}\n if not isinstance(raw, dict):\n raise PlanError(f\"{label} must be an object\")\n unknown = set(raw) - set(BUDGET_FIELDS)\n if unknown:\n raise PlanError(f\"{label} has unknown fields: {', '.join(sorted(unknown))}\")\n budget: Dict[str, Any] = {}\n max_calls = raw.get(\"maxCalls\")\n if max_calls is not None:\n if not isinstance(max_calls, int) or isinstance(max_calls, bool):\n raise PlanError(f\"{label}.maxCalls must be an integer\")\n if not 1 \u003c= max_calls \u003c= MAX_BUDGET_CALLS:\n raise PlanError(f\"{label}.maxCalls must be between 1 and {MAX_BUDGET_CALLS}\")\n budget[\"maxCalls\"] = max_calls\n max_wall = raw.get(\"maxWallSeconds\")\n if max_wall is not None:\n if not isinstance(max_wall, int) or isinstance(max_wall, bool):\n raise PlanError(f\"{label}.maxWallSeconds must be an integer\")\n if not 1 \u003c= max_wall \u003c= 86400:\n raise PlanError(f\"{label}.maxWallSeconds must be between 1 and 86400\")\n budget[\"maxWallSeconds\"] = max_wall\n max_tokens = raw.get(\"maxEstimatedPromptTokens\")\n if max_tokens is not None:\n if not isinstance(max_tokens, int) or isinstance(max_tokens, bool) or max_tokens \u003c 1:\n raise PlanError(f\"{label}.maxEstimatedPromptTokens must be a positive integer\")\n budget[\"maxEstimatedPromptTokens\"] = max_tokens\n max_cost = raw.get(\"maxReportedCostUsd\")\n if max_cost is not None:\n if not isinstance(max_cost, (int, float)) or isinstance(max_cost, bool) or max_cost \u003c= 0:\n raise PlanError(f\"{label}.maxReportedCostUsd must be a positive number\")\n budget[\"maxReportedCostUsd\"] = max_cost\n max_reported = raw.get(\"maxReportedTokens\")\n if max_reported is not None:\n if not isinstance(max_reported, int) or isinstance(max_reported, bool) or max_reported \u003c 1:\n raise PlanError(f\"{label}.maxReportedTokens must be a positive integer\")\n budget[\"maxReportedTokens\"] = max_reported\n return budget\n\n\ndef estimated_prompt_tokens(text: str) -> int:\n \"\"\"Prompt-only chars/4 heuristic; provider-side context is additional and not estimable.\"\"\"\n return max(1, -(-len(text) // 4))\n\n\nRAW_TOKEN_PARTS = (\"input_tokens\", \"output_tokens\", \"cached_input_tokens\", \"cache_creation_input_tokens\", \"cache_read_input_tokens\")\n\n\ndef normalized_usage(usage: Dict[str, Any]) -> Dict[str, Any]:\n if not any(key in usage for key in TOKEN_PARTS + RAW_TOKEN_PARTS):\n return dict(usage)\n for key in TOKEN_PARTS + RAW_TOKEN_PARTS:\n if key not in usage:\n continue\n value = number(usage[key])\n if value is None or not math.isfinite(value) or value \u003c 0 or int(value) != value:\n raise PlanError(f\"usage.{key} must be a nonnegative integer\")\n if any(key in usage for key in TOKEN_PARTS):\n return {key: int(usage.get(key, 0)) for key in TOKEN_PARTS}\n cached = int(usage.get(\"cached_input_tokens\", 0))\n uncached = int(usage.get(\"input_tokens\", 0)) - cached\n if uncached \u003c 0:\n raise PlanError(\"usage.cached_input_tokens exceeds input_tokens\")\n return {\n \"inputTokens\": uncached,\n \"cacheReadTokens\": int(usage.get(\"cache_read_input_tokens\", cached)),\n \"cacheWriteTokens\": int(usage.get(\"cache_creation_input_tokens\", 0)),\n \"outputTokens\": int(usage.get(\"output_tokens\", 0)),\n }\n\n\ndef reported_total_tokens(usage: Any) -> Optional[int]:\n if not isinstance(usage, dict) or not usage:\n return None\n if any(key in usage for key in TOKEN_PARTS + (\"input_tokens\", \"output_tokens\", \"cached_input_tokens\",\n \"cache_creation_input_tokens\", \"cache_read_input_tokens\")):\n parts = normalized_usage(usage)\n return parts[\"inputTokens\"] + parts[\"cacheWriteTokens\"] + parts[\"outputTokens\"]\n for key in (\"total_tokens\", \"totalTokens\"):\n value = number(usage.get(key))\n if value is not None:\n return int(value)\n return None\n\n\ndef validate_model_tiers(raw: Any, label: str) -> Dict[str, Any]:\n if raw is None:\n return {}\n if not isinstance(raw, dict):\n raise PlanError(f\"{label} must be an object\")\n unknown = set(raw) - set(MODEL_TIER_KEYS)\n if unknown:\n raise PlanError(f\"{label} has unknown tiers: {', '.join(sorted(unknown))}\")\n tiers: Dict[str, Any] = {}\n for tier, mapping in raw.items():\n if not isinstance(mapping, dict):\n raise PlanError(f\"{label}.{tier} must be an object\")\n entries: Dict[str, Any] = {}\n for raw_provider, spec in mapping.items():\n provider = PROVIDER_ALIASES.get(str(raw_provider).lower(), str(raw_provider).lower())\n if provider not in BUILTIN_PROVIDERS:\n raise PlanError(f\"{label}.{tier} contains unsupported provider {provider}\")\n if not isinstance(spec, dict):\n raise PlanError(f\"{label}.{tier}.{provider} must be an object\")\n unknown_spec = set(spec) - {\"model\", \"effort\"}\n if unknown_spec:\n raise PlanError(\n f\"{label}.{tier}.{provider} has unknown fields: {', '.join(sorted(unknown_spec))}\"\n )\n model = optional_name(spec.get(\"model\"), f\"{label}.{tier}.{provider}.model\")\n effort = optional_name(spec.get(\"effort\"), f\"{label}.{tier}.{provider}.effort\")\n if not model and not effort:\n raise PlanError(f\"{label}.{tier}.{provider} must set model or effort\")\n entries[provider] = {\n key: value for key, value in ((\"model\", model), (\"effort\", effort)) if value\n }\n if entries:\n tiers[tier] = entries\n return tiers\n\n\ndef resolve_model_tier(\n worker: Dict[str, Any], policy: Dict[str, Any], settings: Dict[str, Any]\n) -> Dict[str, Any]:\n tier = worker[\"tier\"]\n provider = worker[\"provider\"]\n spec = {\n **settings[\"tiers\"].get(tier, {}).get(provider, {}),\n **policy[\"modelTiers\"].get(tier, {}).get(provider, {}),\n **{key: worker[key] for key in (\"model\", \"effort\") if worker[key]},\n }\n parts = \", \".join(f\"{key} {value}\" for key, value in spec.items())\n note = f\"tier {tier} on {provider}: {parts}\" if parts else f\"tier {tier}: no model configured for {provider}\"\n return {**worker, \"model\": spec.get(\"model\"), \"effort\": spec.get(\"effort\"), \"modelNote\": note}\n\n\ndef validate_policy(raw: Any) -> Dict[str, Any]:\n if not isinstance(raw, dict):\n raise PlanError(\"policy must be a JSON object\")\n unknown = set(raw) - set(DEFAULT_POLICY)\n if unknown:\n raise PlanError(f\"unknown policy fields: {', '.join(sorted(unknown))}\")\n if raw.get(\"version\", POLICY_VERSION) != POLICY_VERSION:\n raise PlanError(f\"policy.version must be {POLICY_VERSION}\")\n onboarded = raw.get(\"onboarded\", False)\n if not isinstance(onboarded, bool):\n raise PlanError(\"policy.onboarded must be a boolean\")\n backend = raw.get(\"backend\", DEFAULT_POLICY[\"backend\"])\n if backend not in (\"auto\", \"skill\", \"mcp\"):\n raise PlanError(\"policy.backend must be auto, skill or mcp\")\n provider_switch = raw.get(\"providerSwitch\", DEFAULT_POLICY[\"providerSwitch\"])\n if provider_switch not in (\"ask\", \"automatic\", \"never\"):\n raise PlanError(\"policy.providerSwitch must be ask, automatic or never\")\n max_workers = raw.get(\"maxWorkers\", DEFAULT_POLICY[\"maxWorkers\"])\n max_concurrent = raw.get(\"maxConcurrent\", DEFAULT_POLICY[\"maxConcurrent\"])\n for label, value in ((\"maxWorkers\", max_workers), (\"maxConcurrent\", max_concurrent)):\n if not isinstance(value, int) or isinstance(value, bool) or not 1 \u003c= value \u003c= MAX_WORKERS:\n raise PlanError(f\"policy.{label} must be between 1 and {MAX_WORKERS}\")\n return {\n \"version\": POLICY_VERSION,\n \"onboarded\": onboarded,\n \"backend\": backend,\n \"defaultStrategy\": normalized_strategy(\n raw.get(\"defaultStrategy\", DEFAULT_POLICY[\"defaultStrategy\"]),\n \"policy.defaultStrategy\",\n ),\n \"providerSwitch\": provider_switch,\n \"providerOrder\": validate_provider_list(\n raw.get(\"providerOrder\", DEFAULT_POLICY[\"providerOrder\"]),\n \"policy.providerOrder\",\n allow_empty=False,\n ),\n \"conserveProviders\": validate_provider_list(\n raw.get(\"conserveProviders\", DEFAULT_POLICY[\"conserveProviders\"]),\n \"policy.conserveProviders\",\n allow_empty=True,\n ),\n \"maxWorkers\": max_workers,\n \"maxConcurrent\": max_concurrent,\n \"budget\": validate_budget(raw.get(\"budget\"), \"policy.budget\"),\n \"modelTiers\": validate_model_tiers(raw.get(\"modelTiers\"), \"policy.modelTiers\"),\n }\n\n\ndef load_policy(path: Path) -> Dict[str, Any]:\n if not path.exists():\n return dict(DEFAULT_POLICY)\n return validate_policy(read_json(path))\n\n\ndef save_policy(path: Path, policy: Dict[str, Any]) -> None:\n atomic_write_json(path, validate_policy(policy))\n\n\ndef available_builtin_providers() -> List[str]:\n commands = {\"agy\": \"agy\"}\n return [\n provider\n for provider in BUILTIN_PROVIDERS\n if shutil.which(commands.get(provider, provider)) is not None\n ]\n\n\ndef choose_provider(requested: str, worker: Dict[str, Any], policy: Dict[str, Any]) -> str:\n if requested == \"custom\":\n return requested\n available = available_builtin_providers()\n conserved = set(policy[\"conserveProviders\"])\n read_only = worker.get(\"mode\") == \"read-only\"\n candidates = [\n provider\n for provider in policy[\"providerOrder\"]\n if provider in available\n and provider not in conserved\n and (not read_only or PROVIDER_CAPABILITIES[provider][\"readOnly\"])\n ]\n if worker.get(\"resume\") and requested == \"auto\":\n raise PlanError(\n f\"worker {worker['id']} must name the provider that owns its resumed session\"\n )\n if requested == \"auto\":\n if worker.get(\"model\") or worker.get(\"effort\"):\n raise PlanError(f\"worker {worker['id']} cannot combine provider auto with model or effort\")\n if not candidates:\n needed = \"read-only \" if read_only else \"\"\n raise PlanError(f\"no unconserved built-in provider can run a {needed}worker\")\n return candidates[0]\n if requested not in conserved:\n return requested\n switch = policy[\"providerSwitch\"]\n if switch == \"never\":\n return requested\n if switch == \"ask\":\n raise PlanError(\n f\"provider {requested} is marked for conservation; choose another provider or enable automatic switching\"\n )\n alternatives = [provider for provider in candidates if provider != requested]\n if not alternatives:\n raise PlanError(f\"no available provider can replace conserved provider {requested}\")\n if worker.get(\"model\") or worker.get(\"effort\") or worker.get(\"resume\"):\n raise PlanError(\n f\"worker {worker['id']} pins provider-specific state and cannot switch automatically\"\n )\n return alternatives[0]\n\n\ndef infer_strategy(workers: Sequence[Dict[str, Any]]) -> Tuple[str, str]:\n if len(workers) == 1:\n if workers[0].get(\"resume\"):\n return \"rescue\", \"one resumed worker session\"\n return \"economy\", \"one new worker\"\n if all(worker[\"mode\"] == \"read-only\" for worker in workers):\n return \"research\", f\"{len(workers)} read-only workers\"\n writers = [worker for worker in workers if worker[\"mode\"] == \"write\"]\n if len(writers) >= 2 and all(worker[\"isolate\"] for worker in writers):\n return \"parallel\", f\"{len(writers)} isolated writers\"\n return \"balanced\", \"mixed modes or a non-isolated writer\"\n\n\ndef enforce_strategy(\n requested: str,\n workers: Sequence[Dict[str, Any]],\n requested_concurrency: Optional[int],\n policy: Dict[str, Any],\n) -> Tuple[str, int]:\n strategy = infer_strategy(workers)[0] if requested == \"auto\" else requested\n rules = STRATEGIES[strategy]\n max_workers = min(rules[\"maxWorkers\"], policy[\"maxWorkers\"])\n if len(workers) > max_workers:\n raise PlanError(f\"strategy {strategy} allows at most {max_workers} workers\")\n if strategy == \"research\" and any(worker[\"mode\"] != \"read-only\" for worker in workers):\n raise PlanError(\"strategy research accepts read-only workers only\")\n if strategy == \"parallel\" and any(\n worker[\"mode\"] == \"write\" and not worker[\"isolate\"] for worker in workers\n ):\n raise PlanError(\"strategy parallel requires every writer to use isolate: true\")\n if strategy == \"rescue\" and (len(workers) != 1 or not workers[0].get(\"resume\")):\n raise PlanError(\"strategy rescue requires exactly one worker with resume\")\n concurrency = requested_concurrency or rules[\"maxConcurrent\"]\n concurrency = min(concurrency, rules[\"maxConcurrent\"], policy[\"maxConcurrent\"])\n return strategy, concurrency\n\n\ndef require_string(value: Any, label: str) -> str:\n if not isinstance(value, str) or not value.strip():\n raise PlanError(f\"{label} must be a non-empty string\")\n return value.strip()\n\n\ndef optional_name(value: Any, label: str) -> Optional[str]:\n if value is None:\n return None\n text = require_string(value, label)\n if not SAFE_NAME_PATTERN.fullmatch(text) or text.startswith(\"-\"):\n raise PlanError(f\"{label} contains unsafe characters\")\n return text\n\n\ndef optional_note(value: Any, label: str, limit: int) -> Optional[str]:\n if value is None:\n return None\n text = \" \".join(require_string(value, label).split())\n if len(text) > limit:\n raise PlanError(f\"{label} must be at most {limit} characters\")\n return text\n\n\ndef normalized_provider(value: Any) -> str:\n provider = require_string(value, \"worker.provider\").lower()\n provider = PROVIDER_ALIASES.get(provider, provider)\n if provider != \"auto\" and provider not in PROVIDER_CAPABILITIES:\n raise PlanError(f\"unsupported provider {provider}; use provider custom for another CLI\")\n return provider\n\n\ndef validate_relative_path(value: Any, label: str) -> str:\n text = require_string(value, label).replace(\"\\\\\", \"/\")\n path = Path(text)\n if any(character in text for character in UNSAFE_SEED_CHARACTERS) or path.is_absolute() or any(part in (\"\", \".\", \"..\") for part in path.parts):\n raise PlanError(f\"{label} must be a safe relative path\")\n return text\n\n\ndef load_task(worker: Dict[str, Any], plan_dir: Path) -> str:\n inline = worker.get(\"task\")\n task_file = worker.get(\"taskFile\")\n if (inline is None) == (task_file is None):\n raise PlanError(f\"worker {worker.get('id', '?')} must set exactly one of task or taskFile\")\n if inline is not None:\n task = require_string(inline, \"worker.task\")\n else:\n path = Path(require_string(task_file, \"worker.taskFile\"))\n if not path.is_absolute():\n path = plan_dir / path\n try:\n task = path.read_text(encoding=\"utf-8\").strip()\n except OSError as error:\n raise PlanError(f\"cannot read task file {path}: {error}\") from error\n if not task:\n raise PlanError(f\"task file {path} is empty\")\n return f\"{task.rstrip()}\\n\\n{WORKER_CONTRACT}\\n\"\n\n\ndef validate_plan(\n raw: Any,\n plan_path: Path,\n policy: Optional[Dict[str, Any]] = None,\n strategy_override: Optional[str] = None,\n policy_path: Optional[Path] = None,\n) -> Dict[str, Any]:\n settings = playbook_settings(policy_path)\n policy = validate_policy(policy or DEFAULT_POLICY)\n if not isinstance(raw, dict):\n raise PlanError(\"plan must be a JSON object\")\n unknown = set(raw) - {\n \"version\",\n \"title\",\n \"why\",\n \"playbook\",\n \"phase\",\n \"strategy\",\n \"preset\",\n \"maxConcurrent\",\n \"defaultTimeoutSeconds\",\n \"budget\",\n \"workers\",\n }\n if unknown:\n raise PlanError(f\"unknown plan fields: {', '.join(sorted(unknown))}\")\n if raw.get(\"version\") != PLAN_VERSION:\n raise PlanError(f\"plan.version must be {PLAN_VERSION}\")\n title = optional_note(raw.get(\"title\"), \"plan.title\", NOTE_LIMITS[\"title\"])\n why = optional_note(raw.get(\"why\"), \"plan.why\", NOTE_LIMITS[\"why\"])\n playbook = optional_name(raw.get(\"playbook\"), \"plan.playbook\")\n phase = optional_name(raw.get(\"phase\"), \"plan.phase\")\n if strategy_override is None and \"strategy\" in raw and \"preset\" in raw:\n raise PlanError(\"use strategy, not strategy and legacy preset together\")\n strategy_value = strategy_override or raw.get(\n \"strategy\", raw.get(\"preset\", policy[\"defaultStrategy\"])\n )\n requested_strategy = normalized_strategy(strategy_value, \"plan.strategy\")\n requested_concurrency = raw.get(\"maxConcurrent\")\n if requested_concurrency is not None:\n if not isinstance(requested_concurrency, int) or isinstance(requested_concurrency, bool):\n raise PlanError(\"maxConcurrent must be an integer\")\n if not 1 \u003c= requested_concurrency \u003c= MAX_WORKERS:\n raise PlanError(f\"maxConcurrent must be between 1 and {MAX_WORKERS}\")\n default_timeout = raw.get(\"defaultTimeoutSeconds\", DEFAULT_TIMEOUT_SECONDS)\n if not isinstance(default_timeout, int) or isinstance(default_timeout, bool):\n raise PlanError(\"defaultTimeoutSeconds must be an integer\")\n if not 1 \u003c= default_timeout \u003c= 86400:\n raise PlanError(\"defaultTimeoutSeconds must be between 1 and 86400\")\n workers = raw.get(\"workers\")\n if not isinstance(workers, list) or not workers:\n raise PlanError(\"workers must be a non-empty array\")\n if len(workers) > MAX_WORKERS:\n raise PlanError(f\"a plan may contain at most {MAX_WORKERS} workers\")\n\n seen = set()\n normalized_workers: List[Dict[str, Any]] = []\n for index, item in enumerate(workers):\n if not isinstance(item, dict):\n raise PlanError(f\"workers[{index}] must be an object\")\n unknown = set(item) - {\n \"id\",\n \"why\",\n \"provider\",\n \"mode\",\n \"cwd\",\n \"task\",\n \"taskFile\",\n \"timeoutSeconds\",\n \"isolate\",\n \"seedPaths\",\n \"model\",\n \"effort\",\n \"tier\",\n \"resume\",\n \"command\",\n \"args\",\n \"promptTransport\",\n \"outputFormat\",\n \"allowUnsafe\",\n \"mcp\",\n \"idleTimeoutSeconds\",\n }\n if unknown:\n raise PlanError(\n f\"worker {item.get('id', index)} has unknown fields: {', '.join(sorted(unknown))}\"\n )\n worker_id = require_string(item.get(\"id\"), f\"workers[{index}].id\")\n if not ID_PATTERN.fullmatch(worker_id):\n raise PlanError(f\"worker id {worker_id!r} is not safe\")\n if worker_id in seen:\n raise PlanError(f\"duplicate worker id {worker_id}\")\n seen.add(worker_id)\n requested_provider = normalized_provider(item.get(\"provider\"))\n mode = item.get(\"mode\", \"read-only\")\n if mode not in (\"read-only\", \"write\"):\n raise PlanError(f\"worker {worker_id} mode must be read-only or write\")\n cwd = Path(require_string(item.get(\"cwd\"), f\"worker {worker_id}.cwd\")).expanduser()\n if not cwd.is_absolute():\n cwd = (plan_path.parent / cwd).resolve()\n if not cwd.is_dir():\n raise PlanError(f\"worker {worker_id} cwd is not a directory: {cwd}\")\n timeout = item.get(\"timeoutSeconds\", default_timeout)\n if not isinstance(timeout, int) or isinstance(timeout, bool) or not 1 \u003c= timeout \u003c= 86400:\n raise PlanError(f\"worker {worker_id} timeoutSeconds must be between 1 and 86400\")\n isolate = item.get(\"isolate\", False)\n if not isinstance(isolate, bool):\n raise PlanError(f\"worker {worker_id} isolate must be a boolean\")\n if isolate and mode != \"write\":\n raise PlanError(f\"worker {worker_id} only needs isolate in write mode\")\n seed_paths = item.get(\"seedPaths\", [])\n if not isinstance(seed_paths, list):\n raise PlanError(f\"worker {worker_id} seedPaths must be an array\")\n seed_paths = [\n validate_relative_path(value, f\"worker {worker_id}.seedPaths\") for value in seed_paths\n ]\n if seed_paths and not isolate:\n raise PlanError(f\"worker {worker_id} seedPaths requires isolate\")\n model = optional_name(item.get(\"model\"), f\"worker {worker_id}.model\")\n effort = optional_name(item.get(\"effort\"), f\"worker {worker_id}.effort\")\n tier = item.get(\"tier\", \"standard\")\n if tier not in MODEL_TIERS:\n raise PlanError(f\"worker {worker_id} tier must be one of {', '.join(MODEL_TIERS)}\")\n if tier != \"standard\" and (model or effort):\n raise PlanError(\n f\"worker {worker_id} cannot combine tier {tier} with an explicit model or effort\"\n )\n resume = optional_name(item.get(\"resume\"), f\"worker {worker_id}.resume\")\n if resume and isolate:\n raise PlanError(f\"worker {worker_id} cannot combine resume with isolate\")\n mcp = item.get(\"mcp\", False)\n if not isinstance(mcp, bool):\n raise PlanError(f\"worker {worker_id} mcp must be a boolean\")\n idle_timeout = item.get(\"idleTimeoutSeconds\")\n if idle_timeout is not None and (\n not isinstance(idle_timeout, int) or isinstance(idle_timeout, bool) or not 5 \u003c= idle_timeout \u003c= 86400\n ):\n raise PlanError(f\"worker {worker_id} idleTimeoutSeconds must be between 5 and 86400\")\n task = load_task(item, plan_path.parent)\n normalized = {\n \"id\": worker_id,\n \"why\": optional_note(item.get(\"why\"), f\"worker {worker_id}.why\", NOTE_LIMITS[\"workerWhy\"]),\n \"provider\": requested_provider,\n \"requestedProvider\": requested_provider,\n \"mode\": mode,\n \"cwd\": str(cwd),\n \"timeoutSeconds\": timeout,\n \"isolate\": isolate,\n \"seedPaths\": seed_paths,\n \"model\": model,\n \"effort\": effort,\n \"tier\": tier,\n \"resume\": resume,\n \"mcp\": mcp,\n \"idleTimeoutSeconds\": idle_timeout,\n \"task\": task,\n }\n if requested_provider == \"custom\":\n normalized.update(validate_custom(item, worker_id))\n chosen_provider = choose_provider(requested_provider, normalized, policy)\n normalized[\"provider\"] = chosen_provider\n if chosen_provider == requested_provider:\n normalized[\"routingNote\"] = None\n elif requested_provider == \"auto\":\n normalized[\"routingNote\"] = \"selected by provider order\"\n else:\n normalized[\"routingNote\"] = (\n f\"conserved provider {requested_provider} replaced by providerSwitch=automatic\"\n )\n prompt_in_argument = chosen_provider in (\"agy\", \"kimi\") or (\n chosen_provider == \"custom\" and normalized[\"promptTransport\"] == \"argument\"\n )\n if prompt_in_argument and len(task) > ARGUMENT_PROMPT_LIMIT:\n raise PlanError(\n f\"worker {worker_id} task is {len(task)} characters, but {chosen_provider} takes its prompt \"\n f\"as an argument limited to {ARGUMENT_PROMPT_LIMIT} here; shorten it or point to files\"\n )\n normalized[\"estimatedPromptTokens\"] = estimated_prompt_tokens(task)\n normalized_workers.append(normalized)\n\n budget = {**policy[\"budget\"], **validate_budget(raw.get(\"budget\"), \"plan.budget\")}\n token_limit = budget.get(\"maxEstimatedPromptTokens\")\n if token_limit is not None:\n estimate = sum(worker[\"estimatedPromptTokens\"] for worker in normalized_workers)\n if estimate > token_limit:\n raise PlanError(\n f\"estimated prompt tokens {estimate} exceed budget maxEstimatedPromptTokens {token_limit}\"\n )\n\n strategy, max_concurrent = enforce_strategy(\n requested_strategy, normalized_workers, requested_concurrency, policy\n )\n if strategy == \"economy\":\n worker = normalized_workers[0]\n if \"tier\" not in workers[0] and not worker[\"model\"] and not worker[\"effort\"]:\n normalized_workers = [{**worker, \"tier\": \"cheap\"}]\n normalized_workers = [resolve_model_tier(worker, policy, settings) for worker in normalized_workers]\n\n return {\n \"version\": PLAN_VERSION,\n \"title\": title,\n \"why\": why,\n \"playbook\": playbook,\n \"phase\": phase,\n \"strategy\": strategy,\n \"requestedStrategy\": requested_strategy,\n \"maxConcurrent\": max_concurrent,\n \"defaultTimeoutSeconds\": default_timeout,\n \"budget\": budget,\n \"policy\": policy,\n \"workers\": normalized_workers,\n }\n\n\ndef validate_custom(worker: Dict[str, Any], worker_id: str) -> Dict[str, Any]:\n command = require_string(worker.get(\"command\"), f\"worker {worker_id}.command\")\n args = worker.get(\"args\", [])\n if not isinstance(args, list) or not all(isinstance(arg, str) for arg in args):\n raise PlanError(f\"worker {worker_id}.args must be an array of strings\")\n transport = worker.get(\"promptTransport\", \"stdin\")\n if transport not in (\"stdin\", \"argument\"):\n raise PlanError(f\"worker {worker_id}.promptTransport must be stdin or argument\")\n output_format = worker.get(\"outputFormat\", \"text\")\n if output_format not in (\"text\", \"json\", \"jsonl\"):\n raise PlanError(f\"worker {worker_id}.outputFormat must be text, json or jsonl\")\n allow_unsafe = worker.get(\"allowUnsafe\", False)\n if not isinstance(allow_unsafe, bool):\n raise PlanError(f\"worker {worker_id}.allowUnsafe must be a boolean\")\n if not allow_unsafe and any(arg in UNSAFE_FLAGS for arg in args):\n raise PlanError(f\"worker {worker_id} uses an unsafe flag without allowUnsafe\")\n return {\n \"command\": command,\n \"args\": args,\n \"promptTransport\": transport,\n \"outputFormat\": output_format,\n \"allowUnsafe\": allow_unsafe,\n }\n\n\ndef run_git(\n cwd: Path, args: Sequence[str], input_bytes: Optional[bytes] = None, timeout: float = GIT_TIMEOUT_SECONDS\n) -> bytes:\n try:\n result = subprocess.run(\n [\"git\", \"--literal-pathspecs\", \"-C\", str(cwd), *args],\n input=input_bytes,\n stdout=subprocess.PIPE,\n stderr=subprocess.PIPE,\n timeout=timeout,\n check=False,\n )\n except subprocess.TimeoutExpired as error:\n raise PlanError(f\"git {' '.join(args)} timed out after {timeout:g}s\") from error\n if result.returncode != 0:\n message = result.stderr.decode(\"utf-8\", errors=\"replace\").strip()\n raise PlanError(f\"git {' '.join(args)} failed: {message}\")\n return result.stdout\n\n\ndef is_inside(path: Path, root: Path) -> bool:\n try:\n return os.path.commonpath([str(path.resolve()), str(root.resolve())]) == str(root.resolve())\n except ValueError:\n return False\n\n\ndef reject_seed_symlinks(path: Path) -> None:\n if path.is_symlink():\n raise PlanError(f\"seed path is a symlink: {path}\")\n if not path.is_dir():\n return\n for current, directories, files in os.walk(path, followlinks=False):\n for name in [*directories, *files]:\n candidate = Path(current) / name\n if candidate.is_symlink():\n raise PlanError(f\"seed path contains a symlink: {candidate}\")\n\n\ndef exclude_worktrees(root: Path) -> None:\n with EXCLUDE_LOCK:\n ignored = git_returncode(root, [\"check-ignore\", \"-q\", \".alethe/worktrees/probe\"]) == 0\n if ignored:\n return\n exclude = Path(run_git(root, [\"rev-parse\", \"--git-path\", \"info/exclude\"]).decode().strip())\n if not exclude.is_absolute():\n exclude = root / exclude\n existing = exclude.read_text(encoding=\"utf-8\") if exclude.is_file() else \"\"\n exclude.parent.mkdir(parents=True, exist_ok=True)\n with exclude.open(\"a\", encoding=\"utf-8\") as handle:\n if existing and not existing.endswith(\"\\n\"):\n handle.write(\"\\n\")\n handle.write(WORKTREE_EXCLUDE + \"\\n\")\n\n\ndef prepare_worktree(worker: Dict[str, Any], run_id: str) -> Path:\n origin = Path(worker[\"cwd\"])\n root = Path(run_git(origin, [\"rev-parse\", \"--show-toplevel\"]).decode().strip()).resolve()\n target = root / \".alethe\" / \"worktrees\" / f\"skill-{run_id}-{worker['id']}\"\n if target.exists():\n raise PlanError(f\"worktree target already exists: {target}\")\n exclude_worktrees(root)\n target.parent.mkdir(parents=True, exist_ok=True)\n run_git(root, [\"worktree\", \"add\", \"--detach\", str(target), \"HEAD\"])\n\n try:\n seed_worktree(root, target, worker[\"seedPaths\"])\n except Exception as error:\n raise WorktreeSeedError(target, str(error)) from error\n return target\n\n\nclass WorktreeSeedError(PlanError):\n def __init__(self, worktree: Path, message: str) -> None:\n super().__init__(message)\n self.worktree = worktree\n\n\ndef seed_worktree(root: Path, target: Path, seed_paths: Sequence[str]) -> None:\n if not seed_paths:\n return\n for relative in seed_paths:\n validate_relative_path(relative, \"seedPaths\")\n source = root / relative\n if not is_inside(source, root):\n raise PlanError(f\"seed path escapes the repository: {relative}\")\n reject_seed_symlinks(source)\n patch = run_git(root, [\"diff\", \"--binary\", \"HEAD\", \"--\", *seed_paths])\n if patch.strip():\n run_git(target, [\"apply\", \"-\"], patch)\n untracked = run_git(root, [\"ls-files\", \"--others\", \"--exclude-standard\", \"-z\", \"--\", *seed_paths])\n for raw in untracked.split(b\"\\0\"):\n if not raw:\n continue\n relative = raw.decode(\"utf-8\", errors=\"surrogateescape\")\n source = root / relative\n if not is_inside(source, root):\n raise PlanError(f\"seed path escapes the repository: {relative}\")\n reject_seed_symlinks(source)\n destination = target / relative\n destination.parent.mkdir(parents=True, exist_ok=True)\n shutil.copy2(source, destination, follow_symlinks=False)\n\n\ndef git_returncode(cwd: Path, args: Sequence[str]) -> int:\n try:\n return subprocess.run(\n [\"git\", \"-C\", str(cwd), *args],\n stdout=subprocess.DEVNULL,\n stderr=subprocess.DEVNULL,\n timeout=GIT_TIMEOUT_SECONDS,\n check=False,\n ).returncode\n except subprocess.TimeoutExpired as error:\n raise PlanError(f\"git {' '.join(args)} timed out after {GIT_TIMEOUT_SECONDS:g}s\") from error\n\n\ndef command_output(args: Sequence[str], cache: Dict[str, str], lock: threading.Lock) -> str:\n key = \"\\0\".join(args)\n with lock:\n if key in cache:\n return cache[key]\n try:\n result = subprocess.run(\n list(args),\n stdin=subprocess.DEVNULL,\n stdout=subprocess.PIPE,\n stderr=subprocess.STDOUT,\n text=True,\n encoding=\"utf-8\",\n errors=\"replace\",\n timeout=15,\n check=False,\n )\n output = result.stdout or \"\"\n except (OSError, subprocess.TimeoutExpired):\n output = \"\"\n with lock:\n cache[key] = output\n return output\n\n\ndef command_help(launch: Sequence[str], cache: Dict[str, str], lock: threading.Lock) -> str:\n return command_output([*launch, \"--help\"], cache, lock)\n\n\ndef resolve_executable(name: str) -> str:\n expanded = str(Path(name).expanduser())\n if os.path.dirname(expanded):\n path = Path(expanded)\n if not path.is_file():\n raise PlanError(f\"executable not found: {name}\")\n return str(path.resolve())\n resolved = shutil.which(name)\n if not resolved:\n raise PlanError(f\"executable not found on PATH: {name}\")\n return resolved\n\n\ndef is_batch_file(executable: str) -> bool:\n return os.name == \"nt\" and Path(executable).suffix.lower() in BATCH_SUFFIXES\n\n\ndef shim_target(shim: Path) -> Optional[List[str]]:\n try:\n text = shim.read_text(encoding=\"utf-8\", errors=\"replace\")\n except OSError:\n return None\n lines = [line for line in text.splitlines() if \"%*\" in line]\n if not lines:\n return None\n head = lines[-1].split(\"%*\", 1)[0].rsplit(\"&\", 1)[-1]\n tokens = re.findall(r'\"([^\"]+)\"', head)\n if not tokens or re.sub(r'\"[^\"]*\"', \"\", head).strip():\n return None\n folder = str(shim.parent) + \"\\\\\"\n launch: List[str] = []\n for token in tokens:\n if token == \"%_prog%\":\n bundled = shim.parent / \"node.exe\"\n node = str(bundled) if bundled.is_file() else shutil.which(\"node\")\n if not node:\n return None\n launch.append(node)\n continue\n value = token.replace(\"%~dp0\", folder).replace(\"%dp0%\", folder)\n if \"%\" in value or not Path(value).is_file():\n return None\n launch.append(str(Path(value).resolve()))\n return launch\n\n\ndef launch_command(executable: str) -> List[str]:\n if not is_batch_file(executable):\n return [executable]\n # cmd.exe cuts a batch argument at a newline and runs what follows an unbalanced quote.\n return shim_target(Path(executable)) or [executable]\n\n\ndef refuse_batch_arguments(args: Sequence[str]) -> None:\n if not is_batch_file(args[0]):\n return\n if any(BATCH_UNSAFE.search(arg) for arg in args[1:]):\n raise PlanError(\n f\"{Path(args[0]).name} is a batch file and cmd.exe would rewrite its arguments; \"\n \"point the worker at the real executable or send the prompt through stdin\"\n )\n\n\ndef in_git_worktree(cwd: Path) -> bool:\n return git_returncode(cwd, [\"rev-parse\", \"--is-inside-work-tree\"]) == 0\n\n\ndef build_invocation(\n worker: Dict[str, Any],\n cwd: Path,\n worker_dir: Path,\n help_cache: Dict[str, str],\n help_lock: threading.Lock,\n timeout: int,\n cost_cap: Optional[float] = None,\n) -> Tuple[List[str], bool, str, Path]:\n provider = worker[\"provider\"]\n mode = worker[\"mode\"]\n model = worker[\"model\"]\n effort = worker[\"effort\"]\n resume = worker[\"resume\"]\n result_path = worker_dir / \"result.md\"\n\n if provider == \"codex\":\n launch = launch_command(resolve_executable(\"codex\"))\n features = command_output([*launch, \"features\", \"list\"], help_cache, help_lock)\n args = list(launch)\n for feature in NESTED_AGENT_FEATURES:\n if re.search(rf\"^{feature}\\s\", features, re.MULTILINE):\n args += [\"--disable\", feature]\n sandbox = \"read-only\" if mode == \"read-only\" else \"workspace-write\"\n args += [\"exec\", \"--sandbox\", sandbox]\n if resume:\n args.append(\"resume\")\n args += [\"--json\", \"--output-last-message\", str(result_path)]\n if not resume:\n args += [\"--cd\", str(cwd)]\n if model:\n args += [\"--model\", model]\n if effort:\n args += [\"--config\", f\"model_reasoning_effort={effort}\"]\n if not worker[\"mcp\"]:\n args += [\"--config\", \"mcp_servers={}\"]\n if not in_git_worktree(cwd):\n args.append(\"--skip-git-repo-check\")\n if resume:\n args.append(resume)\n args.append(\"-\")\n return args, True, \"jsonl\", result_path\n\n if provider == \"claude\":\n launch = launch_command(resolve_executable(\"claude\"))\n permission = \"plan\" if mode == \"read-only\" else \"acceptEdits\"\n args = [\n *launch,\n \"--print\",\n \"--output-format\",\n \"stream-json\",\n \"--verbose\",\n \"--permission-mode\",\n permission,\n ]\n help_text = command_help(launch, help_cache, help_lock)\n if \"--permission-prompts\" in help_text:\n args += [\"--permission-prompts\", \"none\"]\n if \"--disable-slash-commands\" in help_text:\n args.append(\"--disable-slash-commands\")\n if \"--disallowedTools\" in help_text:\n args += [\"--disallowedTools\", CLAUDE_AGENT_TOOLS]\n if not worker[\"mcp\"] and \"--strict-mcp-config\" in help_text:\n args.append(\"--strict-mcp-config\")\n if cost_cap is not None and \"--max-budget-usd\" in help_text:\n args += [\"--max-budget-usd\", f\"{cost_cap:.2f}\"]\n if model:\n args += [\"--model\", model]\n if effort:\n args += [\"--effort\", effort]\n if resume:\n args += [\"--resume\", resume]\n return args, True, \"jsonl\", result_path\n\n if provider == \"agy\":\n launch = launch_command(resolve_executable(\"agy\"))\n agent_mode = \"plan\" if mode == \"read-only\" else \"accept-edits\"\n # The prompt must be attached to --print: agy otherwise treats the next flag as the\n # prompt and drops the real one.\n args = [\n *launch,\n \"--output-format\",\n \"json\",\n \"--mode\",\n agent_mode,\n \"--print-timeout\",\n f\"{timeout}s\",\n ]\n help_text = command_help(launch, help_cache, help_lock)\n if \"--disable-slash-commands\" in help_text:\n args.append(\"--disable-slash-commands\")\n if \"--add-dir\" in help_text:\n args += [\"--add-dir\", str(cwd)]\n if model:\n args += [\"--model\", model]\n if effort:\n args += [\"--effort\", effort]\n if resume:\n args += [\"--conversation\", resume]\n args.append(f\"--print={worker['task']}\")\n return args, False, \"json\", result_path\n\n if provider == \"kimi\":\n launch = launch_command(resolve_executable(\"kimi\"))\n # Prompt mode always runs in Kimi's auto permission mode and refuses --plan and --yolo.\n args = [*launch, \"--output-format\", \"stream-json\"]\n if model:\n args += [\"--model\", model]\n if resume:\n args += [\"--session\", resume]\n args.append(f\"--prompt={worker['task']}\")\n return args, False, \"jsonl\", result_path\n\n launch = launch_command(resolve_executable(worker[\"command\"]))\n replacements = {\"{cwd}\": str(cwd), \"{result}\": str(result_path)}\n custom_args = []\n for raw in worker[\"args\"]:\n value = raw\n for marker, replacement in replacements.items():\n value = value.replace(marker, replacement)\n custom_args.append(value)\n stdin_prompt = worker[\"promptTransport\"] == \"stdin\"\n if not stdin_prompt:\n custom_args.append(worker[\"task\"])\n return [*launch, *custom_args], stdin_prompt, worker[\"outputFormat\"], result_path\n\n\ndef worker_environment() -> Dict[str, str]:\n environment = {key: value for key, value in os.environ.items() if key != \"CLAUDECODE\"}\n environment[WORKER_ENV_FLAG] = \"1\"\n return environment\n\n\ndef windows_kernel() -> Any:\n import ctypes\n from ctypes import wintypes\n\n kernel = ctypes.WinDLL(\"kernel32\", use_last_error=True)\n signatures = {\n \"CreateJobObjectW\": ([ctypes.c_void_p, wintypes.LPCWSTR], wintypes.HANDLE),\n \"SetInformationJobObject\": ([wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD], wintypes.BOOL),\n \"AssignProcessToJobObject\": ([wintypes.HANDLE, wintypes.HANDLE], wintypes.BOOL),\n \"CloseHandle\": ([wintypes.HANDLE], wintypes.BOOL),\n \"OpenProcess\": ([wintypes.DWORD, wintypes.BOOL, wintypes.DWORD], wintypes.HANDLE),\n \"GetExitCodeProcess\": ([wintypes.HANDLE, ctypes.POINTER(wintypes.DWORD)], wintypes.BOOL),\n \"TerminateProcess\": ([wintypes.HANDLE, wintypes.UINT], wintypes.BOOL),\n }\n for name, (arguments, result) in signatures.items():\n function = getattr(kernel, name)\n function.argtypes, function.restype = arguments, result\n return kernel\n\n\nclass WindowsJob:\n def __init__(self) -> None:\n import ctypes\n from ctypes import wintypes\n\n class BasicLimits(ctypes.Structure):\n _fields_ = [\n (\"PerProcessUserTimeLimit\", ctypes.c_longlong), (\"PerJobUserTimeLimit\", ctypes.c_longlong),\n (\"LimitFlags\", wintypes.DWORD), (\"MinimumWorkingSetSize\", ctypes.c_size_t),\n (\"MaximumWorkingSetSize\", ctypes.c_size_t), (\"ActiveProcessLimit\", wintypes.DWORD),\n (\"Affinity\", ctypes.c_size_t), (\"PriorityClass\", wintypes.DWORD),\n (\"SchedulingClass\", wintypes.DWORD),\n ]\n\n class IoCounters(ctypes.Structure):\n _fields_ = [(name, ctypes.c_ulonglong) for name in (\n \"ReadOperationCount\", \"WriteOperationCount\", \"OtherOperationCount\",\n \"ReadTransferCount\", \"WriteTransferCount\", \"OtherTransferCount\",\n )]\n\n class ExtendedLimits(ctypes.Structure):\n _fields_ = [\n (\"BasicLimitInformation\", BasicLimits), (\"IoInfo\", IoCounters),\n (\"ProcessMemoryLimit\", ctypes.c_size_t), (\"JobMemoryLimit\", ctypes.c_size_t),\n (\"PeakProcessMemoryUsed\", ctypes.c_size_t), (\"PeakJobMemoryUsed\", ctypes.c_size_t),\n ]\n\n self.lock = threading.Lock()\n self.kernel = windows_kernel()\n self.handle = self.kernel.CreateJobObjectW(None, None)\n if not self.handle:\n raise ctypes.WinError(ctypes.get_last_error())\n limits = ExtendedLimits()\n limits.BasicLimitInformation.LimitFlags = JOB_OBJECT_LIMIT_KILL_ON_JOB_CLOSE\n if not self.kernel.SetInformationJobObject(\n self.handle, JOB_OBJECT_EXTENDED_LIMIT_INFORMATION, ctypes.byref(limits), ctypes.sizeof(limits)\n ):\n error = ctypes.WinError(ctypes.get_last_error())\n self.close()\n raise error\n\n def assign(self, process: subprocess.Popen) -> None:\n import ctypes\n\n if not self.kernel.AssignProcessToJobObject(self.handle, int(process._handle)):\n raise ctypes.WinError(ctypes.get_last_error())\n\n def close(self) -> None:\n with self.lock:\n handle, self.handle = self.handle, None\n if handle:\n self.kernel.CloseHandle(handle)\n\n\ndef resume_windows_process(process: subprocess.Popen) -> None:\n import ctypes\n from ctypes import wintypes\n\n resume = ctypes.WinDLL(\"ntdll\").NtResumeProcess\n resume.argtypes, resume.restype = [wintypes.HANDLE], ctypes.c_long\n if resume(int(process._handle)) != 0:\n raise OSError(\"could not resume suspended worker\")\n\n\ndef close_process_job(process: subprocess.Popen) -> bool:\n job = getattr(process, \"_alethe_job\", None)\n if job is None:\n return False\n job.close()\n return True\n\n\ndef pid_alive(pid: Any) -> bool:\n if not isinstance(pid, int) or isinstance(pid, bool) or pid \u003c= 0:\n return False\n if os.name == \"nt\":\n import ctypes\n from ctypes import wintypes\n\n kernel = windows_kernel()\n handle = kernel.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)\n if not handle:\n return ctypes.get_last_error() == WINDOWS_ACCESS_DENIED\n try:\n code = wintypes.DWORD()\n return not kernel.GetExitCodeProcess(handle, ctypes.byref(code)) or code.value == STILL_ACTIVE\n finally:\n kernel.CloseHandle(handle)\n try:\n os.kill(pid, 0)\n return True\n except ProcessLookupError:\n return False\n except PermissionError:\n return True\n\n\ndef stop_windows_pid(pid: int) -> None:\n try:\n result = subprocess.run(\n [\"taskkill\", \"/PID\", str(pid), \"/T\", \"/F\"],\n stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,\n timeout=PROCESS_STOP_TIMEOUT_SECONDS, check=False,\n )\n if result.returncode == 0:\n return\n except (OSError, subprocess.TimeoutExpired):\n pass\n if not pid_alive(pid):\n return\n import ctypes\n\n kernel = windows_kernel()\n handle = kernel.OpenProcess(PROCESS_TERMINATE, False, pid)\n if not handle:\n raise PlanError(f\"cannot stop worker PID {pid}: {ctypes.WinError(ctypes.get_last_error())}\")\n try:\n if not kernel.TerminateProcess(handle, 1):\n raise PlanError(f\"cannot stop worker PID {pid}: {ctypes.WinError(ctypes.get_last_error())}\")\n finally:\n kernel.CloseHandle(handle)\n\n\ndef stop_recorded_pid(pid: Any) -> None:\n if not pid_alive(pid) or pid == os.getpid():\n return\n if os.name == \"nt\":\n stop_windows_pid(pid)\n return\n try:\n os.killpg(pid, signal.SIGKILL)\n except ProcessLookupError:\n try:\n os.kill(pid, signal.SIGKILL)\n except ProcessLookupError:\n pass\n\n\ndef effective_run_state(state: Dict[str, Any]) -> Dict[str, Any]:\n runner_pid = state.get(\"runnerPid\")\n if state.get(\"status\") != \"running\" or runner_pid is None or pid_alive(runner_pid):\n return state\n workers = {\n key: {**entry, \"status\": \"interrupted\"} if entry.get(\"status\") in (\"running\", \"queued\") else entry\n for key, entry in state[\"workers\"].items()\n }\n return {**state, \"status\": \"interrupted\", \"workers\": workers}\n\n\ndef start_process(\n args: Sequence[str],\n cwd: Path,\n stdin: Any,\n stdout: Any,\n stderr: Any,\n env: Optional[Dict[str, str]] = None,\n) -> subprocess.Popen:\n refuse_batch_arguments(args)\n options: Dict[str, Any] = {\n \"args\": list(args),\n \"cwd\": str(cwd),\n \"stdin\": stdin,\n \"stdout\": stdout,\n \"stderr\": stderr,\n \"shell\": False,\n \"env\": env,\n }\n job = None\n if os.name == \"nt\":\n options[\"creationflags\"] = subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW\n try:\n job = WindowsJob()\n except OSError:\n pass\n if job is not None:\n # Assign before resuming so descendants inherit job ownership.\n options[\"creationflags\"] |= WINDOWS_CREATE_SUSPENDED\n else:\n options[\"start_new_session\"] = True\n try:\n process = subprocess.Popen(**options)\n except BaseException:\n if job is not None:\n job.close()\n raise\n if job is not None:\n try:\n try:\n job.assign(process)\n process._alethe_job = job\n except OSError:\n job.close()\n resume_windows_process(process)\n except BaseException:\n job.close()\n process.kill()\n process.wait()\n raise\n return process\n\n\ndef stop_process_tree(process: subprocess.Popen) -> None:\n if os.name == \"nt\":\n if not close_process_job(process):\n stop_windows_pid(process.pid)\n else:\n try:\n os.killpg(process.pid, signal.SIGTERM)\n except (ProcessLookupError, PermissionError):\n pass\n try:\n process.wait(timeout=3)\n except subprocess.TimeoutExpired:\n pass\n try:\n os.killpg(process.pid, signal.SIGKILL)\n except (ProcessLookupError, PermissionError):\n pass\n try:\n process.wait(timeout=5)\n except subprocess.TimeoutExpired:\n process.kill()\n process.wait()\n\n\ndef feed_stdin(stream: Any, data: bytes) -> None:\n def write() -> None:\n try:\n stream.write(data)\n stream.close()\n except (OSError, ValueError):\n pass\n\n threading.Thread(target=write, daemon=True).start()\n\n\ndef exit_code_of(process: subprocess.Popen) -> int:\n return process.returncode if process.returncode is not None else -1\n\n\ndef watch_process(\n process: subprocess.Popen,\n worker_id: str,\n timeout: int,\n idle_timeout: Optional[int],\n logs: Sequence[Path],\n state: \"BatchState\",\n) -> Tuple[int, Optional[str]]:\n started = time.monotonic()\n last_size = 0\n last_change = last_refresh = started\n stop_path = state.path.parent / STOP_FILE\n while True:\n remaining = timeout - (time.monotonic() - started)\n if remaining \u003c= 0:\n stop_process_tree(process)\n return exit_code_of(process), \"timed_out\"\n try:\n code = process.wait(timeout=min(POLL_SECONDS, remaining))\n return code, \"stopped\" if stop_path.exists() else None\n except subprocess.TimeoutExpired:\n pass\n if stop_path.exists():\n stop_process_tree(process)\n return exit_code_of(process), \"stopped\"\n now = time.monotonic()\n size = sum(file_bytes(path) for path in logs)\n if size != last_size:\n last_size, last_change = size, now\n if idle_timeout is not None and now - last_change >= idle_timeout:\n stop_process_tree(process)\n return exit_code_of(process), \"stalled\"\n if now - last_refresh >= STATE_REFRESH_SECONDS:\n last_refresh = now\n try:\n state.update_worker(worker_id, {\"outputBytes\": size, \"idleSeconds\": round(now - last_change)})\n except OSError:\n pass\n\n\ndef json_values(text: str, output_format: str) -> List[Any]:\n if not text.strip():\n return []\n if output_format == \"json\":\n try:\n return [json.loads(text)]\n except json.JSONDecodeError:\n pass\n for line in reversed(text.splitlines()):\n line = line.strip()\n if line.startswith(\"{\"):\n try:\n return [json.loads(line)]\n except json.JSONDecodeError:\n continue\n return []\n if output_format == \"jsonl\":\n values = []\n for line in text.splitlines():\n line = line.strip()\n if not line:\n continue\n try:\n values.append(json.loads(line))\n except json.JSONDecodeError:\n continue\n return values\n return []\n\n\ndef text_from_value(value: Any) -> str:\n if isinstance(value, str):\n return value.strip()\n if not isinstance(value, dict):\n return \"\"\n for key in (\"result\", \"output\", \"text\", \"content\", \"response\"):\n text = value.get(key)\n if isinstance(text, str) and text.strip():\n return text.strip()\n message = value.get(\"message\")\n if isinstance(message, str):\n return message.strip()\n if isinstance(message, dict):\n content = message.get(\"content\")\n if isinstance(content, str):\n return content.strip()\n if isinstance(content, list):\n parts = []\n for block in content:\n if isinstance(block, dict) and isinstance(block.get(\"text\"), str):\n parts.append(block[\"text\"])\n return \"\".join(parts).strip()\n for key in (\"part\", \"data\", \"response\"):\n nested = value.get(key)\n if isinstance(nested, dict):\n text = text_from_value(nested)\n if text:\n return text\n return \"\"\n\n\ndef result_text_from_values(values: Sequence[Any]) -> str:\n for value in reversed(values):\n if isinstance(value, dict) and value.get(\"type\") == \"result\":\n text = text_from_value(value)\n if text:\n return text\n chunks = []\n for value in values:\n if not isinstance(value, dict):\n continue\n kind = value.get(\"type\") or value.get(\"role\")\n if kind not in (\"assistant\", \"message\", \"text\", \"output\"):\n continue\n text = text_from_value(value)\n if text:\n chunks.append(text)\n if chunks:\n return \"\\n\".join(chunks).strip()\n for value in reversed(values):\n text = text_from_value(value)\n if text:\n return text\n return \"\"\n\n\ndef extract_session(values: Sequence[Any]) -> Optional[str]:\n for value in reversed(values):\n if not isinstance(value, dict):\n continue\n for key in (\n \"session_id\",\n \"sessionId\",\n \"sessionID\",\n \"conversation_id\",\n \"conversationId\",\n \"thread_id\",\n ):\n session = value.get(key)\n if isinstance(session, str) and session.strip():\n return session.strip()\n return None\n\n\ndef extract_metrics(values: Sequence[Any]) -> Dict[str, Any]:\n metrics: Dict[str, Any] = {}\n turns = []\n final = None\n for value in values:\n if not isinstance(value, dict):\n continue\n usage = value.get(\"usage\") or value.get(\"tokenUsage\") or value.get(\"token_usage\")\n if isinstance(usage, dict) and usage:\n metrics[\"usage\"] = normalized_usage(usage)\n if value.get(\"type\") == \"turn.completed\":\n turns.append(normalized_usage(usage))\n if value.get(\"type\") == \"result\":\n final = metrics[\"usage\"]\n cost = number(value.get(\"total_cost_usd\"))\n if cost is not None:\n metrics[\"totalCostUsd\"] = cost\n models = value.get(\"modelUsage\")\n if isinstance(models, dict):\n metrics[\"costByModel\"] = {str(model): data[\"costUSD\"] for model, data in models.items()\n if isinstance(data, dict) and number(data.get(\"costUSD\")) is not None}\n part = value.get(\"part\")\n if isinstance(part, dict):\n tokens = part.get(\"tokens\")\n if isinstance(tokens, dict):\n metrics[\"usage\"] = normalized_usage(tokens)\n part_cost = number(part.get(\"cost\"))\n if part_cost is not None:\n metrics[\"totalCostUsd\"] = part_cost\n if turns:\n metrics[\"usage\"] = {key: sum(turn.get(key, 0) for turn in turns) for key in set().union(*turns)}\n if final is not None:\n metrics[\"usage\"] = final\n return metrics\n\n\ndef classify_result(\n provider: str,\n output_format: str,\n stdout_path: Path,\n result_path: Path,\n exit_code: int,\n timed_out: bool,\n) -> Tuple[str, str, List[Any], Optional[str]]:\n stdout_text = stdout_path.read_text(encoding=\"utf-8\", errors=\"replace\")\n values = json_values(stdout_text, output_format)\n error: Optional[str] = None\n text = \"\"\n\n if result_path.is_file():\n text = result_path.read_text(encoding=\"utf-8\", errors=\"replace\").strip()\n else:\n text = stdout_text.strip() if output_format == \"text\" else result_text_from_values(values)\n if text:\n result_path.write_text(text + \"\\n\", encoding=\"utf-8\")\n\n if timed_out:\n return \"timed_out\", text, values, \"worker exceeded its deadline\"\n if exit_code != 0:\n return \"failed\", text, values, f\"worker exited with status {exit_code}\"\n if output_format in (\"json\", \"jsonl\") and not values:\n return \"failed\", text, values, \"worker output was not valid structured JSON\"\n if provider in (\"claude\", \"agy\") and values:\n final = next((value for value in reversed(values) if not isinstance(value, dict)\n or value.get(\"type\") != \"rate_limit_event\"), None)\n if isinstance(final, dict):\n if final.get(\"is_error\") is True or str(final.get(\"subtype\", \"\")).startswith(\"error\"):\n error = f\"worker returned {final.get('subtype', 'an error result')}\"\n if provider == \"claude\" and final.get(\"type\") != \"result\":\n error = \"Claude output did not end with a result object\"\n if error:\n return \"failed\", text, values, error\n if not text:\n return \"failed\", text, values, \"worker produced no final result\"\n return \"succeeded\", text, values, None\n\n\ndef capture_diff(cwd: Path, worker_dir: Path, isolated: bool) -> Tuple[Optional[Path], Optional[str]]:\n if not in_git_worktree(cwd):\n return None, None\n try:\n run_git(cwd, [\"rev-parse\", \"--verify\", \"--quiet\", \"HEAD\"], timeout=30)\n except PlanError:\n return None, \"the repository has no commits yet, so there is no diff to capture\"\n try:\n if isolated:\n untracked = run_git(cwd, [\"ls-files\", \"--others\", \"--exclude-standard\", \"-z\"])\n paths = [\n value.decode(\"utf-8\", errors=\"surrogateescape\")\n for value in untracked.split(b\"\\0\")\n if value\n ]\n if paths:\n run_git(cwd, [\"add\", \"-N\", \"--\", *paths])\n status = run_git(cwd, [\"status\", \"--short\"])\n (worker_dir / \"status.txt\").write_bytes(status)\n patch = run_git(cwd, [\"diff\", \"--binary\", \"HEAD\"])\n except PlanError as error:\n return None, str(error)\n if not patch.strip():\n return None, None\n diff_path = worker_dir / \"diff.patch\"\n diff_path.write_bytes(patch)\n return diff_path, None\n\n\ndef checkout_root(cwd: Path) -> Optional[Path]:\n try:\n return Path(run_git(cwd, [\"rev-parse\", \"--show-toplevel\"], timeout=30).decode().strip()).resolve()\n except PlanError:\n return None\n\n\ndef file_fingerprint(path: Path) -> str:\n try:\n if path.is_symlink():\n return \"link:\" + os.readlink(path)\n if path.is_dir():\n return \"dir\"\n size = path.stat().st_size\n if size > FINGERPRINT_MAX_BYTES:\n return f\"size:{size}:{path.stat().st_mtime_ns}\"\n return hashlib.sha256(path.read_bytes()).hexdigest()\n except OSError:\n return \"missing\"\n\n\ndef checkout_snapshot(root: Path, ignore: Sequence[Path] = ()) -> Optional[Dict[str, str]]:\n try:\n raw = run_git(root, [\"status\", \"--porcelain=v1\", \"-z\", \"--untracked-files=all\"], timeout=120)\n except PlanError:\n return None\n tokens = raw.decode(\"utf-8\", errors=\"surrogateescape\").split(\"\\0\")\n entries: Dict[str, str] = {}\n index = 0\n while index \u003c len(tokens):\n token = tokens[index]\n index += 1\n if len(token) \u003c 4:\n continue\n code, path = token[:2], token[3:]\n if \"R\" in code or \"C\" in code:\n index += 1\n absolute = root / path\n if any(is_inside(absolute, folder) for folder in ignore):\n continue\n entries[path] = f\"{code}:{file_fingerprint(absolute)}\"\n if len(entries) > SNAPSHOT_LIMIT:\n return None\n return entries\n\n\ndef changed_paths(before: Dict[str, str], after: Dict[str, str]) -> List[str]:\n return sorted(path for path in set(before) | set(after) if before.get(path) != after.get(path))\n\n\nclass CheckoutRegistry:\n def __init__(self) -> None:\n self.lock = threading.Lock()\n self.active: Dict[Path, Dict[str, str]] = {}\n self.overlaps: Dict[str, Dict[str, str]] = {}\n\n def enter(self, root: Path, worker_id: str, mode: str) -> None:\n with self.lock:\n active = self.active.setdefault(root, {})\n mine = self.overlaps.setdefault(worker_id, {})\n for other, other_mode in active.items():\n mine[other] = other_mode\n self.overlaps.setdefault(other, {})[worker_id] = mode\n active[worker_id] = mode\n\n def leave(self, root: Path, worker_id: str) -> Dict[str, str]:\n with self.lock:\n self.active.get(root, {}).pop(worker_id, None)\n return dict(self.overlaps.pop(worker_id, {}))\n\n\ndef parse_handoff(text: str) -> Dict[str, Optional[str]]:\n status = verdict = summary = None\n for match in HANDOFF_LINE.finditer(text or \"\"):\n key, value = match.group(1).upper(), match.group(2).strip()\n if key == \"STATUS\":\n candidate = value.split()[0].strip(\".,;:\").upper()\n status = candidate if candidate in WORKER_STATUSES else status\n elif key == \"VERDICT\":\n candidate = re.sub(r\"[\\s-]+\", \"_\", value.strip(\".,;: \")).upper()\n verdict = next((option for option in VERDICTS if candidate.startswith(option)), verdict)\n else:\n summary = value\n if summary and len(summary) > NOTE_LIMITS[\"workerWhy\"]:\n summary = summary[: NOTE_LIMITS[\"workerWhy\"] - 3].rstrip() + \"...\"\n return {\"workerStatus\": status, \"workerSummary\": summary, \"verdict\": verdict}\n\n\ndef extract_details(values: Sequence[Any]) -> Dict[str, List[str]]:\n denials: List[str] = []\n billed: List[str] = []\n for value in values:\n if not isinstance(value, dict):\n continue\n raw = value.get(\"permission_denials\")\n if isinstance(raw, list):\n denials = [\n str(item.get(\"tool_name\") or item.get(\"toolName\") or \"tool\") if isinstance(item, dict) else \"tool\"\n for item in raw\n ]\n usage = value.get(\"modelUsage\")\n if isinstance(usage, dict) and usage:\n billed = sorted(str(key) for key in usage)\n return {\"permissionDenials\": denials, \"billedModels\": billed}\n\n\ndef tail_text(path: Path, lines: int = 20) -> str:\n size = file_bytes(path)\n if not size:\n return \"\"\n with path.open(\"rb\") as handle:\n handle.seek(max(0, size - 65536))\n text = handle.read().decode(\"utf-8\", errors=\"replace\")\n kept = [line for line in text.splitlines() if line.strip()]\n return \"\\n\".join(kept[-lines:])\n\n\ndef classify_failure(\n status: str, error: Optional[str], stderr_path: Path, result_text: str, stalled: bool\n) -> Tuple[Optional[str], Optional[str]]:\n if status == \"succeeded\":\n return None, None\n if stalled:\n return \"stalled\", None\n if status == \"timed_out\":\n return \"timed_out\", None\n if str(error or \"\").startswith(\"executable not found\"):\n return \"not_installed\", None\n text = \"\\n\".join(part for part in (error or \"\", tail_text(stderr_path), (result_text or \"\")[-4000:]) if part)\n for kind, pattern in FAILURE_PATTERNS:\n if pattern.search(text):\n if kind != \"rate_limited\":\n return kind, None\n reset = re.search(r\"\\b(?:resets?|try again|retry)\\b[^\\n.]{0,60}\", text, re.IGNORECASE)\n return kind, reset.group(0).strip() if reset else None\n if \"structured JSON\" in str(error) or \"result object\" in str(error):\n return \"invalid_output\", None\n if \"no final result\" in str(error):\n return \"no_result\", None\n return \"error\", None\n\n\nclass BatchState:\n def __init__(self, path: Path, initial: Dict[str, Any]) -> None:\n self.path = path\n self.value = initial\n self.lock = threading.Lock()\n atomic_write_json(self.path, self.value)\n\n def update_worker(self, worker_id: str, patch: Dict[str, Any]) -> None:\n with self.lock:\n self.value[\"workers\"][worker_id].update(patch)\n atomic_write_json(self.path, self.value)\n\n def update_budget(self, usage: Dict[str, Any]) -> None:\n with self.lock:\n self.value[\"budgetUsage\"] = usage\n atomic_write_json(self.path, self.value)\n\n def finish(self, status: str, extra: Optional[Dict[str, Any]] = None) -> None:\n with self.lock:\n self.value[\"status\"] = status\n self.value[\"finishedAt\"] = now_iso()\n if extra:\n self.value.update(extra)\n atomic_write_json(self.path, self.value)\n\n\nclass ProcessRegistry:\n def __init__(self) -> None:\n self.processes: Dict[int, subprocess.Popen] = {}\n self.stopping = False\n self.lock = threading.Lock()\n\n def add(self, process: subprocess.Popen) -> None:\n with self.lock:\n self.processes[process.pid] = process\n stopping = self.stopping\n if stopping:\n stop_process_tree(process)\n\n def remove(self, process: subprocess.Popen) -> None:\n with self.lock:\n self.processes.pop(process.pid, None)\n\n def stop_all(self) -> None:\n with self.lock:\n processes = list(self.processes.values())\n self.stopping = True\n for process in processes:\n stop_process_tree(process)\n\n\ndef orchestration_refusal(\n limits: Dict[str, Any],\n used: Dict[str, Any],\n calls: int = 0,\n cost: Optional[float] = None,\n tokens: Optional[int] = None,\n) -> Optional[str]:\n max_calls = limits.get(\"maxCalls\")\n if max_calls is not None and int(used.get(\"calls\") or 0) + calls >= max_calls:\n return f\"orchestration budget maxCalls {max_calls} exhausted\"\n spent = [value for value in (number(used.get(\"reportedCostUsd\")), cost) if value is not None]\n max_cost = limits.get(\"maxReportedCostUsd\")\n if max_cost is not None and spent and sum(spent) >= max_cost:\n return f\"orchestration budget maxReportedCostUsd {max_cost} exhausted\"\n counted = [value for value in (number(used.get(\"reportedTokens\")), tokens) if value is not None]\n max_tokens = limits.get(\"maxReportedTokens\")\n if max_tokens is not None and counted and sum(counted) >= max_tokens:\n return f\"orchestration budget maxReportedTokens {max_tokens} exhausted\"\n return None\n\n\nclass BudgetTracker:\n \"\"\"Thread-safe batch budget. Gates are checked at worker submit time, so with\n maxConcurrent > 1 they are approximate: already-running workers finish.\"\"\"\n\n def __init__(\n self,\n budget: Dict[str, Any],\n outer: Optional[Dict[str, Any]] = None,\n max_concurrent: int = 1,\n workers: int = 1,\n state: Optional[BatchState] = None,\n ) -> None:\n self.budget = budget\n self.state = state\n self.outer_limits: Dict[str, Any] = (outer or {}).get(\"limits\", {})\n self.outer_used: Dict[str, Any] = (outer or {}).get(\"used\", {})\n self.max_concurrent = max_concurrent\n self.workers = workers\n self.finished: set = set()\n self.reserved: Dict[str, float] = {}\n self.lock = threading.Lock()\n self.calls = 0\n self.started_clock = time.monotonic()\n self.reported_cost = 0.0\n self.cost_seen = False\n self.reported_tokens = 0\n self.tokens_seen = False\n\n def remaining_wall(self) -> Optional[float]:\n limit = self.budget.get(\"maxWallSeconds\")\n if limit is None:\n return None\n return limit - (time.monotonic() - self.started_clock)\n\n def remaining_cost(self) -> Optional[float]:\n limits = []\n plan_limit = self.budget.get(\"maxReportedCostUsd\")\n if plan_limit is not None:\n limits.append(plan_limit - self.reported_cost)\n outer_limit = self.outer_limits.get(\"maxReportedCostUsd\")\n if outer_limit is not None:\n limits.append(outer_limit - (number(self.outer_used.get(\"reportedCostUsd\")) or 0) - self.reported_cost)\n return min(limits) if limits else None\n\n def refusal(self) -> Optional[str]:\n max_calls = self.budget.get(\"maxCalls\")\n if max_calls is not None and self.calls >= max_calls:\n return f\"budget maxCalls {max_calls} exhausted\"\n max_wall = self.budget.get(\"maxWallSeconds\")\n if max_wall is not None and time.monotonic() - self.started_clock >= max_wall:\n return f\"budget maxWallSeconds {max_wall} exhausted\"\n max_cost = self.budget.get(\"maxReportedCostUsd\")\n if max_cost is not None and self.cost_seen and self.reported_cost >= max_cost:\n return f\"budget maxReportedCostUsd {max_cost} exhausted\"\n max_tokens = self.budget.get(\"maxReportedTokens\")\n if max_tokens is not None and self.tokens_seen and self.reported_tokens >= max_tokens:\n return f\"budget maxReportedTokens {max_tokens} exhausted\"\n return orchestration_refusal(\n self.outer_limits,\n self.outer_used,\n self.calls,\n self.reported_cost if self.cost_seen else None,\n self.reported_tokens if self.tokens_seen else None,\n )\n\n def admit(self, worker_id: str, wants_cost_cap: bool = False) -> Tuple[Optional[str], Optional[float]]:\n with self.lock:\n refusal = self.refusal()\n if refusal is not None:\n return refusal, None\n cap = None\n remaining = self.remaining_cost() if wants_cost_cap else None\n if remaining is not None:\n free = remaining - sum(self.reserved.values())\n pending = self.workers - len(self.finished)\n slots = max(1, min(self.max_concurrent, pending) - len(self.reserved))\n cap = math.floor(max(0.0, free) / slots * 100) / 100\n if cap \u003c 0.01:\n return \"budget maxReportedCostUsd leaves less than $0.01 for this worker\", None\n self.reserved[worker_id] = cap\n self.calls += 1\n if self.state is not None:\n self.state.update_budget(self.usage_snapshot())\n return None, cap\n\n def finish(self, worker_id: str, metrics: Dict[str, Any]) -> None:\n with self.lock:\n if worker_id in self.finished:\n return\n self.finished.add(worker_id)\n self.reserved.pop(worker_id, None)\n cost = metrics.get(\"totalCostUsd\")\n if isinstance(cost, (int, float)) and not isinstance(cost, bool):\n self.cost_seen = True\n self.reported_cost += cost\n tokens = reported_total_tokens(metrics.get(\"usage\"))\n if tokens is not None:\n self.tokens_seen = True\n self.reported_tokens += tokens\n if self.state is not None:\n self.state.update_budget(self.usage_snapshot())\n\n def usage_snapshot(self) -> Dict[str, Any]:\n return {\n \"calls\": self.calls,\n \"wallSeconds\": round(time.monotonic() - self.started_clock, 3),\n \"reportedCostUsd\": round(self.reported_cost, 6) if self.cost_seen else None,\n \"reportedTokens\": self.reported_tokens if self.tokens_seen else None,\n }\n\n def usage(self) -> Dict[str, Any]:\n with self.lock:\n return self.usage_snapshot()\n\n\ndef run_worker(\n worker: Dict[str, Any],\n run_id: str,\n run_dir: Path,\n state: BatchState,\n help_cache: Dict[str, str],\n help_lock: threading.Lock,\n registry: ProcessRegistry,\n tracker: BudgetTracker,\n checkouts: CheckoutRegistry,\n) -> Dict[str, Any]:\n worker_id = worker[\"id\"]\n worker_dir = run_dir / \"workers\" / worker_id\n worker_dir.mkdir(parents=True, exist_ok=True)\n prompt_path = worker_dir / \"prompt.txt\"\n stdout_path = worker_dir / \"stdout.log\"\n stderr_path = worker_dir / \"stderr.log\"\n outcome_path = worker_dir / \"outcome.json\"\n prompt_path.write_text(worker[\"task\"], encoding=\"utf-8\")\n started_clock = time.monotonic()\n outcome: Dict[str, Any] = {\n \"id\": worker_id,\n \"provider\": worker[\"provider\"],\n \"status\": \"failed\",\n \"exitCode\": None,\n \"timedOut\": False,\n \"startedAt\": now_iso(),\n \"finishedAt\": None,\n \"durationSeconds\": 0.0,\n \"cwd\": worker[\"cwd\"],\n \"worktree\": None,\n \"resultPath\": None,\n \"stdoutPath\": None,\n \"stderrPath\": None,\n \"diffPath\": None,\n \"diffError\": None,\n \"sessionId\": None,\n \"metrics\": {},\n \"costCapUsd\": None,\n \"error\": None,\n \"failureKind\": None,\n \"resetHint\": None,\n \"workerStatus\": None,\n \"workerSummary\": None,\n \"verdict\": None,\n \"permissionDenials\": [],\n \"billedModels\": [],\n \"readOnlyViolation\": None,\n \"touchedFiles\": None,\n \"dirtyBefore\": None,\n \"sharedCheckoutWith\": [],\n }\n\n def settle(**fields: Any) -> Dict[str, Any]:\n outcome.update(fields)\n outcome[\"finishedAt\"] = now_iso()\n outcome[\"durationSeconds\"] = round(time.monotonic() - started_clock, 3)\n try:\n policy_path = Path(state.value[\"policyPath\"])\n if worker[\"provider\"] in QUOTA_PROVIDERS and stdout_path.is_file():\n values = json_values(stdout_path.read_text(encoding=\"utf-8\", errors=\"replace\"), \"jsonl\")\n outcome[\"quota\"] = worker_quota(worker[\"provider\"], values, policy_path)\n else:\n save_usage(policy_path, None)\n except (OSError, PlanError) as error:\n outcome[\"quotaError\"] = str(error)\n atomic_write_json(outcome_path, outcome)\n state.update_worker(worker_id, outcome)\n return outcome\n\n if (run_dir / STOP_FILE).exists():\n tracker.finish(worker_id, {})\n return settle(status=\"skipped\", error=\"stopped on request\", failureKind=\"stopped\")\n if not state.value.get(\"allowLowQuota\"):\n try:\n quota_stop = quota_refusal([worker], Path(state.value[\"policyPath\"]))\n except (OSError, PlanError):\n quota_stop = None\n if quota_stop is not None:\n tracker.finish(worker_id, {})\n return settle(status=\"skipped\", error=quota_stop, failureKind=\"quota_low\")\n refusal, cost_cap = tracker.admit(worker_id, worker[\"provider\"] == \"claude\")\n if refusal is not None:\n tracker.finish(worker_id, {})\n return settle(status=\"skipped\", error=refusal, failureKind=\"budget\")\n\n process: Optional[subprocess.Popen] = None\n root: Optional[Path] = None\n metrics: Dict[str, Any] = {}\n try:\n cwd = prepare_worktree(worker, run_id) if worker[\"isolate\"] else Path(worker[\"cwd\"])\n outcome.update(cwd=str(cwd), worktree=str(cwd) if worker[\"isolate\"] else None)\n timeout = worker[\"timeoutSeconds\"]\n remaining_wall = tracker.remaining_wall()\n if remaining_wall is not None:\n timeout = max(1, int(min(timeout, remaining_wall)))\n args, stdin_prompt, output_format, result_path = build_invocation(\n worker, cwd, worker_dir, help_cache, help_lock, timeout, cost_cap\n )\n applied_cost_cap = cost_cap if \"--max-budget-usd\" in args else None\n outcome[\"costCapUsd\"] = applied_cost_cap\n before = None\n if not worker[\"isolate\"]:\n root = checkout_root(cwd)\n if root is not None:\n checkouts.enter(root, worker_id, worker[\"mode\"])\n before = checkout_snapshot(root, ignore=[run_dir])\n state.update_worker(\n worker_id,\n {\n \"status\": \"running\",\n \"startedAt\": outcome[\"startedAt\"],\n \"cwd\": str(cwd),\n \"worktree\": outcome[\"worktree\"],\n \"costCapUsd\": applied_cost_cap,\n },\n )\n with stdout_path.open(\"wb\") as stdout_file, stderr_path.open(\"wb\") as stderr_file:\n stdin_mode = subprocess.PIPE if stdin_prompt else subprocess.DEVNULL\n if (run_dir / STOP_FILE).exists():\n if root is not None:\n checkouts.leave(root, worker_id)\n tracker.finish(worker_id, {})\n return settle(status=\"skipped\", error=\"stopped on request\", failureKind=\"stopped\")\n process = start_process(args, cwd, stdin_mode, stdout_file, stderr_file, worker_environment())\n registry.add(process)\n outcome[\"pid\"] = process.pid\n state.update_worker(worker_id, {\"pid\": process.pid})\n if stdin_prompt and process.stdin is not None:\n feed_stdin(process.stdin, worker[\"task\"].encode(\"utf-8\"))\n exit_code, ended = watch_process(\n process, worker_id, timeout, None if worker[\"provider\"] == \"agy\" else worker[\"idleTimeoutSeconds\"],\n (stdout_path, stderr_path), state\n )\n close_process_job(process)\n registry.remove(process)\n overlapped = checkouts.leave(root, worker_id) if root is not None else {}\n status, result_text, values, error = classify_result(\n worker[\"provider\"], output_format, stdout_path, result_path, exit_code, ended is not None\n )\n if ended == \"stalled\":\n error = f\"worker printed nothing for {worker['idleTimeoutSeconds']}s\"\n metrics = extract_metrics(values)\n details = extract_details(values)\n handoff = parse_handoff(result_text)\n kind, reset_hint = classify_failure(status, error, stderr_path, result_text, ended == \"stalled\")\n if ended == \"stopped\":\n status, error, kind, reset_hint = \"stopped\", \"stopped on request\", \"stopped\", None\n diff_path, diff_error = (\n capture_diff(cwd, worker_dir, worker[\"isolate\"]) if worker[\"mode\"] == \"write\" else (None, None)\n )\n touched = dirty_before = violation = None\n writer_nearby = any(mode == \"write\" for mode in overlapped.values())\n if before is not None and root is not None and not writer_nearby:\n after = checkout_snapshot(root, ignore=[run_dir])\n if after is not None:\n touched = changed_paths(before, after)[:MAX_LISTED_PATHS]\n if worker[\"mode\"] == \"read-only\":\n violation = bool(touched)\n else:\n diff_paths = []\n if diff_path is not None:\n diff_paths = run_git(cwd, [\"diff\", \"--name-only\", \"-z\", \"HEAD\"]).decode(\n \"utf-8\", errors=\"surrogateescape\"\n ).split(\"\\0\")\n dirty_before = sorted(set(before).intersection(diff_paths))[:MAX_LISTED_PATHS]\n tracker.finish(worker_id, metrics)\n return settle(\n status=status,\n exitCode=exit_code,\n timedOut=ended is not None,\n resultPath=str(result_path),\n stdoutPath=str(stdout_path),\n stderrPath=str(stderr_path),\n diffPath=str(diff_path) if diff_path else None,\n diffError=diff_error,\n sessionId=extract_session(values),\n metrics=metrics,\n error=error,\n failureKind=kind,\n resetHint=reset_hint,\n readOnlyViolation=violation,\n touchedFiles=touched,\n dirtyBefore=dirty_before,\n sharedCheckoutWith=sorted(overlapped),\n **handoff,\n **details,\n )\n except Exception as error: # Every worker failure must settle the batch state.\n if process is not None:\n stop_process_tree(process)\n registry.remove(process)\n if root is not None:\n checkouts.leave(root, worker_id)\n tracker.finish(worker_id, metrics)\n if isinstance(error, WorktreeSeedError):\n outcome.update(worktree=str(error.worktree), cwd=str(error.worktree))\n message = str(error)\n return settle(\n status=\"failed\",\n error=message,\n failureKind=\"not_installed\" if message.startswith(\"executable not found\") else \"error\",\n stdoutPath=str(stdout_path),\n stderrPath=str(stderr_path),\n )\n\n\ndef validate_orchestration_limits(raw: Any, label: str) -> Dict[str, Any]:\n limits = validate_budget(raw, label)\n extra = set(limits) - set(ORCHESTRATION_BUDGET_FIELDS)\n if extra:\n raise PlanError(\n f\"{label} only takes {', '.join(ORCHESTRATION_BUDGET_FIELDS)}, not {', '.join(sorted(extra))}\"\n )\n return limits\n\n\ndef load_orchestration_budget(folder: Path) -> Dict[str, Any]:\n path = folder / ORCHESTRATION_BUDGET_FILE\n if not path.is_file():\n return {}\n raw = read_json(path)\n if not isinstance(raw, dict) or raw.get(\"version\") != 1:\n raise PlanError(f\"{path} must be an object with version 1\")\n return validate_orchestration_limits(raw.get(\"limits\"), f\"{path.name} limits\")\n\n\ndef orchestration_usage(folder: Path, exclude: Optional[Path] = None) -> Dict[str, Any]:\n calls, runs = 0, 0\n cost: Optional[float] = None\n tokens: Optional[int] = None\n if not folder.is_dir():\n return {\"runs\": 0, \"calls\": 0, \"reportedCostUsd\": None, \"reportedTokens\": None}\n for child in sorted(folder.iterdir()):\n if exclude is not None and child.resolve() == exclude.resolve():\n continue\n state = read_optional_json(child / \"state.json\")\n if not isinstance(state, dict):\n continue\n runs += 1\n usage = state.get(\"budgetUsage\") if isinstance(state.get(\"budgetUsage\"), dict) else {}\n calls += int(number(usage.get(\"calls\")) or 0)\n spent = number(usage.get(\"reportedCostUsd\"))\n if spent is not None:\n cost = (cost or 0.0) + spent\n reported = number(usage.get(\"reportedTokens\"))\n if reported is not None:\n tokens = (tokens or 0) + int(reported)\n return {\n \"runs\": runs,\n \"calls\": calls,\n \"reportedCostUsd\": round(cost, 6) if cost is not None else None,\n \"reportedTokens\": tokens,\n }\n\n\ndef budget_payload(folder: Path, updates: Dict[str, Any], clear: bool) -> Dict[str, Any]:\n folder = folder.expanduser().resolve()\n path = folder / ORCHESTRATION_BUDGET_FILE\n if clear:\n if path.exists():\n path.unlink()\n limits: Dict[str, Any] = {}\n else:\n limits = load_orchestration_budget(folder)\n if updates:\n limits = validate_orchestration_limits({**limits, **updates}, \"budget\")\n folder.mkdir(parents=True, exist_ok=True)\n atomic_write_json(path, {\"version\": 1, \"limits\": limits, \"updatedAt\": now_iso()})\n used = orchestration_usage(folder)\n remaining: Dict[str, Any] = {}\n if \"maxCalls\" in limits:\n remaining[\"calls\"] = max(0, limits[\"maxCalls\"] - used[\"calls\"])\n if \"maxReportedCostUsd\" in limits:\n remaining[\"reportedCostUsd\"] = round(max(0.0, limits[\"maxReportedCostUsd\"] - (used[\"reportedCostUsd\"] or 0)), 6)\n if \"maxReportedTokens\" in limits:\n remaining[\"reportedTokens\"] = max(0, limits[\"maxReportedTokens\"] - (used[\"reportedTokens\"] or 0))\n return {\n \"path\": str(path),\n \"limits\": limits,\n \"used\": used,\n \"remaining\": remaining,\n \"note\": \"Reported cost covers only CLIs that report it; token and call limits cover the rest.\",\n }\n\n\ndef run_with_worker(folder: Path, worker_id: str) -> Path:\n for run_dir in reversed(wave_dirs(folder)):\n state = read_optional_json(run_dir / \"state.json\")\n if isinstance(state, dict) and worker_id in state.get(\"workers\", {}):\n return run_dir\n raise PlanError(f\"{folder} has no worker {worker_id}\")\n\n\ndef mark_payload(run_dir: Path, worker_id: str, mark: str, why: Optional[str]) -> Dict[str, Any]:\n run_dir = run_with_worker(run_dir.expanduser().resolve(), worker_id)\n entry = {\n \"worker\": worker_id,\n \"mark\": mark,\n \"why\": optional_note(why, \"mark why\", NOTE_LIMITS[\"workerWhy\"]),\n \"markedAt\": now_iso(),\n }\n atomic_write_json(run_dir / \"workers\" / worker_id / \"mark.json\", entry)\n return entry\n\n\ndef orchestrations_home() -> Path:\n configured = os.environ.get(\"ALETHE_ORCHESTRATE_HOME\")\n if configured:\n return Path(configured).expanduser()\n return default_policy_path().parent / \"orchestrations\"\n\n\ndef slugify(text: str, fallback: str) -> str:\n plain = unicodedata.normalize(\"NFKD\", text).encode(\"ascii\", \"ignore\").decode(\"ascii\")\n value = re.sub(r\"[^a-z0-9]+\", \"-\", plain.lower()).strip(\"-\")[:40].strip(\"-\")\n return value or fallback\n\n\ndef set_current(folder: Path) -> None:\n home = orchestrations_home()\n home.mkdir(parents=True, exist_ok=True)\n atomic_write_json(home / CURRENT_FILE, {\"version\": 1, \"folder\": str(folder), \"updatedAt\": now_iso()})\n\n\ndef current_folder() -> Optional[Path]:\n raw = read_optional_json(orchestrations_home() / CURRENT_FILE)\n folder = raw.get(\"folder\") if isinstance(raw, dict) else None\n if not isinstance(folder, str) or not Path(folder).is_dir():\n return None\n return Path(folder)\n\n\ndef resolve_folder(raw: Optional[Path]) -> Path:\n if raw is not None:\n return raw.expanduser().resolve()\n current = current_folder()\n if current is None:\n raise PlanError(\"no current orchestration; pass its folder or start one with new\")\n return current\n\n\ndef wave_dirs(folder: Path) -> List[Path]:\n if (folder / \"state.json\").is_file():\n return [folder]\n if not folder.is_dir():\n return []\n return sorted(child for child in folder.iterdir() if (child / \"state.json\").is_file())\n\n\ndef next_wave_dir(folder: Path, phase: Any) -> Path:\n if (folder / \"state.json\").is_file():\n raise PlanError(f\"{folder} is a single run; start an orchestration with new to add waves\")\n numbered = [child for child in folder.iterdir() if child.is_dir() and re.match(r\"^\\d{2,}-\", child.name)] if folder.is_dir() else []\n return folder / f\"{len(numbered) + 1:02d}-{slugify(phase if isinstance(phase, str) else '', 'wave')}\"\n\n\ndef lead_identity(spec: Optional[str]) -> Optional[Dict[str, Any]]:\n if spec:\n provider, _, model = spec.partition(\":\")\n return {\"provider\": optional_name(provider.lower(), \"--lead\"), \"model\": optional_name(model or None, \"--lead model\"),\n \"effort\": None, \"source\": \"flag\"}\n if os.environ.get(\"CLAUDECODE\") == \"1\":\n effort = os.environ.get(\"CLAUDE_EFFORT\") or None\n return {\"provider\": \"claude\", \"model\": None,\n \"effort\": effort if effort and SAFE_NAME_PATTERN.fullmatch(effort) else None, \"source\": \"environment\"}\n return None\n\n\ndef lead_text(lead: Optional[Dict[str, Any]]) -> Optional[str]:\n if not lead:\n return None\n detail = \", \".join(str(part) for part in (lead.get(\"model\"), lead.get(\"effort\")) if part)\n return f\"{lead.get('provider')} ({detail})\" if detail else str(lead.get(\"provider\"))\n\n\ndef new_orchestration(title: str, playbook: Optional[str], why: Optional[str], lead: Optional[str] = None) -> Dict[str, Any]:\n meta = {\n \"version\": 1,\n \"title\": optional_note(require_string(title, \"title\"), \"title\", NOTE_LIMITS[\"title\"]),\n \"playbook\": optional_name(playbook, \"playbook\"),\n \"why\": optional_note(why, \"why\", NOTE_LIMITS[\"why\"]),\n \"lead\": lead_identity(lead),\n \"createdAt\": now_iso(),\n }\n stamp = dt.datetime.now().strftime(\"%Y%m%d-%H%M%S\")\n folder = orchestrations_home() / f\"{stamp}-{slugify(meta['title'], 'orchestration')}\"\n folder.mkdir(parents=True)\n atomic_write_json(folder / ORCHESTRATION_FILE, meta)\n set_current(folder)\n return {\"folder\": str(folder), **meta}\n\n\ndef orchestration_meta(folder: Optional[Path]) -> Dict[str, Any]:\n raw = read_optional_json(folder / ORCHESTRATION_FILE) if folder is not None else None\n return raw if isinstance(raw, dict) else {}\n\n\ndef add_note(folder: Path, text: str, kind: str) -> Dict[str, Any]:\n if not folder.is_dir():\n raise PlanError(f\"not an orchestration folder: {folder}\")\n note = optional_note(text, \"note\", NOTE_LIMITS[\"note\"])\n if not note:\n raise PlanError(\"a note needs text\")\n entry = {\"at\": now_iso(), \"kind\": kind, \"text\": note}\n with (folder / JOURNAL_FILE).open(\"a\", encoding=\"utf-8\") as handle:\n handle.write(json.dumps(entry, ensure_ascii=False) + \"\\n\")\n return entry\n\n\ndef read_journal(folder: Optional[Path]) -> List[Dict[str, Any]]:\n if folder is None:\n return []\n try:\n lines = (folder / JOURNAL_FILE).read_text(encoding=\"utf-8\").splitlines()\n except OSError:\n return []\n entries = []\n for line in lines:\n try:\n entry = json.loads(line)\n except ValueError:\n continue\n if isinstance(entry, dict) and entry.get(\"kind\") in NOTE_KINDS and isinstance(entry.get(\"text\"), str):\n entries.append(entry)\n return entries\n\n\ndef stop_payload(folder: Path) -> Dict[str, Any]:\n running = []\n for run_dir in wave_dirs(folder):\n state = read_optional_json(run_dir / \"state.json\")\n if not isinstance(state, dict) or state.get(\"status\") not in (\"running\", \"interrupted\"):\n continue\n atomic_write_text(run_dir / STOP_FILE, now_iso() + \"\\n\")\n for entry in state.get(\"workers\", {}).values():\n stop_recorded_pid(entry.get(\"pid\"))\n if effective_run_state(state)[\"status\"] == \"interrupted\":\n atomic_write_json(run_dir / \"state.json\", effective_run_state(state))\n running.append(str(run_dir))\n return {\n \"folder\": str(folder),\n \"stopping\": running,\n \"note\": \"running workers stop within seconds and queued ones are skipped\" if running else \"nothing is running\",\n }\n\n\ndef report_target(paths: Sequence[Path]) -> Tuple[List[Path], Optional[Path]]:\n if paths:\n folder = paths[0].expanduser().resolve() if len(paths) == 1 else None\n return find_run_dirs(paths), folder\n current = current_folder()\n if current is None:\n return find_run_dirs([]), None\n runs = wave_dirs(current)\n if not runs:\n raise PlanError(f\"{current} has no waves yet; run one with run --in current\")\n return runs, current\n\n\ndef list_payload(limit: int) -> List[Dict[str, Any]]:\n home = orchestrations_home()\n folders = [child for child in home.iterdir() if child.is_dir()] if home.is_dir() else []\n current = current_folder()\n if current is not None and current.resolve() not in {folder.resolve() for folder in folders}:\n folders.append(current)\n entries = []\n for folder in folders:\n meta = orchestration_meta(folder)\n runs = wave_dirs(folder)\n entry = {\n \"folder\": str(folder),\n \"current\": current is not None and folder.resolve() == current.resolve(),\n \"title\": meta.get(\"title\"),\n \"playbooks\": [meta[\"playbook\"]] if meta.get(\"playbook\") else [],\n \"startedAt\": meta.get(\"createdAt\"),\n \"waves\": len(runs),\n \"status\": None,\n \"calls\": 0,\n \"tokens\": None,\n \"costUsd\": None,\n \"planChanges\": len(read_journal(folder)),\n }\n if runs:\n try:\n report = build_report(runs, folder)\n except (PlanError, OSError):\n report = None\n if report is not None:\n totals = report[\"totals\"]\n entry.update(\n title=report[\"title\"],\n playbooks=report[\"playbooks\"],\n startedAt=entry[\"startedAt\"] or report[\"startedAt\"],\n status=report[\"runs\"][-1][\"status\"],\n calls=totals[\"calls\"],\n tokens=totals[\"tokens\"],\n costUsd=totals[\"costUsd\"],\n )\n if runs or meta:\n entries.append(entry)\n entries.sort(key=lambda entry: entry[\"startedAt\"] or \"\", reverse=True)\n return entries[:limit]\n\n\ndef render_list(entries: Sequence[Dict[str, Any]]) -> str:\n if not entries:\n return \"No orchestrations yet.\\n\"\n lines = []\n for entry in entries:\n started = local_time(entry[\"startedAt\"])\n parts = [entry[\"title\"] or Path(entry[\"folder\"]).name]\n if entry[\"playbooks\"]:\n parts.append(\"playbook \" + \", \".join(entry[\"playbooks\"]))\n parts.append(counted(entry[\"waves\"], \"wave\") + (f\", {entry['status']}\" if entry[\"status\"] else \"\"))\n parts.append(f\"{counted(entry['calls'], 'call')}, {spend_text(entry['tokens'], entry['costUsd'])}\")\n if entry[\"planChanges\"]:\n parts.append(counted(entry[\"planChanges\"], \"plan change\"))\n when = started.strftime(\"%Y-%m-%d %H:%M\") if started else \"----------------\"\n lines.append(f\"{'*' if entry['current'] else ' '} {when} \" + \" | \".join(parts))\n lines.append(f\" {entry['folder']}\")\n return \"\\n\".join(lines) + \"\\n\"\n\n\ndef mode_scope(cwd: Path) -> Path:\n cwd = cwd.expanduser().resolve()\n return checkout_root(cwd) or cwd\n\n\ndef read_mode_state(now: float) -> Dict[str, Any]:\n raw = read_optional_json(orchestrations_home() / MODES_FILE)\n raw = raw if isinstance(raw, dict) else {}\n modes = raw.get(\"modes\") if isinstance(raw.get(\"modes\"), dict) else {}\n everywhere = raw.get(\"everywhere\")\n return {\n \"modes\": {\n scope: mode\n for scope, mode in modes.items()\n if isinstance(mode, dict) and now - float(number(mode.get(\"seenAt\")) or 0) \u003c MODE_IDLE_SECONDS\n },\n \"everywhere\": everywhere if isinstance(everywhere, dict) else None,\n \"offIn\": [scope for scope in raw.get(\"offIn\") or [] if isinstance(scope, str)],\n }\n\n\ndef save_mode_state(state: Dict[str, Any]) -> None:\n home = orchestrations_home()\n home.mkdir(parents=True, exist_ok=True)\n atomic_write_json(home / MODES_FILE, {\"version\": 1, **state})\n\n\ndef covers(scope: str, here: Path) -> bool:\n return here == Path(scope) or is_inside(here, Path(scope))\n\n\ndef effective_mode(state: Dict[str, Any], here: Path) -> Tuple[Optional[str], Optional[Dict[str, Any]]]:\n matches = [scope for scope in state[\"modes\"] if covers(scope, here)]\n if matches:\n scope = max(matches, key=len)\n return scope, state[\"modes\"][scope]\n if state[\"everywhere\"] and not any(covers(scope, here) for scope in state[\"offIn\"]):\n return None, state[\"everywhere\"]\n return None, None\n\n\ndef updated_mode(\n mode: Optional[Dict[str, Any]], goal: Optional[str], care: Optional[str], rules: Sequence[str], clear_rules: bool\n) -> Dict[str, Any]:\n mode = dict(mode or {\"goal\": \"balanced\", \"care\": \"normal\", \"rules\": [], \"since\": now_iso()})\n if goal:\n mode[\"goal\"] = goal\n if care:\n mode[\"care\"] = care\n kept = [] if clear_rules else list(mode.get(\"rules\") or [])\n for rule in rules:\n text = optional_note(rule, \"rule\", NOTE_LIMITS[\"note\"])\n if text and text not in kept:\n kept.append(text)\n if len(kept) > MAX_MODE_RULES:\n raise PlanError(f\"keep at most {MAX_MODE_RULES} rules; pass --clear-rules to start over\")\n mode[\"rules\"] = kept\n return mode\n\n\ndef mode_payload(\n action: str,\n cwd: Path,\n goal: Optional[str],\n care: Optional[str],\n rules: Sequence[str],\n clear_rules: bool,\n everywhere: bool = False,\n) -> Dict[str, Any]:\n now = time.time()\n state = read_mode_state(now)\n if everywhere:\n if action == \"off\":\n state.update(everywhere=None, offIn=[])\n save_mode_state(state)\n elif action == \"on\":\n state[\"everywhere\"] = updated_mode(state[\"everywhere\"], goal, care, rules, clear_rules)\n save_mode_state(state)\n return {\"everywhere\": state[\"everywhere\"]}\n scope = str(mode_scope(cwd))\n if action == \"show\":\n found, mode = effective_mode(state, Path(scope))\n source = None if mode is None else \"project\" if found else \"everywhere\"\n return {\"scope\": scope, \"on\": mode is not None, \"source\": source, **(mode or {})}\n if action == \"off\":\n state[\"modes\"].pop(scope, None)\n if state[\"everywhere\"] and scope not in state[\"offIn\"]:\n state[\"offIn\"].append(scope)\n save_mode_state(state)\n return {\"scope\": scope, \"on\": False}\n state[\"offIn\"] = [entry for entry in state[\"offIn\"] if entry != scope]\n mode = updated_mode(state[\"modes\"].get(scope), goal, care, rules, clear_rules)\n mode[\"seenAt\"] = now\n state[\"modes\"][scope] = mode\n save_mode_state(state)\n return {\"scope\": scope, \"on\": True, \"source\": \"project\", **mode}\n\n\ndef mode_reminder(mode: Dict[str, Any], everywhere: bool) -> str:\n goal = mode.get(\"goal\") if mode.get(\"goal\") in MODE_GOALS else \"balanced\"\n where = \"in every project\" if everywhere else \"for this project\"\n parts = [\n f\"Alethe orchestration is on {where}, so the user does not need to call it.\",\n f\"Goal: {goal}, {MODE_GOALS[goal]}.\",\n ]\n if mode.get(\"care\") == \"careful\":\n parts.append(f\"Careful: {MODE_CAREFUL}.\")\n rules = [rule for rule in mode.get(\"rules\") or [] if isinstance(rule, str)]\n if rules:\n parts.append(\"User rules: \" + \"; \".join(rules) + \".\")\n if goal == \"freestyle\":\n parts.append(MODE_FREESTYLE + \" '/alethe off' turns this off here.\")\n return \" \".join(parts)\n parts.append(\n \"Follow the alethe skill for each request: do it yourself when that is cheaper, \"\n \"delegate bounded units when it pays, and leave any other skill or spec workflow in charge of \"\n \"its own steps. '/alethe off' turns this off here.\"\n )\n return \" \".join(parts)\n\n\ndef hook_stdin() -> str:\n stream = getattr(sys.stdin, \"buffer\", None)\n if stream is None:\n return sys.stdin.read()\n return stream.read().decode(\"utf-8\", errors=\"replace\")\n\n\ndef hook_output(stdin_text: str) -> str:\n try:\n event = json.loads(stdin_text) if stdin_text.strip() else {}\n cwd = Path(event.get(\"cwd\") or os.getcwd()) if isinstance(event, dict) else Path(os.getcwd())\n now = time.time()\n state = read_mode_state(now)\n scope, mode = effective_mode(state, cwd.expanduser().resolve())\n if mode is None:\n return \"\"\n if scope is not None and now - float(number(mode.get(\"seenAt\")) or 0) > MODE_REFRESH_SECONDS:\n mode[\"seenAt\"] = now\n save_mode_state(state)\n return mode_reminder(mode, scope is None) + \"\\n\"\n except Exception: # A hook must never block the user's prompt.\n return \"\"\n\n\ndef task_text(worker: Dict[str, Any]) -> str:\n return worker[\"task\"].split(\"\\n\\nRuntime contract:\", 1)[0]\n\n\ndef dirty_checkout(cwd: str) -> bool:\n try:\n return bool(run_git(Path(cwd), [\"status\", \"--porcelain\"], timeout=60).strip())\n except PlanError:\n return False\n\n\ndef plan_warnings(plan: Dict[str, Any]) -> List[str]:\n workers = plan[\"workers\"]\n warnings: List[str] = []\n blind = []\n for worker in workers:\n label = Path(worker[\"command\"]).stem if worker[\"provider\"] == \"custom\" else worker[\"provider\"]\n if worker[\"provider\"] not in COST_REPORTING_PROVIDERS and label not in blind:\n blind.append(label)\n if \"maxReportedCostUsd\" in plan[\"budget\"] and blind:\n warnings.append(\n f\"maxReportedCostUsd only sees cost that Claude reports; {listed(blind)} \"\n f\"{'report' if len(blind) > 1 else 'reports'} none, so cap {'them' if len(blind) > 1 else 'it'} \"\n \"with maxReportedTokens or maxCalls\"\n )\n writers = [worker for worker in workers if worker[\"mode\"] == \"write\"]\n small = [worker for worker in writers if len(task_text(worker).split()) \u003c= 50]\n if len(writers) >= 2 and len(small) == len(writers):\n warnings.append(\n f\"{len(writers)} writers with short briefs: each worker first loads its own CLI context, \"\n \"so these may cost less done directly or as one worker\"\n )\n for worker in workers:\n name = worker[\"id\"]\n if worker[\"provider\"] == \"agy\" and worker[\"idleTimeoutSeconds\"] is not None:\n warnings.append(f\"{name}: agy only emits JSON at completion; idleTimeoutSeconds is ignored\")\n custom_args = worker.get(\"args\", []) if worker[\"provider\"] == \"custom\" else []\n if not (worker[\"model\"] or flag_value(custom_args, \"--model\")):\n warnings.append(\n f\"{name} runs on {worker['provider']}'s default model, which may be its most expensive; set a tier or a model\"\n )\n if worker[\"provider\"] == \"kimi\" and worker[\"mode\"] == \"read-only\":\n warnings.append(\n f\"{name}: Kimi runs prompts with every permission, so read-only rests on the worker contract; \"\n \"the runner checks the checkout afterwards\"\n )\n if worker[\"provider\"] == \"kimi\" and worker[\"mode\"] == \"write\" and not worker[\"isolate\"]:\n warnings.append(f\"{name}: Kimi writes without asking; give it isolate: true\")\n if (\n worker[\"provider\"] in (\"codex\", \"claude\")\n and not worker[\"mcp\"]\n and re.search(r\"\\bmcp\\b\", task_text(worker), re.IGNORECASE)\n ):\n warnings.append(f\"{name} mentions MCP, but workers start without MCP servers; set mcp: true if it needs one\")\n if worker[\"mode\"] == \"write\" and not worker[\"isolate\"] and dirty_checkout(worker[\"cwd\"]):\n warnings.append(\n f\"{name} writes in a checkout with uncommitted changes, so its diff will include them; \"\n \"use isolate: true or commit first\"\n )\n return warnings\n\n\ndef default_run_dir() -> Path:\n stamp = dt.datetime.now().strftime(\"%Y%m%d-%H%M%S\")\n suffix = uuid.uuid4().hex[:8]\n return Path(tempfile.gettempdir()) / f\"alethe-orchestrate-{stamp}-{suffix}\"\n\n\ndef wait_for_workers(futures: Sequence[concurrent.futures.Future]) -> List[Dict[str, Any]]:\n pending = set(futures)\n outcomes = []\n while pending:\n done, pending = concurrent.futures.wait(\n pending, timeout=POLL_SECONDS, return_when=concurrent.futures.FIRST_COMPLETED\n )\n outcomes.extend(future.result() for future in done)\n return outcomes\n\n\ndef install_interruption_handlers() -> Dict[int, Any]:\n if threading.current_thread() is not threading.main_thread():\n return {}\n interrupted = False\n\n def interrupt(signum: int, frame: Any) -> None:\n nonlocal interrupted\n if interrupted:\n return\n interrupted = True\n raise KeyboardInterrupt\n\n names = (\"SIGINT\", \"SIGBREAK\") if os.name == \"nt\" else (\"SIGINT\", \"SIGTERM\", \"SIGHUP\")\n previous = {}\n for name in names:\n signum = getattr(signal, name)\n previous[signum] = signal.signal(signum, interrupt)\n return previous\n\n\ndef refuse_low_quota(workers: Sequence[Dict[str, Any]], policy_path: Path) -> None:\n refusal = quota_refusal(workers, policy_path)\n if refusal is not None:\n raise PlanError(refusal)\n\n\ndef detach_run(\n plan_path: Path, run_dir: Optional[Path], policy_path: Path, strategy: Optional[str], allow_low_quota: bool = False\n) -> Dict[str, Any]:\n if os.environ.get(WORKER_ENV_FLAG):\n raise PlanError(\"this process is an orchestration worker, and workers cannot start orchestrations\")\n plan = validate_plan(read_json(plan_path), plan_path.resolve(), load_policy(policy_path), strategy)\n if not allow_low_quota:\n refuse_low_quota(plan[\"workers\"], policy_path)\n target = (run_dir or default_run_dir()).expanduser().resolve()\n if target.exists() and any(target.iterdir()):\n raise PlanError(f\"run directory is not empty: {target}\")\n target.mkdir(parents=True, exist_ok=True)\n args = [\n sys.executable, str(Path(__file__).resolve()), \"run\", str(plan_path.resolve()),\n \"--run-dir\", str(target), \"--policy\", str(policy_path), \"--_background\",\n ]\n if strategy is not None:\n args += [\"--strategy\", strategy]\n if allow_low_quota:\n args.append(\"--allow-low-quota\")\n options: Dict[str, Any] = {\"shell\": False}\n if os.name == \"nt\":\n options[\"creationflags\"] = (\n subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW\n )\n else:\n options[\"start_new_session\"] = True\n with (target / RUNNER_STDOUT_FILE).open(\"wb\") as stdout, (target / RUNNER_STDERR_FILE).open(\"wb\") as stderr:\n process = subprocess.Popen(args, stdin=subprocess.PIPE, stdout=stdout, stderr=stderr, **options)\n try:\n atomic_write_json(target / \"plan.json\", plan_payload(plan, plan_path.resolve(), policy_path, dry_run=False))\n atomic_write_json(target / \"state.json\", {\n \"version\": STATE_VERSION, \"status\": \"running\", \"runnerPid\": process.pid,\n \"runDir\": str(target), \"startedAt\": now_iso(),\n \"workers\": {worker[\"id\"]: {\n \"id\": worker[\"id\"], \"provider\": worker[\"provider\"], \"status\": \"queued\",\n } for worker in plan[\"workers\"]},\n })\n set_current(target.parent if (target.parent / ORCHESTRATION_FILE).is_file() else target)\n process.stdin.write(b\"1\")\n process.stdin.close()\n except BaseException:\n process.kill()\n process.wait()\n raise\n return {\"status\": \"started\", \"runDir\": str(target), \"pid\": process.pid}\n\n\ndef wait_payload(folder: Optional[Path], timeout: int) -> Dict[str, Any]:\n if not 0 \u003c= timeout \u003c= MAX_WAIT_SECONDS:\n raise PlanError(f\"--timeout must be between 0 and {MAX_WAIT_SECONDS} seconds\")\n paths = [folder] if folder is not None else []\n deadline = time.monotonic() + timeout\n while True:\n report = build_report(*report_target(paths))\n done = all(run[\"status\"] != \"running\" for run in report[\"runs\"])\n remaining = deadline - time.monotonic()\n if done or remaining \u003c= 0:\n return {**report, \"done\": done}\n time.sleep(min(POLL_SECONDS, remaining))\n\n\ndef run_batch(\n plan_path: Path,\n requested_run_dir: Optional[Path],\n policy_path: Path,\n strategy_override: Optional[str],\n background: bool = False,\n allow_low_quota: bool = False,\n) -> int:\n if os.environ.get(WORKER_ENV_FLAG):\n raise PlanError(\"this process is an orchestration worker, and workers cannot start orchestrations\")\n plan = validate_plan(\n read_json(plan_path), plan_path.resolve(), load_policy(policy_path), strategy_override, policy_path\n )\n if not (allow_low_quota or background):\n refuse_low_quota(plan[\"workers\"], policy_path)\n started = dt.datetime.now(dt.timezone.utc)\n try:\n save_usage(policy_path, codex_quota())\n quota_before = saved_usage(policy_path)\n except (OSError, PlanError):\n quota_before = {}\n run_dir = (requested_run_dir or default_run_dir()).expanduser().resolve()\n allowed = {RUNNER_STDOUT_FILE, RUNNER_STDERR_FILE, \"state.json\", \"plan.json\"} if background else set()\n if run_dir.exists() and any(path.name not in allowed for path in run_dir.iterdir()):\n raise PlanError(f\"run directory is not empty: {run_dir}\")\n outer = None\n if requested_run_dir is not None:\n limits = load_orchestration_budget(run_dir.parent)\n if limits:\n used = orchestration_usage(run_dir.parent, exclude=run_dir)\n refusal = orchestration_refusal(limits, used)\n if refusal:\n raise PlanError(f\"{refusal}; raise it with budget {run_dir.parent}\")\n outer = {\n \"path\": str(run_dir.parent / ORCHESTRATION_BUDGET_FILE),\n \"limits\": limits,\n \"used\": used,\n }\n run_dir.mkdir(parents=True, exist_ok=True)\n atomic_write_json(\n run_dir / \"plan.json\", plan_payload(plan, plan_path.resolve(), policy_path, dry_run=False)\n )\n try:\n set_current(run_dir.parent if (run_dir.parent / ORCHESTRATION_FILE).is_file() else run_dir)\n except OSError:\n pass\n suffix = run_dir.name[-8:]\n run_id = suffix if re.fullmatch(r\"[0-9a-f]{8}\", suffix) else uuid.uuid4().hex[:8]\n initial_workers = {\n worker[\"id\"]: {\n \"id\": worker[\"id\"],\n \"provider\": worker[\"provider\"],\n \"status\": \"queued\",\n \"cwd\": worker[\"cwd\"],\n \"worktree\": None,\n }\n for worker in plan[\"workers\"]\n }\n state = BatchState(\n run_dir / \"state.json\",\n {\n \"version\": STATE_VERSION,\n \"runId\": run_id,\n \"runnerPid\": os.getpid(),\n \"status\": \"running\",\n \"strategy\": plan[\"strategy\"],\n \"requestedStrategy\": plan[\"requestedStrategy\"],\n \"maxConcurrent\": plan[\"maxConcurrent\"],\n \"policyPath\": str(policy_path),\n \"allowLowQuota\": allow_low_quota,\n \"planPath\": str(plan_path.resolve()),\n \"runDir\": str(run_dir),\n \"startedAt\": now_iso(),\n \"finishedAt\": None,\n \"budget\": plan[\"budget\"],\n \"orchestrationBudget\": outer,\n \"workers\": initial_workers,\n },\n )\n help_cache: Dict[str, str] = {}\n help_lock = threading.Lock()\n registry = ProcessRegistry()\n checkouts = CheckoutRegistry()\n tracker = BudgetTracker(plan[\"budget\"], outer, plan[\"maxConcurrent\"], len(plan[\"workers\"]), state)\n state.update_budget(tracker.usage())\n previous_signals = install_interruption_handlers()\n executor = concurrent.futures.ThreadPoolExecutor(max_workers=plan[\"maxConcurrent\"])\n futures: List[concurrent.futures.Future] = []\n try:\n futures = [\n executor.submit(\n run_worker,\n worker,\n run_id,\n run_dir,\n state,\n help_cache,\n help_lock,\n registry,\n tracker,\n checkouts,\n )\n for worker in plan[\"workers\"]\n ]\n outcomes = wait_for_workers(futures)\n except KeyboardInterrupt:\n try:\n atomic_write_text(run_dir / STOP_FILE, now_iso() + \"\\n\")\n finally:\n registry.stop_all()\n for future in futures:\n future.cancel()\n executor.shutdown(wait=True, cancel_futures=True)\n state.finish(\"interrupted\", {\"budgetUsage\": tracker.usage()})\n print(json.dumps({\"status\": \"interrupted\", \"runDir\": str(run_dir)}))\n return 130\n finally:\n executor.shutdown(wait=True, cancel_futures=True)\n for signum, previous in previous_signals.items():\n signal.signal(signum, previous)\n\n succeeded = sum(outcome[\"status\"] == \"succeeded\" for outcome in outcomes)\n status = \"done\" if succeeded == len(outcomes) else \"partial\" if succeeded else \"failed\"\n if any(outcome[\"failureKind\"] == \"stopped\" for outcome in outcomes):\n status = \"stopped\"\n try:\n save_quota_samples(policy_path, quota_samples(plan[\"workers\"], outcomes, quota_before, saved_usage(policy_path), started))\n except (OSError, PlanError):\n pass\n budget_usage = tracker.usage()\n state.finish(status, {\"budgetUsage\": budget_usage})\n print(\n json.dumps(\n {\n \"status\": status,\n \"runDir\": str(run_dir),\n \"budget\": plan[\"budget\"],\n \"budgetUsage\": budget_usage,\n \"workers\": outcomes,\n },\n ensure_ascii=False,\n )\n )\n return 0 if status == \"done\" else 1\n\n\ndef flag_value(args: Sequence[str], flag: str) -> Optional[str]:\n for index, arg in enumerate(args):\n if arg == flag and index + 1 \u003c len(args):\n return args[index + 1]\n if arg.startswith(flag + \"=\"):\n return arg[len(flag) + 1 :]\n return None\n\n\ndef plan_payload(\n plan: Dict[str, Any], plan_path: Path, policy_path: Path, dry_run: bool = True\n) -> Dict[str, Any]:\n if plan[\"requestedStrategy\"] == \"auto\":\n strategy_reason = f\"auto inferred: {infer_strategy(plan['workers'])[1]}\"\n else:\n strategy_reason = \"requested explicitly\"\n wall_limit = plan[\"budget\"].get(\"maxWallSeconds\")\n workers = []\n for worker in plan[\"workers\"]:\n effective_timeout = worker[\"timeoutSeconds\"]\n if wall_limit is not None:\n effective_timeout = min(effective_timeout, wall_limit)\n custom_args = worker.get(\"args\", []) if worker[\"provider\"] == \"custom\" else []\n workers.append(\n {\n \"id\": worker[\"id\"],\n \"why\": worker[\"why\"],\n \"requestedProvider\": worker[\"requestedProvider\"],\n \"provider\": worker[\"provider\"],\n \"command\": Path(worker[\"command\"]).stem if worker[\"provider\"] == \"custom\" else None,\n \"switched\": worker[\"provider\"] != worker[\"requestedProvider\"],\n \"routingNote\": worker[\"routingNote\"],\n \"tier\": worker[\"tier\"],\n \"model\": worker[\"model\"] or flag_value(custom_args, \"--model\"),\n \"effort\": worker[\"effort\"] or flag_value(custom_args, \"--effort\"),\n \"modelNote\": worker[\"modelNote\"],\n \"mode\": worker[\"mode\"],\n \"isolate\": worker[\"isolate\"],\n \"seedPaths\": worker[\"seedPaths\"],\n \"timeoutSeconds\": worker[\"timeoutSeconds\"],\n \"effectiveTimeoutSeconds\": effective_timeout,\n \"promptChars\": len(worker[\"task\"]),\n \"estimatedPromptTokens\": worker[\"estimatedPromptTokens\"],\n \"subagents\": \"blocked by the CLI\"\n if worker[\"provider\"] in (\"codex\", \"claude\")\n else \"refused by the worker contract\",\n \"mcp\": worker[\"mcp\"] if worker[\"provider\"] in (\"codex\", \"claude\") else None,\n \"idleTimeoutSeconds\": worker[\"idleTimeoutSeconds\"],\n \"telemetry\": {\n \"tokenUsage\": PROVIDER_CAPABILITIES.get(worker[\"provider\"], {}).get(\n \"tokenUsage\", False\n )\n },\n }\n )\n advisory = preview_usage_cost(workers, policy_path)\n workers = [{**worker, **estimate} for worker, estimate in zip(workers, advisory[\"workers\"])]\n return {\n \"dryRun\": dry_run,\n \"skillVersion\": skill_version(),\n \"planPath\": str(plan_path),\n \"policyPath\": str(policy_path),\n \"title\": plan[\"title\"],\n \"why\": plan[\"why\"],\n \"playbook\": plan[\"playbook\"],\n \"phase\": plan[\"phase\"],\n \"strategy\": plan[\"strategy\"],\n \"requestedStrategy\": plan[\"requestedStrategy\"],\n \"strategyReason\": strategy_reason,\n \"maxConcurrent\": plan[\"maxConcurrent\"],\n \"budget\": plan[\"budget\"],\n \"workers\": workers,\n \"totals\": {\n \"workers\": len(workers),\n \"estimatedPromptTokens\": sum(worker[\"estimatedPromptTokens\"] for worker in workers),\n \"listPriceEstimateUsd\": advisory[\"total\"],\n },\n \"estimates\": (\n \"prompt-only chars/4 heuristic; provider-side context such as repository reads \"\n \"and tool calls is additional and cannot be estimated\"\n ),\n \"listPriceEstimate\": {\"label\": advisory[\"label\"], \"asOf\": advisory[\"asOf\"]},\n \"quotaForecast\": advisory[\"quotaForecast\"],\n \"warnings\": plan_warnings(plan) + advisory[\"warnings\"],\n }\n\n\ndef find_run_dirs(paths: Sequence[Path]) -> List[Path]:\n if not paths:\n runs = [\n candidate\n for candidate in Path(tempfile.gettempdir()).glob(\"alethe-orchestrate-*\")\n if (candidate / \"state.json\").is_file()\n ]\n if not runs:\n raise PlanError(\"no runs found; pass the run directory that run printed\")\n return [max(runs, key=lambda candidate: (candidate / \"state.json\").stat().st_mtime)]\n found: List[Path] = []\n for raw in paths:\n path = raw.expanduser().resolve()\n if (path / \"state.json\").is_file():\n candidates = [path]\n elif path.is_dir():\n candidates = sorted(child for child in path.iterdir() if (child / \"state.json\").is_file())\n else:\n raise PlanError(f\"not a run directory: {path}\")\n if not candidates:\n raise PlanError(f\"no runs found in {path}\")\n for candidate in candidates:\n if candidate not in found:\n found.append(candidate)\n return found\n\n\ndef number(value: Any) -> Optional[float]:\n if isinstance(value, (int, float)) and not isinstance(value, bool):\n return value\n return None\n\n\ndef cached_tokens(usage: Any) -> Optional[int]:\n if not isinstance(usage, dict):\n return None\n for key in (\"cacheReadTokens\", \"cache_read_input_tokens\", \"cached_input_tokens\"):\n value = number(usage.get(key))\n if value is not None:\n return int(value)\n return None\n\n\ndef file_bytes(path: Path) -> int:\n try:\n return path.stat().st_size\n except OSError:\n return 0\n\n\ndef diff_stats(path: Path) -> Optional[Dict[str, int]]:\n if not path.is_file():\n return None\n files = insertions = deletions = 0\n in_hunk = False\n with path.open(\"r\", encoding=\"utf-8\", errors=\"replace\") as handle:\n for line in handle:\n if line.startswith(\"diff --git \"):\n files += 1\n in_hunk = False\n elif line.startswith(\"@@\"):\n in_hunk = True\n elif in_hunk and line.startswith(\"+\"):\n insertions += 1\n elif in_hunk and line.startswith(\"-\"):\n deletions += 1\n return {\"files\": files, \"insertions\": insertions, \"deletions\": deletions}\n\n\ndef result_summary(path: Path, limit: int = 200) -> Optional[str]:\n if not path.is_file():\n return None\n first = None\n for raw in path.read_text(encoding=\"utf-8\", errors=\"replace\").splitlines():\n line = re.sub(r\"^(?:[-*+>]|\\d+[.)])\\s+\", \"\", raw.strip()).replace(\"**\", \"\").strip()\n if not line or line.startswith((\"#\", \"|\", \"```\")) or set(line) \u003c= set(\"-*_=| \") or HANDOFF_LINE.match(line):\n continue\n first = first or line\n if len(line.split()) >= 3 or re.search(r\"[.!?]$\", line):\n first = line\n break\n if first is None:\n return None\n return first if len(first) \u003c= limit else first[: limit - 3].rstrip() + \"...\"\n\n\ndef last_line(path: Path, limit: int = 200) -> Optional[str]:\n size = file_bytes(path)\n if not size:\n return None\n with path.open(\"rb\") as handle:\n handle.seek(max(0, size - 4096))\n tail = handle.read().decode(\"utf-8\", errors=\"replace\")\n lines = [line.strip() for line in tail.splitlines() if line.strip()]\n if not lines:\n return None\n errors = [line for line in lines[-20:] if ERROR_LINE.search(line)]\n line = errors[-1] if errors else lines[-1]\n return line if len(line) \u003c= limit else line[: limit - 3].rstrip() + \"...\"\n\n\ndef seconds_between(started: Any, finished: Any) -> Optional[float]:\n try:\n return (dt.datetime.fromisoformat(finished) - dt.datetime.fromisoformat(started)).total_seconds()\n except (TypeError, ValueError):\n return None\n\n\ndef collect_run(run_dir: Path) -> Dict[str, Any]:\n state = read_json(run_dir / \"state.json\")\n if not isinstance(state, dict) or not isinstance(state.get(\"workers\"), dict):\n raise PlanError(f\"{run_dir} has no valid state.json\")\n state = effective_run_state(state)\n plan = read_optional_json(run_dir / \"plan.json\")\n plan = plan if isinstance(plan, dict) else {}\n planned = {\n entry[\"id\"]: entry\n for entry in plan.get(\"workers\", [])\n if isinstance(entry, dict) and isinstance(entry.get(\"id\"), str)\n }\n workers = []\n for worker_id, entry in state[\"workers\"].items():\n worker_dir = run_dir / \"workers\" / worker_id\n outcome = read_optional_json(worker_dir / \"outcome.json\")\n outcome = outcome if isinstance(outcome, dict) else entry\n intent = planned.get(worker_id, {})\n metrics = outcome.get(\"metrics\") if isinstance(outcome.get(\"metrics\"), dict) else {}\n result_path = worker_dir / \"result.md\"\n stderr_path = worker_dir / \"stderr.log\"\n diff_path = worker_dir / \"diff.patch\"\n provider = outcome.get(\"provider\") or entry.get(\"provider\")\n status = outcome.get(\"status\", \"unknown\")\n handoff = {key: outcome.get(key) for key in (\"workerStatus\", \"workerSummary\", \"verdict\")}\n if not any(handoff.values()) and result_path.is_file():\n handoff = parse_handoff(result_path.read_text(encoding=\"utf-8\", errors=\"replace\"))\n mark = read_optional_json(worker_dir / \"mark.json\")\n mark = mark if isinstance(mark, dict) and mark.get(\"mark\") in MARKS else {}\n workers.append(\n {\n \"id\": worker_id,\n \"why\": intent.get(\"why\"),\n \"provider\": provider,\n \"label\": intent.get(\"command\") or provider,\n \"requestedProvider\": intent.get(\"requestedProvider\"),\n \"routingNote\": intent.get(\"routingNote\"),\n \"tier\": intent.get(\"tier\"),\n \"model\": intent.get(\"model\"),\n \"effort\": intent.get(\"effort\"),\n \"modelNote\": intent.get(\"modelNote\"),\n \"mode\": intent.get(\"mode\"),\n \"isolate\": intent.get(\"isolate\"),\n \"status\": status,\n \"error\": outcome.get(\"error\"),\n \"stderrTail\": last_line(stderr_path) if status != \"succeeded\" else None,\n \"durationSeconds\": number(outcome.get(\"durationSeconds\")),\n \"tokens\": reported_total_tokens(metrics.get(\"usage\")),\n \"cachedTokens\": cached_tokens(metrics.get(\"usage\")),\n \"cacheReadTokens\": cached_tokens(metrics.get(\"usage\")),\n \"costByModel\": metrics.get(\"costByModel\", {}),\n \"quotaError\": outcome.get(\"quotaError\"),\n \"costUsd\": number(metrics.get(\"totalCostUsd\")),\n \"sessionId\": outcome.get(\"sessionId\"),\n \"worktree\": outcome.get(\"worktree\"),\n \"summary\": handoff[\"workerSummary\"] or result_summary(result_path),\n \"workerStatus\": handoff[\"workerStatus\"],\n \"verdict\": handoff[\"verdict\"],\n \"failureKind\": outcome.get(\"failureKind\"),\n \"resetHint\": outcome.get(\"resetHint\"),\n \"permissionDenials\": list(outcome.get(\"permissionDenials\") or []),\n \"billedModels\": list(outcome.get(\"billedModels\") or []),\n \"readOnlyViolation\": outcome.get(\"readOnlyViolation\"),\n \"touchedFiles\": outcome.get(\"touchedFiles\"),\n \"dirtyBefore\": outcome.get(\"dirtyBefore\"),\n \"costCapUsd\": number(outcome.get(\"costCapUsd\")),\n \"diffError\": outcome.get(\"diffError\"),\n \"mark\": mark.get(\"mark\"),\n \"markWhy\": mark.get(\"why\"),\n \"resultPath\": str(result_path) if result_path.is_file() else None,\n \"resultBytes\": file_bytes(result_path),\n \"transcriptBytes\": file_bytes(worker_dir / \"stdout.log\") + file_bytes(stderr_path),\n \"stderrPath\": str(stderr_path) if file_bytes(stderr_path) else None,\n \"diff\": diff_stats(diff_path),\n \"diffPath\": str(diff_path) if diff_path.is_file() else None,\n }\n )\n usage = state.get(\"budgetUsage\") if isinstance(state.get(\"budgetUsage\"), dict) else {}\n wall = number(usage.get(\"wallSeconds\"))\n if wall is None:\n wall = seconds_between(state.get(\"startedAt\"), state.get(\"finishedAt\"))\n return {\n \"runDir\": str(run_dir),\n \"runId\": state.get(\"runId\"),\n \"status\": state.get(\"status\"),\n \"title\": plan.get(\"title\"),\n \"why\": plan.get(\"why\"),\n \"playbook\": plan.get(\"playbook\"),\n \"phase\": plan.get(\"phase\"),\n \"strategy\": state.get(\"strategy\"),\n \"requestedStrategy\": state.get(\"requestedStrategy\"),\n \"strategyReason\": plan.get(\"strategyReason\"),\n \"maxConcurrent\": state.get(\"maxConcurrent\"),\n \"startedAt\": state.get(\"startedAt\"),\n \"finishedAt\": state.get(\"finishedAt\"),\n \"wallSeconds\": wall,\n \"budget\": state.get(\"budget\") if isinstance(state.get(\"budget\"), dict) else {},\n \"budgetUsage\": usage,\n \"workers\": workers,\n }\n\n\ndef concerns(worker: Dict[str, Any]) -> List[str]:\n reasons = []\n said = worker[\"workerStatus\"]\n if said in (\"BLOCKED\", \"NEEDS_CONTEXT\"):\n reasons.append(\n f\"it said {said.replace('_', ' ').lower()}; change its context, tier or scope before sending it again\"\n )\n elif said == \"DONE_WITH_CONCERNS\":\n reasons.append(\"it finished with concerns to review\")\n if worker[\"verdict\"] in (\"FAIL\", \"NOT_PROVEN\"):\n reasons.append(f\"its verdict is {worker['verdict'].replace('_', ' ').lower()}\")\n if worker[\"readOnlyViolation\"]:\n touched = list(worker[\"touchedFiles\"] or [])[:5]\n reasons.append(f\"it changed the checkout while read-only ({listed(touched)})\")\n if worker.get(\"quotaError\"):\n reasons.append(f\"quota cache failed: {worker['quotaError']}\")\n denials = worker[\"permissionDenials\"]\n if denials:\n reasons.append(\n f\"{counted(len(denials), 'tool call')} denied ({listed(sorted(set(denials)))}), so its checks may not have run\"\n )\n return reasons\n\n\ndef value_totals(workers: Sequence[Dict[str, Any]]) -> Optional[Dict[str, Any]]:\n if not any(worker[\"mark\"] for worker in workers):\n return None\n groups: Dict[str, Any] = {}\n for key in (*MARKS, \"unmarked\"):\n members = [\n worker\n for worker in workers\n if (worker[\"mark\"] or \"unmarked\") == key and (key != \"unmarked\" or worker[\"status\"] in RAN_STATUSES)\n ]\n groups[key] = {\n \"workers\": [worker[\"id\"] for worker in members],\n \"tokens\": summed([worker[\"tokens\"] for worker in members]),\n \"costUsd\": summed([worker[\"costUsd\"] for worker in members]),\n }\n return groups\n\n\ndef report_budget(runs: Sequence[Dict[str, Any]]) -> Optional[Dict[str, Any]]:\n folders = {Path(run[\"runDir\"]).parent for run in runs}\n if len(folders) != 1:\n return None\n folder = folders.pop()\n try:\n limits = load_orchestration_budget(folder)\n except PlanError:\n return None\n if not limits:\n return None\n return {\"path\": str(folder / ORCHESTRATION_BUDGET_FILE), \"limits\": limits, \"used\": orchestration_usage(folder)}\n\n\ndef summed(values: Sequence[Any]) -> Optional[float]:\n known = [value for value in values if value is not None]\n return sum(known) if known else None\n\n\ndef build_report(run_dirs: Sequence[Path], folder: Optional[Path] = None) -> Dict[str, Any]:\n meta = orchestration_meta(folder)\n runs = sorted((collect_run(path) for path in run_dirs), key=lambda run: run[\"startedAt\"] or \"\")\n workers = [worker for run in runs for worker in run[\"workers\"]]\n ran = [\n worker\n for worker in workers\n if worker[\"status\"] == \"succeeded\"\n or (worker[\"status\"] in RAN_STATUSES and worker[\"transcriptBytes\"])\n ]\n labels: List[str] = []\n for worker in workers:\n if worker[\"label\"] and worker[\"label\"] not in labels:\n labels.append(worker[\"label\"])\n statuses: Dict[str, int] = {}\n for worker in workers:\n statuses[worker[\"status\"]] = statuses.get(worker[\"status\"], 0) + 1\n by_provider = {\n label: {\n \"workers\": sum(worker[\"label\"] == label for worker in workers),\n \"seconds\": summed([w[\"durationSeconds\"] for w in workers if w[\"label\"] == label]),\n \"tokens\": summed([w[\"tokens\"] for w in workers if w[\"label\"] == label]),\n \"costUsd\": summed([w[\"costUsd\"] for w in workers if w[\"label\"] == label]),\n }\n for label in labels\n }\n\n def silent(key: str) -> List[str]:\n reporting = {w[\"label\"] for w in workers if w[key] is not None}\n return sorted({w[\"label\"] for w in ran if w[\"label\"] not in reporting})\n\n changed = [worker for worker in workers if worker[\"diff\"]]\n attention = [worker for worker in workers if worker[\"status\"] != \"succeeded\" or concerns(worker)]\n said: Dict[str, int] = {}\n for worker in workers:\n if worker[\"status\"] in RAN_STATUSES:\n key = worker[\"workerStatus\"] or \"unreported\"\n said[key] = said.get(key, 0) + 1\n titles = [run[\"title\"] for run in runs if run[\"title\"]]\n if meta.get(\"title\"):\n titles.insert(0, meta[\"title\"])\n playbooks: List[str] = [meta[\"playbook\"]] if meta.get(\"playbook\") else []\n for run in runs:\n if run[\"playbook\"] and run[\"playbook\"] not in playbooks:\n playbooks.append(run[\"playbook\"])\n return {\n \"version\": 1,\n \"generatedAt\": now_iso(),\n \"skillVersion\": skill_version(),\n \"title\": titles[0] if titles else None,\n \"playbooks\": playbooks,\n \"why\": meta.get(\"why\"),\n \"lead\": meta.get(\"lead\") if isinstance(meta.get(\"lead\"), dict) else None,\n \"journal\": read_journal(folder),\n \"startedAt\": runs[0][\"startedAt\"] if runs else None,\n \"finishedAt\": max((run[\"finishedAt\"] or \"\" for run in runs), default=\"\") or None,\n \"totals\": {\n \"runs\": len(runs),\n \"workers\": len(workers),\n \"statuses\": statuses,\n \"wallSeconds\": summed([run[\"wallSeconds\"] for run in runs]),\n \"workerSeconds\": summed([worker[\"durationSeconds\"] for worker in workers]),\n \"calls\": int(sum(number(run[\"budgetUsage\"].get(\"calls\")) or 0 for run in runs)),\n \"workerStatuses\": said,\n \"tokens\": summed([worker[\"tokens\"] for worker in workers]),\n \"cachedTokens\": summed([worker[\"cachedTokens\"] for worker in workers]),\n \"cacheReadTokens\": summed([worker[\"cachedTokens\"] for worker in workers]),\n \"tokensNotReportedBy\": silent(\"tokens\"),\n \"costUsd\": summed([worker[\"costUsd\"] for worker in workers]),\n \"costReportedBy\": sorted({w[\"label\"] for w in workers if w[\"costUsd\"] is not None}),\n \"costNotReportedBy\": silent(\"costUsd\"),\n \"transcriptBytes\": sum(worker[\"transcriptBytes\"] for worker in workers),\n \"resultBytes\": sum(worker[\"resultBytes\"] for worker in workers),\n \"changes\": {\n \"writers\": len(changed),\n \"files\": sum(worker[\"diff\"][\"files\"] for worker in changed),\n \"insertions\": sum(worker[\"diff\"][\"insertions\"] for worker in changed),\n \"deletions\": sum(worker[\"diff\"][\"deletions\"] for worker in changed),\n },\n \"budgetStops\": [\n worker[\"error\"]\n for worker in workers\n if worker[\"status\"] == \"skipped\"\n and str(worker[\"error\"] or \"\").startswith((\"budget\", \"orchestration budget\"))\n ],\n \"value\": value_totals(workers),\n },\n \"orchestrationBudget\": report_budget(runs),\n \"byProvider\": by_provider,\n \"runs\": runs,\n \"followUps\": {\n \"worktrees\": [\n {\"worker\": w[\"id\"], \"path\": w[\"worktree\"], \"diff\": w[\"diff\"]}\n for w in workers\n if w[\"worktree\"]\n ],\n \"mainCheckoutChanges\": [\n {\"worker\": w[\"id\"], \"diffPath\": w[\"diffPath\"], \"diff\": w[\"diff\"]}\n for w in changed\n if not w[\"worktree\"]\n ],\n \"attention\": [\n {\n \"worker\": w[\"id\"],\n \"status\": w[\"status\"],\n \"error\": w[\"error\"],\n \"reasons\": concerns(w),\n \"failureKind\": w[\"failureKind\"],\n \"hint\": FAILURE_HINTS.get(w[\"failureKind\"] or \"\"),\n \"resetHint\": w[\"resetHint\"],\n \"stderrTail\": w[\"stderrTail\"],\n \"stderrPath\": w[\"stderrPath\"],\n \"provider\": w[\"provider\"],\n \"resume\": w[\"sessionId\"]\n if w[\"status\"] in (\"failed\", \"timed_out\", \"stopped\")\n and PROVIDER_CAPABILITIES.get(w[\"provider\"], {}).get(\"resume\")\n else None,\n }\n for w in attention\n ],\n },\n }\n\n\ndef human_seconds(seconds: Optional[float]) -> str:\n if seconds is None:\n return \"-\"\n total = int(round(seconds))\n hours, rest = divmod(total, 3600)\n minutes, secs = divmod(rest, 60)\n if hours:\n return f\"{hours}h {minutes:02d}m\"\n if minutes:\n return f\"{minutes}m {secs:02d}s\"\n return f\"{secs}s\"\n\n\ndef human_bytes(size: float) -> str:\n for unit in (\"B\", \"KB\", \"MB\"):\n if size \u003c 1024:\n return f\"{size:.0f} {unit}\" if unit == \"B\" else f\"{size:.1f} {unit}\"\n size /= 1024\n return f\"{size:.1f} GB\"\n\n\ndef human_cost(value: Optional[float]) -> str:\n if value is None:\n return \"-\"\n return f\"${value:.4f}\" if 0 \u003c value \u003c 0.01 else f\"${value:.2f}\"\n\n\ndef human_count(value: Optional[float]) -> str:\n return \"-\" if value is None else f\"{int(value):,}\"\n\n\ndef local_time(stamp: Any) -> Optional[dt.datetime]:\n try:\n return dt.datetime.fromisoformat(stamp).astimezone()\n except (TypeError, ValueError):\n return None\n\n\ndef cell(value: Any) -> str:\n if value in (None, \"\"):\n return \"-\"\n return \" \".join(str(value).split()).replace(\"|\", \"\\\\|\")\n\n\ndef listed(names: Sequence[str]) -> str:\n if len(names) \u003c= 1:\n return \"\".join(names)\n return \", \".join(names[:-1]) + \" and \" + names[-1]\n\n\ndef counted(count: int, noun: str) -> str:\n return f\"{count} {noun}\" if count == 1 else f\"{count} {noun}s\"\n\n\ndef change_text(diff: Dict[str, int]) -> str:\n return f\"{counted(diff['files'], 'file')} (+{diff['insertions']} -{diff['deletions']})\"\n\n\ndef ran_on(worker: Dict[str, Any]) -> str:\n provider = worker[\"provider\"] or \"?\"\n requested = worker[\"requestedProvider\"]\n head = worker[\"label\"] or provider\n if requested and requested not in (\"auto\", provider):\n head = f\"{requested} -> {head}\"\n parts = [head]\n if worker[\"model\"]:\n parts.append(worker[\"model\"])\n elif worker[\"billedModels\"]:\n parts.append(\", \".join(worker[\"billedModels\"]))\n elif provider != \"custom\":\n parts.append(\"default model\")\n if worker[\"effort\"]:\n parts.append(worker[\"effort\"])\n return \", \".join(parts)\n\n\ndef status_text(worker: Dict[str, Any]) -> str:\n parts = [worker[\"status\"].replace(\"_\", \" \")]\n if worker[\"workerStatus\"] and worker[\"workerStatus\"] != \"DONE\":\n parts.append(worker[\"workerStatus\"].replace(\"_\", \" \").lower())\n if worker[\"verdict\"]:\n parts.append(worker[\"verdict\"].replace(\"_\", \" \").lower())\n if worker[\"readOnlyViolation\"]:\n parts.append(\"wrote while read-only\")\n return \", \".join(parts)\n\n\ndef spend_text(tokens: Optional[float], cost: Optional[float]) -> str:\n parts = [human_cost(cost) if cost is not None else \"no cost reported\"]\n parts.append(f\"{human_count(tokens)} tokens\" if tokens is not None else \"no tokens reported\")\n return \", \".join(parts)\n\n\ndef value_text(value: Dict[str, Any]) -> str:\n labels = {\"used\": \"used\", \"partial\": \"partly used\", \"discarded\": \"discarded\"}\n parts = []\n for key, label in labels.items():\n group = value[key]\n if group[\"workers\"]:\n parts.append(f\"{counted(len(group['workers']), 'result')} {label} ({spend_text(group['tokens'], group['costUsd'])})\")\n text = \"; \".join(parts) or \"nothing marked yet\"\n unmarked = value[\"unmarked\"][\"workers\"]\n if unmarked:\n text += f\"; {counted(len(unmarked), 'result')} not marked\"\n return text\n\n\ndef limits_text(limits: Dict[str, Any], calls: Any, cost: Any, tokens: Any) -> str:\n parts = []\n if \"maxCalls\" in limits:\n parts.append(f\"{int(number(calls) or 0)} of {limits['maxCalls']} calls\")\n if \"maxReportedCostUsd\" in limits:\n spent = number(cost)\n limit = human_cost(limits[\"maxReportedCostUsd\"])\n parts.append(f\"{human_cost(spent)} of {limit}\" if spent is not None else f\"no cost reported against {limit}\")\n if \"maxReportedTokens\" in limits:\n counted_tokens = number(tokens)\n limit = human_count(limits[\"maxReportedTokens\"])\n parts.append(\n f\"{human_count(counted_tokens)} of {limit} reported tokens\"\n if counted_tokens is not None\n else f\"no tokens reported against {limit}\"\n )\n return \", \".join(parts)\n\n\ndef verb_for(names: Sequence[str], singular: str, plural: str) -> str:\n return f\"{listed(names)} {singular if len(names) == 1 else plural}\"\n\n\ndef mode_text(worker: Dict[str, Any]) -> str:\n if worker[\"mode\"] == \"write\":\n return \"write, worktree\" if worker[\"isolate\"] else \"write\"\n return \"read\" if worker[\"mode\"] == \"read-only\" else \"-\"\n\n\ndef budget_text(run: Dict[str, Any]) -> Optional[str]:\n budget, usage = run[\"budget\"], run[\"budgetUsage\"]\n parts = []\n text = limits_text(budget, usage.get(\"calls\"), usage.get(\"reportedCostUsd\"), usage.get(\"reportedTokens\"))\n if text:\n parts.append(text)\n if \"maxWallSeconds\" in budget:\n parts.append(f\"{human_seconds(run['wallSeconds'])} of {human_seconds(budget['maxWallSeconds'])}\")\n if \"maxEstimatedPromptTokens\" in budget:\n parts.append(f\"prompts within {human_count(budget['maxEstimatedPromptTokens'])} estimated tokens\")\n return \", \".join(parts) or None\n\n\ndef render_markdown(report: Dict[str, Any]) -> str:\n totals = report[\"totals\"]\n runs = report[\"runs\"]\n lines = [\"# Orchestration report\", \"\"]\n heading = []\n if report[\"title\"]:\n heading.append(f\"**{cell(report['title'])}**\")\n if report[\"playbooks\"]:\n heading.append(\"playbook \" + \", \".join(f\"`{name}`\" for name in report[\"playbooks\"]))\n heading.append(counted(totals[\"runs\"], \"wave\"))\n started, finished = local_time(report[\"startedAt\"]), local_time(report[\"finishedAt\"])\n if started:\n span = started.strftime(\"%Y-%m-%d %H:%M\")\n if finished:\n span += finished.strftime(\" to %H:%M\" if finished.date() == started.date() else \" to %Y-%m-%d %H:%M\")\n heading.append(span)\n lines += [\", \".join(heading), \"\"]\n if lead_text(report[\"lead\"]):\n lines += [f\"Lead: {cell(lead_text(report['lead']))} planned the waves and wrote the prompts\", \"\"]\n if report[\"why\"]:\n lines += [f\"Why: {cell(report['why'])}\", \"\"]\n elif len(runs) == 1 and runs[0][\"why\"]:\n lines += [f\"Why: {cell(runs[0]['why'])}\", \"\"]\n\n statuses = totals[\"statuses\"]\n others = [\n f\"{count} {status.replace('_', ' ')}\"\n for status, count in sorted(statuses.items())\n if status != \"succeeded\"\n ]\n result = f\"{statuses.get('succeeded', 0)} of {counted(totals['workers'], 'worker')} succeeded\"\n lines.append(f\"- **Result:** {result}\" + (f\", {', '.join(others)}\" if others else \"\"))\n lines.append(\n f\"- **Time:** {human_seconds(totals['wallSeconds'])} running, \"\n f\"{human_seconds(totals['workerSeconds'])} of worker time\"\n )\n if totals[\"tokens\"] is not None:\n tokens = f\"{human_count(totals['tokens'])} reported\"\n if totals[\"cachedTokens\"]:\n tokens += f\"; {human_count(totals['cachedTokens'])} cache reads separately (excluded from budgets and Value)\"\n else:\n tokens = \"none reported\"\n if totals[\"tokensNotReportedBy\"]:\n tokens += f\"; {verb_for(totals['tokensNotReportedBy'], 'reports', 'report')} no tokens\"\n lines.append(f\"- **Tokens:** {tokens}\")\n if totals[\"costUsd\"] is not None:\n cost = f\"{human_cost(totals['costUsd'])} reported by {listed(totals['costReportedBy'])}\"\n else:\n cost = \"none reported\"\n if totals[\"costNotReportedBy\"]:\n cost += f\"; {verb_for(totals['costNotReportedBy'], 'reports', 'report')} no cost\"\n lines.append(f\"- **Cost:** {cost}\")\n lines.append(\n f\"- **Lead context:** {human_bytes(totals['transcriptBytes'])} of worker transcripts stayed on disk, \"\n f\"{human_bytes(totals['resultBytes'])} of results to read\"\n )\n changes = totals[\"changes\"]\n if changes[\"writers\"]:\n lines.append(\n f\"- **Changes:** {change_text(changes)} from {counted(changes['writers'], 'writer')}\"\n )\n if totals[\"budgetStops\"]:\n stops = sorted(set(totals[\"budgetStops\"]))\n lines.append(f\"- **Budget:** stopped {counted(len(totals['budgetStops']), 'worker')} ({'; '.join(stops)})\")\n outer = report[\"orchestrationBudget\"]\n if outer:\n used = outer[\"used\"]\n spent = limits_text(outer[\"limits\"], used[\"calls\"], used[\"reportedCostUsd\"], used[\"reportedTokens\"])\n lines.append(f\"- **Orchestration budget:** {spent}\")\n if totals[\"value\"]:\n lines.append(f\"- **Value:** {value_text(totals['value'])}\")\n\n for index, run in enumerate(runs, 1):\n name = run[\"phase\"] or f\"Wave {index}\"\n lines += [\"\", f\"## {index}. {name}\" if len(runs) > 1 else f\"## {name}\", \"\"]\n if len(runs) > 1 and run[\"why\"]:\n lines += [f\"Why: {cell(run['why'])}\", \"\"]\n strategy = str(run[\"strategy\"] or \"?\")\n reason = run[\"strategyReason\"] or \"\"\n if reason.startswith(\"auto inferred: \"):\n strategy += f\" (auto: {reason[len('auto inferred: '):]})\"\n elif run[\"requestedStrategy\"] and run[\"requestedStrategy\"] == run[\"strategy\"]:\n strategy += \" (requested)\"\n meta = [\n f\"Strategy {strategy}\",\n f\"up to {run['maxConcurrent']} at a time\",\n f\"{run['status']} in {human_seconds(run['wallSeconds'])}\",\n ]\n started = local_time(run[\"startedAt\"])\n if started:\n meta.append(f\"started {started.strftime('%H:%M')}\")\n lines.append(\", \".join(meta) + \".\")\n budget = budget_text(run)\n if budget:\n lines.append(f\"Budget: {budget}.\")\n lines.append(\"\")\n with_role = any(worker[\"why\"] for worker in run[\"workers\"])\n header = [\"Worker\"] + ([\"Role\"] if with_role else []) + [\"Ran on\", \"Mode\", \"Status\", \"Time\", \"Tokens\", \"Cache reads\", \"Cost\"]\n lines.append(\"| \" + \" | \".join(header) + \" |\")\n lines.append(\"|\" + \"|\".join(\" --- \" for _ in header) + \"|\")\n for worker in run[\"workers\"]:\n row = [f\"`{worker['id']}`\"] + ([cell(worker[\"why\"])] if with_role else [])\n row += [\n cell(ran_on(worker)),\n mode_text(worker),\n status_text(worker),\n human_seconds(worker[\"durationSeconds\"]),\n human_count(worker[\"tokens\"]),\n human_count(worker[\"cachedTokens\"]),\n human_cost(worker[\"costUsd\"]),\n ]\n lines.append(\"| \" + \" | \".join(row) + \" |\")\n lines.append(\"\")\n for worker in run[\"workers\"]:\n detail = []\n if worker[\"summary\"]:\n detail.append(cell(worker[\"summary\"]))\n if worker[\"error\"]:\n reason = cell(worker[\"error\"])\n if worker[\"stderrTail\"]:\n reason += f\" ({cell(worker['stderrTail'])})\"\n detail.append(f\"{worker['status'].replace('_', ' ').capitalize()}: {reason}.\")\n if worker[\"diff\"]:\n detail.append(f\"Changed {change_text(worker['diff'])}.\")\n if worker[\"dirtyBefore\"]:\n detail.append(\n f\"{counted(len(worker['dirtyBefore']), 'file')} in its diff already had uncommitted changes.\"\n )\n if worker[\"diffError\"]:\n detail.append(f\"No diff: {cell(worker['diffError'])}.\")\n if worker[\"readOnlyViolation\"]:\n detail.append(f\"Changed while read-only: {cell(listed(list(worker['touchedFiles'] or [])[:5]))}.\")\n if worker[\"permissionDenials\"]:\n denied = listed(sorted(set(worker[\"permissionDenials\"])))\n detail.append(f\"Denied {counted(len(worker['permissionDenials']), 'tool call')} ({cell(denied)}).\")\n if worker[\"mark\"]:\n marked = f\"Marked {worker['mark']}\"\n detail.append(marked + (f\": {cell(worker['markWhy'])}.\" if worker[\"markWhy\"] else \".\"))\n notes = [note for note in (worker[\"modelNote\"], worker[\"routingNote\"]) if note]\n text = \" \".join(detail) or \"No result.\"\n if notes:\n text += f\" _({cell('; '.join(notes))})_\"\n lines.append(f\"- `{worker['id']}`: {text}\")\n\n if report[\"journal\"]:\n lines += [\"\", \"## Changes to the plan\", \"\"]\n for entry in report[\"journal\"]:\n stamp = local_time(entry.get(\"at\"))\n when = stamp.strftime(\"%H:%M \") if stamp else \"\"\n lines.append(f\"- {when}{NOTE_KINDS[entry['kind']]}: {cell(entry['text'])}\")\n\n if len(report[\"byProvider\"]) > 1:\n lines += [\"\", \"## By provider\", \"\", \"| Provider | Workers | Time | Tokens | Cost |\", \"| --- | --- | --- | --- | --- |\"]\n for provider, entry in report[\"byProvider\"].items():\n lines.append(\n f\"| {provider} | {entry['workers']} | {human_seconds(entry['seconds'])} | \"\n f\"{human_count(entry['tokens'])} | {human_cost(entry['costUsd'])} |\"\n )\n\n follow = report[\"followUps\"]\n todo = []\n for item in follow[\"attention\"]:\n text = f\"- `{item['worker']}` {item['status'].replace('_', ' ')}\"\n if item[\"error\"]:\n text += f\": {cell(item['error'])}\"\n if item[\"reasons\"]:\n text += (\"; \" if item[\"error\"] else \": \") + \"; \".join(item[\"reasons\"])\n if item[\"hint\"]:\n text += f\". {item['hint'][0].upper()}{item['hint'][1:]}\"\n if item[\"resetHint\"]:\n text += f\" ({cell(item['resetHint'])})\"\n if item[\"stderrPath\"] and item[\"status\"] != \"succeeded\":\n text += f\". Logs: `{item['stderrPath']}`\"\n if item[\"resume\"]:\n text += f\". Resume {item['provider']} session `{item['resume']}` with the rescue strategy\"\n todo.append(text + \".\")\n for item in follow[\"worktrees\"]:\n changed = f\": {change_text(item['diff'])}\" if item[\"diff\"] else \", no changes\"\n todo.append(f\"- Review worktree `{item['path']}` from `{item['worker']}`{changed}. Remove it once handled.\")\n for item in follow[\"mainCheckoutChanges\"]:\n todo.append(\n f\"- `{item['worker']}` changed the main checkout: {change_text(item['diff'])}, see `{item['diffPath']}`.\"\n )\n value = totals[\"value\"]\n if value and value[\"unmarked\"][\"workers\"]:\n names = \", \".join(f\"`{name}`\" for name in value[\"unmarked\"][\"workers\"])\n todo.append(f\"- Mark {names} as used, partial or discarded.\")\n if todo:\n lines += [\"\", \"## Left to do\", \"\"] + todo\n\n lines += [\"\", \"---\", \"\"]\n note = \"Tokens and cost are only what each CLI reported; nothing is estimated.\"\n if \"claude\" in totals[\"costReportedBy\"]:\n note += \" Claude Code reports cost at API prices, so on a subscription that work used plan limits instead of money.\"\n lines.append(note)\n version = report[\"skillVersion\"]\n lines.append(\n f\"Generated by alethe-orchestrate{' v' + version if version else ''} from {counted(totals['runs'], 'run')}.\"\n )\n return \"\\n\".join(lines) + \"\\n\"\n\n\ndef render_status(report: Dict[str, Any]) -> str:\n totals = report[\"totals\"]\n head = [report[\"title\"] or \"Orchestration\"]\n if report[\"playbooks\"]:\n head.append(\"playbook \" + \", \".join(report[\"playbooks\"]))\n head.append(counted(totals[\"runs\"], \"wave\"))\n if lead_text(report[\"lead\"]):\n head.append(\"lead \" + lead_text(report[\"lead\"]))\n lines = [\" | \".join(head)]\n waves = []\n for index, run in enumerate(report[\"runs\"], 1):\n done = sum(worker[\"status\"] == \"succeeded\" for worker in run[\"workers\"])\n waves.append(f\"{index}. {run['phase'] or 'wave'} {run['status']} ({done}/{len(run['workers'])})\")\n lines.append(\"Waves: \" + \"; \".join(waves))\n lines.append(f\"Spent: {counted(totals['calls'], 'worker call')}, {spend_text(totals['tokens'], totals['costUsd'])}\")\n if report[\"journal\"]:\n last = report[\"journal\"][-1]\n lines.append(f\"Plan changes: {len(report['journal'])}, last: {NOTE_KINDS[last['kind']].lower()} {last['text'][:90]}\")\n outer = report[\"orchestrationBudget\"]\n if outer:\n used = outer[\"used\"]\n lines.append(\"Budget: \" + limits_text(outer[\"limits\"], used[\"calls\"], used[\"reportedCostUsd\"], used[\"reportedTokens\"]))\n follow = report[\"followUps\"]\n if follow[\"attention\"]:\n items = []\n for item in follow[\"attention\"]:\n why = item[\"error\"] or \"; \".join(item[\"reasons\"]) or item[\"status\"]\n items.append(f\"{item['worker']} ({why[:90]})\")\n lines.append(\"Needs attention: \" + \"; \".join(items))\n if follow[\"worktrees\"]:\n lines.append(\"Open worktrees: \" + \", \".join(item[\"worker\"] for item in follow[\"worktrees\"]))\n value = totals[\"value\"]\n if value and value[\"unmarked\"][\"workers\"]:\n lines.append(\"Not marked: \" + \", \".join(value[\"unmarked\"][\"workers\"]))\n return \"\\n\".join(lines) + \"\\n\"\n\n\ndef probe_provider(provider: str) -> Dict[str, Any]:\n command = \"agy\" if provider == \"agy\" else provider\n executable = shutil.which(command)\n launch = launch_command(executable) if executable else None\n version = None\n if launch:\n try:\n result = subprocess.run(\n [*launch, \"--version\"],\n stdout=subprocess.PIPE,\n stderr=subprocess.STDOUT,\n text=True,\n timeout=5,\n check=False,\n )\n version = result.stdout.strip().splitlines()[0] if result.stdout.strip() else None\n except (OSError, subprocess.TimeoutExpired):\n pass\n return {\n \"provider\": provider,\n \"installed\": executable is not None,\n \"executable\": executable,\n \"launch\": launch,\n \"version\": version,\n \"capabilities\": PROVIDER_CAPABILITIES[provider],\n }\n\n\nAUTH_COMMANDS = {\"claude\": (\"auth\", \"status\"), \"codex\": (\"login\", \"status\"), \"kimi\": (\"provider\", \"list\")}\n\n\ndef provider_logged_in(provider: str, launch: Optional[List[str]]) -> Optional[bool]:\n command = AUTH_COMMANDS.get(provider)\n if not launch or not command:\n return None\n try:\n result = subprocess.run(\n [*launch, *command],\n stdin=subprocess.DEVNULL,\n stdout=subprocess.PIPE,\n stderr=subprocess.STDOUT,\n text=True,\n encoding=\"utf-8\",\n errors=\"replace\",\n timeout=15,\n check=False,\n )\n except (OSError, subprocess.TimeoutExpired):\n return None\n output = result.stdout or \"\"\n if provider == \"claude\":\n try:\n return bool(json.loads(output).get(\"loggedIn\"))\n except (json.JSONDecodeError, AttributeError):\n return None\n if provider == \"codex\":\n if re.search(r\"not logged in\", output, re.IGNORECASE):\n return False\n return True if result.returncode == 0 and re.search(r\"logged in\", output, re.IGNORECASE) else None\n return bool(re.search(r\"source=(oauth|api)\", output))\n\n\ndef doctor_payload(policy_path: Path) -> Dict[str, Any]:\n python_ok = sys.version_info >= (3, 9)\n exists = policy_path.exists()\n policy: Optional[Dict[str, Any]] = None\n error: Optional[str] = None\n try:\n policy = load_policy(policy_path)\n except PlanError as plan_error:\n error = str(plan_error)\n playbooks = playbook_doctor(policy_path)\n providers = [probe_provider(provider) for provider in BUILTIN_PROVIDERS]\n for entry in providers:\n entry[\"loggedIn\"] = provider_logged_in(entry[\"provider\"], entry[\"launch\"]) if entry[\"installed\"] else None\n installed = [entry[\"provider\"] for entry in providers if entry[\"installed\"]]\n hints: List[str] = []\n for entry in providers:\n if entry[\"loggedIn\"] is False:\n hints.append(f\"{entry['provider']} is installed but not logged in\")\n if not python_ok:\n hints.append(\"python 3.9 or newer is required\")\n if not installed:\n hints.append(\"no built-in provider is installed; install one or use provider custom\")\n if not policy or not policy[\"onboarded\"]:\n hints.append(\"orchestration onboarding is incomplete; offer the setup choices\")\n if not playbooks[\"valid\"]:\n hints.append(playbooks[\"error\"])\n if policy and policy[\"modelTiers\"]:\n hints.append(f\"policy.modelTiers is deprecated; move it into {user_playbooks_path(policy_path)} tiers\")\n if error:\n hints.append(f\"saved policy is invalid: {error}\")\n elif policy and not policy[\"budget\"]:\n hints.append(\"no default budget; policy set can add hard limits such as --max-calls\")\n usage = usage_payload(policy_path, refresh=False)\n return {\n \"usage\": usage,\n \"usageSummary\": usage_summary(usage),\n \"skill\": {\"version\": skill_version(), \"path\": str(Path(__file__).resolve().parent.parent)},\n \"python\": {\"version\": sys.version.split()[0], \"ok\": python_ok},\n \"policyPath\": str(policy_path),\n \"policy\": {\n \"exists\": exists,\n \"valid\": error is None,\n \"error\": error,\n \"configured\": exists and error is None and bool(policy and policy[\"onboarded\"]),\n \"budget\": policy[\"budget\"] if policy else None,\n },\n \"providers\": providers,\n \"playbooks\": playbooks,\n \"installedProviders\": installed,\n \"hints\": hints,\n }\n\n\ndef scaffold_plan(strategy: str, workers: int, cwd: str, provider: str) -> Dict[str, Any]:\n strategy = normalized_strategy(strategy, \"scaffold strategy\")\n provider = normalized_provider(provider)\n if strategy == \"auto\":\n strategy = \"economy\"\n workers = 1\n rules = STRATEGIES[strategy]\n if not 1 \u003c= workers \u003c= rules[\"maxWorkers\"]:\n raise PlanError(f\"strategy {strategy} allows between 1 and {rules['maxWorkers']} workers\")\n if strategy == \"rescue\" and provider == \"auto\":\n raise PlanError(\"a rescue scaffold needs --provider naming the session owner\")\n write_mode = strategy in (\"parallel\", \"quality\")\n entries: List[Dict[str, Any]] = []\n for index in range(workers):\n entry: Dict[str, Any] = {\n \"id\": f\"worker-{index + 1}\",\n \"provider\": provider,\n \"mode\": \"write\" if write_mode else \"read-only\",\n \"cwd\": cwd,\n \"task\": \"TODO: one bounded unit — goal, definition of done, owned paths, checks.\",\n }\n if strategy == \"parallel\":\n entry[\"isolate\"] = True\n if strategy == \"rescue\":\n entry[\"resume\"] = \"TODO-session-id\"\n if provider == \"custom\":\n entry[\"command\"] = \"TODO-executable\"\n entries.append(entry)\n return {\"version\": PLAN_VERSION, \"strategy\": strategy, \"workers\": entries}\n\n\ndef strategies_payload() -> List[Dict[str, Any]]:\n return [\n {\n \"strategy\": name,\n \"maxWorkers\": rules[\"maxWorkers\"],\n \"maxConcurrent\": rules[\"maxConcurrent\"],\n \"description\": rules[\"description\"],\n }\n for name, rules in STRATEGIES.items()\n ]\n\n\ndef update_policy(args: argparse.Namespace, policy_path: Path) -> Dict[str, Any]:\n policy = load_policy(policy_path)\n if args.policy_action == \"reset\":\n policy = dict(DEFAULT_POLICY)\n save_policy(policy_path, policy)\n return policy\n if args.policy_action == \"show\":\n return policy\n if args.strategy:\n policy[\"defaultStrategy\"] = normalized_strategy(args.strategy, \"policy strategy\")\n if args.backend:\n policy[\"backend\"] = args.backend\n if args.provider_switch:\n policy[\"providerSwitch\"] = args.provider_switch\n if args.provider_order:\n policy[\"providerOrder\"] = validate_provider_list(\n args.provider_order, \"provider order\", allow_empty=False\n )\n if args.conserve_providers is not None:\n policy[\"conserveProviders\"] = validate_provider_list(\n args.conserve_providers, \"conserve providers\", allow_empty=True\n )\n if args.max_workers is not None:\n policy[\"maxWorkers\"] = args.max_workers\n if args.max_concurrent is not None:\n policy[\"maxConcurrent\"] = args.max_concurrent\n budget = dict(policy.get(\"budget\", {}))\n for key, value in (\n (\"maxCalls\", args.max_calls),\n (\"maxWallSeconds\", args.max_wall_seconds),\n (\"maxEstimatedPromptTokens\", args.max_estimated_prompt_tokens),\n (\"maxReportedCostUsd\", args.max_reported_cost_usd),\n (\"maxReportedTokens\", args.max_reported_tokens),\n ):\n if value is not None:\n budget[key] = value\n policy[\"budget\"] = validate_budget(budget, \"policy.budget\")\n tiers = {\n tier: {provider: dict(spec) for provider, spec in mapping.items()}\n for tier, mapping in policy.get(\"modelTiers\", {}).items()\n }\n for tier, key, values in (\n (\"cheap\", \"model\", args.cheap_model),\n (\"cheap\", \"effort\", args.cheap_effort),\n (\"strong\", \"model\", args.strong_model),\n ):\n if values:\n for pair in values:\n provider, separator, value = pair.partition(\"=\")\n if not separator or not provider.strip() or not value.strip():\n raise PlanError(f\"expected PROVIDER={key.upper()}, got {pair!r}\")\n tiers.setdefault(tier, {}).setdefault(provider.strip(), {})[key] = value.strip()\n policy[\"modelTiers\"] = validate_model_tiers(tiers, \"policy.modelTiers\")\n policy[\"onboarded\"] = True\n policy = validate_policy(policy)\n save_policy(policy_path, policy)\n return policy\n\n\ndef main(argv: Optional[Sequence[str]] = None) -> int:\n reconfigure = getattr(sys.stdout, \"reconfigure\", None)\n if reconfigure and (sys.stdout.encoding or \"\").lower().replace(\"-\", \"\") != \"utf8\":\n reconfigure(encoding=\"utf-8\")\n parser = argparse.ArgumentParser(description=__doc__)\n parser.add_argument(\n \"--version\", action=\"version\", version=f\"alethe-orchestrate {skill_version() or 'unknown'}\"\n )\n subparsers = parser.add_subparsers(dest=\"command\", required=True)\n subparsers.add_parser(\"strategies\", help=\"list orchestration strategies and limits\")\n playbook_parser = subparsers.add_parser(\n \"playbook\", help=\"print the strategy playbooks the planner follows\"\n )\n playbook_parser.add_argument(\"id\", nargs=\"?\", help=\"print one playbook and its active settings\")\n playbook_parser.add_argument(\"--pick\", action=\"store_true\", help=\"print a compact playbook selection index\")\n playbook_parser.add_argument(\"--json\", action=\"store_true\", help=\"print structured playbook data\")\n playbook_parser.add_argument(\"--policy\", type=Path)\n playbook_parser.add_argument(\"--path\", action=\"store_true\", help=\"print only the effective file path\")\n playbook_parser.add_argument(\"--init\", action=\"store_true\", help=\"copy the bundled playbooks for editing\")\n playbook_parser.add_argument(\"--force\", action=\"store_true\", help=\"with --init, replace an existing copy\")\n editor_parser = subparsers.add_parser(\n \"editor\", help=\"write the playbooks as an interactive HTML flow editor and open it\"\n )\n editor_parser.add_argument(\"--policy\", type=Path)\n editor_parser.add_argument(\"--out\", type=Path, help=f\"where to write the page; default {EDITOR_FILE} next to the policy\")\n editor_parser.add_argument(\"--no-open\", action=\"store_true\", help=\"write the page without opening a browser\")\n probe_parser = subparsers.add_parser(\"probe\", help=\"report available built-in providers\")\n probe_parser.add_argument(\"providers\", nargs=\"*\", choices=[\"codex\", \"claude\", \"agy\", \"kimi\"])\n doctor_parser = subparsers.add_parser(\"doctor\", help=\"check the environment and saved policy\")\n doctor_parser.add_argument(\"--policy\", type=Path)\n for command in (\"usage\", \"models\"):\n information_parser = subparsers.add_parser(command, help=f\"show provider {command} as JSON\")\n information_parser.add_argument(\"--policy\", type=Path)\n information_parser.add_argument(\"--json\", action=\"store_true\", help=\"JSON output (the default)\")\n if command == \"usage\":\n information_parser.add_argument(\n \"--refresh-claude\", action=\"store_true\", help=\"read Claude's quota now with a one-line Haiku call\"\n )\n scaffold_parser = subparsers.add_parser(\n \"scaffold\", help=\"print a starter plan for a strategy\"\n )\n scaffold_parser.add_argument(\"strategy\", choices=list(STRATEGIES))\n scaffold_parser.add_argument(\"--workers\", type=int, default=1)\n scaffold_parser.add_argument(\"--cwd\", default=\".\")\n scaffold_parser.add_argument(\"--provider\", default=\"auto\")\n scaffold_parser.add_argument(\"--write\", type=Path)\n validate_parser = subparsers.add_parser(\"validate\", help=\"validate a batch plan without running it\")\n validate_parser.add_argument(\"plan\", type=Path)\n validate_parser.add_argument(\"--policy\", type=Path)\n validate_parser.add_argument(\"--strategy\", choices=list(STRATEGIES))\n dry_run_parser = subparsers.add_parser(\n \"dry-run\", help=\"explain the strategy, providers, and budget a plan would use\"\n )\n dry_run_parser.add_argument(\"plan\", type=Path)\n dry_run_parser.add_argument(\"--policy\", type=Path)\n dry_run_parser.add_argument(\"--strategy\", choices=list(STRATEGIES))\n dry_run_parser.add_argument(\"--write\", type=Path, help=\"save the resolved preview as JSON\")\n run_parser = subparsers.add_parser(\"run\", help=\"run a batch plan\")\n run_parser.add_argument(\"plan\", type=Path)\n run_parser.add_argument(\"--run-dir\", type=Path)\n run_parser.add_argument(\"--detach\", action=\"store_true\", help=\"start the run in the background\")\n run_parser.add_argument(\"--_background\", action=\"store_true\", help=argparse.SUPPRESS)\n run_parser.add_argument(\n \"--in\", dest=\"into\", metavar=\"FOLDER\", help=\"add the run as the next wave of an orchestration folder, or 'current'\"\n )\n run_parser.add_argument(\"--policy\", type=Path)\n run_parser.add_argument(\"--strategy\", choices=list(STRATEGIES))\n run_parser.add_argument(\n \"--allow-low-quota\", action=\"store_true\", help=\"start workers past quota.stopAt; only after the user agrees\"\n )\n report_parser = subparsers.add_parser(\n \"report\", help=\"explain what runs did, why, and what they reported spending\"\n )\n report_parser.add_argument(\n \"runs\",\n nargs=\"*\",\n type=Path,\n help=\"run directories, or a folder holding one per wave; defaults to the latest run\",\n )\n report_parser.add_argument(\"--format\", choices=[\"markdown\", \"json\", \"html\"], default=\"markdown\")\n report_parser.add_argument(\n \"--write\", type=Path, help=f\"also save the report to this file; html defaults to {REPORT_PAGE_FILE} in the folder\"\n )\n report_parser.add_argument(\"--no-open\", action=\"store_true\", help=\"with --format html, write the page without opening it\")\n status_parser = subparsers.add_parser(\n \"status\", help=\"print a few lines on where an orchestration stands, to pick it up again\"\n )\n status_parser.add_argument(\"runs\", nargs=\"*\", type=Path)\n status_parser.add_argument(\"--policy\", type=Path)\n status_parser.add_argument(\"--format\", choices=[\"text\", \"json\"], default=\"text\")\n wait_parser = subparsers.add_parser(\"wait\", help=\"wait for running waves to finish\")\n wait_parser.add_argument(\"folder\", nargs=\"?\", type=Path)\n wait_parser.add_argument(\"--timeout\", type=int, default=DEFAULT_WAIT_SECONDS)\n new_parser = subparsers.add_parser(\"new\", help=\"start an orchestration folder and make it the current one\")\n new_parser.add_argument(\"title\")\n new_parser.add_argument(\"--playbook\")\n new_parser.add_argument(\"--why\")\n new_parser.add_argument(\"--lead\", help=\"the CLI and model leading it, as provider:model, such as claude:opus\")\n list_parser = subparsers.add_parser(\"list\", help=\"list recent orchestrations; * marks the current one\")\n list_parser.add_argument(\"--limit\", type=int, default=10)\n list_parser.add_argument(\"--format\", choices=[\"text\", \"json\"], default=\"text\")\n use_parser = subparsers.add_parser(\"use\", help=\"make an orchestration folder the current one\")\n use_parser.add_argument(\"folder\", type=Path)\n note_parser = subparsers.add_parser(\n \"note\", help=\"record a change to the plan, such as a skipped phase or a switched playbook\"\n )\n note_parser.add_argument(\"text\")\n note_parser.add_argument(\"--kind\", choices=list(NOTE_KINDS), default=\"change\")\n note_parser.add_argument(\"--in\", dest=\"into\", type=Path, metavar=\"FOLDER\")\n stop_parser = subparsers.add_parser(\"stop\", help=\"stop the running waves of an orchestration\")\n stop_parser.add_argument(\"folder\", nargs=\"?\", type=Path)\n mode_parser = subparsers.add_parser(\n \"mode\", help=\"keep orchestration on for a project, with a goal, care level and the user's standing rules\"\n )\n mode_parser.add_argument(\"action\", choices=[\"on\", \"off\", \"show\"])\n mode_parser.add_argument(\"--goal\", choices=list(MODE_GOALS))\n mode_parser.add_argument(\"--care\", choices=[\"normal\", \"careful\"])\n mode_parser.add_argument(\"--rule\", action=\"append\", default=[], help=\"a standing instruction, in the user's words\")\n mode_parser.add_argument(\"--clear-rules\", action=\"store_true\")\n mode_parser.add_argument(\"--cwd\", type=Path, default=Path(\".\"))\n mode_parser.add_argument(\n \"--everywhere\", action=\"store_true\", help=\"the default for every project; mode off still turns it off in one\"\n )\n subparsers.add_parser(\"hook\", help=\"print the reminder for an agent hook while a mode is on; never fails\")\n budget_parser = subparsers.add_parser(\n \"budget\", help=\"show or set the limits for every wave in an orchestration folder\"\n )\n budget_parser.add_argument(\"folder\", nargs=\"?\", type=Path)\n budget_parser.add_argument(\"--max-calls\", type=int)\n budget_parser.add_argument(\"--max-reported-cost-usd\", type=float)\n budget_parser.add_argument(\"--max-reported-tokens\", type=int)\n budget_parser.add_argument(\"--clear\", action=\"store_true\", help=\"remove the orchestration budget\")\n mark_parser = subparsers.add_parser(\n \"mark\", help=\"record whether a worker's result was used, so the report shows what the spend bought\"\n )\n mark_parser.add_argument(\"target\", nargs=\"+\", metavar=\"[FOLDER] WORKER MARK\")\n mark_parser.add_argument(\"--why\")\n policy_parser = subparsers.add_parser(\"policy\", help=\"show or change skill-only preferences\")\n policy_parser.add_argument(\"policy_action\", choices=[\"show\", \"set\", \"reset\"])\n policy_parser.add_argument(\"strategy\", nargs=\"?\", choices=list(STRATEGIES))\n policy_parser.add_argument(\"--file\", type=Path)\n policy_parser.add_argument(\"--backend\", choices=[\"auto\", \"skill\", \"mcp\"])\n policy_parser.add_argument(\"--provider-switch\", choices=[\"ask\", \"automatic\", \"never\"])\n policy_parser.add_argument(\"--provider-order\", nargs=\"+\", choices=list(BUILTIN_PROVIDERS))\n policy_parser.add_argument(\n \"--conserve-providers\", nargs=\"*\", choices=list(BUILTIN_PROVIDERS)\n )\n policy_parser.add_argument(\"--max-workers\", type=int)\n policy_parser.add_argument(\"--max-concurrent\", type=int)\n policy_parser.add_argument(\"--max-calls\", type=int)\n policy_parser.add_argument(\"--max-wall-seconds\", type=int)\n policy_parser.add_argument(\"--max-estimated-prompt-tokens\", type=int)\n policy_parser.add_argument(\"--max-reported-cost-usd\", type=float)\n policy_parser.add_argument(\"--max-reported-tokens\", type=int)\n policy_parser.add_argument(\"--cheap-model\", nargs=\"*\", metavar=\"PROVIDER=MODEL\")\n policy_parser.add_argument(\"--cheap-effort\", nargs=\"*\", metavar=\"PROVIDER=EFFORT\")\n policy_parser.add_argument(\"--strong-model\", nargs=\"*\", metavar=\"PROVIDER=MODEL\")\n args = parser.parse_args(argv)\n\n try:\n if args.command == \"strategies\":\n print(json.dumps(strategies_payload(), indent=2))\n return 0\n if args.command == \"playbook\":\n policy_path = (args.policy or default_policy_path()).expanduser().resolve()\n if (args.init or args.path) and (args.id or args.pick or args.json):\n raise PlanError(\"--init and --path cannot be combined with an id, --pick or --json\")\n if args.init:\n print(str(init_playbooks(policy_path, args.force)))\n return 0\n path = effective_playbooks_path(policy_path)\n if args.path:\n print(str(path))\n return 0\n print(playbook_output(policy_path, args.id, args.pick, args.json))\n return 0\n if args.command == \"editor\":\n policy_path = (args.policy or default_policy_path()).expanduser().resolve()\n payload = write_editor(policy_path, args.out)\n payload[\"opened\"] = not args.no_open and open_page(Path(payload[\"path\"]))\n print(json.dumps(payload, indent=2, ensure_ascii=False))\n return 0\n if args.command == \"probe\":\n providers = args.providers or [\"codex\", \"claude\", \"agy\", \"kimi\"]\n print(json.dumps([probe_provider(provider) for provider in providers], indent=2))\n return 0\n if args.command in (\"usage\", \"models\"):\n policy_path = (args.policy or default_policy_path()).expanduser().resolve()\n if args.command == \"usage\" and args.refresh_claude and probe_claude_quota(policy_path) is None:\n print(\"warning: Claude printed no rate-limit event, so its quota stays unknown\", file=sys.stderr)\n payload = usage_payload(policy_path) if args.command == \"usage\" else models_payload(policy_path)\n print(json.dumps(payload, indent=2, ensure_ascii=False))\n return 0\n if args.command == \"doctor\":\n policy_path = (args.policy or default_policy_path()).expanduser().resolve()\n print(json.dumps(doctor_payload(policy_path), indent=2, ensure_ascii=False))\n return 0\n if args.command == \"report\":\n runs, folder = report_target(args.runs)\n report = build_report(runs, folder)\n if args.format == \"html\":\n path = write_report_page(report, args.write or (folder or runs[-1]) / REPORT_PAGE_FILE)\n opened = not args.no_open and open_page(path)\n print(json.dumps({\"path\": str(path), \"opened\": opened}, indent=2))\n return 0\n if args.no_open:\n raise PlanError(\"--no-open only applies to --format html\")\n if args.format == \"json\":\n text = json.dumps(report, indent=2, ensure_ascii=False) + \"\\n\"\n else:\n text = render_markdown(report)\n if args.write:\n atomic_write_text(args.write.expanduser(), text)\n print(text, end=\"\")\n return 0\n if args.command == \"status\":\n report = build_report(*report_target(args.runs))\n if args.format == \"json\":\n print(json.dumps(report, indent=2, ensure_ascii=False))\n else:\n print(render_status(report), end=\"\")\n policy_path = (args.policy or default_policy_path()).expanduser().resolve()\n print(\"\\n\".join(usage_summary(usage_payload(policy_path, refresh=False))))\n return 0\n if args.command == \"wait\":\n print(json.dumps(wait_payload(args.folder, args.timeout), indent=2, ensure_ascii=False))\n return 0\n if args.command == \"new\":\n print(json.dumps(new_orchestration(args.title, args.playbook, args.why, args.lead), indent=2, ensure_ascii=False))\n return 0\n if args.command == \"list\":\n entries = list_payload(max(1, args.limit))\n if args.format == \"json\":\n print(json.dumps(entries, indent=2, ensure_ascii=False))\n else:\n print(render_list(entries), end=\"\")\n return 0\n if args.command == \"use\":\n folder = args.folder.expanduser().resolve()\n if not folder.is_dir():\n raise PlanError(f\"not an orchestration folder: {folder}\")\n set_current(folder)\n print(json.dumps({\"current\": str(folder)}, indent=2))\n return 0\n if args.command == \"note\":\n print(json.dumps(add_note(resolve_folder(args.into), args.text, args.kind), indent=2, ensure_ascii=False))\n return 0\n if args.command == \"mode\":\n if args.action != \"on\" and (args.goal or args.care or args.rule or args.clear_rules):\n raise PlanError(\"--goal, --care and --rule go with mode on\")\n payload = mode_payload(\n args.action, args.cwd, args.goal, args.care, args.rule, args.clear_rules, args.everywhere\n )\n print(json.dumps(payload, indent=2, ensure_ascii=False))\n return 0\n if args.command == \"hook\":\n print(hook_output(hook_stdin()), end=\"\")\n return 0\n if args.command == \"stop\":\n print(json.dumps(stop_payload(resolve_folder(args.folder)), indent=2))\n return 0\n if args.command == \"budget\":\n updates = {\n key: value\n for key, value in (\n (\"maxCalls\", args.max_calls),\n (\"maxReportedCostUsd\", args.max_reported_cost_usd),\n (\"maxReportedTokens\", args.max_reported_tokens),\n )\n if value is not None\n }\n if args.clear and updates:\n raise PlanError(\"pass --clear or new limits, not both\")\n print(json.dumps(budget_payload(resolve_folder(args.folder), updates, args.clear), indent=2))\n return 0\n if args.command == \"mark\":\n if len(args.target) not in (2, 3) or args.target[-1] not in MARKS:\n raise PlanError(f\"usage: mark [FOLDER] WORKER {{{','.join(MARKS)}}}\")\n folder = resolve_folder(Path(args.target[0]) if len(args.target) == 3 else None)\n entry = mark_payload(folder, args.target[-2], args.target[-1], args.why)\n print(json.dumps(entry, indent=2, ensure_ascii=False))\n return 0\n if args.command == \"scaffold\":\n plan = scaffold_plan(args.strategy, args.workers, args.cwd, args.provider)\n if args.write:\n atomic_write_json(args.write, plan)\n print(json.dumps(plan, indent=2))\n return 0\n if args.command == \"policy\":\n policy_path = (args.file or default_policy_path()).expanduser().resolve()\n policy = update_policy(args, policy_path)\n print(json.dumps({\"path\": str(policy_path), \"policy\": policy}, indent=2))\n return 0\n if args.command == \"validate\":\n policy_path = (args.policy or default_policy_path()).expanduser().resolve()\n plan = validate_plan(\n read_json(args.plan),\n args.plan.resolve(),\n load_policy(policy_path),\n args.strategy,\n policy_path,\n )\n summary = {\n \"valid\": True,\n \"strategy\": plan[\"strategy\"],\n \"requestedStrategy\": plan[\"requestedStrategy\"],\n \"maxConcurrent\": plan[\"maxConcurrent\"],\n \"budget\": plan[\"budget\"],\n \"policyPath\": str(policy_path),\n \"warnings\": plan_warnings(plan),\n \"workers\": [\n {\n \"id\": worker[\"id\"],\n \"provider\": worker[\"provider\"],\n \"requestedProvider\": worker[\"requestedProvider\"],\n \"mode\": worker[\"mode\"],\n \"isolate\": worker[\"isolate\"],\n }\n for worker in plan[\"workers\"]\n ],\n }\n print(json.dumps(summary, indent=2))\n return 0\n if args.command == \"dry-run\":\n policy_path = (args.policy or default_policy_path()).expanduser().resolve()\n plan = validate_plan(\n read_json(args.plan),\n args.plan.resolve(),\n load_policy(policy_path),\n args.strategy,\n policy_path,\n )\n preview = plan_payload(plan, args.plan.resolve(), policy_path)\n if args.write:\n atomic_write_json(args.write, preview)\n print(\n json.dumps(\n preview,\n indent=2,\n ensure_ascii=False,\n )\n )\n return 0\n policy_path = (args.policy or default_policy_path()).expanduser().resolve()\n run_dir = args.run_dir\n if args.into:\n if run_dir is not None:\n raise PlanError(\"pass --run-dir or --in, not both\")\n raw = read_json(args.plan)\n folder = resolve_folder(None if args.into == \"current\" else Path(args.into))\n run_dir = next_wave_dir(folder, raw.get(\"phase\") if isinstance(raw, dict) else None)\n if args.detach:\n print(json.dumps(detach_run(args.plan, run_dir, policy_path, args.strategy, args.allow_low_quota)))\n return 0\n if args._background and sys.stdin.buffer.read(1) != b\"1\":\n raise PlanError(\"detached launcher exited before preparing the run\")\n return run_batch(args.plan, run_dir, policy_path, args.strategy, args._background, args.allow_low_quota)\n except PlanError as error:\n print(f\"error: {error}\", file=sys.stderr)\n return 2\n\n\nif __name__ == \"__main__\":\n raise SystemExit(main())\n","content_type":"text/x-python; charset=utf-8","language":"python","size":251454,"content_sha256":"54bbf1e57fdd86c04e5c853ec9ddac769865d22259f1f2a13b8d3960ff85224d"}],"content_json":{"type":"doc","content":[{"type":"heading","attrs":{"level":1},"content":[{"text":"Orchestrate CLI workers","type":"text"}]},{"type":"paragraph","content":[{"text":"Finish the work with the least lead context and the cheapest capable workers, or the strongest mix when the user asks. Plan, delegate bounded units, and integrate results supported by evidence.","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Playbook","type":"text","marks":[{"type":"strong"}]},{"text":" = the phases and roles.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Profile","type":"text","marks":[{"type":"strong"}]},{"text":" = which models each tier uses (","type":"text"},{"text":"base","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"savings","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"aggressive","type":"text","marks":[{"type":"code_inline"}]},{"text":").","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Goal","type":"text","marks":[{"type":"strong"}]},{"text":" = how eagerly to delegate while a mode is on (","type":"text"},{"text":"save","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"balanced","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"max","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"freestyle","type":"text","marks":[{"type":"code_inline"}]},{"text":").","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Runner strategy","type":"text","marks":[{"type":"strong"}]},{"text":" = how one wave runs (","type":"text"},{"text":"economy","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"balanced","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"quality","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"research","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"parallel","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"rescue","type":"text","marks":[{"type":"code_inline"}]},{"text":").","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Choose the backend","type":"text"}]},{"type":"paragraph","content":[{"text":"Honor an explicit request for the local skill or no MCP. It always wins. Otherwise run ","type":"text"},{"text":"scripts/orchestrate.py policy show","type":"text","marks":[{"type":"code_inline"}]},{"text":" and use its ","type":"text"},{"text":"backend","type":"text","marks":[{"type":"code_inline"}]},{"text":": ","type":"text"},{"text":"skill","type":"text","marks":[{"type":"code_inline"}]},{"text":" selects this runner, ","type":"text"},{"text":"mcp","type":"text","marks":[{"type":"code_inline"}]},{"text":" selects ","type":"text"},{"text":"alethe_*","type":"text","marks":[{"type":"code_inline"}]},{"text":" tools when present, and ","type":"text"},{"text":"auto","type":"text","marks":[{"type":"code_inline"}]},{"text":" prefers those tools when available.","type":"text"}]},{"type":"paragraph","content":[{"text":"Python is optional. If ","type":"text"},{"text":"python --version","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"python3 --version","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"py --version","type":"text","marks":[{"type":"code_inline"}]},{"text":" all fail or report below 3.9, or the user asks, follow ","type":"text"},{"text":"references/manual.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/manual.md","title":null}}]},{"text":" for detection, options, background workers and reporting. All other sections still apply through manual equivalents.","type":"text"}]},{"type":"paragraph","content":[{"text":"The skill path needs no MCP. It runs bounded concurrent workers with optional worktree isolation and resumable sessions. CLI workers do not appear on Alethe's board.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"First run and configuration","type":"text"}]},{"type":"paragraph","content":[{"text":"Run ","type":"text"},{"text":"scripts/orchestrate.py doctor","type":"text","marks":[{"type":"code_inline"}]},{"text":" when the environment may be unprepared or the user asks what is set up. It checks Python, providers, the saved policy and the user's playbooks copy, and summarizes quota readings with actionable hints.","type":"text"}]},{"type":"paragraph","content":[{"text":"When ","type":"text"},{"text":"doctor","type":"text","marks":[{"type":"code_inline"}]},{"text":" reports no configured policy, or the user asks to set up orchestration, follow ","type":"text"},{"text":"references/onboarding.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/onboarding.md","title":null}}]},{"text":": inventory the CLIs, models and routes, build profiles ","type":"text"},{"text":"base","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"savings","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"aggressive","type":"text","marks":[{"type":"code_inline"}]},{"text":", and get approval before saving them to the playbooks copy. Save only confirmed preferences. A proxy such as 9router carries model traffic; it is not a worker.","type":"text"}]},{"type":"paragraph","content":[{"text":"Setup never blocks work. Mid-task with nothing configured, run ","type":"text"},{"text":"doctor","type":"text","marks":[{"type":"code_inline"}]},{"text":", use its logged-in CLIs with the bundled playbooks, say so in one line and continue. Offer onboarding after the current step; save nothing until the user agrees.","type":"text"}]},{"type":"paragraph","content":[{"text":"Translate loose configuration requests into actions:","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"The user says","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Do","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"save provider X\" as an ongoing preference","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"confirm, then ","type":"text"},{"text":"policy set --conserve-providers X","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"use cheap models for the small stuff\"","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"confirm the provider and model, then edit ","type":"text"},{"text":"tiers","type":"text","marks":[{"type":"code_inline"}]},{"text":" in the playbooks copy","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"at most N workers / T minutes / C dollars\" as an ongoing preference","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"confirm, then ","type":"text"},{"text":"policy set --max-calls N --max-wall-seconds T --max-reported-cost-usd C","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"any of the above \"for this run only\"","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"put it in the plan's ","type":"text"},{"text":"budget","type":"text","marks":[{"type":"code_inline"}]},{"text":" or use ","type":"text"},{"text":"--strategy","type":"text","marks":[{"type":"code_inline"}]},{"text":"; never touch the policy","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"at most N calls / C dollars / T tokens for this whole task\"","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"budget","type":"text","marks":[{"type":"code_inline"}]},{"text":" with those limits before the first wave","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"where were we\", or after the conversation was compacted","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"status","type":"text","marks":[{"type":"code_inline"}]},{"text":" before anything else","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"back to defaults\"","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"confirm, then ","type":"text"},{"text":"policy reset","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"change the playbooks\" or roles per phase","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"playbook --init","type":"text","marks":[{"type":"code_inline"}]},{"text":", then edit the copy it prints; the UI exports the same file","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"use savings\" or \"go aggressive\" as an ongoing preference","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"confirm, then set ","type":"text"},{"text":"profile","type":"text","marks":[{"type":"code_inline"}]},{"text":" in the playbooks copy","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"\"what did it do\", \"how much did it cost\"","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"report","type":"text","marks":[{"type":"code_inline"}]},{"text":"; no confirmation needed","type":"text"}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Short commands","type":"text"}]},{"type":"paragraph","content":[{"text":"/alethe \u003cargs>","type":"text","marks":[{"type":"code_inline"}]},{"text":" in Claude Code, or the same words after \"alethe\" anywhere else, map to one action. Run-specific commands without a folder use the current orchestration: the one ","type":"text"},{"text":"new","type":"text","marks":[{"type":"code_inline"}]},{"text":" started, ","type":"text"},{"text":"use","type":"text","marks":[{"type":"code_inline"}]},{"text":" picked, or the last ","type":"text"},{"text":"run","type":"text","marks":[{"type":"code_inline"}]},{"text":" wrote to.","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Words","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Do","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"a task, in plain words","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"orchestrate it as the rest of this skill describes","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"nothing","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"status","type":"text","marks":[{"type":"code_inline"}]},{"text":"; with no current orchestration, list these commands in a few lines","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"on","type":"text","marks":[{"type":"code_inline"}]},{"text":", or a standing instruction (\"from now on...\")","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"stay on, as below","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"always","type":"text","marks":[{"type":"code_inline"}]},{"text":", or \"always on\", \"in every project\"","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"mode on --everywhere","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"off","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"mode off","type":"text","marks":[{"type":"code_inline"}]},{"text":"; with the default on, this turns it off in this project only","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"off everywhere","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"mode off --everywhere","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"save","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"balanced","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"max","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"freestyle","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"mode on --goal \u003cit>","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"careful","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"mode on --care careful","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"report [folder]","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"report","type":"text","marks":[{"type":"code_inline"}]},{"text":", and show its output as is, without a summary of your own","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"report html","type":"text","marks":[{"type":"code_inline"}]},{"text":", or \"show the report\"","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"report --format html","type":"text","marks":[{"type":"code_inline"}]},{"text":", which writes ","type":"text"},{"text":"report.html","type":"text","marks":[{"type":"code_inline"}]},{"text":" in the folder and opens it; without Python, fill ","type":"text"},{"text":"assets/report.html","type":"text","marks":[{"type":"code_inline"}]},{"text":" as in ","type":"text"},{"text":"references/manual.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/manual.md","title":null}}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"status","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"status","type":"text","marks":[{"type":"code_inline"}]},{"text":", and show its output as is","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"usage","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"usage","type":"text","marks":[{"type":"code_inline"}]},{"text":", or ","type":"text"},{"text":"usage --refresh-claude","type":"text","marks":[{"type":"code_inline"}]},{"text":" when Claude's reading is missing or stale; show windows, resets and freshness","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"models","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"models","type":"text","marks":[{"type":"code_inline"}]},{"text":"; show model prices, ranks and price freshness","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"editor","type":"text","marks":[{"type":"code_inline"}]},{"text":", or \"show the playbooks\"","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"editor","type":"text","marks":[{"type":"code_inline"}]},{"text":", which writes the playbooks as a flow page and opens it; without Python, fill ","type":"text"},{"text":"assets/editor.html","type":"text","marks":[{"type":"code_inline"}]},{"text":" as in ","type":"text"},{"text":"references/manual.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/manual.md","title":null}}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"list","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"list","type":"text","marks":[{"type":"code_inline"}]},{"text":"; ","type":"text"},{"text":"use \u003cfolder>","type":"text","marks":[{"type":"code_inline"}]},{"text":" makes one of them current again","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"stop","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"stop","type":"text","marks":[{"type":"code_inline"}]},{"text":", then ","type":"text"},{"text":"status","type":"text","marks":[{"type":"code_inline"}]},{"text":" once the waves have ended","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"budget \u003climit>","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"budget --max-calls N","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"--max-reported-cost-usd C","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"--max-reported-tokens T","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"skip \u003cphase>","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"switch \u003cplaybook>","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"freeform","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"add \u003cphase>","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"change course as below","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"note \u003ctext>","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"note \"\u003ctext>\"","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"mark \u003cworker> used|partial|discarded","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"mark \u003cworker> \u003cmark>","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"setup","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"doctor","type":"text","marks":[{"type":"code_inline"}]},{"text":", then onboarding when no policy is saved","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"help","type":"text","marks":[{"type":"code_inline"}]}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"list these commands in a few lines","type":"text"}]}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Stay on","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use it proactively whenever a request splits into bounded independent units, even when nobody mentions orchestration, as the description says. Small or tightly coupled work stays with you.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"mode on --everywhere","type":"text","marks":[{"type":"code_inline"}]},{"text":" makes it the default in every project, with no expiry. ","type":"text"},{"text":"mode off","type":"text","marks":[{"type":"code_inline"}]},{"text":" in a project turns it off there only, and ","type":"text"},{"text":"mode off --everywhere","type":"text","marks":[{"type":"code_inline"}]},{"text":" removes the default.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"mode on","type":"text","marks":[{"type":"code_inline"}]},{"text":" keeps it on for the current project (the repository root, or the folder) until ","type":"text"},{"text":"mode off","type":"text","marks":[{"type":"code_inline"}]},{"text":", or after 12 hours without a prompt. A project mode wins over the default.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Both modes take the same options:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"--goal save","type":"text","marks":[{"type":"code_inline"}]},{"text":" spends as little as possible, ","type":"text"},{"text":"max","type":"text","marks":[{"type":"code_inline"}]},{"text":" gets the most out of the available agents, and ","type":"text"},{"text":"balanced","type":"text","marks":[{"type":"code_inline"}]},{"text":" (the default) delegates only when the playbook says it pays.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"--goal freestyle","type":"text","marks":[{"type":"code_inline"}]},{"text":" drops playbooks, plans and waves: you work as usual and only watch over cost. Check quota before long or heavy steps. When a cheaper or better route exists, say so in one line, at most once per reply and not again once declined: \"This rename is mechanical, so I'll hand it to a Sonnet worker to keep it cheap.\" Take a cheaper route that keeps the quality the step needs and say so in that line; ask before anything that costs more, starts several workers or passes the quota stop line. Your own model is the user's to change: suggest ","type":"text"},{"text":"/model \u003cname>","type":"text","marks":[{"type":"code_inline"}]},{"text":". It is not ","type":"text"},{"text":"freeform","type":"text","marks":[{"type":"code_inline"}]},{"text":", which drops the playbook of a running orchestration.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"--care careful","type":"text","marks":[{"type":"code_inline"}]},{"text":" reads before writing, isolates every writer, reviews every diff before integrating it, and asks before anything hard to undo.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"--rule \"\u003cthe user's words>\"","type":"text","marks":[{"type":"code_inline"}]},{"text":" keeps a standing instruction, up to 8; ","type":"text"},{"text":"--clear-rules","type":"text","marks":[{"type":"code_inline"}]},{"text":" drops them.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Turn it on when the user says ","type":"text"},{"text":"/alethe on","type":"text","marks":[{"type":"code_inline"}]},{"text":", gives a standing instruction, or asks for orchestration \"from now on\". Map the instruction to a goal or care level when one fits, and keep the user's own words as a rule. Confirm in one line what is now on, then go back to what they were doing.","type":"text"}]},{"type":"paragraph","content":[{"text":"The Claude Code hook reminds you while a mode is on; without it, run ","type":"text"},{"text":"mode show","type":"text","marks":[{"type":"code_inline"}]},{"text":" at session start and after compaction. Delegate only where the overhead pays under the goal; keep small work yourself.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Join mid-task","type":"text"}]},{"type":"paragraph","content":[{"text":"Join at any point, including inside another skill's workflow. Do not start over:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Take what already exists as the plan: the conversation, a spec, a task list, a design doc, the diff so far. Point workers at those files instead of copying them.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Keep the other workflow's steps and gates: pending tasks become units and its checks become gates. Rewrite its files only when one of its steps says to.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Use ","type":"text"},{"text":"new","type":"text","marks":[{"type":"code_inline"}]},{"text":" only for delegated work. Keep small, coupled or ambiguous work with the lead; say why in one line.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Pick a playbook, then adapt","type":"text"}]},{"type":"paragraph","content":[{"text":"Before planning, run ","type":"text"},{"text":"scripts/orchestrate.py playbook --pick","type":"text","marks":[{"type":"code_inline"}]},{"text":" for the active profile, provider order, pick conditions and playbook summaries. Then run ","type":"text"},{"text":"playbook \u003cid>","type":"text","marks":[{"type":"code_inline"}]},{"text":" for the chosen playbook, resolved tiers and quota and budget rules. Both accept ","type":"text"},{"text":"--json","type":"text","marks":[{"type":"code_inline"}]},{"text":"; plain ","type":"text"},{"text":"playbook","type":"text","marks":[{"type":"code_inline"}]},{"text":" prints the whole file for editing or onboarding only. See ","type":"text"},{"text":"references/playbooks.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/playbooks.md","title":null}}]},{"text":".","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Pick a playbook with ","type":"text"},{"text":"pickOrder","type":"text","marks":[{"type":"code_inline"}]},{"text":". A playbook the user names wins; so does a runner strategy name, which keeps the single-wave behavior described below. ","type":"text"},{"text":"complete","type":"text","marks":[{"type":"code_inline"}]},{"text":" adapts per unit when nothing else fits.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Run the phases in order. ","type":"text"},{"text":"who: lead","type":"text","marks":[{"type":"code_inline"}]},{"text":" is you, in this session. Every other phase is one wave of workers: its ","type":"text"},{"text":"runner","type":"text","marks":[{"type":"code_inline"}]},{"text":" is the plan ","type":"text"},{"text":"strategy","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"workers","type":"text","marks":[{"type":"code_inline"}]},{"text":" the unit count, ","type":"text"},{"text":"readOnly","type":"text","marks":[{"type":"code_inline"}]},{"text":" means ","type":"text"},{"text":"mode: read-only","type":"text","marks":[{"type":"code_inline"}]},{"text":", and ","type":"text"},{"text":"isolate","type":"text","marks":[{"type":"code_inline"}]},{"text":" gives each writer its own worktree.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Skip a phase whose ","type":"text"},{"text":"if","type":"text","marks":[{"type":"code_inline"}]},{"text":" does not hold; among paths joined by ","type":"text"},{"text":"with","type":"text","marks":[{"type":"code_inline"}]},{"text":", the first true one runs. Check a phase's ","type":"text"},{"text":"gate","type":"text","marks":[{"type":"code_inline"}]},{"text":" yourself; on fail, run its ","type":"text"},{"text":"after","type":"text","marks":[{"type":"code_inline"}]},{"text":" chain once per unit, then check again.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Resolve each worker's tier through the playbooks file and active profile, then apply the first matching rule. Deprecated ","type":"text"},{"text":"policy.modelTiers","type":"text","marks":[{"type":"code_inline"}]},{"text":" overrides still win when set; save new tiers only in playbooks. Pass model and effort explicitly; keep read-only work on a capable provider.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Say which playbook you picked and why in one line before the first wave.","type":"text"}]}]}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Watch the quota","type":"text"}]},{"type":"paragraph","content":[{"text":"Never let a provider run out, under every goal, profile and runner strategy, asked or not. Before every wave:","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Run ","type":"text"},{"text":"scripts/orchestrate.py usage","type":"text","marks":[{"type":"code_inline"}]},{"text":". Codex readings come from its session files, Claude readings from each Claude worker's ","type":"text"},{"text":"rate_limit_event","type":"text","marks":[{"type":"code_inline"}]},{"text":". When the plan or you use Claude and its reading is missing or stale (>30 minutes), run ","type":"text"},{"text":"usage --refresh-claude","type":"text","marks":[{"type":"code_inline"}]},{"text":", a one-line Haiku call.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Read ","type":"text"},{"text":"quotaForecast","type":"text","marks":[{"type":"code_inline"}]},{"text":" in ","type":"text"},{"text":"dry-run","type":"text","marks":[{"type":"code_inline"}]},{"text":": what the wave may take of each window, learned from past waves. With too few samples it is unknown; say so instead of guessing.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"If a provider is low (at or above ","type":"text"},{"text":"quota.criticalThreshold","type":"text","marks":[{"type":"code_inline"}]},{"text":") and the next provider in the order runs the same tier, move the work and tell the user in one line with the numbers and reset time: \"Moving the scouts to Codex: your Claude is at 86% of its 5-hour window (resets 14:30), Codex is at 33%.\" Spend quota that would be lost per ","type":"text"},{"text":"quota.whenSpare","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Stop and ask before launching when the only way around a low provider lowers the tier the work needs, every provider is low, or the forecast passes ","type":"text"},{"text":"quota.stopAt","type":"text","marks":[{"type":"code_inline"}]},{"text":". Say what the work needs, what is left and when it resets, then offer two to four options with a recommendation, using the agent's question tool when it has one:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"wait for the reset;","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"run it on another provider, saying what changes;","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"a cheaper tier only for the units that allow it, saying the risk;","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"run now only the units that fit;","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"connect another CLI or account (","type":"text"},{"text":"setup","type":"text","marks":[{"type":"code_inline"}]},{"text":");","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"go on anyway, only on the user's word, with ","type":"text"},{"text":"run --allow-low-quota","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"\"This migration needs the strong tier (Opus). Your Claude is at 91% of its 5-hour window and resets at 14:30, in 2h10; this wave may take about 15%. 1) wait until 14:30 (recommended), 2) Codex high now, at 40%, 3) Sonnet for the two mechanical units only, 4) connect another CLI.\"","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"run","type":"text","marks":[{"type":"code_inline"}]},{"text":" refuses a wave, and skips queued workers (","type":"text"},{"text":"failureKind: quota_low","type":"text","marks":[{"type":"code_inline"}]},{"text":"), on a provider at or past ","type":"text"},{"text":"quota.stopAt","type":"text","marks":[{"type":"code_inline"}]},{"text":". A worker that hit a limit fails as ","type":"text"},{"text":"rate_limited","type":"text","marks":[{"type":"code_inline"}]},{"text":". Handle both as step 4.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"The user's pinned provider, and their answer, win over these rules. When you run on a low provider yourself, say so, keep your own turns short and hand work to a provider with room. If a reading is stale, say so instead of asserting its number. Record every change with ","type":"text"},{"text":"note \"\u003creason and readings>\" --kind switch","type":"text","marks":[{"type":"code_inline"}]},{"text":". See the reference for unknown readings.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Change course","type":"text"}]},{"type":"paragraph","content":[{"text":"Between waves, adapt the plan to results or the user's request: skip, repeat or add phases, do a step yourself, change providers, models or budget, switch playbooks, or plan freeform waves. The user's request wins over the playbook.","type":"text"}]},{"type":"paragraph","content":[{"text":"Record each change in one line before the next wave, so the report can explain it:","type":"text"}]},{"type":"table","attrs":{"layout":null},"content":[{"type":"tr","content":[{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Change","type":"text"}]}]},{"type":"th","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"Command","type":"text"}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"skip a phase","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"note \"\u003cphase>: \u003cwhy>\" --kind skip","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"add a phase or a worker","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"note \"\u003cwhat>: \u003cwhy>\" --kind add","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"switch playbook, or go freeform","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"note \"to \u003cplaybook or freeform>: \u003cwhy>\" --kind switch","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"do a step yourself","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"note \"\u003cwhat you did>\" --kind lead","type":"text","marks":[{"type":"code_inline"}]}]}]}]},{"type":"tr","content":[{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"anything else","type":"text"}]}]},{"type":"td","attrs":{"colspan":1,"rowspan":1,"colwidth":null,"alignment":""},"content":[{"type":"paragraph","content":[{"text":"note \"\u003cwhat changed and why>\"","type":"text","marks":[{"type":"code_inline"}]}]}]}]}]},{"type":"paragraph","content":[{"text":"A freeform wave is an ordinary plan: leave ","type":"text"},{"text":"playbook","type":"text","marks":[{"type":"code_inline"}]},{"text":" out and name the ","type":"text"},{"text":"phase","type":"text","marks":[{"type":"code_inline"}]},{"text":" after what the wave does. Stop a wave that is no longer useful with ","type":"text"},{"text":"stop","type":"text","marks":[{"type":"code_inline"}]},{"text":"; workers that already ran keep their results.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Orchestrate","type":"text"}]},{"type":"ordered_list","attrs":{"order":1,"listStyle":"number"},"content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Delegate when the overhead pays; runner strategies ","type":"text"},{"text":"economy","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"rescue","type":"text","marks":[{"type":"code_inline"}]},{"text":" allow one worker.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Read ","type":"text"},{"text":"references/routing.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/routing.md","title":null}}]},{"text":" for the runner strategies each wave can use: ","type":"text"},{"text":"auto","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"economy","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"balanced","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"quality","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"research","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"parallel","type":"text","marks":[{"type":"code_inline"}]},{"text":", or ","type":"text"},{"text":"rescue","type":"text","marks":[{"type":"code_inline"}]},{"text":". An explicit runner strategy in the current request wins over saved preferences. Treat the word after an explicit skill invocation as the playbook or runner strategy when it matches a name.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Run ","type":"text"},{"text":"scripts/orchestrate.py probe","type":"text","marks":[{"type":"code_inline"}]},{"text":" and use only available providers. Read ","type":"text"},{"text":"references/providers.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/providers.md","title":null}}]},{"text":" when using Agy, Kimi, resuming a worker, or adding a custom CLI.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Give writers disjoint ownership. Use an isolated worktree for concurrent writers; a single writer may use the main checkout when existing user changes are preserved.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Create a version-1 plan JSON outside the repository. ","type":"text"},{"text":"scripts/orchestrate.py scaffold \u003cstrategy> --workers N","type":"text","marks":[{"type":"code_inline"}]},{"text":" prints a valid skeleton to fill in. Use ","type":"text"},{"text":"strategy","type":"text","marks":[{"type":"code_inline"}]},{"text":", and use ","type":"text"},{"text":"provider: auto","type":"text","marks":[{"type":"code_inline"}]},{"text":" only when saved provider switching preferences may decide the provider. Set the plan's ","type":"text"},{"text":"budget","type":"text","marks":[{"type":"code_inline"}]},{"text":" when one wave needs hard limits, and run ","type":"text"},{"text":"budget","type":"text","marks":[{"type":"code_inline"}]},{"text":" when a limit covers every wave of the orchestration. Give the plan a ","type":"text"},{"text":"title","type":"text","marks":[{"type":"code_inline"}]},{"text":", the ","type":"text"},{"text":"playbook","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"phase","type":"text","marks":[{"type":"code_inline"}]},{"text":" it runs, and a one-line ","type":"text"},{"text":"why","type":"text","marks":[{"type":"code_inline"}]},{"text":"; give each worker a short ","type":"text"},{"text":"why","type":"text","marks":[{"type":"code_inline"}]},{"text":" naming its role and the reason for its model. The report shows them. Write each task as ","type":"text"},{"text":"references/briefs.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/briefs.md","title":null}}]},{"text":" describes. Codex and Claude workers start without MCP servers; set ","type":"text"},{"text":"mcp: true","type":"text","marks":[{"type":"code_inline"}]},{"text":" only for a brief that needs one. Validate the plan against ","type":"text"},{"text":"schemas/plan.schema.json","type":"text","marks":[{"type":"link","attrs":{"href":"schemas/plan.schema.json","title":null}}]},{"text":" or with ","type":"text"},{"text":"scripts/orchestrate.py validate","type":"text","marks":[{"type":"code_inline"}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Run ","type":"text"},{"text":"models","type":"text","marks":[{"type":"code_inline"}]},{"text":". For a needed missing or stale price (>60 days), look it up only on the vendor's official pricing page, tell the user, and save the entry with ","type":"text"},{"text":"source","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"asOf","type":"text","marks":[{"type":"code_inline"}]},{"text":" in ","type":"text"},{"text":"alethe-prices.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" next to the policy; never guess. See ","type":"text"},{"text":"references/operations.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/operations.md","title":null}}]},{"text":". Preview with ","type":"text"},{"text":"dry-run","type":"text","marks":[{"type":"code_inline"}]},{"text":": runner strategy, providers, concurrency, prompt tokens, budget and ","type":"text"},{"text":"listPriceEstimateUsd","type":"text","marks":[{"type":"code_inline"}]},{"text":" (an estimate) and ","type":"text"},{"text":"quotaForecast","type":"text","marks":[{"type":"code_inline"}]},{"text":". Fix warnings, including default models, low providers and the stop line, or explain why they do not apply. When the user wants to see the architecture, save the preview with ","type":"text"},{"text":"dry-run --write","type":"text","marks":[{"type":"code_inline"}]},{"text":", then render it with ","type":"text"},{"text":"scripts/diagram.py","type":"text","marks":[{"type":"code_inline"}]},{"text":" as described in ","type":"text"},{"text":"references/diagram.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/diagram.md","title":null}}]},{"text":". Show the SVG before running workers and use only confirmed proxy routes.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Start the orchestration once with ","type":"text"},{"text":"scripts/orchestrate.py new \"\u003ctitle>\" --playbook \u003cname> --why \"\u003cone line>\" --lead \u003cyour CLI>:\u003cyour model>","type":"text","marks":[{"type":"code_inline"}]},{"text":", such as ","type":"text"},{"text":"--lead claude:opus","type":"text","marks":[{"type":"code_inline"}]},{"text":"; it creates the folder, makes it current and names you as the lead in the report. Start every wave with ","type":"text"},{"text":"run \u003cplan> --in current --detach","type":"text","marks":[{"type":"code_inline"}]},{"text":", then call ","type":"text"},{"text":"wait [--timeout N]","type":"text","marks":[{"type":"code_inline"}]},{"text":" (at most 540 seconds) until ","type":"text"},{"text":"done","type":"text","marks":[{"type":"code_inline"}]},{"text":" is true; it returns status too. Claude Code may instead use the shell tool's background mode. Never run a wave in the foreground of a shell tool with a short timeout. Interrupted runners are detected; ","type":"text"},{"text":"stop","type":"text","marks":[{"type":"code_inline"}]},{"text":" also kills leftover workers. The runner caps concurrency and budgets and leaves worktrees for review. It uses argument arrays; an unreadable ","type":"text"},{"text":".cmd","type":"text","marks":[{"type":"code_inline"}]},{"text":" shim runs through ","type":"text"},{"text":"cmd.exe","type":"text","marks":[{"type":"code_inline"}]},{"text":" only under the argument restrictions in ","type":"text"},{"text":"references/providers.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/providers.md","title":null}}]},{"text":".","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Read ","type":"text"},{"text":"state.json","type":"text","marks":[{"type":"code_inline"}]},{"text":", each ","type":"text"},{"text":"outcome.json","type":"text","marks":[{"type":"code_inline"}]},{"text":" (its ","type":"text"},{"text":"workerStatus","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"failureKind","type":"text","marks":[{"type":"code_inline"}]},{"text":" and ","type":"text"},{"text":"readOnlyViolation","type":"text","marks":[{"type":"code_inline"}]},{"text":" first), the concise ","type":"text"},{"text":"result.md","type":"text","marks":[{"type":"code_inline"}]},{"text":", and writer ","type":"text"},{"text":"diff.patch","type":"text","marks":[{"type":"code_inline"}]},{"text":" files. Do not load raw logs unless diagnosing a failure.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Integrate reviewed changes, run validation in the combined checkout, ","type":"text"},{"text":"mark","type":"text","marks":[{"type":"code_inline"}]},{"text":" each worker's result ","type":"text"},{"text":"used","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"partial","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"discarded","type":"text","marks":[{"type":"code_inline"}]},{"text":", then remove only temporary data and worktrees whose useful changes were handled. ","type":"text"},{"text":"mark \u003cworker> used","type":"text","marks":[{"type":"code_inline"}]},{"text":" finds the worker in the current orchestration.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Close with ","type":"text"},{"text":"scripts/orchestrate.py report","type":"text","marks":[{"type":"code_inline"}]},{"text":" and show its output: work, reasons, changes, reported spend and what remains. Use it for user cost questions; no folder means the current orchestration.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"For plan examples, result semantics, selective dirty-state seeding, and cleanup, read ","type":"text"},{"text":"references/operations.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/operations.md","title":null}}]},{"text":".","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Worker contract","type":"text"}]},{"type":"paragraph","content":[{"text":"Each brief must state the task goal, definition of done, owned paths, relevant repository rules, checks, forbidden changes, and desired handoff. The runner automatically appends these invariants:","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"do not delegate, spawn subagents, or launch another agent CLI","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"do not commit, push, tag, release, or open a pull request","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"return a concise outcome, evidence, changed files, checks, failures, and risks","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"finish with a ","type":"text"},{"text":"STATUS:","type":"text","marks":[{"type":"code_inline"}]},{"text":" line (","type":"text"},{"text":"DONE","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"DONE_WITH_CONCERNS","type":"text","marks":[{"type":"code_inline"}]},{"text":", ","type":"text"},{"text":"BLOCKED","type":"text","marks":[{"type":"code_inline"}]},{"text":" or ","type":"text"},{"text":"NEEDS_CONTEXT","type":"text","marks":[{"type":"code_inline"}]},{"text":") and a ","type":"text"},{"text":"SUMMARY:","type":"text","marks":[{"type":"code_inline"}]},{"text":" line","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Codex and Claude workers cannot start subagents: the runner turns that off in the CLI. Every worker runs with ","type":"text"},{"text":"ALETHE_ORCHESTRATE_WORKER=1","type":"text","marks":[{"type":"code_inline"}]},{"text":", so it cannot start an orchestration of its own. A worker's status and summary are claims; check its diff. For review, validation, judge and panel briefs, follow ","type":"text"},{"text":"references/briefs.md","type":"text","marks":[{"type":"link","attrs":{"href":"references/briefs.md","title":null}}]},{"text":".","type":"text"}]},{"type":"paragraph","content":[{"text":"Do not put secrets in plans, tasks, arguments, or logs. Do not use bypass flags unless the user explicitly requested that loss of isolation; custom workers require ","type":"text"},{"text":"allowUnsafe: true","type":"text","marks":[{"type":"code_inline"}]},{"text":" for known unsafe flags.","type":"text"}]},{"type":"heading","attrs":{"level":2},"content":[{"text":"Token discipline","type":"text"}]},{"type":"bullet_list","content":[{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Treat statements such as \"low on tokens\", \"save quota\", or \"economize\" as runner strategy ","type":"text"},{"text":"economy","type":"text","marks":[{"type":"code_inline"}]},{"text":" for the current run. Persist it with ","type":"text"},{"text":"policy set economy","type":"text","marks":[{"type":"code_inline"}]},{"text":" only when the user asks for an ongoing default.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"A plan ","type":"text"},{"text":"budget","type":"text","marks":[{"type":"code_inline"}]},{"text":" turns saving into a hard limit: cap worker calls, wall time, estimated prompt tokens, reported tokens and reported cost. Use ","type":"text"},{"text":"dry-run","type":"text","marks":[{"type":"code_inline"}]},{"text":" to see the estimate before anything runs. Only Claude reports cost, so ","type":"text"},{"text":"maxReportedTokens","type":"text","marks":[{"type":"code_inline"}]},{"text":" caps the others. Cache reads are reported separately and excluded from ","type":"text"},{"text":"maxReportedTokens","type":"text","marks":[{"type":"code_inline"}]},{"text":" and the report's Value token split.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"More workers increase total provider usage. ","type":"text"},{"text":"auto","type":"text","marks":[{"type":"code_inline"}]},{"text":" must choose no orchestration when the task is cheaper to finish directly; if orchestration was explicitly requested, it chooses the smallest valid runner strategy.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Point to files rather than copying them; read concise results and diffs, not transcripts.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"Reuse ","type":"text"},{"text":"sessionId","type":"text","marks":[{"type":"code_inline"}]},{"text":" only for focused follow-ups. Use cheap workers for bounded discovery and mechanical work; keep ambiguity and integration with the lead or a capable reviewer.","type":"text"}]}]},{"type":"list_item","content":[{"type":"paragraph","content":[{"text":"A failed worker may be retried once only when the error has a specific correctable cause.","type":"text"}]}]}]},{"type":"paragraph","content":[{"text":"Do not commit, push, tag, release, or open a pull request unless the user explicitly asked for it.","type":"text"}]}]},"metadata":{"date":"2026-11-02","name":"alethe","author":"@skillopedia","source":{"stars":99,"repo_name":"alethe-orchestrate-skill","origin_url":"https://github.com/Kc1t/alethe-orchestrate-skill/blob/HEAD/skills/alethe/SKILL.md","repo_owner":"Kc1t","body_sha256":"7932b8f9a174610e09b704914b160bef9430351b0a59cd7834ce58adc0a7951a","cluster_key":"6b4cc194abc1363b956b41de25cc0bbd98b81a93a3ea0465c6471936ea6fe2b5","clean_bundle":{"format":"clean-skill-bundle-v1","source":"Kc1t/alethe-orchestrate-skill/skills/alethe/SKILL.md","text_bytes":622522,"source_commit":"460ba5ed3646bee45a7fadcbbe199d571582a145","skills_sh_rank":17333,"attachment_count":15,"text_attachments":15,"attachment_storage":"skill_attachments","binary_attachments":0},"cluster_size":1,"skill_md_path":"skills/alethe/SKILL.md","import_metadata":{"date":"2026-11-02","author":"@skillopedia","version":"v1","category":"general","category_label":"General"},"exact_dupes_collapsed_into_this":0},"version":"v1","category":"general","metadata":{"version":"21"},"import_tag":"skills-sh-top20k-2026-11","description":"Orchestrate bounded Codex, Claude Code, Agy, Kimi, or custom CLI workers to finish work with the least lead context and provider usage, following playbooks such as plan big, execute small. Use when the user asks to orchestrate, delegate, compare agents, conserve agent tokens or quota, recover a worker, report what an orchestration did and cost, or work without MCP or Python, when the user types /alethe or a short command such as \"alethe report\" or \"alethe stop\", and when they want it on from now on, at any point of a task and alongside other skills. Also use it proactively, without being asked, when a request is large enough to split into bounded independent units, such as a multi-file refactor, a migration, an audit or research across many files. Do not use for a small mechanical task unless explicitly requested.","argument-hint":"[task or standing instruction] | on | off | save | max | freestyle | careful | report | status | usage | models | editor | list | stop | skip \u003cphase> | switch \u003cplaybook> | freeform | budget \u003climit> | mark \u003cworker> \u003cused|partial|discarded> | setup | help"}},"renderedAt":1793666106961}

Important: agents should read /llm.txt, /llms.txt, or /.well-known/skills.json to discover the public Skillopedia API.