"""Canonical ⇄ runtime sub-agent fan-out.

Phase 2 of the "memtomem as canonical context gateway" plan. A sub-agent lives
at ``.memtomem/agents/<name>.md`` with YAML frontmatter (Claude Code-compatible
superset) and a Markdown body that acts as the system prompt. From that single
canonical source we fan out to:

* ``.claude/agents/<name>.md`` — Claude Code (project-scope)
* ``.gemini/agents/<name>.md`` — Gemini CLI (project-scope)
* ``.codex/agents/<name>.toml`` — OpenAI Codex CLI (project-scope)
* ``.kimi/agents/<name>.yaml`` — Kimi CLI agent-file YAML (project-scope)

Codex CLI accepts both ``~/.codex/agents/`` (user-scope) and ``.codex/agents/``
(project-scope) per the official subagents docs. memtomem fans out to the
project-scope path so a single repository's `.memtomem/agents/` source tree
stays contained within the project — no host-home pollution, worktrees isolate
naturally, and the layout matches Claude / Gemini.

Unlike Phase 1 skills, sub-agents have genuine format divergence:

* Claude and Gemini share Markdown + YAML frontmatter but disagree on fields
  (Gemini has no ``isolation``/``skills``, Claude has no ``kind``/``temperature``).
* Codex uses a TOML schema (``name``, ``description``, ``developer_instructions``,
  ``model``, ...) — our Markdown body becomes ``developer_instructions``. Tools
  are dropped because Codex models capabilities through ``mcp_servers`` +
  ``skills.config`` rather than a flat tool list.

Every conversion reports its ``dropped`` fields so the user can see what was
lost. ``--strict`` promotes any drop to an error. Nested Claude fields
(``hooks``, ``codex.*`` overrides, full ``mcp_servers`` tables) are out of
scope for Phase 2 — the canonical frontmatter is intentionally flat.
"""

from __future__ import annotations

import logging
import json
import re
from dataclasses import dataclass, field, replace
from pathlib import Path
from typing import Any, Protocol, cast

from memtomem.context import _skip_reasons as skip_codes
from memtomem.context import override as _override
from memtomem.context._atomic import atomic_write_bytes
from memtomem.context._gate_a import GateABlocked, apply_gate_a
from memtomem.config import TargetScope
from memtomem.context._names import GENERATOR_VENDOR, InvalidNameError, Layout, validate_name
from memtomem.context._runtime_targets import (
    DiffRow,
    runtime_artifact_listing,
    runtime_artifact_names,
    runtime_fanout_root,
)
from memtomem.context._sync_atomic import (
    ON_DROP_LEVELS,
    AtomicSyncAdapter,
    AtomicSyncResult,
    StrictDropError as _EngineStrictDropError,
    sync_atomic_artifact,
)
from memtomem.context.scope_resolver import canonical_artifact_dir
from memtomem.context.versioning import make_label_resolver

logger = logging.getLogger(__name__)

CANONICAL_AGENT_ROOT = ".memtomem/agents"
AGENT_DIR_FILENAME = "agent.md"


def canonical_agent_name(path: Path, layout: Layout) -> str:
    """Single source of truth for agent path → name dispatch.

    Used by :func:`list_canonical_agents` consumers, the web routes import
    handler, and any other place that needs the canonical name without
    re-implementing the layout fallback. The brittle
    ``path.name == "agent.md"`` heuristic is intentionally avoided —
    callers must pass the layout tag they got from
    :func:`list_canonical_agents` or :func:`extract_agents_to_canonical`.
    """
    return path.parent.name if layout == "dir" else path.stem


# Reuse the same frontmatter regex used by the markdown chunker so canonical
# agent files parse consistently with the rest of memtomem.
_FRONT_MATTER_RE = re.compile(r"^---\n(.*?)\n---\n?", re.DOTALL)
_KEY_VALUE_RE = re.compile(r"^([A-Za-z_][\w-]*)\s*:\s*(.*)$")


# ── Canonical dataclass ──────────────────────────────────────────────


@dataclass
class SubAgent:
    """In-memory canonical representation of a sub-agent.

    Fields mirror the intersection/union of Claude Code and Gemini CLI
    sub-agent schemas; Codex-specific keys are derived at render time.
    """

    name: str
    description: str
    body: str  # system prompt (markdown)
    tools: list[str] = field(default_factory=list)
    model: str | None = None
    skills: list[str] = field(default_factory=list)
    isolation: str | None = None
    kind: str | None = None
    temperature: float | None = None


class AgentParseError(ValueError):
    """Raised when a canonical agent file cannot be parsed."""


# ── Minimal flat-YAML parser ─────────────────────────────────────────


def _strip_quotes(value: str) -> str:
    if len(value) >= 2 and value[0] == value[-1] and value[0] in ("'", '"'):
        return value[1:-1]
    return value


def _parse_flat_yaml(text: str) -> dict[str, Any]:
    """Parse a minimal flat YAML subset.

    Supported forms:

    * ``key: value`` (string / number / bool)
    * ``key: [a, b, c]`` (inline list)
    * ``key:`` followed by indented ``  - item`` lines (block list)

    Nested dicts, anchors, multi-doc separators, and other advanced YAML
    features are **not** supported — unsupported lines are silently skipped.
    That is intentional for Phase 2 so we don't take a pyyaml dependency.
    """
    result: dict[str, Any] = {}
    lines = text.splitlines()
    i = 0
    while i < len(lines):
        line = lines[i]
        stripped = line.strip()
        if not stripped or stripped.startswith("#"):
            i += 1
            continue
        m = _KEY_VALUE_RE.match(line)
        if not m:
            i += 1
            continue
        key, value = m.group(1), m.group(2).strip()

        if value == "":
            # Possibly a block list.
            block_items: list[str] = []
            j = i + 1
            while j < len(lines):
                nxt = lines[j]
                if nxt.strip().startswith("- "):
                    block_items.append(_strip_quotes(nxt.strip()[2:].strip()))
                    j += 1
                elif nxt.strip() == "":
                    j += 1
                    continue
                else:
                    break
            if block_items:
                result[key] = block_items
                i = j
                continue
            result[key] = None
            i += 1
            continue

        if value.startswith("[") and value.endswith("]"):
            inner = value[1:-1]
            items = [_strip_quotes(tok.strip()) for tok in inner.split(",") if tok.strip()]
            result[key] = items
            i += 1
            continue

        result[key] = _strip_quotes(value)
        i += 1
    return result


