#!/usr/bin/env python3
"""Python multiprocessing deadlock risk validator.

This utility provides a practical pre-deployment checklist for common
multiprocessing deadlock patterns. It does not analyze source code directly;
instead, it helps teams review runtime design choices, coordination paths,
and shutdown behavior before release.

Usage examples:
  python multiprocessing_deadlock_validator.py
  python multiprocessing_deadlock_validator.py --profile batch-job
  python multiprocessing_deadlock_validator.py --json
"""

from __future__ import annotations

import argparse
import json
import sys
from dataclasses import dataclass, asdict
from typing import List, Dict


@dataclass
class CheckItem:
    id: str
    severity: str
    question: str
    why_it_matters: str


CHECKS: List[CheckItem] = [
    CheckItem(
        id="boundaries",
        severity="high",
        question="Are process boundaries and ownership clearly defined?",
        why_it_matters=(
            "Deadlocks often come from unclear parent/worker responsibilities, "
            "especially when workers wait on the parent or on each other."
        ),
    ),
    CheckItem(
        id="timeouts",
        severity="high",
        question="Does every blocking queue, pipe, socket, or join operation have a timeout or shutdown path?",
        why_it_matters=(
            "A blocking call with no escape path can prevent clean termination and make recovery impossible."
        ),
    ),
    CheckItem(
        id="queue-drain",
        severity="high",
        question="Does the parent consume worker output before waiting for worker exit?",
        why_it_matters=(
            "Calling join too early can leave queues full, block writers, and freeze the process tree."
        ),
    ),
    CheckItem(
        id="fork-state",
        severity="high",
        question="Are locks, semaphores, open handles, and import-time side effects safe across the chosen start method?",
        why_it_matters=(
            "Inherited locked state or repeated startup code can create circular waits or hidden blocking."
        ),
    ),
    CheckItem(
        id="nested-pool",
        severity="medium",
        question="Do workers avoid submitting more work to the same pool unless ownership and capacity are explicitly designed?",
        why_it_matters=(
            "Nested pool submission can cause pool starvation when all workers wait on work that cannot start."
        ),
    ),
    CheckItem(
        id="shared-state",
        severity="medium",
        question="Is shared mutable state minimized and protected by a simple coordination model?",
        why_it_matters=(
            "The more shared state a worker needs, the more likely blocking and circular dependencies become."
        ),
    ),
    CheckItem(
        id="slow-paths",
        severity="medium",
        question="Have slow paths such as logging, file writes, retries, and network calls been reviewed inside workers?",
        why_it_matters=(
            "A slow side effect inside a worker can hold locks long enough to stall the rest of the system."
        ),
    ),
    CheckItem(
        id="platform-test",
        severity="medium",
        question="Has the exact start method and target platform been tested under realistic load?",
        why_it_matters=(
            "Behavior can differ across Linux, macOS, and Windows, and between fork, spawn, and forkserver."
        ),
    ),
    CheckItem(
        id="constrained-load",
        severity="medium",
        question="Has the design been load-tested with full queues, limited workers, and slow consumers?",
        why_it_matters=(
            "Deadlocks frequently appear only when buffers fill or progress slows under realistic saturation."
        ),
    ),
]

PROFILE_GUIDANCE: Dict[str, List[str]] = {
    "batch-job": [
        "Use bounded queues and drain results before joining workers.",
        "Add a timeout for every wait and define a retry or rollback path.",
        "Verify that partial work can be resumed safely after failure.",
    ],
    "service-worker": [
        "Keep worker tasks small and independent.",
        "Avoid long-lived locks across file, network, or logging operations.",
        "Test shutdown behavior under cancellation and backpressure.",
    ],
    "etl": [
        "Confirm producer and consumer rates remain stable at peak volume.",
        "Watch for queue saturation and pipeline stalls.",
        "Validate that checkpointing does not depend on hidden worker state.",
    ],
    "default": [
        "Prefer simple parent/worker ownership models.",
        "Keep blocking operations bounded with timeouts.",
        "Reproduce startup behavior on the same OS and Python version used in production.",
    ],
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Validate common multiprocessing deadlock risks before deployment."
    )
    parser.add_argument(
        "--profile",
        choices=sorted(PROFILE_GUIDANCE.keys() - {"default"}),
        help="Optional workload profile to print targeted guidance.",
    )
    parser.add_argument(
        "--json",
        action="store_true",
        help="Output machine-readable JSON instead of formatted text.",
    )
    return parser.parse_args()


def build_report(profile: str | None) -> Dict[str, object]:
    selected_profile = profile or "default"
    return {
        "title": "Python multiprocessing deadlock risk review",
        "summary": (
            "Use this checklist to validate coordination, startup behavior, and shutdown paths before release."
        ),
        "profile": selected_profile,
        "guidance": PROFILE_GUIDANCE[selected_profile],
        "checks": [asdict(item) for item in CHECKS],
        "notes": [
            "This tool is a review aid, not a static analyzer.",
            "If a worker can block indefinitely, add a timeout or explicit shutdown signal.",
            "If results must be collected, make sure the parent drains outputs before joining workers.",
        ],
    }


def validate_report(report: Dict[str, object]) -> None:
    if not report.get("checks"):
        raise ValueError("No checks defined.")
    if not isinstance(report.get("guidance"), list):
        raise ValueError("Guidance must be a list.")


def render_text(report: Dict[str, object]) -> str:
    lines = [report["title"], "=" * len(report["title"]), "", report["summary"], ""]
    lines.append("Targeted guidance:")
    for item in report["guidance"]:
        lines.append(f"- {item}")
    lines.append("")
    lines.append("Checks:")
    for check in report["checks"]:
        lines.append(f"- [{check['severity']}] {check['question']}")
        lines.append(f"  Why it matters: {check['why_it_matters']}")
    lines.append("")
    lines.append("Notes:")
    for note in report["notes"]:
        lines.append(f"- {note}")
    return "\n".join(lines)


def main() -> int:
    args = parse_args()
    report = build_report(args.profile)
    validate_report(report)

    if args.json:
        print(json.dumps(report, indent=2, ensure_ascii=False))
    else:
        print(render_text(report))

    return 0


if __name__ == "__main__":
    raise SystemExit(main())