"""OCTAVE schema validator (P1.5).

Validates AST against schema definitions with:
- Required field checking
- Type validation
- Enum validation (with prefix matching and ambiguity detection)
- Regex pattern validation
- Unknown field detection
- Constraint chain evaluation
- Target routing with audit trail (Issue #103)
- Target validation with registry (Issue #188)
- Policy enforcement (Issue #190): UNKNOWN_FIELDS policy, custom targets

ADR-0006 SR1-T1 Step 6 (R2 closure)
-----------------------------------
This module exposes EXACTLY ONE public validation surface: the class
``Validator`` (which is a structural ``visitor.Visitor[None]``). The
historical module-level ``validate()`` function and the thin delegator
``core/schema.py`` have been removed. Callers MUST use
``Validator(schema).validate(doc, ...)``.

``validate_frontmatter`` has been relocated to
``octave_mcp.core.grammar.entry`` as a parse-stage hook. The legacy
location is intentionally absent — no shim — per design §3 row 6.

See ``docs/adr/adr-0006-sr1-t1-grammar-core-design.md`` §3 row 6, §2.2.
"""

from dataclasses import dataclass, field
from enum import Enum
from typing import Any

from octave_mcp.core.constraints import EnumConstraint, RequiredConstraint
from octave_mcp.core.grammar.cst import (
    Assignment,
    ASTNode,
    Block,
    Document,
    InlineMap,
    ListValue,
    LiteralZoneValue,
    NodeKind,
    Section,
)
from octave_mcp.core.routing import InvalidTargetError, RoutingLog, TargetRegistry, TargetRouter
from octave_mcp.core.schema_extractor import (
    InheritanceResolver,
    SchemaDefinition,
    extract_block_targets,
)

# ADR-0006 SR2-T3 / GH-384 (sub-spec ``docs/adr/adr-0006-g3-meta-audit-markers.md``):
# Bounded-prefix admission policy for META audit-marker keys.
#
# Keys whose name begins with any prefix in this tuple are:
#   1. Admitted in STRICT mode without raising E007 (the unknown-field check
#      skips them — see ``_validate_meta``).
#   2. Surfaced via a single informational ``W_META_AUDIT`` warning in BOTH
#      LENIENT and STRICT modes, regardless of schema presence (mirrors the
#      unconditional W_META_001/W_META_002 pattern — see
#      ``_check_meta_warnings``).
#
# Why a tuple, not a frozenset:
#   The values are prefix strings consumed by ``str.startswith()`` (an
#   ordered-iterable input). The sub-spec line 86 names the constant a
#   "tuple". The GH-386 ``SUPPRESSIBLE_NORMALIZATION_CODES`` frozenset
#   precedent applies to warning-code identity (exact-match set membership),
#   not prefix matching, so it is not the right shape here. Immutability of
#   ``tuple`` carries the closed-set intent.
#
# Adding a new prefix:
#   1. Append to this tuple.
#   2. Update ``META.oct.md`` §4::AUDIT_MARKERS via ``mcp__octave__octave_write``.
#   3. Add an acceptance test in ``test_meta_audit_admission.py``.
META_AUDIT_ADMIT_PATTERNS: tuple[str, ...] = (
    "NON_CANONICAL_",
    "DEGRADED_",
    "NORMALIZED_",
    "ROUNDTRIP_",
)


def _matches_audit_pattern(field_name: str) -> bool:
    """Return True iff ``field_name`` matches any prefix in
    ``META_AUDIT_ADMIT_PATTERNS``.

    Module-private helper centralising the discriminant so the admit-path
    (in ``_validate_meta``) and the warning-emit path (in
    ``_check_meta_warnings``) cannot drift apart over time. See the
    GH-386 ``is_destructive_normalization_repair`` precedent in
    ``core/repair_log.py``.
    """
    return any(field_name.startswith(prefix) for prefix in META_AUDIT_ADMIT_PATTERNS)


class UnknownFieldPolicy(Enum):
    """Policy for handling unknown fields during validation.

    Issue #190: Spec §5::POLICY_BLOCK defines UNKNOWN_FIELDS::REJECT∨IGNORE∨WARN

    Attributes:
        REJECT: Error on unknown fields (E007)
        IGNORE: Skip unknown fields silently
        WARN: Add warning but don't fail (W001)
    """

    REJECT = "REJECT"
    IGNORE = "IGNORE"
    WARN = "WARN"


@dataclass
class ValidationError:
    """Validation error with context.

    Attributes:
        code: Error/warning code (E007, W001, etc.)
        message: Human-readable description
        field_path: Dot-separated path to field (e.g., "SECTION.FIELD")
        line: Line number in source (0 if unknown)
        severity: "error" or "warning" (default: "error")
    """

    code: str
    message: str
    field_path: str = ""
    line: int = 0
    severity: str = field(default="error")