def _coerce_float(value: Any) -> float | None:
    if value is None or value == "":
        return None
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def _coerce_list(value: Any) -> list[str]:
    if value is None:
        return []
    if isinstance(value, list):
        return [str(v) for v in value if str(v).strip()]
    return [str(value).strip()] if str(value).strip() else []


_KNOWN_AGENT_KEYS = frozenset(
    {"name", "description", "tools", "model", "skills", "isolation", "kind", "temperature"}
)


def _parse_canonical_agent_text(
    content: str,
    *,
    source: Path,
    layout: Layout = "flat",
) -> SubAgent:
    """Parse already-loaded canonical agent text. Used by both the path-based
    :func:`parse_canonical_agent` (for backwards compat) and the sync flow
    that captures bytes once to close the scan→write TOCTOU window
    (PR-E3 Codex review fold).
    """
    # Normalize a leading UTF-8 BOM and CRLF → LF so ``_FRONT_MATTER_RE``
    # (which anchors at position 0 and on ``\n``) matches files authored on
    # Windows or by editors that emit a BOM/CRLF. Exactly one leading BOM is
    # stripped (mirroring the ``utf-8-sig`` decoder); a mid-file U+FEFF is a
    # legitimate zero-width no-break space and is left alone.
    content = content.removeprefix("﻿").replace("\r\n", "\n")
    m = _FRONT_MATTER_RE.match(content)
    if not m:
        raise AgentParseError(f"missing YAML frontmatter: {source}")
    frontmatter = _parse_flat_yaml(m.group(1))

    unknown = sorted(set(frontmatter) - _KNOWN_AGENT_KEYS)
    if unknown:
        logger.warning("unknown frontmatter keys %s in %s (ignored)", unknown, source)

    body = content[m.end() :].lstrip("\n").rstrip() + "\n"

    default_name = source.parent.name if layout == "dir" else source.stem
    name = frontmatter.get("name") or default_name
    try:
        name = validate_name(str(name), kind="agent name")
    except InvalidNameError as exc:
        raise AgentParseError(f"{exc} (source: {source})") from exc
    description = frontmatter.get("description") or ""
    return SubAgent(
        name=name,
        description=str(description),
        body=body,
        tools=_coerce_list(frontmatter.get("tools")),
        model=(str(frontmatter["model"]) if frontmatter.get("model") else None),
        skills=_coerce_list(frontmatter.get("skills")),
        isolation=(str(frontmatter["isolation"]) if frontmatter.get("isolation") else None),
        kind=(str(frontmatter["kind"]) if frontmatter.get("kind") else None),
        temperature=_coerce_float(frontmatter.get("temperature")),
    )


def parse_canonical_agent(path: Path, *, layout: Layout = "flat") -> SubAgent:
    """Parse a canonical agent file into a :class:`SubAgent`.

    ``layout`` selects the default-name fallback when the frontmatter omits
    ``name``: ``"flat"`` (legacy ``agents/<name>.md``) uses ``path.stem``;
    ``"dir"`` (ADR-0008 ``agents/<name>/agent.md``) uses
    ``path.parent.name``. Callers normally get ``layout`` from
    :func:`list_canonical_agents`.
    """
    content = path.read_text(encoding="utf-8")
    return _parse_canonical_agent_text(content, source=path, layout=layout)


def _resolve_agent_under_root(canonical_root: Path, agent_name: str) -> tuple[Path, Layout] | None:
    dir_target = canonical_root / agent_name / AGENT_DIR_FILENAME
    flat_target = canonical_root / f"{agent_name}.md"
    has_dir = dir_target.is_file()
    has_flat = flat_target.is_file()
    if has_dir and has_flat:
        logger.warning(
            "agents/%s: reverse-sync updates dir layout (%s/agent.md); the flat "
            "file (%s.md) is now silently divergent. Remove it or run "
            "`mm context migrate` (PR-D).",
            agent_name,
            agent_name,
            agent_name,
        )
        return dir_target, "dir"
    if has_dir:
        return dir_target, "dir"
    if has_flat:
        return flat_target, "flat"
    return None


def resolve_canonical_agent(
    project_root: Path, name: str, *, scope: TargetScope = "project_shared"
) -> tuple[Path, Layout] | None:
    """Return the canonical ``(path, layout)`` for ``name`` if it exists.

    Directory layout wins when both the legacy flat file and ADR-0008
    directory layout are present. Name validation is intentionally left to
    callers so existing 400 behavior remains at the route/CLI boundary.

    ``scope`` selects the canonical residency tier (ADR-0016). Default
    ``project_shared`` preserves pre-#940 behavior.
    """
    return _resolve_agent_under_root(canonical_artifact_dir("agents", scope, project_root), name)


