"""Multi-agent partition manifest: structured work partitioning with conflict analysis."""

from __future__ import annotations

import os
import sqlite3
from collections import Counter, defaultdict

import click

from roam.capability import roam_capability
from roam.commands.resolve import ensure_index
from roam.db.connection import batched_in, open_db
from roam.output.formatter import abbrev_kind, json_envelope, to_json
from roam.output.metric_definitions import COGNITIVE_COMPLEXITY_DEFINITION

# ---------------------------------------------------------------------------
# Conflict-risk classifier (extracted for unit-testing — fleet planner
# uses this label to decide parallel-vs-serial work order)
# ---------------------------------------------------------------------------


def _classify_conflict_risk(cross_edges: int, cochange_score: int) -> str:
    """Return ``"LOW"`` / ``"MEDIUM"`` / ``"HIGH"`` for a partition.

    Both bands are conjunctive: a single high signal escalates the label.
    The pre-v12 version used ``OR`` on the medium band, which silently
    classified ``cross_edges=2, cochange_score=200`` as MEDIUM when it
    should be HIGH — fleet output then chose parallel work for what was
    actually conflict-heavy. Caught by the v12 dogfood pass.
    """
    if cross_edges <= 3 and cochange_score <= 2:
        return "LOW"
    if cross_edges <= 10 and cochange_score <= 5:
        return "MEDIUM"
    return "HIGH"


# ---------------------------------------------------------------------------
# Role classification heuristics
# ---------------------------------------------------------------------------

# Map dominant file extensions / directory patterns to human-readable roles.
_EXTENSION_ROLES: dict[str, str] = {
    ".py": "Python",
    ".js": "JavaScript",
    ".ts": "TypeScript",
    ".tsx": "React/TypeScript",
    ".jsx": "React/JavaScript",
    ".go": "Go",
    ".rs": "Rust",
    ".java": "Java",
    ".rb": "Ruby",
    ".php": "PHP",
    ".c": "C",
    ".cpp": "C++",
    ".cs": "C#",
    ".swift": "Swift",
    ".kt": "Kotlin",
    ".scala": "Scala",
    ".sql": "SQL/Database",
    ".html": "Template/HTML",
    ".css": "Styling",
    ".scss": "Styling",
    ".vue": "Vue",
    ".svelte": "Svelte",
}

_DIR_ROLES: list[tuple[str, str]] = [
    ("api", "API Layer"),
    ("routes", "API Layer"),
    ("handler", "API Layer"),
    ("controller", "Controller Layer"),
    ("view", "View/UI Layer"),
    ("component", "UI Component Layer"),
    ("model", "Data Model Layer"),
    ("schema", "Data Schema Layer"),
    ("db", "Database Layer"),
    ("database", "Database Layer"),
    ("migration", "Migration Layer"),
    ("service", "Service Layer"),
    ("domain", "Domain Logic"),
    ("core", "Core Logic"),
    ("util", "Utility Layer"),
    ("helper", "Utility Layer"),
    ("lib", "Library Layer"),
    ("middleware", "Middleware Layer"),
    ("auth", "Auth Layer"),
    ("security", "Security Layer"),
    ("test", "Test Layer"),
    ("tests", "Test Layer"),
    ("spec", "Test Layer"),
    ("specs", "Test Layer"),
    ("config", "Configuration"),
    ("infra", "Infrastructure"),
    ("deploy", "Infrastructure"),
    ("cli", "CLI Layer"),
    ("cmd", "CLI Layer"),
    ("graph", "Graph/Analysis Layer"),
    ("index", "Indexing Layer"),
    ("search", "Search Layer"),
    ("output", "Output/Formatting Layer"),
]


def _suggest_role(files: list[str], language_counts: Counter) -> str:
    """Suggest a human-readable role label from file paths and language mix."""
    # Try directory-based role first (most descriptive)
    dir_parts: list[str] = []
    for f in files:
        parts = f.replace("\\", "/").split("/")
        dir_parts.extend(p.lower() for p in parts[:-1])

    dir_counter = Counter(dir_parts)
    for dir_name, role in _DIR_ROLES:
        if dir_counter.get(dir_name, 0) > 0:
            return role

    # Fall back to dominant language
    if language_counts:
        top_lang = language_counts.most_common(1)[0][0]
        if top_lang:
            return f"{top_lang} Module"

    return "General Module"


# ---------------------------------------------------------------------------
# Difficulty scoring
# ---------------------------------------------------------------------------


def _difficulty_label(score: float) -> str:
    """Map a 0-100 difficulty score to a human-readable label."""
    if score >= 75:
        return "Critical"
    if score >= 50:
        return "Hard"
    if score >= 25:
        return "Medium"
    return "Easy"