class Validator:
    """OCTAVE AST validator."""

    def __init__(self, schema: dict[str, Any] | None = None):
        """Initialize validator with optional schema."""
        self.schema = schema or {}
        self.errors: list[ValidationError] = []
        self.routing_log: RoutingLog = RoutingLog()
        self._target_registry: TargetRegistry | None = None
        self._target_router: TargetRouter | None = None
        # Block inheritance support (Issue #189, M3 CE violations)
        self._block_targets: dict[str, str] = {}
        self._inheritance_resolver: InheritanceResolver = InheritanceResolver()

    def _to_python_value(self, value: Any) -> Any:
        """Convert AST values to Python primitives for constraint evaluation.

        LiteralZoneValue is returned as-is: TYPE[LITERAL] and LANG[] constraints
        operate directly on the object (Issue #235 T12).
        """
        if isinstance(value, ListValue):
            return [self._to_python_value(item) for item in value.items]
        if isinstance(value, InlineMap):
            return {k: self._to_python_value(v) for k, v in value.pairs.items()}
        if isinstance(value, LiteralZoneValue):
            # Literal zones remain as-is for constraint evaluation.
            # TYPE[LITERAL] and LANG[] constraints operate on the object directly.
            # D4: content is opaque; no conversion or normalization applied.
            return value
        return value

    def validate(
        self,
        doc: Document,
        strict: bool = False,
        section_schemas: dict[Any, SchemaDefinition] | None = None,
    ) -> list[ValidationError]:
        """Validate document against schema.

        Args:
            doc: Document AST
            strict: If True, reject unknown fields
            section_schemas: Optional dict mapping section names to SchemaDefinition.
                            When provided, sections with matching keys will be validated
                            against their schema's constraints. Sections without a
                            matching entry are skipped (I5: schema sovereignty).

        Returns:
            List of validation errors (empty if valid)
        """
        self.errors = []
        self.routing_log = RoutingLog()  # Reset routing log for each validation

        # Shared per-document setup (block targets + META + duplicate-target
        # safety net). Centralised in ``_prepare_document_validation`` so
        # both ``validate()`` and ``visit_document()`` reach the same checks
        # — the R2 closure goal extends to internal symmetry between entry
        # points (per CRS review of PR #401).
        self._prepare_document_validation(doc, strict=strict)

        # Validate sections
        # Issue #325: Walk document tree recursively so nested blocks (e.g., NATURE:
        # inside §1::COGNITIVE_IDENTITY) are also validated against section_schemas.
        self._validate_sections_recursive(doc.sections, strict, section_schemas)

        # GH-428: §-section body coverage warnings.
        # Run once per document against any schema in section_schemas that
        # declares POLICY.REQUIRED_SECTION_IDS or
        # POLICY.SECTION_CONDITIONAL_REQUIRED. The policy lives on the full
        # SchemaDefinition (and is preserved through ``dataclasses.replace``
        # in ``_build_deep_section_schemas``), so the same emit-once
        # contract holds for both envelope and document-type schemas.
        # PROD::I4 (TRANSFORM_AUDITABILITY: stable warning codes
        # W_MISSING_REQUIRED_SECTION, W_INCOMPLETE_SECTION_FIELDS).
        # PROD::I5 (SCHEMA_SOVEREIGNTY: surfaces what the Zone 2 frontmatter
        # check surfaces for #244, but for Zone 1 §-section content).
        if section_schemas is not None:
            self._validate_section_body_coverage(doc, section_schemas)

        # Issue #244: Validate Zone 2 (YAML frontmatter) when schema defines frontmatter
        # This is opt-in: only schemas with frontmatter defs trigger validation.
        # I1: Read-only inspection, never alters Zone 2 content.
        # ADR-0006 SR1-T1 Step 6: validate_frontmatter now lives at
        # ``octave_mcp.core.grammar.entry`` (parse-stage hook). Imported
        # locally to keep the seam crisp and to avoid a circular import
        # at module load time.
        if section_schemas is not None:
            from octave_mcp.core.grammar.entry import validate_frontmatter

            for schema_def in section_schemas.values():
                if schema_def.frontmatter:
                    fm_errors = validate_frontmatter(doc.raw_frontmatter, schema_def)
                    self.errors.extend(fm_errors)

        return self.errors

    # ------------------------------------------------------------------
    # ADR-0006 SR1-T1 Step 6: Visitor[None] protocol surface
    # ------------------------------------------------------------------
    #
    # ``Validator`` is a structural ``visitor.Visitor[None]``. The four
    # typed visit methods plus the ``visit`` dispatcher delegate to the
    # existing recursive validation machinery so a single canonical
    # surface — the class — owns ALL grammar-vs-schema decisions. This
    # is the R2 closure: there is no second validate path.
    #
    # Side effects: visit methods APPEND to ``self.errors``. Callers
    # interested in the accumulated diagnostics should read ``self.errors``
    # or use the orchestrating ``validate()`` method, which resets state,
    # walks the document, and returns the accumulated list. The visit
    # methods themselves return ``None`` (the ``Visitor[None]`` contract).
    #
    # See ``docs/adr/adr-0006-sr1-t1-grammar-core-design.md`` §3 row 6
    # and ``src/octave_mcp/core/grammar/visitor.py`` (``Visitor[T_co]``).

    def _prepare_document_validation(self, doc: Document, *, strict: bool) -> None:
        """Shared per-document setup for ``validate`` and ``visit_document``.

        Performs the four steps that BOTH entry points need before any
        section walk: block-target extraction (feudal inheritance),
        schema-aware META validation, schema-independent loss-accounting
        warnings, and the GH#369 duplicate-target safety net.

        Centralising the setup is the internal-symmetry guard the PR's R2
        closure goal demands: a new check added here is automatically
        consulted by every public entry point, so the two surfaces cannot
        drift apart over time. (Identified by CRS during PR #401 review.)

        Args:
            doc: Document AST.
            strict: If True, ``_validate_meta`` rejects unknown META fields.
        """
        # Extract block-level targets from document AST (Issue #189, M3 CE
        # violations). This enables feudal inheritance: children inherit
        # parent block targets.
        self._block_targets = extract_block_targets(doc)

        # Validate META if schema defines it.
        if "META" in self.schema and doc.meta:
            self._validate_meta(doc.meta, strict)

        # Loss-accounting warnings fire on document content, not schema
        # presence. PR#315 fix: moved outside the ``"META" in self.schema``
        # guard so that W_META_001/W_META_002 fire regardless of whether a
        # schema is provided.
        if doc.meta:
            self._check_meta_warnings(doc.meta)

        # GH#369: structural duplicate-target safety net. If the AST
        # contains both Block(K) and a flat Assignment("K.X", ...) at the
        # same level, that is the silent-corruption fingerprint left by a
        # regressed writer (or by hand-edited input). Surface it under
        # STANDARD profile so I5 (Schema Sovereignty) does not stay silent.
        self._check_duplicate_semantic_targets(doc.sections)

    def visit_document(self, node: Document, /) -> None:
        """Visit a Document node — orchestrates the full validation walk.

        This is the visitor-protocol entry point. It performs the
        schema-aware work that does NOT require a ``section_schemas`` map
        (META validation, loss-accounting warnings, duplicate-target
        check, recursive section walk) but does NOT reset
        ``self.errors``; composition with other visitors that share the
        same accumulator is therefore safe. Frontmatter validation is
        performed only by :meth:`validate`, because it requires the
        ``section_schemas`` argument which is not part of the bare
        ``Visitor[None]`` contract. Callers wanting the legacy "reset +
        return" semantics, frontmatter validation, or schema-aware
        section recursion should use :meth:`validate` instead.
        """
        # Delegate shared setup to ``_prepare_document_validation`` so the
        # two entry points (``validate`` and ``visit_document``) cannot
        # drift apart — internal-symmetry counterpart to the PR's R2
        # closure (per CRS review of PR #401).
        self._prepare_document_validation(node, strict=False)

        # Recurse into the section tree.
        for child in node.sections:
            self.visit(child)

    def visit_section(self, node: Section, /) -> None:
        """Visit a Section node — schema-less by default at this surface.

        Section validation requires a ``section_schemas`` map, which is
        not part of the bare ``Visitor[None]`` contract. The dispatcher
        therefore performs the structural recursion only; callers that
        need schema-aware section validation should use the orchestrating
        :meth:`validate` method (which threads the schema map through
        :meth:`_validate_sections_recursive`).
        """
        for child in node.children:
            self.visit(child)

    def visit_block(self, node: Block, /) -> None:
        """Visit a Block node — recurse into children.

        Schema-aware block validation lives in :meth:`_validate_section`,
        invoked by :meth:`_validate_sections_recursive` when a
        ``section_schemas`` map is supplied. The bare visitor walk only
        descends the structural tree.
        """
        for child in node.children:
            self.visit(child)

    def visit_assignment(self, node: Assignment, /) -> None:
        """Visit an Assignment node — terminal leaf at this surface.

        Field-level constraint evaluation happens inside
        :meth:`_validate_section` (which has the matching schema field
        definition in context). At the bare visitor surface, an
        Assignment is a leaf — nothing to recurse into.
        """
        # Intentional no-op: leaves are handled by the schema-aware
        # walk in ``_validate_section``. The structural visit contract
        # for ``Visitor[None]`` is satisfied by being callable and
        # returning ``None``.
        return None

    def visit(self, node: ASTNode, /) -> None:
        """Fallback dispatcher — routes on ``node.kind`` to a typed visit.

        Per ``grammar/visitor.py``'s ``Visitor`` protocol, ``visit`` is
        the entry point for unknown-kind nodes. Implementations dispatch
        on the ``NodeKind`` discriminant to avoid ``isinstance`` chains
        at the call site.
        """
        kind = getattr(node, "kind", None)
        if kind is NodeKind.DOCUMENT and isinstance(node, Document):
            self.visit_document(node)
        elif kind is NodeKind.SECTION and isinstance(node, Section):
            self.visit_section(node)
        elif kind is NodeKind.BLOCK and isinstance(node, Block):
            self.visit_block(node)
        elif kind is NodeKind.ASSIGNMENT and isinstance(node, Assignment):
            self.visit_assignment(node)
        # NodeKind.COMMENT and any future kinds: no-op at validator surface.

    def _validate_meta(self, meta: dict[str, Any], strict: bool) -> None:
        """Validate META block."""
        schema_meta = self.schema.get("META", {})

        # Check required fields
        required = schema_meta.get("required", [])
        for field_name in required:
            if field_name not in meta:
                self.errors.append(
                    ValidationError(
                        code="E003",
                        message=f"Cannot auto-fill missing required field '{field_name}'. Author must provide value.",
                        field_path=f"META.{field_name}",
                    )
                )

        # Check unknown fields in strict mode
        if strict:
            allowed = schema_meta.get("fields", {}).keys()
            for field_name in meta.keys():
                if field_name not in allowed and allowed:
                    # ADR-0006 SR2-T3 / GH-384: META audit-marker keys
                    # matching ``META_AUDIT_ADMIT_PATTERNS`` are admitted
                    # without E007. The matching ``W_META_AUDIT`` warning
                    # is emitted unconditionally by ``_check_meta_warnings``
                    # so the admission decision remains visible in the
                    # validator output (PROD::I5 SCHEMA_SOVEREIGNTY).
                    if _matches_audit_pattern(field_name):
                        continue
                    self.errors.append(
                        ValidationError(
                            code="E007",
                            message=f"Unknown field '{field_name}' not allowed in STRICT mode.",
                            field_path=f"META.{field_name}",
                        )
                    )

        # Type validation
        fields_schema = schema_meta.get("fields", {})
        for field_name, value in meta.items():
            if field_name in fields_schema:
                self._validate_type(field_name, value, fields_schema[field_name])

    def _check_meta_warnings(self, meta: dict[str, Any]) -> None:
        """Emit loss accounting consistency warnings based on META content.

        PR#315 fix: These warnings are based on document content, not schema
        presence.  Runs unconditionally when doc.meta exists.

        I4 (Transform Auditability): COMPRESSION_TIER without LOSS_PROFILE
        means loss accounting is incomplete.
        """
        compression_tier = meta.get("COMPRESSION_TIER")
        loss_profile = meta.get("LOSS_PROFILE")

        if compression_tier and not loss_profile:
            self.errors.append(
                ValidationError(
                    code="W_META_001",
                    message="COMPRESSION_TIER declared but LOSS_PROFILE absent — loss accounting incomplete",
                    field_path="META.LOSS_PROFILE",
                    severity="warning",
                )
            )

        if (
            loss_profile is not None
            and str(loss_profile).lower() == "none"
            and compression_tier
            and str(compression_tier).upper() != "LOSSLESS"
        ):
            self.errors.append(
                ValidationError(
                    code="W_META_002",
                    message=(
                        f"LOSS_PROFILE is 'none' but COMPRESSION_TIER is " f"{compression_tier} — verify accuracy"
                    ),
                    field_path="META.LOSS_PROFILE",
                    severity="warning",
                )
            )

        # ADR-0006 SR2-T3 / GH-384: emit W_META_AUDIT for every META key
        # matching ``META_AUDIT_ADMIT_PATTERNS``. Runs unconditionally on
        # ``doc.meta`` content — independent of mode (LENIENT/STRICT) and
        # of schema presence — mirroring the W_META_001/W_META_002
        # pattern above. PROD::I4 (TRANSFORM_AUDITABILITY: stable-ID
        # audit receipt) + PROD::I5 (SCHEMA_SOVEREIGNTY: validation
        # status visible in output).
        for field_name in meta.keys():
            if _matches_audit_pattern(field_name):
                self.errors.append(
                    ValidationError(
                        code="W_META_AUDIT",
                        message=(
                            f"Audit marker '{field_name}' admitted "
                            f"(matches META_AUDIT_ADMIT_PATTERNS — see "
                            f"ADR-0006 G3 / GH-384)"
                        ),
                        field_path=f"META.{field_name}",
                        severity="warning",
                    )
                )

    def _check_duplicate_semantic_targets(self, nodes: list[ASTNode]) -> None:
        """Emit W_DUPLICATE_TARGET when a Block(K) coexists with a flat
        Assignment("K.X", ...) at the same nesting level.

        GH#369: Structural safety net. The writer-side fix prevents new
        duplicates from being created, but pre-existing files (or any future
        writer regression) can still contain this corruption fingerprint.
        Surfacing it as a warning fulfils I5 (Schema Sovereignty: validation
        status visible) and I3 (Mirror Constraint: do not silently accept
        fabricated structure).

        Walks the given nodes list and recurses into Block.children and
        Section.children so the check fires at every nesting level.
        """
        # Collect Block keys and flat dotted Assignment keys at this level.
        block_keys: set[str] = set()
        dotted_assignments: list[Assignment] = []
        for node in nodes:
            if isinstance(node, Block):
                block_keys.add(node.key)
            elif isinstance(node, Assignment) and "." in node.key:
                dotted_assignments.append(node)

        for assignment in dotted_assignments:
            parent_key, _, child_key = assignment.key.partition(".")
            # Only flag if PARENT collides with an actual Block at the same
            # level. Single-identifier dotted keys like "P1.1" without a
            # corresponding Block remain valid (GH#347 carve-out).
            if parent_key in block_keys:
                self.errors.append(
                    ValidationError(
                        code="W_DUPLICATE_TARGET",
                        message=(
                            f"Duplicate semantic target: Block '{parent_key}' coexists with "
                            f"flat assignment '{assignment.key}' at the same level. This is "
                            f"the corruption fingerprint of a writer that failed to resolve "
                            f"a nested path (GH#369). Modify '{child_key}' inside block "
                            f"'{parent_key}' or remove one of the two."
                        ),
                        field_path=assignment.key,
                        severity="warning",
                    )
                )

        # Recurse into block/section children.
        for node in nodes:
            if isinstance(node, Block):
                self._check_duplicate_semantic_targets(node.children)
            elif isinstance(node, Section):
                self._check_duplicate_semantic_targets(node.children)

    def _validate_section_body_coverage(
        self,
        doc: Document,
        section_schemas: dict[Any, SchemaDefinition],
    ) -> None:
        """Emit W_MISSING_REQUIRED_SECTION / W_INCOMPLETE_SECTION_FIELDS (GH-428).

        Two complementary section-body checks driven entirely by POLICY:

        1. ``REQUIRED_SECTION_IDS``: For each id (e.g. ``"1"``) declared in
           any schema's POLICY, surface ``W_MISSING_REQUIRED_SECTION`` when
           no top-level :class:`Section` in ``doc.sections`` carries that
           ``section_id``. The id is a string to support both plain numbers
           and suffix forms (e.g. ``"2b"``); see
           :class:`octave_mcp.core.grammar.cst.Section.section_id`.

        2. ``SECTION_CONDITIONAL_REQUIRED``: For each ``(section_key,
           required_fields)`` entry, IF a section/block with matching key
           is present in the document, ensure every required field appears
           as an Assignment among its direct children. Missing members
           surface ``W_INCOMPLETE_SECTION_FIELDS`` naming the gap. The
           check is conditional — documents without the section are
           unaffected (the policy expresses *coverage when present*, not
           *presence requirement*).

        Emit-once contract: each policy aggregate is unioned across all
        section_schemas before walking, so duplicated schemas (envelope
        plus per-section copies from ``_build_deep_section_schemas``) do
        not produce duplicate warnings.

        PROD::I4 stable warning codes; PROD::I5 schema-sovereignty surface.
        """
        # Aggregate policy fields across all schemas in the map. Duplicates
        # collapse via set / dict semantics so we emit once.
        required_ids: set[str] = set()
        conditional: dict[str, list[str]] = {}
        allows_empty: set[str] = set()
        for schema_def in section_schemas.values():
            policy = getattr(schema_def, "policy", None)
            if policy is None:
                continue
            for sid in getattr(policy, "required_section_ids", []) or []:
                required_ids.add(str(sid))
            for section_key, fields in (getattr(policy, "section_conditional_required", {}) or {}).items():
                # Later entries overwrite earlier — schemas are not expected
                # to declare conflicting field sets for the same section key
                # within a single validation pass.
                if fields:
                    conditional[section_key] = list(fields)
            # GH-426 cubic P2 fix on PR #446: per-schema empty-section opt-in.
            # Aggregated across schemas with the same set-union semantic as
            # required_ids / conditional (if any schema in the map opts a
            # section in, that section is treated as empty-allowed for this
            # validation pass).
            for section_key in getattr(policy, "section_allows_empty", set()) or set():
                allows_empty.add(section_key)

        if not required_ids and not conditional:
            return

        # Collect the section_ids present at the document top level. We do
        # not recurse: REQUIRED_SECTION_IDS expresses envelope-level
        # presence (e.g. "§1 must exist"), not nested presence.
        present_section_ids: set[str] = set()

        # Collect, for each section_key, the set of Assignment keys that
        # belong to that section. The OCTAVE parser models §-sections as
        # flat siblings in ``doc.sections``: a ``Section`` node carries
        # only its header (key + section_id), and the assignments that
        # appear *after* it (until the next ``Section``) are its body.
        # The same flat model applies inside any Block/Section ``children``
        # list (we recurse for completeness, though the SKILL §5 pattern
        # is exclusively top-level today).
        #
        # PR #444 cubic P2 #2 rework: previously this walker registered
        # only top-level Section keys as buckets and reset the active
        # section context on every recursive call into a Block's
        # children. That meant a SECTION_CONDITIONAL_REQUIRED entry for a
        # key authored as a nested Block (e.g., ``§1::WRAPPER`` then
        # ``ANCHOR_KERNEL:`` block carrying the quartet) silently passed
        # because ``ANCHOR_KERNEL not in section_field_keys``. The walker
        # now registers both Section and keyed-Block nodes as buckets and
        # threads the active key into the recursive call so nested
        # Assignments attribute to their enclosing container.
        section_field_keys: dict[str, set[str]] = {}

        def _walk_flat(nodes: list[ASTNode], enclosing_key: str | None = None) -> None:
            current_section_key: str | None = enclosing_key
            for node in nodes:
                if isinstance(node, Section):
                    current_section_key = node.key
                    # First sight wins — duplicate §-headers within one
                    # parent are pathological; later occurrences union
                    # into the same bucket so a single Section "body"
                    # collects every assignment authored under that key.
                    section_field_keys.setdefault(current_section_key, set())
                elif isinstance(node, Block):
                    # PR #444 cubic P2 #2: a keyed Block is a legitimate
                    # SECTION_CONDITIONAL_REQUIRED target. Register its
                    # key as a bucket so the conditional check fires.
                    section_field_keys.setdefault(node.key, set())
                    # Record any direct Assignment children of the Block.
                    for child in getattr(node, "children", []) or []:
                        if isinstance(child, Assignment):
                            section_field_keys[node.key].add(child.key)
                    # Recurse with the Block's key as the enclosing
                    # context so deeper Assignments attribute correctly.
                    if hasattr(node, "children") and node.children:
                        _walk_flat(node.children, enclosing_key=node.key)
                    # A keyed Block does not change the enclosing Section
                    # bucket for subsequent siblings — restore tracker.
                    continue
                elif isinstance(node, Assignment) and current_section_key is not None:
                    section_field_keys[current_section_key].add(node.key)
                # Recurse into nested Section children, threading the
                # enclosing Section key so any deeper Assignments still
                # attribute to it.
                if isinstance(node, Section) and hasattr(node, "children") and node.children:
                    _walk_flat(node.children, enclosing_key=node.key)

        for node in doc.sections:
            if isinstance(node, Section):
                present_section_ids.add(str(node.section_id))

        _walk_flat(doc.sections)

        # REQUIRED_SECTION_IDS check.
        for sid in sorted(required_ids):
            if sid not in present_section_ids:
                self.errors.append(
                    ValidationError(
                        code="W_MISSING_REQUIRED_SECTION",
                        message=(
                            f"Required §-section id '{sid}' is missing from the document body. "
                            f"The schema declares POLICY.REQUIRED_SECTION_IDS — without this "
                            f"section the Zone 1 body is structurally incomplete (GH-428)."
                        ),
                        field_path=f"§{sid}",
                        severity="warning",
                    )
                )

        # SECTION_CONDITIONAL_REQUIRED check (only fires when section is present).
        for section_key, required_fields in conditional.items():
            if section_key not in section_field_keys:
                continue  # Section absent — conditional check does not fire.

            present_field_keys = section_field_keys[section_key]

            # GH-426 cubic P2 fix on PR #446 — per-schema empty-section opt-in.
            #
            # Background: PR #446 first attempt added a universal
            # ``if not present_field_keys: continue`` bypass here to
            # support the legitimate CRS_REVIEW empty-§3-FINDINGS case
            # (an APPROVED review with TOTAL::0 authors an empty
            # findings section meaning "no findings to report"). cubic
            # correctly flagged this as a P2: the universal bypass
            # weakened SECTION_CONDITIONAL_REQUIRED enforcement for
            # SKILL — an author writing an empty §5::ANCHOR_KERNEL
            # block almost certainly meant to populate it; silent-pass
            # is wrong.
            #
            # Fix: scope the empty-pass per-schema via the POLICY opt-in
            # field SECTION_ALLOWS_EMPTY. A section is treated as
            # empty-allowed iff at least one schema in this validation
            # pass's section_schemas declared the section in
            # POLICY.SECTION_ALLOWS_EMPTY. CRS_REVIEW declares
            # ``SECTION_ALLOWS_EMPTY::["FINDINGS"]``; SKILL does not
            # declare the field at all, so its ANCHOR_KERNEL stays
            # strict.
            #
            # The REQUIRED_SECTION_IDS check above still fires
            # independently (the section MUST exist — empty is fine
            # only if opted in, absent is not allowed regardless).
            # PROD::I5 SCHEMA_SOVEREIGNTY preserved on both axes:
            # missing-finding fields are surfaced when entries are
            # present, AND the SKILL strict-empty contract is restored.
            if not present_field_keys and section_key in allows_empty:
                continue  # GH-426 opt-in: schema declared this section empty-allowed.

            missing = [f for f in required_fields if f not in present_field_keys]
            if missing:
                self.errors.append(
                    ValidationError(
                        code="W_INCOMPLETE_SECTION_FIELDS",
                        message=(
                            f"Section '{section_key}' present but missing required field(s): "
                            f"{', '.join(missing)}. POLICY.SECTION_CONDITIONAL_REQUIRED declares "
                            f"the full set as {', '.join(required_fields)} (GH-428)."
                        ),
                        field_path=section_key,
                        severity="warning",
                    )
                )

    def _validate_unknown_fields(
        self,
        document_fields: set[str],
        schema_fields: set[str],
        policy: UnknownFieldPolicy,
        section_key: str,
    ) -> list[ValidationError]:
        """Check for unknown fields per UNKNOWN_FIELDS policy.

        Issue #190: Implements §5::POLICY_BLOCK UNKNOWN_FIELDS enforcement.

        Args:
            document_fields: Set of field names present in the document section
            schema_fields: Set of field names defined in the schema
            policy: How to handle unknown fields (REJECT, WARN, or IGNORE)
            section_key: Section name for error path construction

        Returns:
            List of ValidationErrors (errors for REJECT, warnings for WARN, empty for IGNORE)
        """
        unknown = document_fields - schema_fields
        if not unknown:
            return []

        errors: list[ValidationError] = []

        if policy == UnknownFieldPolicy.REJECT:
            for field_name in sorted(unknown):  # Sort for deterministic output
                errors.append(
                    ValidationError(
                        code="E007",
                        message=f"Unknown field '{field_name}' not allowed per UNKNOWN_FIELDS::REJECT policy",
                        field_path=f"{section_key}.{field_name}",
                        severity="error",
                    )
                )
        elif policy == UnknownFieldPolicy.WARN:
            for field_name in sorted(unknown):
                errors.append(
                    ValidationError(
                        code="W001",
                        message=f"Unknown field '{field_name}' (UNKNOWN_FIELDS::WARN policy)",
                        field_path=f"{section_key}.{field_name}",
                        severity="warning",
                    )
                )
        # IGNORE: return empty list (no errors/warnings)

        return errors

    def _validate_sections_recursive(
        self,
        nodes: list[ASTNode],
        strict: bool,
        section_schemas: dict[Any, SchemaDefinition] | None,
    ) -> None:
        """Walk document tree and validate each section/block against section_schemas.

        Issue #325: For document-type schemas (e.g., COGNITION_DEFINITION), fields
        may be nested inside sections and sub-blocks. This method recursively walks
        the AST so that nested blocks (e.g., NATURE: inside §1::COGNITIVE_IDENTITY)
        are also matched against section_schemas and validated.

        Args:
            nodes: List of AST nodes to walk
            strict: Whether strict mode is enabled
            section_schemas: Optional dict mapping section/block names to SchemaDefinition
        """
        for node in nodes:
            # Look up schema for this node by key
            section_schema = None
            if section_schemas is not None:
                section_key = getattr(node, "key", None)
                if section_key is not None:
                    section_schema = section_schemas.get(section_key)
            self._validate_section(node, strict, section_schema)

            # Recurse into children if section_schemas is provided and node has children
            if section_schemas is not None:
                children = getattr(node, "children", None)
                if children:
                    self._validate_sections_recursive(children, strict, section_schemas)

    def _validate_section(self, section: ASTNode, strict: bool, section_schema: SchemaDefinition | None = None) -> None:
        """Validate a section against its schema definition.

        Args:
            section: The section AST node to validate
            strict: Whether strict mode is enabled
            section_schema: Optional schema definition for this section.
                           If None, skip content validation (I5: schema-less = UNVALIDATED).

        Validates:
        - Holographic patterns: ["example"∧CONSTRAINT→§TARGET]
        - Constraint chains: REQ∧ENUM[A,B]∧REGEX["^[a-z]+$"]
        - Target routing with audit trail (Issue #103)
        - Target validation with registry (Issue #188)
        - Unknown fields per UNKNOWN_FIELDS policy (Issue #190)
        """
        # I5: Schema-less sections skip content validation
        if section_schema is None:
            return

        # Only Block and Section nodes have children to validate
        # Issue #325: Section nodes (§N::NAME) also need validation for
        # document-type schemas like COGNITION_DEFINITION where fields are
        # inside numbered sections, not at the envelope level.
        if not isinstance(section, Block | Section):
            return

        # Build map of present fields for REQ constraint checking
        present_fields: dict[str, Any] = {}
        for child in section.children:
            if isinstance(child, Assignment):
                present_fields[child.key] = self._to_python_value(child.value)

        # Issue #190: UNKNOWN_FIELDS policy enforcement
        # Parse policy from schema, defaulting to REJECT
        policy_str = section_schema.policy.unknown_fields if section_schema.policy else "REJECT"
        try:
            unknown_policy = UnknownFieldPolicy(policy_str)
        except ValueError:
            # Invalid policy value, default to REJECT (fail-safe)
            unknown_policy = UnknownFieldPolicy.REJECT

        # Check for unknown fields
        document_fields = set(present_fields.keys())
        schema_fields = set(section_schema.fields.keys())
        unknown_errors = self._validate_unknown_fields(document_fields, schema_fields, unknown_policy, section.key)
        self.errors.extend(unknown_errors)

        # Get block-level default target (feudal inheritance)
        default_target = getattr(section_schema, "default_target", None)

        # Initialize target registry with custom targets from POLICY.TARGETS (Issue #190)
        registry = TargetRegistry()
        # Get custom targets from policy
        policy_targets = section_schema.policy.targets if section_schema.policy else []
        for custom_target in policy_targets:
            registry.register_custom(custom_target)

        # Register default_target as custom if not already a builtin (M3 fix)
        # This ensures policy default targets are valid routing destinations
        if default_target and default_target not in registry.BUILTINS:
            registry.register_custom(default_target)

        # Register block-level targets as custom targets (M3 fix for CE violation #1)
        # Block targets from [->TARGET] syntax are implicitly valid routing destinations
        for block_target in self._block_targets.values():
            if block_target not in registry.BUILTINS:
                registry.register_custom(block_target)

        # Create router for target validation and routing
        router = TargetRouter(registry=registry, routing_log=self.routing_log)

        # Validate each field defined in schema
        for field_name, field_def in section_schema.fields.items():
            # Check if field is present
            value = present_fields.get(field_name)
            field_path = f"{section.key}.{field_name}"
            constraint_passed = True

            # Evaluate constraints if pattern exists
            if field_def.pattern and field_def.pattern.constraints:
                # Check REQ constraint first for missing fields
                has_req = any(isinstance(c, RequiredConstraint) for c in field_def.pattern.constraints.constraints)
                if has_req and value is None:
                    self.errors.append(
                        ValidationError(
                            code="E003",
                            message=f"Field '{field_name}' is required but missing",
                            field_path=field_path,
                        )
                    )
                    continue

                # Skip other validation if field is absent (and not required)
                if value is None:
                    continue

                # Evaluate full constraint chain
                result = field_def.pattern.constraints.evaluate(value=value, path=field_path)
                constraint_passed = result.valid

                # Convert constraint errors to validator errors
                for error in result.errors:
                    self.errors.append(
                        ValidationError(
                            code=error.code,
                            message=error.message,
                            field_path=error.path,
                        )
                    )

            # Target routing with audit trail (Issue #103) and validation (Issue #188)
            # Feudal inheritance hierarchy (closest wins):
            # 1. Field explicit target (highest priority)
            # 2. Block-level target [->TARGET] from AST (Issue #189, M3 CE violations)
            # 3. POLICY.DEFAULT_TARGET (lowest priority)
            if field_def.pattern:
                target: str | None = field_def.pattern.target  # Field's explicit target
                if target is None:
                    # Try block inheritance: resolve from block hierarchy
                    field_path_parts = field_path.split(".")
                    inherited_target = self._inheritance_resolver.resolve_target(field_path_parts, self._block_targets)
                    # Use inherited target if found, otherwise fall back to policy default
                    target = inherited_target if inherited_target is not None else default_target

                if target is not None and value is not None:
                    # Route through TargetRouter for validation and multi-target support
                    try:
                        router.route(
                            source_path=field_path,
                            target_spec=target,
                            value=value,
                            constraint_passed=constraint_passed,
                        )
                    except InvalidTargetError as e:
                        # E009: Invalid target error
                        self.errors.append(
                            ValidationError(
                                code="E009",
                                message=f"Invalid target '{e.target_name}': not a builtin, registered custom, or file path target",
                                field_path=field_path,
                            )
                        )

    def _validate_type(self, field: str, value: Any, field_schema: dict[str, Any]) -> None:
        """Validate value type and enum constraints."""
        expected_type = field_schema.get("type")
        if not expected_type:
            return

        # Handle ENUM type with constraint evaluation
        if expected_type == "ENUM":
            enum_values = field_schema.get("values", [])
            constraint = EnumConstraint(allowed_values=enum_values)
            result = constraint.evaluate(value, path=f"META.{field}")

            if not result.valid:
                for error in result.errors:
                    self.errors.append(
                        ValidationError(
                            code=error.code,
                            message=error.message,
                            field_path=error.path,
                        )
                    )
            return

        # Handle standard types
        # I1: Reject booleans for NUMBER type (bool inherits from int in Python)
        if expected_type == "NUMBER":
            if isinstance(value, bool):
                self.errors.append(
                    ValidationError(
                        code="E007",
                        message=f"Field '{field}' expected NUMBER, got {type(value).__name__}",
                        field_path=field,
                    )
                )
                return
            if not isinstance(value, int | float):
                self.errors.append(
                    ValidationError(
                        code="E007",
                        message=f"Field '{field}' expected NUMBER, got {type(value).__name__}",
                        field_path=field,
                    )
                )
            return

        # Handle other types
        type_map: dict[str, type | tuple[type, ...]] = {
            "STRING": str,
            "BOOLEAN": bool,
            "LIST": list,
        }

        expected_python_type = type_map.get(expected_type)
        if expected_python_type and not isinstance(value, expected_python_type):
            self.errors.append(
                ValidationError(
                    code="E007",  # I4: Use E007 for type errors (consistency with constraints.py)
                    message=f"Field '{field}' expected {expected_type}, got {type(value).__name__}",
                    field_path=field,
                )
            )