def list_canonical_agents(
    project_root: Path,
    *,
    scope: TargetScope = "project_shared",
) -> list[tuple[Path, Layout]]:
    """Enumerate canonical agents in both flat and directory layouts.

    Flat layout (legacy): ``agents/<name>.md``. Directory layout (ADR-0008
    PR-C+): ``agents/<name>/agent.md``. When the same name has both forms,
    the directory layout wins and a WARNING is logged so the silent flat
    file is visible. ``mm context migrate`` (PR-D) is the supported way
    to consolidate.

    ADR-0011 PR-E3: ``scope`` selects the canonical root via
    :func:`canonical_artifact_dir` (default ``project_shared`` preserves
    pre-PR-E3 behavior of reading ``<project_root>/.memtomem/agents``).
    """
    root = canonical_artifact_dir("agents", scope, project_root)
    if not root.is_dir():
        return []

    flat: dict[str, Path] = {p.stem: p for p in sorted(root.glob("*.md")) if p.is_file()}
    dirs: dict[str, Path] = {}
    for entry in sorted(root.iterdir()):
        if entry.is_dir():
            agent_md = entry / AGENT_DIR_FILENAME
            if agent_md.is_file():
                dirs[entry.name] = agent_md

    for name in sorted(set(flat) & set(dirs)):
        logger.warning(
            "agents/%s: both flat (%s.md) and dir (%s/agent.md) layouts present; "
            "using dir. Remove the flat file or run `mm context migrate` (PR-D).",
            name,
            name,
            name,
        )

    merged_paths = {**flat, **dirs}  # dir overrides flat on collision
    layouts: dict[str, Layout] = {**dict.fromkeys(flat, "flat"), **dict.fromkeys(dirs, "dir")}
    return [(merged_paths[k], layouts[k]) for k in sorted(merged_paths)]


# ── Renderers ────────────────────────────────────────────────────────


def _yaml_inline_list(items: list[str]) -> str:
    return "[" + ", ".join(items) + "]"


def _render_markdown_agent(
    agent: SubAgent,
    include_fields: list[str],
) -> str:
    """Render an agent as Markdown + YAML frontmatter, emitting only the
    frontmatter keys listed in ``include_fields`` (in order)."""
    lines: list[str] = [f"name: {agent.name}", f"description: {agent.description}"]
    for key in include_fields:
        if key in ("name", "description"):
            continue
        if key == "tools" and agent.tools:
            lines.append(f"tools: {_yaml_inline_list(agent.tools)}")
        elif key == "model" and agent.model:
            lines.append(f"model: {agent.model}")
        elif key == "skills" and agent.skills:
            lines.append(f"skills: {_yaml_inline_list(agent.skills)}")
        elif key == "isolation" and agent.isolation:
            lines.append(f"isolation: {agent.isolation}")
        elif key == "kind" and agent.kind:
            lines.append(f"kind: {agent.kind}")
        elif key == "temperature" and agent.temperature is not None:
            lines.append(f"temperature: {agent.temperature}")
    frontmatter = "\n".join(lines)
    body = agent.body if agent.body.endswith("\n") else agent.body + "\n"
    return f"---\n{frontmatter}\n---\n\n{body}"


_CLAUDE_FIELDS = ["tools", "model", "skills", "isolation"]
_GEMINI_FIELDS = ["tools", "model", "kind", "temperature"]


def _subagent_to_claude_md(agent: SubAgent) -> tuple[str, list[str]]:
    dropped: list[str] = []
    if agent.kind is not None:
        dropped.append("kind")
    if agent.temperature is not None:
        dropped.append("temperature")
    return _render_markdown_agent(agent, _CLAUDE_FIELDS), dropped


def _subagent_to_gemini_md(agent: SubAgent) -> tuple[str, list[str]]:
    dropped: list[str] = []
    if agent.skills:
        dropped.append("skills")
    if agent.isolation is not None:
        dropped.append("isolation")
    return _render_markdown_agent(agent, _GEMINI_FIELDS), dropped


# ── TOML writer (hand-rolled, no pyyaml / tomli-w dependency) ────────


def _toml_escape_basic_string(s: str) -> str:
    """Escape ``s`` for a TOML basic (single-line, ``"``-delimited) string.

    TOML basic strings require ``\\b \\t \\n \\f \\r \\" \\\\`` for those
    characters and ``\\uXXXX`` for any other C0 control or DEL. Leaving raw
    control chars produces TOML that ``tomllib.loads`` rejects.
    """
    out: list[str] = []
    for ch in s:
        if ch == "\\":
            out.append("\\\\")
        elif ch == '"':
            out.append('\\"')
        elif ch == "\b":
            out.append("\\b")
        elif ch == "\t":
            out.append("\\t")
        elif ch == "\n":
            out.append("\\n")
        elif ch == "\f":
            out.append("\\f")
        elif ch == "\r":
            out.append("\\r")
        else:
            code = ord(ch)
            if code < 0x20 or code == 0x7F:
                out.append(f"\\u{code:04x}")
            else:
                out.append(ch)
    return "".join(out)


def _toml_escape_multiline_string(s: str) -> str:
    """Escape ``s`` for a TOML multi-line basic (``\"\"\"``-delimited) string.

    Literal newlines and tabs are permitted; ``\\r`` and other C0 controls
    still need escaping, and any stray ``\"\"\"`` must be broken up.
    """
    out: list[str] = []
    for ch in s:
        if ch == "\\":
            out.append("\\\\")
        elif ch == "\n" or ch == "\t":
            out.append(ch)
        elif ch == "\b":
            out.append("\\b")
        elif ch == "\f":
            out.append("\\f")
        elif ch == "\r":
            out.append("\\r")
        else:
            code = ord(ch)
            if code < 0x20 or code == 0x7F:
                out.append(f"\\u{code:04x}")
            else:
                out.append(ch)
    return "".join(out).replace('"""', '""\\"')