def compute_difficulty_score(
    partitions: list[dict],
    *,
    complexity_weight: float = 0.3,
    coupling_weight: float = 0.25,
    churn_weight: float = 0.25,
    size_weight: float = 0.2,
) -> list[dict]:
    """Compute a composite difficulty score for each partition.

    Each metric is normalized to 0-100 relative to the max across all
    partitions before applying weights.  The result is a 0-100 score
    with a human-readable label.

    Parameters
    ----------
    partitions:
        List of partition dicts (must already contain ``complexity``,
        ``cross_partition_edges``, ``churn``, and ``symbol_count``).
    complexity_weight, coupling_weight, churn_weight, size_weight:
        Relative importance of each factor (should sum to 1.0).

    Returns
    -------
    The same list with ``difficulty_score`` (float 0-100) and
    ``difficulty_label`` (str) added to each dict.
    """
    if not partitions:
        return partitions

    # Collect raw values
    complexities = [p.get("complexity", 0) for p in partitions]
    cross_edges = [p.get("cross_partition_edges", 0) for p in partitions]
    churns = [p.get("churn", 0) for p in partitions]
    sizes = [p.get("symbol_count", 0) for p in partitions]

    max_complexity = max(complexities) or 1
    max_cross = max(cross_edges) or 1
    max_churn = max(churns) or 1
    max_size = max(sizes) or 1

    for i, p in enumerate(partitions):
        norm_complexity = (complexities[i] / max_complexity) * 100
        norm_cross = (cross_edges[i] / max_cross) * 100
        norm_churn = (churns[i] / max_churn) * 100
        norm_size = (sizes[i] / max_size) * 100

        score = (
            complexity_weight * norm_complexity
            + coupling_weight * norm_cross
            + churn_weight * norm_churn
            + size_weight * norm_size
        )

        p["difficulty_score"] = round(score, 1)
        p["difficulty_label"] = _difficulty_label(score)

    return partitions


# ---------------------------------------------------------------------------
# Partition analysis engine
# ---------------------------------------------------------------------------


def _accumulate_partition_meta(nodes, node_meta) -> tuple[set[str], Counter, float, set[str]]:
    """Single pass over a partition's nodes — collect file set, per-language
    counts, total complexity, and the lowercased names of test-file symbols."""
    files_set: set[str] = set()
    language_counter: Counter = Counter()
    total_complexity = 0.0
    test_names: set[str] = set()
    for n in nodes:
        meta = node_meta.get(n)
        if not meta:
            continue
        files_set.add(meta["path"])
        if meta["language"]:
            lang_label = _EXTENSION_ROLES.get(
                "." + (meta["language"] or "").lower().split("/")[-1],
                meta["language"],
            )
            language_counter[lang_label] += 1
        total_complexity += meta["complexity"]
        if meta["file_role"] == "test":
            test_names.add(meta["name"].lower())
    return files_set, language_counter, total_complexity, test_names


def _top_key_symbols(nodes, node_meta, limit: int = 5) -> list[dict]:
    """Top-N partition members by PageRank, formatted for the manifest."""
    ranked = sorted(
        [n for n in nodes if n in node_meta],
        key=lambda n: node_meta[n]["pagerank"],
        reverse=True,
    )
    out: list[dict] = []
    for n in ranked[:limit]:
        m = node_meta[n]
        out.append(
            {
                "name": m["name"],
                "kind": abbrev_kind(m["kind"]),
                # W-dogfood (W336 sibling family): round to 6 decimals.
                # On graphs >~5K symbols the per-node PR floor sits around
                # 1.4e-05, so 4-decimal rounding collapses ~72% of nonzero
                # values to 0.0 (12,833 of 17,710 on roam-code itself).
                # The displayed PR becomes useless for ranking introspection.
                # Matches the 6-decimal precedent in cmd_search / cmd_intent /
                # cmd_hover / cmd_metrics.
                "pagerank": round(m["pagerank"], 6),
                "file": m["path"],
            }
        )
    return out


def _partition_test_coverage(nodes, node_meta, test_names: set[str]) -> float:
    """Ratio of source symbols within the partition whose name matches a
    test symbol (or test_<name>) in the same partition."""
    total_source = sum(1 for n in nodes if n in node_meta and node_meta[n]["file_role"] != "test")
    if total_source == 0 or not test_names:
        return 0.0
    symbols_with_tests = 0
    for n in nodes:
        meta = node_meta.get(n)
        if not (meta and meta["file_role"] != "test"):
            continue
        lowered = meta["name"].lower()
        if lowered in test_names or f"test_{lowered}" in test_names:
            symbols_with_tests += 1
    return round(symbols_with_tests / total_source, 2)


def _partition_cross_edges(idx: int, nodes, node_part, G) -> int:
    """Count edges crossing this partition's boundary (in either direction)."""
    cross = 0
    for n in nodes:
        for succ in G.successors(n):
            if succ in node_part and node_part[succ] != idx:
                cross += 1
        for pred in G.predecessors(n):
            if pred in node_part and node_part[pred] != idx:
                cross += 1
    return cross


def _partition_label(idx: int, files_set: set[str]) -> str:
    """Cluster label from directory majority — falls back to partition-N or
    root-N when no directory information is available."""
    dirs = [os.path.dirname(f).replace("\\", "/") for f in files_set if f]
    if not dirs:
        return f"partition-{idx + 1}"
    label = Counter(dirs).most_common(1)[0][0]
    label = label.rstrip("/").rsplit("/", 1)[-1] if label else f"partition-{idx + 1}"
    if not label:
        return f"root-{idx + 1}"
    return label