def _count_literal_zones(doc: Document) -> list[dict[str, Any]]:
    """Return per-zone metadata for all LiteralZoneValue instances in a Document.

    Issue #235 T12: Shared utility used by the validator and MCP tools (T13, T14).

    Returns per-zone metadata rather than a plain count so that MCP tools can
    populate ``zone_report.literal.zones`` with key/info_tag/line entries.

    D4: Content is opaque -- only envelope metadata (key, info_tag, line) is
    reported.  I5: Caller is responsible for setting ``literal_zones_validated=False``.

    **Line semantics**: The ``line`` field records the source line of the
    *assignment* (the ``KEY::`` line), not the line of the opening fence
    marker (the ` ``` ` line).  ``LiteralZoneValue`` does not store its own
    source position; only the containing ``Assignment`` node carries a line
    number (inherited from ``ASTNode.line``).  The fence opening always
    appears on ``assignment.line + 1`` in well-formed OCTAVE documents, but
    this function does not compute that offset -- it exposes the assignment
    line directly.  Consumers that need the fence line should add 1 to the
    reported ``line`` value.

    Args:
        doc: Parsed Document AST to inspect.

    Returns:
        List of dicts, each with keys:
          - "key":      Assignment key (str)
          - "info_tag": Language tag from fence opener, or None
          - "line":     Source line number of the *assignment* node (the
                        ``KEY::`` line), NOT the fence opening line.

        Returns an empty list when no literal zones are present.
    """
    zones: list[dict[str, Any]] = []

    def _collect_from_value(key: str, value: Any, line: int) -> None:
        """Recursively collect LiteralZoneValue instances from a value.

        Descends into ListValue.items and InlineMap.pairs so that literal
        zones nested inside list or map assignments are not silently missed
        (Fix for CE review finding: T13/T14 zone_report metadata correctness).

        Args:
            key:   Assignment key (used as the zone key in the report entry).
            value: The assignment value to inspect (may be any AST value type).
            line:  Source line of the containing assignment.
        """
        if isinstance(value, LiteralZoneValue):
            zones.append(
                {
                    "key": key,
                    "info_tag": value.info_tag,
                    "line": line,
                }
            )
        elif isinstance(value, ListValue):
            for item in value.items:
                _collect_from_value(key, item, line)
        elif isinstance(value, InlineMap):
            for v in value.pairs.values():
                _collect_from_value(key, v, line)

    def _traverse(nodes: list[ASTNode]) -> None:
        for node in nodes:
            if isinstance(node, Assignment):
                _collect_from_value(node.key, node.value, node.line)
            elif isinstance(node, Block):
                _traverse(node.children)
            elif hasattr(node, "children"):
                # Handles Section nodes and any future container nodes.
                _traverse(node.children)

    _traverse(doc.sections)
    return zones