def _toml_scalar(value: Any) -> str:
    if isinstance(value, bool):
        return "true" if value else "false"
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, str):
        if "\n" in value:
            return f'"""\n{_toml_escape_multiline_string(value)}"""'
        return f'"{_toml_escape_basic_string(value)}"'
    raise TypeError(f"unsupported TOML scalar: {type(value).__name__}")


def _subagent_to_codex_toml(agent: SubAgent) -> tuple[str, list[str]]:
    dropped: list[str] = []
    if agent.tools:
        dropped.append("tools")
    if agent.skills:
        dropped.append("skills")
    if agent.isolation is not None:
        dropped.append("isolation")
    if agent.kind is not None:
        dropped.append("kind")
    if agent.temperature is not None:
        dropped.append("temperature")

    parts: list[str] = [
        f"name = {_toml_scalar(agent.name)}",
        f"description = {_toml_scalar(agent.description)}",
        f"developer_instructions = {_toml_scalar(agent.body.rstrip())}",
    ]
    if agent.model:
        parts.append(f"model = {_toml_scalar(agent.model)}")
    return "\n".join(parts) + "\n", dropped


def _yaml_json_scalar(value: str) -> str:
    """Return a YAML-compatible quoted scalar without adding a dependency."""
    return json.dumps(value, ensure_ascii=False)


def _yaml_block(value: str, indent: int) -> str:
    value = value.rstrip("\n")
    pad = " " * indent
    if not value:
        return f"{pad}\n"
    return "\n".join(f"{pad}{line}" if line else pad for line in value.split("\n")) + "\n"


def _subagent_to_kimi_yaml(agent: SubAgent) -> tuple[str, list[str]]:
    dropped: list[str] = []
    if agent.tools:
        dropped.append("tools")
    if agent.model:
        dropped.append("model")
    if agent.skills:
        dropped.append("skills")
    if agent.isolation is not None:
        dropped.append("isolation")
    if agent.kind is not None:
        dropped.append("kind")
    if agent.temperature is not None:
        dropped.append("temperature")

    role_lines: list[str] = []
    if agent.description:
        role_lines.append(f"Description: {agent.description}")
        role_lines.append("")
    role_lines.append(agent.body.rstrip())
    role_additional = "\n".join(role_lines).rstrip()
    return (
        "version: 1\n"
        "agent:\n"
        "  extend: default\n"
        f"  name: {_yaml_json_scalar(agent.name)}\n"
        "  system_prompt_args:\n"
        "    ROLE_ADDITIONAL: |-\n"
        f"{_yaml_block(role_additional, 6)}",
        dropped,
    )


# ── Generator registry ───────────────────────────────────────────────


class AgentGenerator(Protocol):
    """Protocol for runtime-specific sub-agent generators.

    ``target_file`` accepts an ADR-0011 ``scope`` keyword (default
    ``project_shared`` preserves pre-PR-E behavior). Returns ``None``
    when the (runtime, scope) tuple has no fan-out by design — see
    :data:`memtomem.context._runtime_targets.RUNTIME_FANOUT_TABLE`.
    Callers must handle ``None`` and emit
    ``skip_codes.NO_PROJECT_FANOUT_FOR_RUNTIME``.
    """

    name: str

    def target_file(
        self,
        project_root: Path,
        agent_name: str,
        *,
        scope: TargetScope = "project_shared",
    ) -> Path | None:
        """Return the file that should hold the rendered agent (or ``None``)."""
        ...

    def render(self, agent: SubAgent) -> tuple[str, list[str]]:
        """Return ``(file_content, dropped_field_names)``."""
        ...


AGENT_GENERATORS: dict[str, AgentGenerator] = {}


def _register(gen: AgentGenerator) -> AgentGenerator:
    AGENT_GENERATORS[gen.name] = gen
    return gen


@dataclass
class ClaudeAgentsGenerator:
    name: str = "claude_agents"
    output_root: str = ".claude/agents"

    def target_file(
        self,
        project_root: Path,
        agent_name: str,
        *,
        scope: TargetScope = "project_shared",
    ) -> Path | None:
        root = runtime_fanout_root("agents", "claude", scope, project_root)
        return None if root is None else root / f"{agent_name}.md"

    def render(self, agent: SubAgent) -> tuple[str, list[str]]:
        return _subagent_to_claude_md(agent)


@dataclass
class GeminiAgentsGenerator:
    name: str = "gemini_agents"
    output_root: str = ".gemini/agents"

    def target_file(
        self,
        project_root: Path,
        agent_name: str,
        *,
        scope: TargetScope = "project_shared",
    ) -> Path | None:
        root = runtime_fanout_root("agents", "gemini", scope, project_root)
        return None if root is None else root / f"{agent_name}.md"

    def render(self, agent: SubAgent) -> tuple[str, list[str]]:
        return _subagent_to_gemini_md(agent)


@dataclass
class CodexAgentsGenerator:
    name: str = "codex_agents"
    output_root: str = ".codex/agents"

    def target_file(
        self,
        project_root: Path,
        agent_name: str,
        *,
        scope: TargetScope = "project_shared",
    ) -> Path | None:
        root = runtime_fanout_root("agents", "codex", scope, project_root)
        return None if root is None else root / f"{agent_name}.toml"

    def render(self, agent: SubAgent) -> tuple[str, list[str]]:
        return _subagent_to_codex_toml(agent)


@dataclass
class KimiAgentsGenerator:
    name: str = "kimi_agents"
    output_root: str = ".kimi/agents"

    def target_file(
        self,
        project_root: Path,
        agent_name: str,
        *,
        scope: TargetScope = "project_shared",
    ) -> Path | None:
        root = runtime_fanout_root("agents", "kimi", scope, project_root)
        return None if root is None else root / f"{agent_name}.yaml"

    def render(self, agent: SubAgent) -> tuple[str, list[str]]:
        return _subagent_to_kimi_yaml(agent)