def _build_partition_descriptor(
    idx: int,
    p: dict,
    node_meta: dict,
    node_part: dict,
    G,
    file_path_to_id: dict,
    cochange_map: dict,
    churn_by_file_id: dict,
) -> tuple[dict, set[str]]:
    """Build the per-partition manifest entry. Returns ``(descriptor, files_set)``.

    The descriptor matches the schema documented in
    ``compute_partition_manifest`` — keeping ``files_set`` separate so
    callers can still maintain the ``all_partition_files`` index for
    downstream conflict-hotspot analysis without iterating the descriptor.
    """
    nodes = p["nodes"]
    files_set, language_counter, total_complexity, test_names = _accumulate_partition_meta(nodes, node_meta)
    key_symbols = _top_key_symbols(nodes, node_meta)
    test_coverage = _partition_test_coverage(nodes, node_meta, test_names)
    cross_edges = _partition_cross_edges(idx, nodes, node_part, G)

    partition_file_ids = {file_path_to_id[f] for f in files_set if f in file_path_to_id}
    cochange_score = sum(
        count
        for (fid_a, fid_b), count in cochange_map.items()
        if fid_a in partition_file_ids and fid_b not in partition_file_ids
    )
    conflict_risk = _classify_conflict_risk(cross_edges, cochange_score)
    partition_churn = sum(churn_by_file_id.get(pfid, 0) for pfid in partition_file_ids)
    role = _suggest_role(sorted(files_set), language_counter)
    label = _partition_label(idx, files_set)

    descriptor = {
        "id": idx + 1,
        "label": label,
        "role": role,
        "files": sorted(files_set),
        "file_count": len(files_set),
        "symbol_count": len(nodes),
        "key_symbols": key_symbols,
        "complexity": round(total_complexity, 1),
        "churn": partition_churn,
        "test_coverage": test_coverage,
        "conflict_risk": conflict_risk,
        "cross_partition_edges": cross_edges,
        "cochange_score": cochange_score,
    }
    return descriptor, files_set


def _edge_sample_preserves_dependency_direction(u_meta: dict, v_meta: dict) -> str:
    """Format an edge sample without hiding source -> target direction."""
    edge_parts = [u_meta.get("path", "?"), v_meta.get("path", "?")]
    return " -> ".join(edge_parts)


def _dependencies_preserve_handoff_direction(
    G,
    node_part: dict[int, int],
    node_meta: dict[int, dict],
    all_partition_files: list[set[str]],
) -> list[dict]:
    """Build cross-partition dependencies while preserving handoff direction."""
    dep_counter: dict[tuple[int, int], list[str]] = defaultdict(list)
    for u, v in G.edges:
        pu = node_part.get(u)
        pv = node_part.get(v)
        if pu is None or pv is None or pu == pv:
            continue
        edge_desc = _edge_sample_preserves_dependency_direction(
            node_meta.get(u, {}),
            node_meta.get(v, {}),
        )
        dep_counter[(pu + 1, pv + 1)].append(edge_desc)

    dependencies = []
    for (from_id, to_id), edge_descs in sorted(dep_counter.items()):
        shared = all_partition_files[from_id - 1] & all_partition_files[to_id - 1]
        dependencies.append(
            {
                "from": from_id,
                "to": to_id,
                "edge_count": len(edge_descs),
                "sample_edges": edge_descs[:5],
                "shared_files": sorted(shared),
            }
        )
    return dependencies


def _derive_conflict_topology(
    G,
    node_part: dict[int, int],
    node_meta: dict[int, dict],
    all_partition_files: list[set[str]],
) -> tuple[list[dict], list[dict]]:
    """Derive cross-partition dependencies and shared-file conflict hotspots.

    Walking the symbol graph once per edge builds the dependency topology;
    a second pass over node -> partition membership surfaces files that
    straddle partition boundaries. Returning both together keeps the two
    conflict-relevant views co-located so callers do not re-derive them.
    """
    dependencies = _dependencies_preserve_handoff_direction(
        G,
        node_part,
        node_meta,
        all_partition_files,
    )

    # Files referenced by symbols in multiple partitions
    file_partitions: dict[str, set[int]] = defaultdict(set)
    for n, pidx in node_part.items():
        meta = node_meta.get(n)
        if meta:
            file_partitions[meta["path"]].add(pidx + 1)

    conflict_hotspots = []
    for fpath, part_ids in sorted(file_partitions.items()):
        if len(part_ids) >= 2:
            conflict_hotspots.append(
                {
                    "file": fpath,
                    "partition_count": len(part_ids),
                    "partitions": sorted(part_ids),
                }
            )
    conflict_hotspots.sort(key=lambda h: -h["partition_count"])
    conflict_hotspots = conflict_hotspots[:20]  # cap

    return dependencies, conflict_hotspots


def _load_node_meta(conn: sqlite3.Connection, node_ids: list[int]) -> dict[int, dict]:
    """Bulk-load symbol metadata needed by partition descriptors.

    Joins symbols, files, graph_metrics and symbol_metrics once for the
    whole partition set so downstream descriptor construction is O(1) per
    symbol lookup instead of hitting the DB per node.
    """
    node_meta: dict[int, dict] = {}
    if not node_ids:
        return node_meta
    rows = batched_in(
        conn,
        "SELECT s.id, s.name, s.kind, s.qualified_name, f.path, f.language, "
        "f.file_role, COALESCE(gm.pagerank, 0) AS pagerank, "
        "COALESCE(sm.cognitive_complexity, 0) AS complexity "
        "FROM symbols s "
        "JOIN files f ON s.file_id = f.id "
        "LEFT JOIN graph_metrics gm ON s.id = gm.symbol_id "
        "LEFT JOIN symbol_metrics sm ON s.id = sm.symbol_id "
        "WHERE s.id IN ({ph})",
        node_ids,
    )
    for r in rows:
        node_meta[r["id"]] = {
            "name": r["name"],
            "kind": r["kind"],
            "qualified_name": r["qualified_name"],
            "path": r["path"].replace("\\", "/"),
            "language": r["language"],
            "file_role": r["file_role"],
            "pagerank": r["pagerank"],
            "complexity": r["complexity"],
        }
    return node_meta


