#!/usr/bin/env python3 """CartoonOS Operations & Planning CLI. This module provides local planning, validation, roadmap generation, snapshot planning, and production unit economics utilities for CartoonOS. It runs purely locally without external network calls, provider mutations, or secret exposure. Core Invariants: 1. Top 50 Task Graph: Strictly exactly 50 unique tasks (CO-001 through CO-050) ordered topologically with acyclic dependencies. 2. Evidence Enforcement: Done tasks must link to existing local files on disk. 3. Financial Accuracy: Economics calculations use exact Python Decimal math to prevent floating-point rounding drift in production budget estimates. 4. Timezone Strictness: Snapshot checkpoints enforce explicit UTC offsets to prevent silent scheduling skew across international distribution windows. """ import argparse import json import math import re import shutil import sys from datetime import datetime, timedelta, timezone from decimal import Decimal from pathlib import Path from typing import Any # Global directory roots and file locations ROOT: Path = Path(__file__).resolve().parents[1] TASK_FILE: Path = ROOT / "data/operations/priority-tasks.json" # Canonical performance snapshot checkpoints: D1, D3, D7, D14, D30, D90 CHECKPOINTS: tuple[int, ...] = (1, 3, 7, 14, 30, 90) # Valid workflow task statuses STATUSES: frozenset[str] = frozenset({"backlog", "in_progress", "blocked", "done"}) def read_plan() -> dict[str, Any]: """Reads and parses the operational priority tasks JSON plan from disk. Returns: dict[str, Any]: The parsed priority-tasks structure. Raises: FileNotFoundError: If data/operations/priority-tasks.json does not exist. json.JSONDecodeError: If the plan file is malformed JSON. """ base = json.loads(TASK_FILE.read_text(encoding="utf-8")) patch = json.loads((ROOT / "data/operations/priority-task-corrections-2026-09-08.json").read_text(encoding="utf-8")) return apply_corrections(base, patch) def apply_corrections(base: dict[str, Any], patch: dict[str, Any]) -> dict[str, Any]: known = {task["id"] for task in base["tasks"]} overrides: dict[str, Any] = {} for item in patch["task_overrides"]: if item["id"] not in known or item["id"] in overrides: raise ValueError(f"Unknown or repeated readiness correction: {item['id']}") overrides[item["id"]] = item return { **base, "as_of": patch["as_of"], "scope": patch["scope_override"], "tasks": [{**task, **overrides.get(task["id"], {}), "id": task["id"]} for task in base["tasks"]], } def validate_plan(plan: dict[str, Any], root: Path = ROOT) -> list[str]: """Validates the operational plan against topological, evidentiary, and structural rules. Validation Rules: 1. Exactly 50 unique tasks CO-001 through CO-050. 2. Ranks must be unique integers from 1 to 50. 3. Mandatory fields: title, owner, effort, acceptance, evidence. 4. Valid status (backlog, in_progress, blocked, done) and priority (P0, P1, P2). 5. Blocked tasks must supply a blocked_reason. 6. Done tasks cannot have unfinished dependencies. 7. Evidence files must exist on disk and be relative to repository root. 8. The dependency graph must be a Directed Acyclic Graph (DAG) with zero cycles. Args: plan: Parsed task plan dictionary. root: Base repository root path for resolving evidence files. Returns: list[str]: A list of human-readable error descriptions. Empty if valid. """ errors: list[str] = [] tasks: list[dict[str, Any]] = plan.get("tasks", []) # 1. Structural count and ID checks expected_ids = {f"CO-{n:03d}" for n in range(1, 51)} actual_ids = {t.get("id") for t in tasks} if len(tasks) != 50 or actual_ids != expected_ids: errors.append("Exactly 50 unique task IDs CO-001 through CO-050 are required.") # 2. Sequential ranking check actual_ranks = sorted(t.get("rank", 0) for t in tasks) if actual_ranks != list(range(1, 51)): errors.append("Task ranks must be unique integers 1 through 50.") by_id: dict[str, dict[str, Any]] = {task.get("id", ""): task for task in tasks} # 3. Individual task field, status, dependency, and evidence validation for task in tasks: task_id = task.get("id", "missing") # Mandatory metadata keys for key in ("title", "owner", "effort", "acceptance", "evidence"): if not task.get(key): errors.append(f"{task_id}: missing {key}") # Status and priority invariants if task.get("status") not in STATUSES: errors.append(f"{task_id}: invalid status") if task.get("priority") not in {"P0", "P1", "P2"}: errors.append(f"{task_id}: invalid priority") if task.get("status") == "blocked" and not task.get("blocked_reason"): errors.append(f"{task_id}: blocked task needs a reason") # Dependency integrity for dependency in task.get("depends_on", []): if dependency not in by_id or dependency == task_id: errors.append(f"{task_id}: invalid dependency {dependency}") elif task.get("status") == "done" and by_id[dependency].get("status") != "done": errors.append(f"{task_id}: done with unfinished dependency {dependency}") # Evidence file existence on disk for evidence in task.get("evidence", []): candidate = (root / evidence).resolve() if not candidate.is_relative_to(root.resolve()) or not candidate.is_file(): errors.append(f"{task_id}: missing/local-invalid evidence {evidence}") # 4. Cycle Detection using Depth-First Search (DFS) visited: set[str] = set() visiting: set[str] = set() def walk(current_id: str) -> None: """DFS recursive walk to detect cycles in the dependency graph.""" if current_id in visiting: errors.append(f"Dependency cycle at {current_id}") return if current_id in visited or current_id not in by_id: return visiting.add(current_id) for dependency_id in by_id[current_id].get("depends_on", []): walk(dependency_id) visiting.remove(current_id) visited.add(current_id) for task_id in by_id: walk(task_id) return errors def eligible_tasks(plan: dict[str, Any]) -> list[dict[str, Any]]: """Returns unblocked tasks whose dependencies are completely done. Tasks are filtered to only those in 'backlog' or 'in_progress' states where every dependency ID is marked 'done', sorted by rank. Args: plan: The parsed task plan. Returns: list[dict[str, Any]]: Unblocked tasks ordered by rank. """ done_ids = {t["id"] for t in plan["tasks"] if t.get("status") == "done"} return [ t for t in sorted(plan["tasks"], key=lambda task: task.get("rank", 0)) if t.get("status") in {"backlog", "in_progress"} and all(dep in done_ids for dep in t.get("depends_on", [])) ] def snapshot_plan( organization: str, channel: str, asset: str, published_at: str ) -> list[dict[str, Any]]: """Generates the D1, D3, D7, D14, D30, D90 performance snapshot collection windows. Enforces strict identifier sanitization and explicit timezone offsets (ISO 8601). Args: organization: Organization identifier (e.g. 'ORG-CARTOONOS'). channel: Channel identifier (e.g. 'CH-MAIN'). asset: Asset identifier (e.g. 'S01E01'). published_at: ISO 8601 publication timestamp with timezone offset or 'Z'. Returns: list[dict[str, Any]]: List of snapshot plan checkpoint records. Raises: ValueError: If identifiers contain unsafe characters or timestamp lacks timezone. """ for identifier in (organization, channel, asset): if not re.fullmatch(r"[A-Za-z0-9_-]+", identifier): raise ValueError("IDs must contain only letters, digits, underscores or hyphens.") # Parse and normalize published_at to UTC published_iso = published_at.replace("Z", "+00:00") published = datetime.fromisoformat(published_iso) if published.tzinfo is None or published.utcoffset() is None: raise ValueError("published-at requires an explicit timezone offset or Z.") published_utc = published.astimezone(timezone.utc) # Build checkpoint records for D1, D3, D7, D14, D30, D90 return [ { "key": f"{organization}:{channel}:{asset}:D{day}:v1", "organization_id": organization, "channel_id": channel, "asset_id": asset, "checkpoint": f"D{day}", "window_start_at": published_utc.isoformat(), "due_at": (published_utc + timedelta(days=day)).isoformat(), "state": "planning", "value": None, "note": "Due time only; collector must record actual source window and data-through time.", } for day in CHECKPOINTS ] def format_currency(value: Decimal) -> str: """Formats a Decimal number as standard currency with two decimal places.""" return str(value.quantize(Decimal("0.01"))) def economics( seconds: Any, shot_seconds: Any, acceptance_rate: Any, credits_per_attempt: Any, usd_per_credit: Any, labor_hours: Any, hourly_usd: Any, overhead_usd: Any, ) -> dict[str, Any]: """Calculates production unit economics and cost per accepted second. Uses high-precision Decimal arithmetic to avoid floating-point drift. Args: seconds: Total target video duration in seconds. shot_seconds: Average shot duration in seconds. acceptance_rate: Estimated shot acceptance probability (0.0 to 1.0]. credits_per_attempt: Provider generation credits per attempt. usd_per_credit: Cost in USD per provider credit. labor_hours: Production review and assembly hours required. hourly_usd: Hourly labor rate in USD. overhead_usd: Fixed overhead costs (music, assets, licenses) in USD. Returns: dict[str, Any]: Detailed breakdown of shots, attempts, credits, cash, and total costs. Raises: ValueError: If numeric values are negative, infinite, or acceptance_rate is outside (0, 1]. """ raw_inputs = ( seconds, shot_seconds, acceptance_rate, credits_per_attempt, usd_per_credit, labor_hours, hourly_usd, overhead_usd, ) values = [Decimal(str(x)) for x in raw_inputs] if not all(x.is_finite() for x in values): raise ValueError("Inputs must be finite numbers.") sec, length, rate, credits_val, fx, hours, hourly, overhead = values if any(x <= 0 for x in (sec, length, credits_val, fx)): raise ValueError("Duration, shot length, credits and USD/credit must be positive.") if not 0 < rate <= 1 or any(x < 0 for x in (hours, hourly, overhead)): raise ValueError("Acceptance must be in (0, 1]; labor and overhead must be non-negative.") # Mathematical shot modeling shots_needed = math.ceil(sec / length) expected_attempts = math.ceil(Decimal(shots_needed) / rate) total_credits = expected_attempts * credits_val media_usd = total_credits * fx cash_usd = media_usd + overhead economic_usd = cash_usd + (hours * hourly) return { "state": "planning", "accepted_shots_needed": shots_needed, "expected_attempts_rounded_up": expected_attempts, "credits": str(total_credits), "media_usd": format_currency(media_usd), "cash_cost_usd": format_currency(cash_usd), "economic_cost_usd": format_currency(economic_usd), "economic_usd_per_accepted_second": format_currency(economic_usd / sec), "assumptions": [ "Every shot has the supplied cost and acceptance probability; no variance guarantee.", "USD/credit must come from your quote or invoice; this utility has no current prices.", "Add subscription waste, music, storage, contractors and contingency in overhead as appropriate.", "Cash cost excludes founder labor; economic cost includes the supplied labor rate.", ], } def render_roadmap(plan: dict[str, Any]) -> str: """Renders the Markdown roadmap table for docs/readiness/TOP_50_TASKS.md. Args: plan: The parsed priority tasks plan. Returns: str: Fully formatted Markdown content. """ def clean_table_cell(value: Any) -> str: """Sanitizes strings for markdown table cells.""" return str(value).replace("|", "\\|").replace("\n", " ") lines = [ "# Top 50 CartoonOS priorities", "", f"Reviewed plan as of {plan['as_of']}. Generated from the base register and active correction overlay (also used by Studio).", "", "The base register is a historical reviewed snapshot; this page applies the current correction overlay.", "", "Order follows risk and dependencies. Effort is a rough focused person-day estimate, not a deadline.", "Done means this task's acceptance criteria are implemented/documented; it does not imply production readiness.", "Blocked includes external input; a backlog item may also depend on unfinished work. Evidence links point to code or specifications, not necessarily completed runtime proof.", "", "| Rank / ID | Priority / state | Task and acceptance | Owner / effort | Dependencies / evidence |", "|---|---|---|---|---|", ] for task in sorted(plan["tasks"], key=lambda t: t.get("rank", 0)): evidence_links = " · ".join(f"[{Path(p).name}](../../{p})" for p in task.get("evidence", [])) deps = ", ".join(task.get("depends_on", [])) or "None" rank = task.get("rank", 0) task_id = task.get("id", "") priority = task.get("priority", "") status = task.get("status", "") title = clean_table_cell(task.get("title", "")) acceptance = clean_table_cell(task.get("acceptance", "")) owner = clean_table_cell(task.get("owner", "")) effort = clean_table_cell(task.get("effort", "")) lines.append( f"| {rank} · {task_id} | {priority} / {status} | " f"**{title}** — {acceptance} | " f"{owner} / {effort} | {deps}; {evidence_links} |" ) lines.extend(["", "## Next unblocked work", ""]) for task in eligible_tasks(plan)[:5]: lines.append(f"- {task['id']}: {task['title']} — {task['owner']}.") return "\n".join(lines) + "\n" def doctor() -> dict[str, Any]: """Performs prerequisite diagnostic health checks for the local environment.""" checks = [ {"check": tool, "state": "present" if shutil.which(tool) else "missing"} for tool in ("node", "pnpm", "uv", "python3") ] checks += [ {"check": str(path), "state": "present" if (ROOT / path).is_file() else "missing"} for path in ("pnpm-lock.yaml", "backend/uv.lock", ".env.example") ] return { "scope": "Local prerequisites only; no network, secrets or production health checks.", "checks": checks, "plan_errors": validate_plan(read_plan()), "production_state": "not_verified", "remaining_live_checks": [ "database migration/restore", "authenticated private dashboard", "worker activity assembly", "funded provider and exact references", "durable media export", "owner YouTube analytics OAuth", ], } def main() -> int: """Main CLI entrypoint.""" parser = argparse.ArgumentParser(description=__doc__) commands = parser.add_subparsers(dest="command", required=True) commands.add_parser("validate", help="Validate the priority tasks DAG and evidence") commands.add_parser("roadmap", help="Regenerate docs/readiness/TOP_50_TASKS.md") commands.add_parser("doctor", help="Run local prerequisite diagnostics") snapshots_parser = commands.add_parser("snapshots", help="Plan D1-D90 performance snapshot windows") snapshots_parser.add_argument("--organization", default="ORG-CARTOONOS") for name in ("channel", "asset", "published-at"): snapshots_parser.add_argument(f"--{name}", required=True) economics_parser = commands.add_parser("economics", help="Calculate shot and episode economics") for name in ( "seconds", "shot-seconds", "acceptance-rate", "credits-per-attempt", "usd-per-credit", "labor-hours", "hourly-usd", "overhead-usd", ): economics_parser.add_argument(f"--{name}", required=True, type=Decimal) args = vars(parser.parse_args()) command = args.pop("command") try: if command == "validate": errors = validate_plan(read_plan()) print(json.dumps({"valid": not errors, "errors": errors}, indent=2)) return 1 if errors else 0 if command == "roadmap": plan = read_plan() errors = validate_plan(plan) if errors: raise ValueError("; ".join(errors)) path = ROOT / "docs/readiness/TOP_50_TASKS.md" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(render_roadmap(plan), encoding="utf-8") print(f"Updated {path.relative_to(ROOT)}") return 0 if command == "doctor": result = doctor() elif command == "snapshots": result = snapshot_plan( organization=args["organization"], channel=args["channel"], asset=args["asset"], published_at=args["published_at"], ) elif command == "economics": result = economics(**args) else: parser.error(f"Unknown command {command}") print(json.dumps(result, indent=2)) return 0 except (ValueError, KeyError, TypeError, ArithmeticError) as exc: parser.error(str(exc)) if __name__ == "__main__": sys.exit(main())