_register(ClaudeAgentsGenerator())
_register(GeminiAgentsGenerator())
_register(CodexAgentsGenerator())
_register(KimiAgentsGenerator())


# ── Fan-out: canonical → runtimes ───────────────────────────────────


# Sister subclass (issue #900): inherits the engine's result shape but
# stays a distinct class so ``AgentSyncResult is CommandSyncResult`` is
# False and ``type(result).__name__ == "AgentSyncResult"`` matches the
# pre-extraction API. External callers consume it by attribute access
# (``.generated``, ``.dropped``, ``.skipped``) — those attributes come
# from the parent dataclass.
@dataclass
class AgentSyncResult(AtomicSyncResult):
    """Module-specific result subclass — see :class:`AtomicSyncResult`."""


# Sister subclass for the same reason — ``agents.StrictDropError is
# commands.StrictDropError`` stays False so an ``except StrictDropError``
# block imported from ``memtomem.context.agents`` only catches the
# agents-flavored raise.
class StrictDropError(_EngineStrictDropError):
    """Module-specific strict-drop error — see :class:`_EngineStrictDropError`."""


@dataclass
class ExtractResult:
    """Result of a reverse (runtime → canonical) import.

    Each entry in ``imported`` is ``(path, layout)`` so consumers can use
    :func:`canonical_agent_name` without re-deriving the layout from the
    path. ``layout`` is whichever form the canonical now lives in on disk
    (preserving an existing flat file or writing a new dir entry — see
    :func:`_resolve_agent_extract_target`).
    """

    imported: list[tuple[Path, Layout]]
    # (item_name, human_reason, reason_code) — see :mod:`memtomem.context._skip_reasons`.
    skipped: list[tuple[str, str, skip_codes.SkipCode]] = field(default_factory=list)


# Issue #900 extraction: the previously per-module ``generate_all_agents``
# body is now :func:`memtomem.context._sync_atomic.sync_atomic_artifact`.
# The adapter below bundles the agent-specific callables the engine plugs
# into. Behavior is byte-for-byte identical (TOCTOU close on canonical and
# override bytes, Phase 1 raise-early atomicity for ``project_shared``,
# Phase 2 strict-drop partial-write boundary pinned by #908).
_AGENT_ADAPTER: AtomicSyncAdapter[SubAgent] = AtomicSyncAdapter(
    kind="agent",
    artifact_label="agents",
    list_canonical=list_canonical_agents,
    parse_canonical_text=_parse_canonical_agent_text,
    parse_error_type=AgentParseError,
    name_of=lambda a: a.name,
    generators=AGENT_GENERATORS,
    result_type=AgentSyncResult,
    strict_drop_error_type=StrictDropError,
    logger=logger,
)


def generate_all_agents(
    project_root: Path,
    runtimes: list[str] | None = None,
    strict: bool = False,
    on_drop: str = "ignore",
    *,
    scope: TargetScope = "project_shared",
    label: str | None = None,
    surface: str = "cli_context_sync",
) -> AgentSyncResult:
    """Fan out every canonical sub-agent to the requested runtimes.

    Thin wrapper that binds the agent-specific adapter and delegates to
    :func:`memtomem.context._sync_atomic.sync_atomic_artifact` — see that
    function for the full Phase 1 / Phase 2 contract.

    Args:
        on_drop: Severity when fields are dropped during conversion.
            ``"ignore"`` (default) — silently record in ``result.dropped``.
            ``"warn"``  — log a warning per dropped-field set.
            ``"error"`` — raise :class:`StrictDropError` immediately.
        strict: Legacy alias for ``on_drop="error"``. If *both* are supplied,
            ``on_drop`` takes precedence unless it is still the default.
        scope: ADR-0011 PR-E3 — selects canonical root and runtime
            fan-out destination. Default ``project_shared`` preserves
            pre-PR-E3 behavior.
        label: ADR-0022 — when set to a non-``latest`` value, fan out the
            version pointed at by this label (or a bare ``vN`` tag) instead of
            the working ``agent.md``. ``None`` / ``"latest"`` preserve today's
            behavior byte-for-byte (the working file). Per-artifact resolution
            failures isolate as skips; they never abort the whole fan-out.
        surface: Gate A audit identifier, forwarded verbatim to the
            engine's privacy scans. The CLI relies on the default
            ``"cli_context_sync"``; the Web sync route passes
            ``"web_context_agents_sync"`` and the MCP tools pass
            ``"mcp_context_generate"`` / ``"mcp_context_sync"`` (#1246).
    """
    adapter = _AGENT_ADAPTER
    if label is not None and label != "latest":
        adapter = replace(_AGENT_ADAPTER, resolve_canonical_bytes=make_label_resolver(label))
    # Runtime type matches AgentSyncResult because _AGENT_ADAPTER's
    # ``result_type`` is AgentSyncResult; mypy can't infer that through the
    # engine's AtomicSyncResult signature, so cast to preserve the public
    # type annotation.
    return cast(
        "AgentSyncResult",
        sync_atomic_artifact(
            adapter,
            project_root,
            runtimes,
            strict=strict,
            on_drop=on_drop,
            scope=scope,
            surface=surface,
        ),
    )


# ── Reverse: runtime → canonical ────────────────────────────────────