def _assign_agents_by_load(result_partitions: list[dict], n_agents: int) -> None:
    """Assign each partition to the least-loaded worker by complexity.

    Sorting partitions by descending complexity before round-robin
    assignment keeps the max agent load low (a greedy load-balancing
    heuristic). The ``agent`` key is written in-place on each partition.
    """
    sorted_parts = sorted(
        range(len(result_partitions)),
        key=lambda i: result_partitions[i]["complexity"],
        reverse=True,
    )
    agent_loads = [0.0] * n_agents
    for pi in sorted_parts:
        min_agent = min(range(n_agents), key=lambda a: agent_loads[a])
        result_partitions[pi]["agent"] = f"Worker-{min_agent + 1}"
        agent_loads[min_agent] += result_partitions[pi]["complexity"]


def compute_partition_manifest(
    conn: sqlite3.Connection,
    n_agents: int | None = None,
) -> dict:
    """Build a detailed partition manifest from the symbol graph.

    Parameters
    ----------
    conn:
        Open (readonly) connection to the roam index DB.
    n_agents:
        Number of agents.  ``None`` means auto-detect from cluster count.

    Returns
    -------
    dict with keys: partitions, dependencies, conflict_hotspots,
    overall_conflict_probability, verdict.
    """
    from roam.graph.builder import build_symbol_graph
    from roam.graph.clusters import detect_clusters
    from roam.graph.partition import (
        _adjust_cluster_count,
        compute_conflict_probability,
        compute_merge_order,
    )

    G = build_symbol_graph(conn)
    if len(G) == 0:
        return _empty_manifest(n_agents or 2)

    # -- 1. Detect communities -----------------------------------------------
    cluster_map = detect_clusters(G)
    if not cluster_map:
        cluster_map = {n: 0 for n in G.nodes}

    groups: dict[int, set[int]] = defaultdict(set)
    for node_id, cid in cluster_map.items():
        groups[cid].add(node_id)

    # Auto-detect agent count from natural cluster count
    if n_agents is None:
        n_agents = max(2, len(groups))

    partitions = _adjust_cluster_count(G, groups, n_agents)

    # -- 2. Gather node metadata in bulk -------------------------------------
    all_node_ids = list(set().union(*(p["nodes"] for p in partitions)))
    node_meta = _load_node_meta(conn, all_node_ids)

    # -- 3. Build node → partition index ---------------------------------------
    node_part: dict[int, int] = {}
    for idx, p in enumerate(partitions):
        for n in p["nodes"]:
            node_part[n] = idx

    # -- 4. Co-change data for conflict analysis -----------------------------
    # Build file_path → file_id mapping
    file_path_to_id: dict[str, int] = {}
    file_rows = conn.execute("SELECT id, path FROM files").fetchall()
    for fr in file_rows:
        file_path_to_id[fr["path"].replace("\\", "/")] = fr["id"]

    cochange_map: dict[tuple[int, int], int] = {}
    cochange_rows = conn.execute("SELECT file_id_a, file_id_b, cochange_count FROM git_cochange").fetchall()
    for cr in cochange_rows:
        cochange_map[(cr["file_id_a"], cr["file_id_b"])] = cr["cochange_count"]
        cochange_map[(cr["file_id_b"], cr["file_id_a"])] = cr["cochange_count"]

    # Pre-load total_churn for every file in one pass — the per-partition
    # loop below would otherwise do one ``SELECT total_churn FROM file_stats``
    # per file in each partition (an N+1 against file_stats).
    churn_by_file_id: dict[int, int] = {}
    for row in conn.execute("SELECT file_id, total_churn FROM file_stats").fetchall():
        if row["total_churn"]:
            churn_by_file_id[row["file_id"]] = row["total_churn"]

    # -- 5. Build per-partition descriptors ----------------------------------
    result_partitions: list[dict] = []
    all_partition_files: list[set[str]] = []
    for idx, p in enumerate(partitions):
        descriptor, files_set = _build_partition_descriptor(
            idx, p, node_meta, node_part, G, file_path_to_id, cochange_map, churn_by_file_id
        )
        result_partitions.append(descriptor)
        all_partition_files.append(files_set)

    # -- 6. Balance partitions by complexity → assign to agents ------------
    _assign_agents_by_load(result_partitions, n_agents)

    # -- 6b. Compute composite difficulty scores ---------------------------
    compute_difficulty_score(result_partitions)

    # -- 7. Cross-partition topology (dependencies + hotspots) -------------
    dependencies, conflict_hotspots = _derive_conflict_topology(G, node_part, node_meta, all_partition_files)

    # -- 8. Overall conflict probability -------------------------------------
    overall_cp = compute_conflict_probability(G, partitions)

    # -- 10. Merge order -----------------------------------------------------
    merge_order = compute_merge_order(G, partitions)

    # LAW 6 verdict-first + LAW 4 concrete-noun anchor: terminal token
    # must be a concrete noun. ``agents`` is in the anchor set; the prior
    # form terminated on ``9%`` which trips both the auto-derive humanizer
    # and the LAW 4 lint heuristics.
    verdict = (
        f"conflict probability {int(overall_cp * 100)}% across "
        f"{len(result_partitions)} partitions for {n_agents} agents"
    )

    return {
        "verdict": verdict,
        "total_partitions": len(result_partitions),
        "n_agents": n_agents,
        "overall_conflict_probability": round(overall_cp, 4),
        "merge_order": merge_order,
        "partitions": result_partitions,
        "dependencies": dependencies,
        "conflict_hotspots": conflict_hotspots,
    }


def _empty_manifest(n_agents: int) -> dict:
    """Return a valid but empty manifest."""
    return {
        "verdict": f"conflict probability 0% across 0 partitions for {n_agents} agents",
        "total_partitions": 0,
        "n_agents": n_agents,
        "overall_conflict_probability": 0.0,
        "merge_order": [],
        "partitions": [],
        "dependencies": [],
        "conflict_hotspots": [],
    }