def _resolve_agent_extract_target(canonical_root: Path, agent_name: str) -> tuple[Path, Layout]:
    """Decide where reverse-sync writes the canonical for ``agent_name``.

    Truth table (ADR-0008 PR-C):
      dir+flat both → dir wins, flat is silently divergent → WARN
      dir only      → dir
      flat only     → flat (preserve existing layout; PR-C does not migrate)
      neither       → dir (ADR-0008 layout for new agents)
    """
    resolved = _resolve_agent_under_root(canonical_root, agent_name)
    if resolved is not None:
        return resolved
    return canonical_root / agent_name / AGENT_DIR_FILENAME, "dir"


def extract_agents_to_canonical(
    project_root: Path,
    overwrite: bool = False,
    only_name: str | None = None,
    *,
    scope: TargetScope = "project_shared",
    force_unsafe_import: bool = False,
    dry_run: bool = False,
    surface: str = "cli_context_init",
) -> ExtractResult:
    """Import existing Claude / Gemini agent files into ``.memtomem/agents/``.

    ``dry_run`` (rank-10 import preview) runs the full scan + validation +
    Gate A re-scan + cross-runtime dedup + canonical-exists check, then
    **skips only the write**: ``imported`` lists the destinations that *would*
    be written and ``skipped`` carries the same reasons a real run would, with
    nothing touching disk.

    Codex TOML is **not** imported (one-way conversion; too lossy to round-trip
    without reconstructing fields we dropped on the way out). Kimi agent YAML
    is likewise export-only: the renderer drops tools/model/skills/isolation/
    kind/temperature and wraps description+body into
    ``system_prompt_args.ROLE_ADDITIONAL`` — not round-trippable. First
    occurrence wins across runtimes (Claude before Gemini — deterministic
    order).

    ADR-0011 PR-E2: ``scope`` selects both the canonical destination and the
    runtime source (per-scope import — ``scope="user"`` reads
    ``~/.claude/agents`` etc. via :func:`runtime_fanout_root` and writes into
    ``~/.memtomem/agents/``). ``project_local`` has no runtime fan-out by
    design (ADR §3) and short-circuits to an empty result.

    Gate A (``privacy.enforce_write_guard``) re-scans every source file's
    bytes before write. ``project_shared`` destinations hard-abort via
    :class:`click.ClickException` on any blocked content (with or without
    ``force_unsafe_import``); ``user`` / ``project_local`` destinations
    skip-and-warn unless ``force_unsafe_import=True`` flips the decision
    to ``"bypassed"`` and writes raw bytes through.

    Returns an :class:`ExtractResult` with both imported paths and skipped
    items so the caller can warn the user about silent deduplication.

    When ``only_name`` is set, every runtime file with a different stem is
    silently skipped before any validation/dedupe work. Callers (e.g. the
    single-item import route) can detect "no such runtime artifact" by
    inspecting an empty ``imported`` + ``skipped``.

    Layout policy: new agents (no existing canonical) land in directory
    layout per ADR-0008. Existing flat-layout entries are preserved by
    PR-C — migration to directory layout is a separate command (PR-D).
    """
    if scope == "project_local":
        # ADR §3 — gitignored draft tier has no runtime fan-out, so there
        # is nothing to import. Loud-emit (NO_PROJECT_FANOUT_FOR_RUNTIME)
        # rather than returning silently empty so callers/tests can pin
        # the explicit reason.
        return ExtractResult(
            imported=[],
            skipped=[
                (
                    "<all>",
                    "project_local has no runtime fan-out (ADR-0011 §3)",
                    skip_codes.NO_PROJECT_FANOUT_FOR_RUNTIME,
                )
            ],
        )

    canonical_root = canonical_artifact_dir("agents", scope, project_root)
    imported: list[tuple[Path, Layout]] = []
    skipped: list[tuple[str, str, skip_codes.SkipCode]] = []
    seen: dict[str, str] = {}  # agent_name → first runtime label

    for runtime in ("claude", "gemini"):
        try:
            runtime_dir = runtime_fanout_root("agents", runtime, scope, project_root)
        except KeyError:
            # Defensive — every (agents, claude|gemini, scope) tuple is in
            # the table at PR-E1 land time, so this only fires on a future
            # runtime added without table churn.
            continue
        if runtime_dir is None or not runtime_dir.is_dir():
            continue
        runtime_label = f"{runtime} ({runtime_dir})"
        for md_file in sorted(runtime_dir.glob("*.md")):
            agent_name = md_file.stem
            if only_name is not None and agent_name != only_name:
                continue
            try:
                validate_name(agent_name, kind="agent name")
            except InvalidNameError as exc:
                skipped.append((agent_name, f"invalid name: {exc}", skip_codes.INVALID_NAME))
                logger.warning(
                    "skip %r from %s: invalid name",
                    agent_name,
                    runtime_label,
                )
                continue
            if agent_name in seen:
                reason = f"already imported from {seen[agent_name]}"
                skipped.append((agent_name, reason, skip_codes.ALREADY_IMPORTED))
                logger.warning("skip %s from %s: %s", agent_name, runtime_label, reason)
                continue
            dst, layout = _resolve_agent_extract_target(canonical_root, agent_name)
            if dst.exists() and not overwrite:
                reason = "canonical exists (use --overwrite)"
                skipped.append((agent_name, reason, skip_codes.CANONICAL_EXISTS))
                logger.warning("skip %s from %s: %s", agent_name, runtime_label, reason)
                seen[agent_name] = runtime_label
                continue

            # Gate A — re-scan the source bytes before any write. Use
            # ``errors="replace"`` so non-UTF8 bytes cannot mask an
            # embedded ASCII secret (the replacement char `�` does
            # not overlap with any provider-token alphanumeric pattern).
            # Project_shared hard-abort (git-history-is-forever) is
            # raised inside ``apply_gate_a`` — Files imported earlier in
            # this run that passed Gate A stay (each was scanned
            # independently).
            try:
                content_bytes = md_file.read_bytes()
            except OSError as exc:
                skipped.append((agent_name, f"unreadable: {exc}", skip_codes.PARSE_ERROR))
                continue
            content_text = content_bytes.decode("utf-8", errors="replace")
            outcome = apply_gate_a(
                content_text=content_text,
                src=md_file,
                scope=scope,
                force_unsafe_import=force_unsafe_import,
                surface=surface,
                audit_context={
                    "source": str(md_file),
                    "target": str(dst),
                    "kind": "agents",
                    "agent_name": agent_name,
                },
                message_kind="agent",
                imported_so_far=len(imported),
            )
            if isinstance(outcome, GateABlocked):
                skipped.append(
                    (
                        agent_name,
                        f"blocked: {outcome.hits_count} privacy pattern hit(s){outcome.hint}",
                        outcome.code,
                    )
                )
                seen[agent_name] = runtime_label
                continue
            # outcome is GateAProceed — write. ``dry_run`` records the
            # would-import target but skips the write so the preview never
            # mutates disk (rank-10).
            if not dry_run:
                dst.parent.mkdir(parents=True, exist_ok=True)
                atomic_write_bytes(dst, content_bytes)
            imported.append((dst, layout))
            seen[agent_name] = runtime_label

    return ExtractResult(imported=imported, skipped=skipped)


# ── Diff: canonical ↔ runtimes ──────────────────────────────────────


# Per-runtime file suffix for agents fan-out. Used by ``diff_agents`` to
# delegate to ``runtime_artifact_names``. Kept inline (rather than on each
# generator) because (a) the suffix is a property of the on-disk format, not
# of the runtime conversion, and (b) it's referenced from one site only.
_AGENT_RUNTIME_SUFFIX: dict[str, str] = {
    "claude": ".md",
    "gemini": ".md",
    "codex": ".toml",
    "kimi": ".yaml",
}


def _runtime_agent_names(gen_name: str, project_root: Path) -> set[str]:
    """Thin wrapper around :func:`runtime_artifact_names` for back-compat.

    Pre-PR-E3 callers and tests pass ``gen_name`` (e.g. ``"claude_agents"``)
    and an implicit ``project_shared`` scope. New callers should consume
    :func:`runtime_artifact_names` directly with explicit scope.
    """
    runtime = gen_name.split("_", 1)[0]
    suffix = _AGENT_RUNTIME_SUFFIX.get(runtime)
    if suffix is None:
        return set()
    return runtime_artifact_names(
        "agents", runtime, project_root, "project_shared", file_suffix=suffix
    )


def diff_agents(
    project_root: Path,
    *,
    scope: TargetScope = "project_shared",
) -> list[tuple[str, str, str]]:
    """Compare canonical agents against every registered runtime.

    Returns a list of ``(runtime, agent_name, status)`` where status is one of
    ``"in sync"``, ``"out of sync"``, ``"missing target"``, ``"missing canonical"``,
    ``"parse error"``, ``"invalid name"``.

    ADR-0011 PR-E3: ``scope`` selects both the canonical source and the
    runtime fan-out roots. Default ``project_shared`` preserves
    pre-PR-E3 behavior.
    """
    results: list[tuple[str, str, str]] = []
    # Key canonicals by the parsed frontmatter ``name:`` — the identity sync
    # targets (``_sync_atomic`` Phase 1: bytes → decode errors="replace" →
    # parse → ``name_of``). Keying by file stem reported permanent phantom
    # drift after a successful sync whenever stem and frontmatter name
    # disagree: ('<stem>', 'missing target') + ('<name>', 'missing
    # canonical') on every runtime, forever (#1229). The lenient decode also
    # keeps a stray non-UTF-8 byte from aborting the whole diff with an
    # uncaught UnicodeDecodeError while sync handles the same file fine.
    # Unreadable / unparseable canonicals keep the stem / dir-name key
    # (``None`` value) so their "parse error" row still names the file.
    canonical_index: dict[str, SubAgent | None] = {}
    # Parse failures keep the exception text (it embeds the source path) so
    # the "parse error" row can carry a diagnostic reason — pre-#1229 the
    # exception was swallowed here without even a log line.
    parse_failures: dict[str, str] = {}
    for path, layout in list_canonical_agents(project_root, scope=scope):
        fallback_name = path.parent.name if layout == "dir" else path.stem
        try:
            text = path.read_bytes().decode("utf-8", errors="replace")
            parsed = _parse_canonical_agent_text(text, source=path, layout=layout)
        except OSError as exc:
            # Same first-parsed-wins precedence as below (#1247 Codex impl
            # round): a fallback-name entry must not SHADOW an earlier
            # successfully parsed canonical that claimed the same effective
            # name — sync writes that name fine (the unreadable file never
            # claims a name there), so a None overwrite here would report
            # permanent phantom "parse error" drift no sync can clear. The
            # warning below still fires, so the broken file stays loud.
            if canonical_index.get(fallback_name) is None:
                canonical_index[fallback_name] = None
                parse_failures[fallback_name] = f"unreadable: {exc}"
            logger.warning("canonical agent %s unreadable: %s", path, exc)
        except AgentParseError as exc:
            if canonical_index.get(fallback_name) is None:
                canonical_index[fallback_name] = None
                parse_failures[fallback_name] = str(exc)
            logger.warning("canonical agent %s failed to parse: %s", path, exc)
        else:
            # First-parsed-wins on a frontmatter-name collision, matching the
            # sync engine's duplicate-name dedupe (#1247) — last-wins here
            # would diff the runtime file against the canonical sync never
            # wrote, reporting permanent phantom drift. The loser surfaces in
            # the sync result as a ``duplicate_name`` skip, not as a diff row
            # (flat-vs-dir precedent: log-only in diff). A ``None`` entry
            # (parse-failure fallback name) stays overwritable — pre-existing
            # interplay, unchanged.
            if canonical_index.get(parsed.name) is not None:
                logger.warning(
                    "duplicate agent name %r: keeping first-seen canonical, ignoring %s",
                    parsed.name,
                    path,
                )
            else:
                canonical_index[parsed.name] = parsed
    canonical_names = set(canonical_index)

    for gen_name, gen in AGENT_GENERATORS.items():
        # ADR-0011 PR-E3 cleanup item #1: query the table directly via
        # ``runtime_fanout_root``. Earlier code probed with a fixed agent
        # name (``__probe_891__``) which leaked the table-shape assumption
        # into the call shape — call-shape fragility, not name-independence.
        runtime = gen_name.split("_", 1)[0]
        if runtime_fanout_root("agents", runtime, scope, project_root) is None:
            continue
        suffix = _AGENT_RUNTIME_SUFFIX.get(runtime, ".md")
        runtime_names, invalid_runtime_names = runtime_artifact_listing(
            "agents", runtime, project_root, scope, file_suffix=suffix
        )
        # Invalid-named runtime files used to vanish from diff entirely
        # (log-only) — the dashboard read fully in-sync while an unmanaged
        # runtime artifact existed (#1229). Surface them as a dedicated row;
        # a canonical "parse error" row of the same fallback name wins
        # (only possible when the canonical stem itself is invalid).
        for raw_name, invalid_reason in invalid_runtime_names:
            if raw_name not in canonical_names:
                results.append(DiffRow(gen_name, raw_name, "invalid name", invalid_reason))
        for name in sorted(canonical_names | runtime_names):
            # Unparseable canonical (including an invalid *effective* name —
            # frontmatter name, or stem fallback) checked BEFORE the
            # missing-target branch: "missing target" implied sync would
            # create the runtime file, which it never can — the row showed
            # a permanent drift no sync clears (#1229).
            if name in canonical_names and canonical_index[name] is None:
                results.append(DiffRow(gen_name, name, "parse error", parse_failures.get(name)))
                continue
            if name in canonical_names and name not in runtime_names:
                results.append((gen_name, name, "missing target"))
                continue
            if name in runtime_names and name not in canonical_names:
                results.append((gen_name, name, "missing canonical"))
                continue

            agent = canonical_index[name]
            assert agent is not None  # consumed by the parse-error branch above
            # Cleanup item #2: the upstream ``runtime_fanout_root`` guard
            # above guarantees this runtime+scope has a fan-out root, so
            # ``gen.target_file`` cannot return ``None`` for any name.
            # Earlier defensive ``if target is None: continue`` removed.
            target = gen.target_file(project_root, name, scope=scope)
            assert target is not None  # narrowed by upstream NO_FANOUT guard

            # A per-vendor override replaces the rendered runtime file at sync
            # time (``_sync_atomic`` Phase 2 / ADR-0008 Invariant 4) via
            # ``atomic_write_bytes`` — raw bytes, no strip. Compare byte-exact
            # against the same source so an override-carrying agent isn't
            # reported permanently "out of sync"; decoding as text would crash
            # on a non-UTF-8 override and would mask whitespace-only byte drift.
            vendor = GENERATOR_VENDOR.get(gen_name)
            override_path = (
                _override.resolve(project_root, "agents", name, vendor, scope=scope)
                if vendor is not None
                else None
            )
            if override_path is not None:
                try:
                    expected_bytes = override_path.read_bytes()
                except OSError:
                    # Sync skips an unreadable override (no effective fan-out),
                    # so we can't assert parity — report drift, never mask it.
                    results.append((gen_name, name, "out of sync"))
                    continue
                try:
                    actual_bytes = target.read_bytes() if target.is_file() else b""
                except OSError:
                    # Unreadable runtime file — same contract as above.
                    results.append((gen_name, name, "out of sync"))
                    continue
                status = "in sync" if expected_bytes == actual_bytes else "out of sync"
                results.append((gen_name, name, status))
                continue

            expected, _ = gen.render(agent)
            # Byte-exact compare against what sync would write (Phase 2
            # ``atomic_write_bytes`` of ``content.encode("utf-8")`` verbatim)
            # — a ``.strip()`` compare reported whitespace-padded runtime
            # files "in sync" while sync would rewrite them (#1229), and a
            # lenient text decode collapses distinct invalid byte sequences
            # to the same U+FFFD. Bytes cannot raise UnicodeDecodeError. An
            # unreadable runtime file can't assert parity — report drift,
            # never mask it.
            try:
                actual_bytes = target.read_bytes() if target.is_file() else b""
            except OSError:
                results.append((gen_name, name, "out of sync"))
                continue
            status = "in sync" if expected.encode("utf-8") == actual_bytes else "out of sync"
            results.append((gen_name, name, status))

    return results


__all__ = [
    "AGENT_DIR_FILENAME",
    "AGENT_GENERATORS",
    "AgentGenerator",
    "AgentParseError",
    "AgentSyncResult",
    "CANONICAL_AGENT_ROOT",
    "ExtractResult",
    "ClaudeAgentsGenerator",
    "CodexAgentsGenerator",
    "GeminiAgentsGenerator",
    "KimiAgentsGenerator",
    "Layout",
    "ON_DROP_LEVELS",
    "StrictDropError",
    "SubAgent",
    "canonical_agent_name",
    "diff_agents",
    "extract_agents_to_canonical",
    "generate_all_agents",
    "list_canonical_agents",
    "parse_canonical_agent",
    "resolve_canonical_agent",
]