def _partition_agent_contract(manifest: dict) -> dict:
    """Build a curated agent_contract for the partition envelope.

    The auto-derive humanizer (see :func:`roam.output.formatter._humanize_summary_fact`)
    produces garbage facts on the ``n_agents`` /
    ``overall_conflict_probability`` summary keys ("9092 n agents",
    "0.0971 overall conflict probability findings"). Override the block
    explicitly with LAW 4-anchored facts so agents reading the envelope
    get readable, concrete-noun-anchored signal.
    """
    pct = int(round(manifest["overall_conflict_probability"] * 100))
    n_parts = manifest["total_partitions"]
    n_agents = manifest["n_agents"]
    cp = manifest["overall_conflict_probability"]
    # LAW 4 anchored: each fact terminates on a concrete-noun token
    # (``agents`` / ``hotspots``) or is a measurement-suffix form
    # (``conflict score 0.0971`` matches the ``score`` suffix in the
    # auto-derive humaniser). Avoid bare ``"N partitions"`` because
    # ``partitions`` is not yet in the formatter's anchor set.
    facts = [
        manifest["verdict"],
        f"partitioned into {n_parts} clusters for {n_agents} agents",
        f"conflict score {cp:.4f}",
    ]
    n_hotspots = len(manifest.get("conflict_hotspots") or [])
    if n_hotspots:
        facts.append(f"{n_hotspots} conflict hotspots")
    next_commands = []
    if pct >= 25 or n_hotspots >= 5:
        next_commands.append("roam clusters")
    return {
        "facts": facts,
        "risks": [],
        "next_commands": next_commands,
        "confidence": None,
    }


# ---------------------------------------------------------------------------
# claude-teams format helpers
# ---------------------------------------------------------------------------


def _to_claude_teams(manifest: dict) -> dict:
    """Convert the manifest into a format optimized for Claude Agent Teams SDK.

    Produces a structure that can be directly fed into a multi-agent
    orchestrator:
      - agents: list of agent configs with scope and constraints
      - coordination: shared interfaces and merge strategy
    """
    agents = []
    for p in manifest["partitions"]:
        agents.append(
            {
                "agent_id": p.get("agent", f"Worker-{p['id']}"),
                "role": p["role"],
                "scope": {
                    "write_files": p["files"],
                    "read_only_deps": [],  # populated below
                },
                "constraints": {
                    "conflict_risk": p["conflict_risk"],
                    "estimated_complexity": p["complexity"],
                    "test_coverage": p["test_coverage"],
                },
            }
        )

    # Fill read-only deps from dependencies
    part_id_to_idx = {p["id"]: i for i, p in enumerate(manifest["partitions"])}
    for dep in manifest["dependencies"]:
        from_idx = part_id_to_idx.get(dep["from"])
        if from_idx is not None and from_idx < len(agents):
            # The "from" partition depends on files in the "to" partition
            to_idx = part_id_to_idx.get(dep["to"])
            if to_idx is not None and to_idx < len(manifest["partitions"]):
                to_files = manifest["partitions"][to_idx]["files"]
                existing = set(agents[from_idx]["scope"]["read_only_deps"])
                for f in to_files:
                    if f not in existing:
                        agents[from_idx]["scope"]["read_only_deps"].append(f)

    coordination = {
        "merge_order": manifest["merge_order"],
        "conflict_hotspots": [h["file"] for h in manifest["conflict_hotspots"]],
        "overall_conflict_probability": manifest["overall_conflict_probability"],
    }

    return {
        "agents": agents,
        "coordination": coordination,
    }


# ---------------------------------------------------------------------------
# CLI command
# ---------------------------------------------------------------------------


@roam_capability(
    name="partition",
    category="architecture",
    summary="Generate a multi-agent partition manifest with conflict analysis",
    maturity="stable",
    mcp_expose=True,
    mcp_preset=("core", "architecture"),
    side_effect=False,
    task_required=False,
    destructive=False,
    stale_sensitive=True,
    ai_safe=True,
    requires_index=True,
)
@click.command("partition")
@click.option(
    "--agents",
    "n_agents",
    type=int,
    default=None,
    help="Number of agents (default: auto-detect from cluster count)",
)
@click.option(
    "--format",
    "output_format",
    type=click.Choice(["plain", "json", "claude-teams"]),
    default="plain",
    help="Output format: plain (human readable), json, claude-teams",
)
@click.pass_context
def partition(ctx, n_agents, output_format):
    """Generate a multi-agent partition manifest with conflict analysis.

    Partitions the codebase into non-overlapping work zones using community
    detection, then enriches each partition with conflict probability,
    test coverage, estimated complexity, and a suggested agent role.

    Unlike ``orchestrate`` (which assigns scoped write zones and interface
    contracts for immediate multi-agent dispatch), this command provides
    deeper analytical metrics: difficulty scores, churn, co-change coupling,
    and the ``--format claude-teams`` output for SDK integration.

    \b
    Examples:
      roam partition --n-agents 4
      roam partition --n-agents 6 --format json
      roam partition --n-agents 3 --format claude-teams
      roam --json partition --n-agents 5

    See also ``orchestrate`` (operational dispatch with interface
    contracts), ``agent-plan`` (dependency-ordered phases), and
    ``fleet`` (graph-aware planner for external orchestrators).
    """
    json_mode = ctx.obj.get("json") if ctx.obj else False
    sarif_mode = ctx.obj.get("sarif") if ctx.obj else False
    ensure_index()

    # W607-DU -- substrate-boundary plumbing for cmd_partition.
    # ``_run_check_du`` wraps each substrate helper so an uncaught raise
    # in any one boundary degrades to a sensible empty-floor default
    # AND surfaces a marker in ``_w607du_warnings_out`` rather than
    # crashing the partition (multi-agent partition manifest) command
    # outright. cmd_partition is the "deeper analytical metrics +
    # claude-teams output" sibling to cmd_orchestrate -- both run
    # Louvain community detection over the symbol graph and assign
    # partitions to worker buckets, but partition layers on difficulty
    # scores, churn, co-change coupling, and the ``--format claude-teams``
    # output for SDK integration. A raise inside
    # ``compute_partition_manifest`` (DB reads, Louvain, worker
    # assignment, conflict scoring), ``_to_claude_teams`` (output
    # projection), or the downstream verdict / envelope composers used
    # to crash the partition command outright.
    # Marker family ``partition_<phase>_failed:<exc_class>:<detail>``.
    # Substrates wrapped:
    #
    #   * resolve_target_files               -- n_agents normalisation
    #   * build_dependency_graph             -- build_symbol_graph(conn)
    #   * compute_louvain_partitions         -- manifest construction
    #                                           (Louvain + metrics)
    #   * assign_workers                     -- partition -> worker bucket
    #                                           ("agent" key + difficulty)
    #   * extract_claude_teams_descriptor    -- _to_claude_teams projection
    #   * compose_verdict                    -- LAW 6 single-line floor
    #   * compose_facts                      -- agent_contract.facts list
    #   * compose_next_commands              -- agent_contract.next_commands
    #   * serialize_envelope                 -- JSON envelope emission
    #   * format_text_output                 -- text path partition printing
    #
    # W978 7-discipline applied: (1) f-string verdict floor uses literal
    # zero-count text -- no Name references, (2) default={...} carries
    # plain literals, (3) no json.dumps(default=str) needed (no
    # datetimes), (4) ``partition_*`` prefix is unique (collision-checked
    # by cross-prefix-discipline test), (5) len() at kwarg-bind is
    # gated by the manifest fallback, (6) len() / if x: on a poisoned
    # object only runs after the empty-floor guard, (7) no dict.get(key,
    # expensive_default) calls -- all defaults are immutable literals.
    _w607du_warnings_out: list[str] = []

    def _run_check_du(phase, fn, *args, default=None, **kwargs):
        """Run one substrate helper with W607-DU marker emission.

        On a clean call the result is returned as-is. On an uncaught
        exception, surface a ``partition_<phase>_failed:<exc_class>:<detail>``
        marker via ``_w607du_warnings_out`` and return *default* -- the
        envelope still emits cleanly with the remaining substrates.
        """
        try:
            return fn(*args, **kwargs)
        except Exception as exc:  # noqa: BLE001 -- top-level disclosure
            _w607du_warnings_out.append(f"partition_{phase}_failed:{type(exc).__name__}:{exc}")
            return default

    # W607-DU: ``resolve_target_files`` substrate -- normalise the
    # n_agents click option. cmd_partition has no --file/--staged
    # selector (whole-codebase only) so the substrate effectively
    # passes through the user input; an empty-floor default of 2
    # keeps the manifest call shape stable on a degraded path.
    def _resolve_target_files():
        if n_agents is None:
            return None
        return int(n_agents)

    resolved_n_agents = _run_check_du(
        "resolve_target_files",
        _resolve_target_files,
        default=2,
    )

    # W607-DU: empty-floor manifest reused by every degraded path so the
    # envelope still composes a coherent verdict. The literal zero counts
    # avoid the 7919-partition CATASTROPHE shape (CONSTRAINT 12 first-token
    # executability) -- "0 partitions for 0 agents" is the executable
    # empty-state, not "7919 partitions for 7919 agents" echoing the raw
    # user input.
    empty_manifest_floor: dict = {
        "verdict": "conflict probability 0% across 0 partitions for 0 agents",
        "total_partitions": 0,
        "n_agents": 0,
        "overall_conflict_probability": 0.0,
        "merge_order": [],
        "partitions": [],
        "dependencies": [],
        "conflict_hotspots": [],
    }

    with open_db(readonly=True) as conn:
        # W607-DU: ``build_dependency_graph`` substrate -- the DB ->
        # networkx graph build happens inside ``compute_partition_manifest``
        # but we ALSO probe ``build_symbol_graph`` directly so a raise
        # in the graph builder surfaces a distinct marker. The probe
        # result is discarded; the canonical engine call below re-runs
        # the build via the manifest engine and benefits from any graph
        # cache that the probe warmed.
        from roam.graph.builder import build_symbol_graph

        _run_check_du(
            "build_dependency_graph",
            build_symbol_graph,
            conn,
            default=None,
        )

        # W607-DU: ``compute_louvain_partitions`` substrate -- the
        # canonical Louvain community-detection + per-partition descriptor
        # engine. A raise here degrades to the empty-floor manifest so the
        # envelope still composes the LAW-6 floor.
        manifest = _run_check_du(
            "compute_louvain_partitions",
            compute_partition_manifest,
            conn,
            resolved_n_agents,
            default=empty_manifest_floor,
        )
        if manifest is None:
            manifest = empty_manifest_floor

    # W607-DU: ``assign_workers`` substrate -- the manifest already
    # carries ``agent: Worker-N`` and ``difficulty_score`` per partition
    # (assigned inside ``compute_partition_manifest`` via the
    # least-loaded round-robin + ``compute_difficulty_score``). This
    # substrate validates the assignment by checking every partition
    # descriptor has an ``agent`` key -- a malformed manifest with
    # missing assignments degrades to the empty floor so the text /
    # claude-teams paths don't KeyError downstream.
    def _assign_workers():
        partitions_list = manifest.get("partitions") or []
        for p in partitions_list:
            if "agent" not in p:
                raise KeyError(f"partition descriptor missing ``agent`` key: id={p.get('id')!r}")
        return partitions_list

    _run_check_du(
        "assign_workers",
        _assign_workers,
        default=[],
    )

    # -- SARIF format --------------------------------------------------------
    # W1159: SARIF projection for CI / GitHub Code Scanning integration.
    # Branches BEFORE json / claude-teams / text paths so those legacy
    # paths stay byte-identical to pre-W1159 output. The claude-teams
    # format is an external-SDK contract (not a CI report) so a
    # ``--sarif --format claude-teams`` combo still routes to SARIF;
    # consumers wanting the teams shape should drop ``--sarif``.
    if sarif_mode:
        from roam.output.sarif import partition_to_sarif, write_sarif

        sarif_summary: dict = {
            "verdict": manifest["verdict"],
            "total_partitions": manifest["total_partitions"],
            "n_agents": manifest["n_agents"],
            "overall_conflict_probability": manifest["overall_conflict_probability"],
            # W1298 Pattern-3a: each partition's ``complexity`` (and the
            # downstream ``estimated_complexity`` constraint) is a sum
            # of cognitive_complexity from symbol_metrics.
            "complexity_definition": COGNITIVE_COMPLEXITY_DEFINITION,
        }
        sarif_envelope_kwargs: dict = dict(
            summary=sarif_summary,
            agent_contract=_partition_agent_contract(manifest),
            partitions=manifest["partitions"],
            dependencies=manifest["dependencies"],
            conflict_hotspots=manifest["conflict_hotspots"],
            merge_order=manifest["merge_order"],
        )
        if _w607du_warnings_out:
            sarif_summary["partial_success"] = True
            sarif_summary["warnings_out"] = list(_w607du_warnings_out)
            sarif_envelope_kwargs["warnings_out"] = list(_w607du_warnings_out)
        envelope = json_envelope("partition", **sarif_envelope_kwargs)
        click.echo(write_sarif(partition_to_sarif(envelope)))
        return

    # -- claude-teams format -------------------------------------------------
    if output_format == "claude-teams":
        # W607-DU: ``extract_claude_teams_descriptor`` substrate -- the
        # output-format projection that converts the manifest into the
        # Claude Agent Teams SDK shape. A raise here degrades to the
        # canonical empty teams_data so the envelope still composes.
        empty_teams_floor: dict = {"agents": [], "coordination": {}}
        teams_data = _run_check_du(
            "extract_claude_teams_descriptor",
            _to_claude_teams,
            manifest,
            default=empty_teams_floor,
        )
        if teams_data is None:
            teams_data = empty_teams_floor

        if json_mode:
            teams_summary: dict = {
                "verdict": manifest["verdict"],
                "total_partitions": manifest["total_partitions"],
                "overall_conflict_probability": manifest["overall_conflict_probability"],
                # W1298 Pattern-3a: estimated_complexity in
                # teams constraints is a sum of cognitive_complexity.
                "complexity_definition": COGNITIVE_COMPLEXITY_DEFINITION,
            }
            teams_envelope_kwargs: dict = dict(
                summary=teams_summary,
                agent_contract=_partition_agent_contract(manifest),
                format="claude-teams",
            )
            teams_envelope_kwargs.update(teams_data)
            if _w607du_warnings_out:
                teams_summary["partial_success"] = True
                teams_summary["warnings_out"] = list(_w607du_warnings_out)
                teams_envelope_kwargs["warnings_out"] = list(_w607du_warnings_out)

            def _serialize_envelope_teams():
                click.echo(to_json(json_envelope("partition", **teams_envelope_kwargs)))

            _run_check_du("serialize_envelope", _serialize_envelope_teams, default=None)
        else:

            def _serialize_envelope_teams_plain():
                click.echo(to_json(teams_data))

            _run_check_du("serialize_envelope", _serialize_envelope_teams_plain, default=None)
        return

    # -- JSON format ---------------------------------------------------------
    if json_mode or output_format == "json":
        # W607-DU: ``compose_verdict`` substrate -- LAW 6 single-line
        # verdict + LAW 4 concrete-noun terminal ("agents"). A raise
        # inside the f-string degrades to a literal floor string with
        # explicit zero counts -- the W811/W817 Pattern-2 guard: never
        # collapse to a SAFE/passed verdict on the degraded path.
        # W978 #1: f-string verdict floor is plain text, no Name
        # references inside the literal.
        def _compose_verdict():
            mv = manifest.get("verdict")
            if isinstance(mv, str) and mv:
                return mv
            cp = manifest.get("overall_conflict_probability", 0.0)
            tp = manifest.get("total_partitions", 0)
            na = manifest.get("n_agents", 0)
            return f"conflict probability {int(cp * 100)}% across {tp} partitions for {na} agents"

        verdict = _run_check_du(
            "compose_verdict",
            _compose_verdict,
            default="conflict probability 0% across 0 partitions for 0 agents",
        )
        if not isinstance(verdict, str) or not verdict:
            verdict = "conflict probability 0% across 0 partitions for 0 agents"

        # W607-DU: ``compose_facts`` substrate -- curated
        # ``agent_contract.facts`` list. A raise degrades to a single
        # verdict-only fact so LAW 6 verdict-first invariant holds.
        def _compose_facts():
            ac = _partition_agent_contract(manifest)
            facts_local = ac.get("facts") or [verdict]
            return facts_local

        facts = _run_check_du(
            "compose_facts",
            _compose_facts,
            default=[verdict],
        )
        if facts is None:
            facts = [verdict]

        # W607-DU: ``compose_next_commands`` substrate -- the conditional
        # advisory next-step suggestions. A raise degrades to an empty
        # list so the agent_contract still composes.
        def _compose_next_commands():
            ac = _partition_agent_contract(manifest)
            return ac.get("next_commands") or []

        next_commands = _run_check_du(
            "compose_next_commands",
            _compose_next_commands,
            default=[],
        )
        if next_commands is None:
            next_commands = []

        # W607-DU: ``serialize_envelope`` substrate -- json_envelope
        # construction + click.echo emission. The wrap protects against
        # crashes inside the formatter call so the marker surfaces and
        # the function returns cleanly.
        envelope_summary: dict = {
            "verdict": verdict,
            "total_partitions": manifest["total_partitions"],
            "n_agents": manifest["n_agents"],
            "overall_conflict_probability": manifest["overall_conflict_probability"],
            # W1298 Pattern-3a: each partition's ``complexity``
            # (and the downstream agents[*].constraints
            # ``estimated_complexity``) is a sum of
            # cognitive_complexity from symbol_metrics.
            "complexity_definition": COGNITIVE_COMPLEXITY_DEFINITION,
        }
        envelope_kwargs: dict = dict(
            summary=envelope_summary,
            agent_contract={
                "facts": facts,
                "risks": [],
                "next_commands": next_commands,
                "confidence": None,
            },
            partitions=manifest["partitions"],
            dependencies=manifest["dependencies"],
            conflict_hotspots=manifest["conflict_hotspots"],
            merge_order=manifest["merge_order"],
        )
        # W607-DU: mirror substrate markers into BOTH the top-level
        # envelope ``warnings_out`` AND ``summary.warnings_out`` so MCP
        # consumers see disclosure regardless of which surface they
        # read. Flipping ``partial_success: True`` is the Pattern-2
        # silent-fallback guard.
        if _w607du_warnings_out:
            envelope_summary["partial_success"] = True
            envelope_summary["warnings_out"] = list(_w607du_warnings_out)
            envelope_kwargs["warnings_out"] = list(_w607du_warnings_out)

        def _serialize_envelope():
            click.echo(to_json(json_envelope("partition", **envelope_kwargs)))

        _run_check_du("serialize_envelope", _serialize_envelope, default=None)
        return

    # -- Plain text format ---------------------------------------------------
    # W607-DU: ``format_text_output`` substrate -- the human-readable
    # text emission path. A raise inside the loop (e.g. KeyError on a
    # malformed partition dict missing ``files`` / ``key_symbols`` /
    # ``conflict_risk``) degrades to a verdict-only emission so the user
    # still sees the LAW 6 floor.
    def _format_text_output():
        click.echo(f"VERDICT: {manifest['verdict']}")
        click.echo()

        for p in manifest["partitions"]:
            agent_label = p.get("agent", f"Worker-{p['id']}")
            click.echo(f'PARTITION {p["id"]} -- "{p["role"]}" (Agent: {agent_label})')
            diff_label = p.get("difficulty_label", "?")
            diff_score = p.get("difficulty_score", 0)
            click.echo(
                f"  Files: {p['file_count']} | "
                f"Symbols: {p['symbol_count']} | "
                f"Complexity: {p['complexity']} | "
                f"Churn: {p.get('churn', 0)} | "
                f"Difficulty: {diff_label} ({diff_score})"
            )
            click.echo(f"  Test coverage: {int(p['test_coverage'] * 100)}%")
            # Key files (top 3)
            if p["files"]:
                top_files = p["files"][:3]
                more_files = f" (+{len(p['files']) - 3} more)" if len(p["files"]) > 3 else ""
                click.echo(f"  Key files: {', '.join(top_files)}{more_files}")
            # Key symbols
            if p["key_symbols"]:
                key_symbols = ", ".join(
                    f"{s['kind']} {s['name']} (PageRank {s['pagerank']:.4f})" for s in p["key_symbols"][:3]
                )
                click.echo(f"  Key symbols: {key_symbols}")
            click.echo(f"  Conflict risk: {p['conflict_risk']} ({p['cross_partition_edges']} cross-partition edges)")
            click.echo()

        # Cross-partition dependencies
        if manifest["dependencies"]:
            click.echo("CROSS-PARTITION DEPENDENCIES:")
            for dep in manifest["dependencies"]:
                sample_edges = dep["sample_edges"]
                sample = f": {sample_edges[0]}{', ...' if len(sample_edges) > 1 else ''}" if sample_edges else ""
                click.echo(f"  Partition {dep['from']} -> Partition {dep['to']} ({dep['edge_count']} edges{sample})")
            click.echo()

        # Conflict hotspots
        if manifest["conflict_hotspots"]:
            click.echo("CONFLICT HOTSPOTS:")
            for h in manifest["conflict_hotspots"][:10]:
                click.echo(f"  {h['file']} -- referenced by {h['partition_count']} partitions (assign to Coordinator)")
            click.echo()

        # Merge order
        if manifest["merge_order"]:
            order_str = " -> ".join(f"Partition {pid}" for pid in manifest["merge_order"])
            click.echo(f"Merge order: {order_str}")

    _run_check_du("format_text_output", _format_text_output, default=None)
    # Marker accumulator handles disclosure on the text path -- the
    # warning rides into ``_w607du_warnings_out`` even when text-mode
    # output is human-targeted (JSON mode carries the structured
    # disclosure surface).
