stable_tasks.py

"""Pure acceptance contracts for changing, structured page content.

Values are parsed, never compared to a snapshot. Supported Markdown/plain-text
layouts preserve record boundaries; fields are never searched across cards.
Passing establishes the declared output structure, not factual source fidelity.
"""
from __future__ import annotations

import hashlib
import math
import re

import soupsieve
import unicodedata
from datetime import date, datetime
from html import unescape
from urllib.parse import urlsplit

from .dynamic_lists import _label_key, extract_named_entries, named_entry_identity, named_entry_source_pair_hash
from .reference_quality import (
    clean_reference_text, is_markdown_image_start, markdown_delimiter_pairs,
    visible_markdown_links, without_markdown_images,
)

ACCEPTANCE_CONTRACT_FIELDS = {
    'version', 'schema', 'minimum_records',
    'maximum_records', 'page_patterns', 'stable_text_anchors',
    'stable_text_role', 'hosts', 'entry_path',
    'entry_identity_pattern', 'entry_context', 'entry_scope',
    'name_validation', 'identity_query_keys', 'identity_query_paths',
    'field_minimum_records', 'record_name_exclusions', 'title_exclusion_patterns',
    'price_patterns', 'price_exclusion_patterns', 'price_unavailable_patterns',
    'record_count_policy', 'source_record_pair_hashes', 'source_record_pairs_sha256',
    'table_columns', 'record_identity_fields', 'table_scope',
    'duplicate_policy', 'link_base_url', 'text_exclusions',
    'content_scope', 'source_membership',
}

SCHEMA_FIELDS = {
    "table_row": (),
    "earthquake": ("magnitude", "location", "time", "depth"),
    "body_text": (),
    "quote": ("quote", "author", "tags"),
    "product": ("name", "price"),
    "catalog_product": ("title", "price", "description", "reviews"),
    "review": ("date", "text"),
    "name_description": ("name", "description"),
    "api_operation": ("method", "path", "summary"),
    "taxonomy_family": ("family",),
    "named_link": ("title", "url"),
    "linked_product": ("title", "url", "price"),
    "page_identity": (),
}
_NATIVE_CARD_SCHEMAS = frozenset({
    'product', 'catalog_product', 'earthquake', 'quote', 'api_operation',
})
UNIQUE_FIELDS = {
    "table_row": (),
    "earthquake": ("time", "location"),
    "body_text": (),
    "quote": ("quote",), "product": ("name",),
    "catalog_product": ("title", "description"), "review": ("date", "text"),
    "name_description": ("name",), "api_operation": ("method", "path"),
    "taxonomy_family": ("family",), "named_link": ("url",),
    "linked_product": ("url",), "page_identity": (),
}
_AMOUNT = r"(?:\d{1,3}(?:,\d{3})+|\d+)(?:\.\d{1,2})?"
_MONEY = rf"(?:[$£€¥]\s*{_AMOUNT}|(?:USD|EUR|GBP|CNY)\s+{_AMOUNT}|{_AMOUNT}\s+(?:USD|EUR|GBP|CNY))"
_METHOD = r"(?:GET|POST|PUT|PATCH|DELETE|HEAD|OPTIONS|TRACE|CONNECT)"
_REVIEW_COUNT = r"\d+\s+reviews?"
_PLACEHOLDER = re.compile(r"(?:n/?a|none|null|undefined|unknown|tbd|loading[. …]*|[-—]+|add to cart|read more|view details|description|name|title)", re.I)
_LABELS = {"quote": "quote", "author": "(?:author|by)", "tags": "tags",
    "magnitude": "magnitude", "location": "location", "time": "time", "depth": "depth",
    "name": "name", "title": "title", "price": "price", "description": "description",
    "reviews": "reviews", "date": "date", "text": "(?:text|review)", "method": "method",
    "path": "path", "summary": "summary", "family": "family", "financing": "financing", "shipping": "shipping"}


def _key(text):
    return re.sub(r"\s+", " ", unicodedata.normalize("NFKC", text).casefold()).strip()


_CARD_BOUNDARY = "\ue000"
_MEDIA_BOUNDARY = "\ue001"
_HTML_HEADING = "\ue002"


def _unwrap_links(text, schema):
    """Unwrap balanced inline links, including multiline card labels.

    Only explicit markup creates boundaries; text from separate cards is never
    flattened together. Image alt text is not a substitute for a visible title.
    """
    pairs = markdown_delimiter_pairs(text)
    output, i = [], 0
    while i < len(text):
        image = is_markdown_image_start(text, i)
        start = i + 1 if image else i
        if text[start:start + 1] != "[":
            output.append(text[i])
            i += 1
            continue
        end = pairs.get(start)
        destination = end + 1 if end is not None else len(text)
        close = pairs.get(destination) if text[destination:destination + 1] == "(" else None
        if close is None:
            output.append(text[i])
            i += 1
            continue
        raw_label = text[start + 1:end]
        label = "" if image else _unwrap_links(raw_label, None).strip()
        image_only = image or (not label and raw_label.lstrip().startswith("!["))
        if image_only and schema == "name_description":
            prefix = "".join(output).rsplit("\n", 1)[-1]
            if re.fullmatch(r"\s*(?:[-*+]|\d+[.)])\s*", prefix):
                output.append(_CARD_BOUNDARY)
            elif image and not prefix.strip():
                line_end = text.find("\n", close + 1)
                suffix = text[close + 1:line_end if line_end >= 0 else len(text)]
                if not suffix.strip():
                    # A standalone image is an explicit local card boundary.
                    # Keep it as ownership evidence without treating alt text as
                    # the record name or joining prose across adjacent cards.
                    output.append(_MEDIA_BOUNDARY)
        if label:
            rendered = "".join(output)
            prefix = rendered.rsplit("\n", 1)[-1]
            if schema in {"product", "catalog_product", "earthquake"} and "\n" in text[start + 1:end]:
                label = f"\n{_CARD_BOUNDARY}\n{label}\n{_CARD_BOUNDARY}\n"
                output.append(label)
            elif schema == "name_description" and (re.fullmatch(r"\s*(?:[-*+]|\d+[.)])\s*", prefix)
                    or re.fullmatch(rf"\s*(?:[-*+]|\d+[.)])\s*{_CARD_BOUNDARY}\s*", prefix)
                    or re.search(rf"(?m)^\s*(?:[-*+]|\d+[.)])\s*(?:{_CARD_BOUNDARY})?\s*\n\s*$", rendered)):
                # A bullet's linked title owns the following prose, even when
                # both share one physical line. Some converters put an
                # image-only bullet on the previous line; that orphan bullet
                # still marks the following visible title as the card owner.
                suffix = re.match(r'[\w-]+', text[close + 1:])
                if suffix:
                    joined = visible_markdown_links(text[i:close + 1] + suffix[0])
                    if joined == label + suffix[0]:
                        label = joined
                        close += len(suffix[0])
                output.append(label + "\n")
            else:
                # The shared visible-link owner preserves word fragments,
                # adjacent labels and native path/multi-word boundaries.
                output.append(text[i:close + 1])
        i = close + 1
    return visible_markdown_links("".join(output))


def _html_record_owners(soup):
    """Keep semantic cards; nested feature lists remain presentation content."""
    return [node for node in soup.select('article, li, [role="listitem"]')
            if node.name == 'article' or node.get('role') == 'listitem' or not node.find_parent(
                lambda parent: parent.name in {'article', 'li'}
                or parent.get('role') == 'listitem')]


def _mark_html_headings(soup, *, include=None):
    """Keep real heading levels available to the shared record parsers."""
    for heading in soup.select('h1,h2,h3,h4,h5,h6'):
        if include is None or include(heading):
            heading.insert(0, _HTML_HEADING + heading.name[1] + _HTML_HEADING + ' ')


def _visible(text, schema=None, *, normalized_html=False, representation=None):
    if representation == 'plain_text':
        return clean_reference_text(text).replace('\r\n', '\n')
    from .reference_quality import is_html_document_or_fragment

    projected_html = (
        not normalized_html and schema in SCHEMA_FIELDS
        and schema not in {"named_link", "linked_product", "table_row"}
        and (representation == 'html' or is_html_document_or_fragment(text))
    )
    if projected_html:
        from .dynamic_lists import visible_html_text

        if schema == "review":
            from .dynamic_lists import scoped_html
            soup = scoped_html(text)
            # Nested presentation blocks stay inside their single-review owner.
            containers = 'section,article,li,div'
            candidates = [node for node in soup.select(containers)
                      if len(re.findall(r"(?m)^\s*(?:date\s*:\s*)?\d{4}-\d{2}-\d{2}(?=\s|$)",
                                        visible_html_text(str(node)), re.I)) == 1]
            dated = {id(node) for node in candidates}
            owners = {id(node) for node in candidates
                      if not any(id(child) in dated for child in node.select(containers))}
            # Presentation wrappers do not own records. Independent articles and
            # auxiliary blocks cannot supply an enclosing review's missing prose.
            for node in candidates:
                if (id(node) not in owners or node.name == 'article'
                        or not re.fullmatch(r"(?:date\s*:\s*)?\d{4}-\d{2}-\d{2}",
                                            visible_html_text(str(node)), re.I)):
                    continue
                for parent in node.parents:
                    if id(parent) not in dated:
                        continue
                    own_text = ' '.join(str(text) for text in parent.strings
                        if next((p for p in text.parents if p is parent or p is node
                                 or p.name in {'article', 'footer', 'nav', 'aside'}), None) is parent)
                    if own_text.strip():
                        owners.remove(id(node))
                        owners.add(id(parent))
                        break
                    if parent.name == 'article':
                        break
            def inside_record(node):
                return any(id(parent) in owners for parent in node.parents)
            for node in list(soup.select(containers + ',main,header,footer,nav,aside,hr')):
                if inside_record(node) and node.name not in {'article', 'footer', 'nav', 'aside'}:
                    continue
                node.insert_before("\n" + _CARD_BOUNDARY + "\n")
                node.insert_after("\n" + _CARD_BOUNDARY + "\n")
            _mark_html_headings(soup, include=lambda node: not inside_record(node))
            text = str(soup)
        elif schema == 'name_description':
            from .dynamic_lists import scoped_html
            soup = scoped_html(text)
            _mark_html_headings(soup)
            text = str(soup)
        elif schema in _NATIVE_CARD_SCHEMAS:
            from .dynamic_lists import scoped_html
            soup = scoped_html(text)
            if schema in {'catalog_product', 'api_operation'}:
                _mark_html_headings(soup)
            owners = _html_record_owners(soup)
            identities = {id(node) for node in owners}
            groups = {}
            for node in owners:
                ancestors = [parent for parent in node.parents if id(parent) in identities]
                root = ancestors[-1] if ancestors else node
                groups.setdefault(id(root), (root, []))[1].append(node)
            for root, cards in groups.values():
                chunks = []
                for card in cards:
                    copy = scoped_html(str(card))
                    owner = copy.find()
                    for nested in _html_record_owners(copy):
                        if nested is not owner and nested.parent is not None:
                            nested.decompose()
                    chunks.append(visible_html_text(str(copy)))
                # Preformatted projection keeps field lines intact through the
                # shared HTML reader; each card still uses the existing parser.
                projected = soup.new_tag('pre')
                projected.string = ('\n' + _CARD_BOUNDARY + '\n'
                                    + ('\n' + _CARD_BOUNDARY + '\n').join(chunks)
                                    + '\n' + _CARD_BOUNDARY + '\n')
                root.replace_with(projected)
            text = str(soup)
        text = visible_html_text(text)
    if schema == "name_description" and not normalized_html and not projected_html and re.search(r"<(?:div|section|article|span|script|template)\b", text, re.I):
        from .dynamic_lists import scoped_html
        text = scoped_html(text).get_text(separator="\n")
        projected_html = True
    text = clean_reference_text(text)
    if projected_html or representation == 'html':
        return text.replace('\r\n', '\n')
    literals = {}
    marker = '\ue026stable-code'
    while marker in text or marker in unescape(text):
        marker += '-'
    body_view = schema in {None, 'body_text', 'page_identity'}
    spans = _markdown_code_payloads(text, preserve_delimiters=not body_view)
    for start, end, payload in reversed(spans):
        token = marker + str(len(literals)) + '\ue027'
        literals[token] = payload
        text = text[:start] + token + text[end:]
    from .reference_quality import _markdown_unescape
    def rendered_literal(match):
        raw = match[0]
        punctuation = match[1]
        if punctuation is not None:
            if schema == 'table_row' and punctuation in {'|', '\\'}:
                return raw
            if punctuation == '\\' and re.match(r'[ \t]*\n', text[match.end():]):
                return raw
        elif normalized_html:
            return raw
        value = _markdown_unescape(raw)
        if value == raw:
            return raw
        token = marker + str(len(literals)) + '\ue027'
        # Keep emitted delimiters opaque until an owned record scalar is read.
        literals[token] = '\\' + value if not body_view and value in {'`', '\\'} else value
        return token
    text = re.sub(r'\\([!"#$%&\'()*+,\-./:;<=>?@\[\]\\^_`{|}~])'
                  r'|&(?:#[xX][0-9a-fA-F]+|#\d+|[A-Za-z][A-Za-z0-9]*);',
                  rendered_literal, text)
    # An actual inline HTML break in Markdown is layout. Preserve escaped
    # markup and code spans, and decode character references only afterward.
    text = re.sub(r'(`+|~{3,}).*?\1|(?<!\\)<br\s*/?>' if schema != "table_row" and not normalized_html else r'(`+|~{3,}).*?\1',
                  lambda m: m.group(0) if m.group(1) else " ", text,
                  flags=re.I | re.S)
    # Escaped punctuation and references above are decoded once and restored
    # after formatting; their emitted characters cannot introduce Markdown.
    # Extractors represent Markdown hard breaks using one or two backslashes.
    text = re.sub(r"\\{1,2}[ \t]*(?=\n)", "", text)
    text = (visible_markdown_links(without_markdown_images(text))
            if schema in {None, "body_text", "page_identity"} else _unwrap_links(text, schema))
    # CommonMark punctuation escapes preserve the same visible characters.
    # Remove them before date/field parsing (for example ``2023\-05\-18``).
    text = re.sub(r'\\([!"#$%&\'()*+,\-./:;<=>?@\[\]\\^_`{|}~])', lambda m: m[0] if schema == 'table_row' and m[1] in {'|', '\\'} else m[1], text)
    from .metrics import native_punctuation_boundaries
    text = native_punctuation_boundaries(text)
    if schema == "name_description":
        # Image-only card links intentionally contribute no visible title.
        # Remove their now-empty bullet line so blank-delimited card parsing
        # sees the following visible title and description as one record.
        text = re.sub(r"(?m)^\s*(?:[-*+]|\d+[.)])\s*$", "", text)
        text = re.sub(rf"(?m)^\s*(?:[-*+]|\d+[.)])\s*{_CARD_BOUNDARY}\s*$",
                      f"\n{_CARD_BOUNDARY}\n", text)
        boundaries = {_CARD_BOUNDARY, _MEDIA_BOUNDARY}
        text = "\n".join(line if line.strip() in boundaries
                         else line.replace(_CARD_BOUNDARY, "").replace(_MEDIA_BOUNDARY, "")
                         for line in text.splitlines())
    text = re.sub(r"\*\*|__" if schema in {None, "body_text", "page_identity"} else r"\*\*|__|`", "", text)
    # Link unwrapping may insert boundary spaces inside surrounding emphasis
    # (``*[label](url)*`` becomes ``* label *``). Treat that as the same visible
    # text as ordinary emphasis, without flattening other punctuation.
    text = re.sub(r"(?<!\w)([*_])\s*([^\n]*?\S)\s*\1(?!\w)", r"\2", text)
    # Preserve newlines: flattening them would let one card borrow another's fields.
    for token, literal in literals.items():
        text = text.replace(token, literal)
    return text.replace("\r\n", "\n")


def _line(text, *, literal=False):
    if literal:
        return re.sub(r'^' + _HTML_HEADING + r'[1-6]' + _HTML_HEADING + r'\s*', '', text).strip()
    text = re.sub(r"^(?:\s*[#>]+\s*|\s*[-*+]\s+|\s*\d+[.)]\s+)", "", text)
    return _render_inline_scalar(text)


def _render_inline_scalar(text, *, pipe_cell=False):
    """Render an owned value after layout, without reinterpreting code payloads."""
    from .article_text import mask_markdown_inline_code

    text, restore_code = mask_markdown_inline_code(text)
    text = re.sub(r"</?(?:strong|b|em|i)>", "", text, flags=re.I)
    text = re.sub(r'\\(`)' if pipe_cell else r'\\([`\\])', r'\1', text)
    return restore_code(text).strip()


def _field(text, name):
    return re.sub(rf"^(?:{_LABELS.get(name, name)})\s*:\s*", "", text, flags=re.I).strip()


def _text(value, minimum=2):
    return (len(value.strip()) >= minimum and any(c.isalpha() for c in value)
            and not _PLACEHOLDER.fullmatch(value.strip()) and not re.match(r"https?://", value)
            and not re.fullmatch(_MONEY, value, re.I))


def _valid_field(name, value):
    if not isinstance(value, str) or not value.strip():
        return False
    if name == "price":
        return bool(re.fullmatch(_MONEY, value, re.I))
    if name == "magnitude":
        return bool(re.fullmatch(r"-?\d{1,2}(?:\.\d+)?", value))
    if name == "depth":
        return bool(re.fullmatch(r"-?\d+(?:\.\d+)?\s*km", value, re.I))
    if name == "time":
        # Accept source-local UTC offsets and ISO timestamps without freezing
        # a date. A date alone is not an event timestamp.
        normalized = re.sub(r"\s*\(UTC([+-]\d{2}:\d{2})\)$", r"\1", value)
        normalized = re.sub(r"\s*\(UTC\)$", "+00:00", normalized)
        if not re.fullmatch(r"\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})?", normalized):
            return False
        try:
            datetime.fromisoformat(normalized.replace("Z", "+00:00"))
            return True
        except ValueError:
            return False
    if name == "reviews":
        return bool(re.fullmatch(r"\d+(?:\s*reviews?)?", value, re.I))
    if name == "date":
        try:
            return bool(re.fullmatch(r"\d{4}-\d{2}-\d{2}", value) and date.fromisoformat(value))
        except ValueError:
            return False
    if name == "method":
        return bool(re.fullmatch(_METHOD, value))
    if name == "path":
        return bool(re.fullmatch(r"/(?:[A-Za-z0-9_{}./~%:@+!$&'()*,;=\-]*)", value)) and not value.startswith("//")
    if name == "family":
        return bool(re.fullmatch(r"[A-Za-z]+idae", value))
    if name == "financing":
        return bool(re.fullmatch(rf"(?:or\s+)?[1-9]\d*\s*[x×]\s*{_MONEY}", value, re.I))
    if name == "shipping":
        return bool(re.fullmatch(rf"(?:free shipping|shipping\s*:?\s*(?:free|{_MONEY}))", value, re.I))
    return _text(value, 3 if name in {"description", "text", "summary", "quote"} else 2)


def _record(fields, schema):
    errors = [f"invalid_field:{field}" for field in SCHEMA_FIELDS[schema]
              if not _valid_field(field, fields.get(field))]
    for field in ("financing", "shipping"):
        if field in fields and not _valid_field(field, fields[field]):
            errors.append(f"invalid_field:{field}")
    if schema == "name_description" and _key(fields.get("name", "")) == _key(fields.get("description", "")):
        errors.append("description_repeats_name")
    return {"fields": fields, "valid": not errors, "errors": errors}


def _column_record(headers, values, schema):
    """Apply the same field and column rules to owned Markdown or HTML rows."""
    record = _record(dict(zip(headers, values)), schema)
    if len(values) != len(headers) or len(headers) != len(set(headers)):
        record["valid"] = False
        record["errors"].append("invalid_table_columns")
    return record


def _html_record_tables(content, schema, *, allow_closed_title_cards=False):
    """Read native cell ownership before any line/card projection."""
    from .dynamic_lists import scoped_html
    from .structured_tables import _native_cell_text, _spanning_rows, _visible_markdown_scalar

    soup = scoped_html(content)
    tables = list(soup.find_all('table'))
    if not tables:
        return None
    required = set(SCHEMA_FIELDS[schema])
    blocked = _spanning_rows(soup)
    records = []
    for table in tables:
        headers = None
        rows = [(row, row.find_all(['th', 'td'], recursive=False))
                for row in table.find_all('tr') if row.find_parent('table') is table]
        def header_row(row, cells):
            return bool(cells) and (all(cell.name == 'th' for cell in cells)
                or (row.find_parent(['thead', 'table']) or table).name == 'thead')
        has_header = any(header_row(row, cells) for row, cells in rows)
        for row, cells in rows:
            if not cells:
                continue
            values = [_visible_markdown_scalar(_native_cell_text(cell, table)) for cell in cells]
            if header_row(row, cells):
                labels = [_key(value).replace('review count', 'reviews') for value in values]
                headers = labels if required <= set(labels) and id(row) not in blocked else None
                continue
            if headers is not None:
                record = _column_record(headers, values, schema)
                if id(row) in blocked:
                    record['valid'] = False
                    record['errors'].append('invalid_table_columns')
                records.append(record)
            elif not has_header and id(row) not in blocked:
                # Headerless layout tables can contain existing native cards.
                # Keep each cell isolated and let the established literal card
                # parser recognize it without interpreting cell text as markup.
                for cell in cells:
                    card = scoped_html(str(cell))
                    for nested in card.find_all('table'):
                        nested.decompose()
                    descriptions = (_html_list_descriptions(str(card),
                        allow_closed_title_cards=allow_closed_title_cards)
                        if schema == 'name_description' else None)
                    records.extend(descriptions if descriptions is not None else _parse_records(
                        _visible(str(card), schema, representation='html'), schema, literal=True,
                        allow_closed_title_cards=allow_closed_title_cards))
    # Tables own their rows, including incomplete or unqualified rows. Their
    # flattened text cannot supply an independent card or another table's data.
    for table in reversed(tables):
        if table.parent is not None:
            table.decompose()
    return records, str(soup)


def _table_records(lines, schema):
    """Explicit Markdown table headers own their columns, including empty cells."""
    from .structured_tables import _pipe_cells, _table_header_at
    from .article_text import mask_markdown_inline_code

    masked = [mask_markdown_inline_code(line) for line in lines]
    records, occupied = [], set()
    required = set(SCHEMA_FIELDS[schema])
    for i in range(len(lines) - 1):
        if i in occupied or "|" not in lines[i]:
            continue
        headers = [_key(masked[i][1](_render_inline_scalar(x, pipe_cell=True))).replace("review count", "reviews")
                   for x in _pipe_cells(masked[i][0])]
        separator = _pipe_cells(masked[i + 1][0])
        if (not required <= set(headers) or len(separator) != len(headers)
                or not all(re.fullmatch(r"\s*:?-{3,}:?\s*", x) for x in separator)):
            continue
        occupied.update((i, i + 1))
        j = i + 2
        while j < len(lines) and "|" in lines[j] and lines[j].strip():
            if _table_header_at(lines, j):
                break
            values = [masked[j][1](_render_inline_scalar(x, pipe_cell=True)).strip()
                      for x in _pipe_cells(masked[j][0])]
            records.append(_column_record(headers, values, schema))
            occupied.add(j)
            j += 1
    return records, occupied


def _product_start(lines, index, schema, levels):
    """Recognize the same supported title/price layouts at every card boundary."""
    if re.match(r"description\s*:", lines[index], re.I):
        return None
    current = _field(lines[index], "price")
    price_first = bool(re.fullmatch(_MONEY, current, re.I))
    inline = None if price_first else re.fullmatch(rf"(.+?)\s*\|?\s+({_MONEY})", lines[index], re.I)
    # In a catalog, heading price then a title heading no deeper than it is its own card;
    # a plain line above is not its title. A deeper heading is the description.
    next_price = (index + 1 < len(lines)
                  and not (schema == "catalog_product"
                           and levels[index] is None and levels[index + 1] is not None
                           and index + 2 < len(lines) and levels[index + 2] is not None
                           and levels[index + 2] <= levels[index + 1])
                  and re.fullmatch(_MONEY, _field(lines[index + 1], "price"), re.I))
    if not price_first and not inline and not next_price:
        return None
    title_field = "title" if schema == "catalog_product" else "name"
    if inline:
        return {title_field: _field(inline[1].rstrip(" |"), title_field), "price": inline[2]}, index + 1
    if price_first:
        fields = {"price": current}
        if schema == "catalog_product" and index + 1 < len(lines):
            following = lines[index + 1]
            if (not re.fullmatch(_MONEY, _field(following, 'price'), re.I)
                    and not re.fullmatch(rf"(.+?)\s*\|?\s+({_MONEY})", following, re.I)):
                fields[title_field] = _field(following, title_field)
                return fields, index + 2
        return fields, index + 1
    return {title_field: _field(lines[index], title_field),
            "price": _field(lines[index + 1], "price")}, index + 2


def _products(lines, schema, *, literal=False):
    raw_lines = [line for line in lines if line.strip()]
    heading_pattern = (r'^' + _HTML_HEADING + r'([1-6])' + _HTML_HEADING + r'\s+'
                       if literal else r'^ {0,3}(#{1,6})[ \t]+')
    headings = [re.match(heading_pattern, line) for line in raw_lines]
    levels = [None if heading is None else int(heading[1]) if literal else len(heading[1])
              for heading in headings]
    lines = [_line(line, literal=literal) for line in raw_lines]
    records = []
    claimed_shipping = set()
    i = 0
    while i < len(lines):
        start = _product_start(lines, i, schema, levels)
        if start is None:
            i += 1
            continue
        fields, end = start
        if schema == "catalog_product":
            level = next((value for value in levels[i:end] if value is not None), 6)
            description = []
            while end < len(lines):
                if levels[end] is not None and levels[end] <= level:
                    break
                line = lines[end]
                count = _field(line, 'reviews')
                labeled = re.match(rf"^(?:{_LABELS['reviews']})\s*:", line, re.I)
                if (_valid_field('reviews', count)
                        and (labeled or re.fullmatch(r"\d+\s*reviews?", line, re.I))):
                    fields["reviews"] = re.match(r"\d+", count)[0]
                    end += 1
                    break
                # Stop before any subsequent card, rather than borrowing its
                # description or review count to repair this card.
                if _product_start(lines, end, schema, levels) is not None:
                    break
                description.append(_field(line, "description"))
                end += 1
            fields["description"] = " ".join(description)
        else:
            if i and i-1 not in claimed_shipping and re.fullmatch(r'free shipping|shipping\s*:\s*free', lines[i-1], re.I):
                fields['shipping'] = lines[i-1]
                claimed_shipping.add(i-1)
            while end < len(lines) and re.match(r"(?:or(?=\s|\d)|financing\s*:|free shipping|shipping\s*:)", lines[end], re.I):
                field = "shipping" if "shipping" in lines[end].casefold() else "financing"
                fields[field] = re.sub(r'^or(?=\d)', 'or ', _field(lines[end], field)) if field == "financing" else lines[end]
                if field == 'shipping':
                    claimed_shipping.add(end)
                end += 1
        records.append(_record(fields, schema))
        i = end
    return records


def _product_records(text, schema, *, literal=False):
    """Parse product cards without erasing explicit record ownership.

    A blank-line block is a record boundary only when every nonempty block is
    independently a single complete card.  This preserves extractors that put
    an optional badge before the product name, while retaining the permissive
    line-layout parser for prose/headings that use blank lines decoratively.
    """
    records = []
    for marked_block in text.split(_CARD_BOUNDARY):
        blocks = [block for block in re.split(r"\n\s*\n", marked_block)
                  if block.strip()]
        # A separate prose preface cannot own a price-first catalog card.
        # Require price and title in the same following block before trimming.
        if schema == "catalog_product":
            while (len(blocks) > 1 and not _products(blocks[0].splitlines(), schema, literal=literal)
                   and re.match(rf"^{_MONEY}\n[^\n]+", blocks[1], re.I)):
                blocks.pop(0)
            marked_block = "\n\n".join(blocks)
        parsed = [
            _products(block.splitlines(), schema, literal=literal)
            for block in blocks
        ]
        if (len(parsed) > 1
                and all(len(block_records) == 1 and block_records[0]["valid"]
                        for block_records in parsed)):
            for block_records in parsed:
                records.extend(block_records)
        else:
            records.extend(_products(
                marked_block.splitlines(), schema, literal=literal
            ))
    return records


def _quotes(lines):
    records, i = [], 0
    while i < len(lines):
        line = lines[i]
        if not re.match(r'[“"‘]|quote\s*:', line, re.I):
            i += 1
            continue
        parts = [line]
        j = i + 1
        while j < len(lines) and not re.match(r'[“"‘]|quote\s*:', lines[j], re.I):
            if re.match(r'(?:#{1,6}\s*)?(?:loading[. …]*|quotes by\s*:|made with\b|'
                        r'(?:next|previous)(?:\s+(?:page\s*)?\d+)?\s*[→←»«]?\s*$|'
                        r'page\s+\d+(?:\s+of\s+\d+)?\s*$)',
                        lines[j], re.I):
                break
            parts.append(lines[j])
            j += 1
        text = " ".join(parts)
        quote_text = re.sub(r'^quote\s*:\s*', '', text, flags=re.I)
        closing = {'“': '”', '"': '"', '‘': '’'}.get(quote_text[:1])
        suffix = r'\s*(?:[-–—]\s*)?(?:by\s*|author\s*:\s*)(.+?)\s*tags\s*:\s*(.+)'
        match = (re.fullmatch(re.escape(quote_text[0]) + r'(.*)' + re.escape(closing) + suffix, quote_text, re.I)
                 if closing else re.fullmatch(r'quote\s*:\s*(.+)\s+' + suffix, text, re.I))
        fields = {"quote": match[1].strip(), "author": match[2], "tags": match[3]} if match else {}
        if fields:
            fields["author"] = re.sub(r"\s*\(about\)\s*$", "", fields["author"], flags=re.I)
        records.append(_record(fields, "quote"))
        i = j
    return records


def _reviews(lines, *, literal=False):
    lines = [line.strip() for line in lines if line.strip()]
    records, i, context_level = [], 0, 6
    while i < len(lines):
        pattern = (_HTML_HEADING + r'([1-6])' + _HTML_HEADING + r'\s+'
                   if literal else r'(#{1,6})\s+')
        heading = re.match(pattern, lines[i])
        if heading:
            context_level = int(heading[1]) if literal else len(heading[1])
        match = re.match(r"(?:date\s*:\s*)?(\d{4}-\d{2}-\d{2})(?:\s+(.+))?$", _line(lines[i], literal=literal), re.I)
        if not match:
            i += 1
            continue
        fields = {"date": match[1]}
        text = [match[2]] if match[2] else []
        j = i + 1
        while j < len(lines):
            following = _line(lines[j], literal=literal)
            heading = re.match(pattern, lines[j])
            if (re.match(r"(?:date\s*:\s*)?\d{4}-\d{2}-\d{2}", following, re.I)
                    or (heading and (int(heading[1]) if literal else len(heading[1])) <= context_level)):
                break
            text.append(_field(following, "text"))
            j += 1
        fields["text"] = " ".join(text)
        records.append(_record(fields, "review"))
        i = j
    return records


def _operations(lines):
    records, i = [], 0
    while i < len(lines):
        # Some native Markdown serializers emit METHOD[/path](target)Summary;
        # visible-link normalization leaves METHOD/path Summary without a gap.
        match = re.match(rf"(?:method\s*:\s*)?({_METHOD})(?:(?:\s+|(?=/))(.+))?$", lines[i])
        if not match:
            i += 1
            continue
        fields = {"method": match[1]}
        rest = [match[2]] if match[2] else []
        j = i + 1
        while j < len(lines) and not re.match(rf"(?:method\s*:\s*)?{_METHOD}\b", lines[j]):
            rest.append(lines[j])
            j += 1
        if rest:
            path = re.fullmatch(r"(?:path\s*:\s*)?(/\S*)(?:\s+(.+))?", rest[0], re.I)
            if path:
                fields["path"] = path[1]
                summary = ([path[2]] if path[2] else []) + rest[1:]
                fields["summary"] = " ".join(_field(s, "summary") for s in summary)
        records.append(_record(fields, "api_operation"))
        i = j
    return records


def _descriptions(text, *, literal=False):
    records = []
    if _CARD_BOUNDARY in text:
        # A marker comes from an image-only bullet immediately before the
        # visible linked title. It is a stronger ownership boundary than blank
        # lines, which may legitimately occur inside a card description.
        prefix, *marked = text.split(_CARD_BOUNDARY)
        records.extend(_descriptions(prefix, literal=literal))
        for block in marked:
            following = None if literal else re.search(r"(?m)^\s*(?:[-*+]|\d+[.)])[ \t]+\S.*$", block)
            end = following.start() if following else len(block)
            lines = [_line(line, literal=literal) for line in block[:end].splitlines() if line.strip()]
            if len(lines) >= 2:
                records.append(_record({"name": _field(lines[0], "name"),
                    "description": " ".join(_field(line, "description") for line in lines[1:])},
                    "name_description"))
            if following:
                records.extend(_descriptions(block[end:], literal=literal))
        return records
    # A whitespace-only line is a blank line for the setext check below.
    text = re.sub(r'(?m)^[ \t]+$', '', text)
    bullets = [] if literal else list(re.finditer(r"(?m)^\s*(?:[-*+]|\d+[.)])[ \t]+\S.*$", text))
    if bullets:
        # Fences are consumed so their code lines cannot declare a section. A
        # setext title needs a blank line before it; otherwise it is a lazy
        # continuation of the item and the underline is not a heading.
        sections = [match for match in re.finditer(
            r'(?ms)^ {0,3}(`{3,}(?=[^`\n]*$)|~{3,}).*?(?:^ {0,3}\1[`~]*[ \t]*$|\Z)'
            r'|^(?:(?= {0,3}#{1,6}(?:[ \t]+\S|[ \t]*$))'
            r'|(?<![^\n]\n)(?= {0,3}\S[^\n]*\n {0,3}(?:=+|-+)[ \t]*$))( {0,3})', text)
            if match[2] is not None]
        for index, bullet in enumerate(bullets):
            end = bullets[index + 1].start() if index + 1 < len(bullets) else len(text)
            # A heading left of the item's content column may be a later section or
            # the card's own title. It ends the description when nothing follows it
            # before the next bullet, or after the last bullet unless it repeats the
            # bullet's title. A heading indented into the item remains a continuation.
            column = re.match(r'\s*(?:[-*+]|\d+[.)])[ \t]+', bullet[0].rsplit('\n', 1)[-1].expandtabs(4)).end()
            section = next((section.start() for section in sections
                            if bullet.end() <= section.start() < end and len(section[2]) < column), None)
            if section is not None:
                heading, _, body = text[section:end].partition('\n')
                title = _line(heading)
                if not body.strip() or (index + 1 == len(bullets) and not (title and title in bullet[0])):
                    end = section
            lines = [_line(line, literal=literal) for line in text[bullet.start():end].splitlines() if line.strip()]
            records.append(_record({"name": _field(lines[0], "name"),
                "description": " ".join(_field(line, "description") for line in lines[1:])}, "name_description"))
        return records
    # Both Markdown headings and blank-delimited plain cards express ownership.
    pattern = (r'(?m)^' + _HTML_HEADING + r'([1-6])' + _HTML_HEADING + r'\s+(.+)$'
               if literal else r'(?m)^(#{1,6})\s+(.+)$')
    headings = list(re.finditer(pattern, text))
    if headings:
        # Repeated deepest-level headings are cards; enclosing page/section
        # headings are not records. Blank lines within a card do not erase its
        # ownership boundary.
        depth = max(int(h[1]) if literal else len(h[1]) for h in headings)
        for index, heading in enumerate(headings):
            if (int(heading[1]) if literal else len(heading[1])) != depth:
                continue
            end = headings[index + 1].start() if index + 1 < len(headings) else len(text)
            body = [_line(line, literal=literal) for line in text[heading.end():end].splitlines() if line.strip()]
            records.append(_record({"name": _field(_line(heading[2], literal=literal), "name"),
                "description": " ".join(_field(line, "description") for line in body)}, "name_description"))
        return records
    blocks = re.split(r"\n\s*\n", text)
    for block in blocks:
        lines = [_line(line, literal=literal) for line in block.splitlines() if line.strip()]
        if len(lines) < 2:
            continue
        fields = {"name": _field(lines[0], "name"),
                  "description": " ".join(_field(line, "description") for line in lines[1:])}
        records.append(_record(fields, "name_description"))
    return records


def _html_list_descriptions(content, *, allow_closed_title_cards=False):
    """Preserve native leaf-list ownership before flattening visible HTML."""
    from .dynamic_lists import visible_html_text
    from .document_structure import _content_root
    from .reference_quality import is_html_document_or_fragment
    if not is_html_document_or_fragment(content):
        return None
    from .dynamic_lists import scoped_html
    soup = scoped_html(content)
    # Keep record-local titles before the document helper removes page chrome.
    for header in soup.select('article header,section header,li header'):
        header.unwrap()
    root = _content_root(str(soup), single_article=False)
    items = [node for node in root.select('li') if not node.select('li')]
    records = []
    for item in items:
        lines = visible_html_text(str(item)).splitlines()
        if len(lines) >= 2:
            records.append(_record({'name': _field(lines[0], 'name'),
                                    'description': ' '.join(_field(line, 'description') for line in lines[1:])},
                                   'name_description'))
    # List ownership includes incomplete items and enclosing list labels.
    # Remove those subtrees before parsing independently owned non-list cards.
    for item in list(root.select('li')):
        if item.parent is not None:
            item.decompose()
    def bounded_records(container):
        original = str(container)
        children = [child for child in container.select('section')
                    if child.find_parent(['section', 'article']) is container]
        for child in children:
            child.extract()
        # An empty wrapper or a heading over sibling sections is a group.
        # A record title with one presentation section retains its own prose.
        own_text = visible_html_text(str(container)).strip()
        headings = '\n'.join(visible_html_text(str(h)) for h in container.select('h1,h2,h3,h4,h5,h6'))
        if children and (not own_text or own_text == headings):
            child_records = [record for child in children for record in bounded_records(child)]
            if not own_text or len(children) > 1 or len(child_records) > 1:
                return child_records
        return _parse_records(_visible(original, 'name_description', representation='html'), 'name_description',
                              allow_closed_title_cards=allow_closed_title_cards, literal=True)
    # Consume nested articles first; retain presentation sections with their title.
    for article in reversed(list(root.select('article'))):
        if article.parent is not None:
            records.extend(bounded_records(article))
            article.decompose()
    for section in list(root.select('section')):
        if section.parent is not None:
            records.extend(bounded_records(section))
            section.decompose()
    return records + _parse_records(_visible(str(root), 'name_description', representation='html'), 'name_description',
                                    allow_closed_title_cards=allow_closed_title_cards, literal=True)


def _closed_title_descriptions(text, *, line_delimited=False, literal=False):
    """Treat title-description-title triples as explicitly closed cards."""
    blocks = text.splitlines() if line_delimited else re.split(r"\n\s*\n", text)
    paragraphs = [block.strip() for block in blocks if block.strip()]
    records = []
    for first, description, repeated in zip(
        paragraphs, paragraphs[1:], paragraphs[2:], strict=False
    ):
        if any("\n" in value for value in (first, description, repeated)):
            continue
        name = _line(first, literal=literal)
        if _key(name) != _key(_line(repeated, literal=literal)):
            continue
        record = _record(
            {"name": name, "description": _line(description, literal=literal)},
            "name_description",
        )
        if record["valid"]:
            records.append(record)
    return records


def _media_closed_title_descriptions(text, *, literal=False):
    """Read title-description-title closures only inside image-owned blocks."""
    if _MEDIA_BOUNDARY not in text:
        return []
    return [
        record
        for block in text.split(_MEDIA_BOUNDARY)[1:]
        for record in _closed_title_descriptions(block, line_delimited=True, literal=literal)
    ]


def _earthquakes(text, *, literal=False):
    """Read contiguous magnitude/location/time/depth cards, never bag-of-fields.

    Four-line ordering is independently observed on the source list. Explicit
    Markdown card/link boundaries cannot be crossed to repair a missing field.
    Tables are handled separately with column ownership.
    """
    records = []
    for block in text.split(_CARD_BOUNDARY):
        lines = [_line(line, literal=literal) for line in block.splitlines() if line.strip()]
        # Inline custom elements can serialize location and timestamp on one
        # line. Split only a complete trailing timestamp, within this card.
        separated = []
        for line in lines:
            compact = re.fullmatch(
                r"(.+?)(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}"
                r"(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2}|\s*\(UTC(?:[+-]\d{2}:\d{2})?\))?)"
                r"\s+(-?\d+(?:\.\d+)?\s*km)", line, re.I,
            )
            if (compact and _text(compact[1])
                    and _valid_field("time", compact[2])
                    and _valid_field("depth", compact[3])):
                separated.extend((compact[1].strip(), compact[2], compact[3]))
                continue
            joined = re.fullmatch(
                r"(.+?)(\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}"
                r"(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2}|\s*\(UTC(?:[+-]\d{2}:\d{2})?\))?)",
                line,
            )
            if joined and _valid_field("time", joined[2]) and _text(joined[1]):
                separated.extend((joined[1].strip(), joined[2]))
            else:
                separated.append(line)
        lines = separated
        for i, line in enumerate(lines):
            magnitude = _field(line, "magnitude")
            if not _valid_field("magnitude", magnitude):
                continue
            fields = {"magnitude": magnitude}
            for offset, name in enumerate(("location", "time", "depth"), 1):
                if i + offset < len(lines):
                    fields[name] = _field(lines[i + offset], name)
            records.append(_record(fields, "earthquake"))
    return records


def _parse_records(visible, schema, *, allow_closed_title_cards=False, literal=False):
    raw_lines = visible.splitlines()
    records, occupied = ([], set()) if literal else _table_records(raw_lines, schema)
    remainder = "\n".join(line if i not in occupied else _CARD_BOUNDARY for i, line in enumerate(raw_lines))
    lines = [_line(line, literal=literal) for line in remainder.splitlines() if line.strip()]
    if schema == "earthquake":
        records.extend(_earthquakes(remainder, literal=literal))
    elif schema in {"product", "catalog_product"}:
        records.extend(_product_records(remainder, schema, literal=literal))
    elif schema == "quote":
        for block in remainder.split(_CARD_BOUNDARY):
            records.extend(_quotes([_line(line, literal=literal)
                                    for line in block.splitlines() if line.strip()]))
    elif schema == "review":
        for block in remainder.split(_CARD_BOUNDARY):
            records.extend(_reviews(block.splitlines(), literal=literal))
    elif schema == "api_operation":
        # Keep section boundaries until after parsing ownership. Otherwise the
        # last operation in each group absorbs the next heading, its overview,
        # or the models/footer section into its summary. A heading may itself
        # contain METHOD /path, so retain its text within the new section.
        boundary = (r'(?m)(?=^' + _HTML_HEADING + r'[1-6]' + _HTML_HEADING + r'\s+)'
                    if literal else r'(?m)(?=^[ \t]{0,3}#{1,6}[ \t]+)')
        sections = [section for block in remainder.split(_CARD_BOUNDARY)
                    for section in re.split(boundary, block)]
        for section in sections:
            records.extend(_operations([_line(line, literal=literal) for line in section.splitlines() if line.strip()]))
    elif schema == "name_description":
        # Scoped title-description-title cards and image-owned cards are two
        # ways to establish the same ownership boundary.  When the caller has
        # already bounded the target section, parsing both routes would emit
        # every image-backed card twice and reject an otherwise valid answer as
        # a duplicate-name list.
        if _MEDIA_BOUNDARY in remainder:
            closed = _media_closed_title_descriptions(remainder, literal=literal)
            if allow_closed_title_cards:
                closed += _closed_title_descriptions(remainder.split(_MEDIA_BOUNDARY, 1)[0], literal=literal)
        else:
            closed = _closed_title_descriptions(remainder, literal=literal) if allow_closed_title_cards else []
        # A page heading cannot own descriptions across separate media cards.
        weak = [record for record in _descriptions(remainder, literal=literal)
                if not any(_MEDIA_BOUNDARY in value
                           for value in record['fields'].values())]
        for record in weak:
            description = _key(record["fields"].get("description", ""))
            name = _key(record["fields"].get("name", ""))
            if any(
                name != _key(owner["fields"]["name"])
                and _key(owner["fields"]["name"]) in description
                and _key(owner["fields"]["description"]) in description
                for owner in closed
            ):
                record["valid"] = False
                record["errors"].append("description_contains_separate_owned_record")
        records.extend(weak)
        records.extend(closed)
    elif schema == "taxonomy_family":
        records.extend(_record({"family": _field(line, "family")}, schema) for line in lines
                       if re.fullmatch(r"(?:family\s*:\s*)?[A-Za-z]+idae", line, re.I))
    return records


def _non_record_visible(content, contract, records, base_url, *, representation=None):
    """Remove owned record links, without erasing identical navigation text."""
    if representation == 'plain_text':
        return None
    from .response_input import join_answer_url

    from .dynamic_lists import scoped_html, named_entry_identity, canonical_entry_path, visible_html_text
    from .reference_quality import markdown_links, visible_markdown_links, visible_mixed_html_text

    identities = {named_entry_identity(r['fields']['url'], contract)
                  for r in records if r.get('valid') and r.get('fields', {}).get('url')}

    def is_record(url):
        try:
            absolute = join_answer_url(base_url, url)
            if urlsplit(absolute).scheme not in {'http', 'https'} or not urlsplit(absolute).hostname:
                return False
            parts = urlsplit(absolute)
            return (named_entry_identity(absolute, contract) in identities
                    or (parts.hostname in contract['hosts']
                        and re.fullmatch(contract['entry_path'],
                                         canonical_entry_path(parts.path).rstrip('/')) is not None))
        except ValueError:
            return False

    from .reference_quality import is_html_document_or_fragment
    if is_html_document_or_fragment(content):
        from bs4 import BeautifulSoup
        soup = BeautifulSoup(content, 'html.parser')
        # A linked logo may supply page identity, but never an owned record
        # title. Preserve that label before the record parser removes image
        # metadata. Keeping the element's attributes until visibility cleanup
        # also excludes hidden images and their hidden ancestors.
        for image in soup.select('a[href] img[alt]'):
            owner = image.find_parent('a', href=True)
            try:
                target = urlsplit(join_answer_url(base_url, owner['href']))
            except ValueError:
                continue
            if target.scheme in {'http', 'https'} and target.hostname and not is_record(owner['href']):
                image.name = 'span'
                image.string = image['alt']
        soup = scoped_html(str(soup))
        for node in list(soup.select('a[href]')):
            if node.parent is None or node.attrs is None:
                continue
            if is_record(node['href']):
                node.decompose()
            else:
                # Separate distinct HTML link labels before flattening, just
                # as adjacent Markdown links keep their visible boundaries.
                node.insert_before(' ')
                node.insert_after(' ')
        return _visible(visible_html_text(str(soup)), normalized_html=True)
    links = markdown_links(content)
    if links:
        for link in reversed(links):
            if link.is_image or is_record(link.destination):
                content = content[:link.start] + ' ' + content[link.end:]
            else:
                try:
                    target = urlsplit(join_answer_url(base_url, link.destination))
                    web_link = target.scheme in {'http', 'https'} and bool(target.hostname)
                except ValueError:
                    web_link = False
                if not web_link:
                    content = content[:link.start] + _visible(link.label) + content[link.end:]
        # A linked logo can be the only reader-visible site identity. Preserve
        # its alt text for page-level anchors after owned record links are
        # removed. Record parsing still uses ``_visible`` above, so image alt
        # text cannot manufacture a named-link record title.
        return _visible(visible_markdown_links(visible_mixed_html_text(content), include_image_alt=True))
    # Plain text has no URL ownership; retain the conservative fallback.
    return None


# A label letter-spaced into single CJK characters. Words of two or more stay separate.
_LETTER_SPACED_CJK = re.compile(r'(?<![\u3400-\u9fff])[\u3400-\u9fff](?:[ \t]+[\u3400-\u9fff](?![\u3400-\u9fff]))+')


def _stable_literal_match(surface, value, registered_words=(), *, phrase_boundaries=False, representation=None, rendered=False, literal=False):
    """An atomic Latin label is a word, not part of another word.

    Native camel boundaries and fully covered chains of registered labels
    preserve adjacent UI text without accepting arbitrary word fragments.
    Longer declared text retains the existing exact-phrase containment rule.
    """
    from .metrics import WORD_CHARACTERS, fact_in_text, normalize_for_match as normalize, quoted_scalar_equivalence
    from .reference_quality import visible_markdown_links
    def normalize_for_match(text):
        if rendered and representation not in {'plain_text', 'html'}:
            text = quoted_scalar_equivalence(unicodedata.normalize('NFKC', text).casefold())
        text = normalize(text, representation='plain_text' if rendered or representation == 'html' else representation)
        return _LETTER_SPACED_CJK.sub(lambda match: re.sub(r'[ \t]+', '', match[0]), text)
    if not rendered and representation not in {'plain_text', 'html'}:
        surface = visible_markdown_links(surface)
    surface = re.sub(r'(?<=[^\W\d_])(?=\d+\s+of\s+\d+\b)', ' ', surface)
    expected_value = value if literal else _visible(value, representation='plain_text' if representation == 'html' else representation)
    expected = normalize_for_match(expected_value)
    # Only an explicit heading with a trailing fragment URL can supply this
    # additional surface; ordinary word suffixes remain significant.
    if not rendered and representation not in {'plain_text', 'html'}:
        surface = re.sub(r'(?m)^(#{1,6}\s+[^\n]+?)https?://[^\s]+#[^\s]*[ \t]*$',
                         r'\1', surface)
    folded = normalize_for_match(surface)
    from .reference_quality import separate_case_labels as camel
    alphabetic_boundaries = (not re.fullmatch(r'[A-Za-z]+', value)
                             and (phrase_boundaries or bool(re.search(r'[^\x00-\x7f]', value))))
    surfaces = [(folded, expected)]
    if alphabetic_boundaries:
        surfaces.append((normalize_for_match(camel(surface)), normalize_for_match(camel(expected_value))))
    # Native serializers can join an all-caps preceding label to a complete
    # notice. Apply the same case boundary used for atomic labels before the
    # Unicode phrase gate, while retaining complete-word matching on both ends.
    if not any(fact_in_text(text, phrase, latin_boundaries=alphabetic_boundaries)
               for text, phrase in surfaces):
        return False
    # Arabic labels are whitespace-delimited words too. A longer inflected
    # word cannot supply an independently registered ordinary navigation label.
    if value.isalpha() and re.fullmatch(r'[\u0600-\u06ff]+', value):
        return bool(re.search(r'(?<![^\W\d_])' + re.escape(expected)
                              + r'(?![^\W\d_])', folded))
    if not re.fullmatch(r"[A-Za-z]+", value):
        return True
    boundary = lambda needle, text: bool(re.search(
        rf"(?<![{WORD_CHARACTERS}])" + re.escape(needle) + rf"(?![{WORD_CHARACTERS}])", text))
    if boundary(expected, folded):
        return True
    if boundary(normalize_for_match(camel(value)),
                normalize_for_match(camel(surface))):
        return True
    labels = {normalize_for_match(word) for word in registered_words}
    for token in set(re.findall(rf"[{WORD_CHARACTERS}]+", folded)):
        if expected not in token:
            continue
        states = {(0, False)}
        for position in range(len(token)):
            for start, found in tuple(states):
                if start != position:
                    continue
                for label in labels:
                    if token.startswith(label, position):
                        states.add((position + len(label), found or label == expected))
        if (len(token), True) in states:
            return True
    return False


def _record_field_key(value):
    """Normalize rendered field text without interpreting Markdown syntax."""
    from .reference_quality import clean_reference_text
    import unicodedata
    value = value.translate(str.maketrans({"’": "'", "‘": "'", "“": '"', "”": '"',
                                         "–": "-", "—": "-", "−": "-"}))
    # Frozen record hashes use literal whitespace, not evolving prose matching.
    value = clean_reference_text(unicodedata.normalize('NFKC', value).casefold())
    value = re.sub(r'\s+', ' ', value).strip('“”"‘’ ')
    return re.sub(r'([$£€¥]) +(?=[+-]?\d)', r'\1', value)


def source_record_hash(fields, names, schema):
    """Hash already-extracted literal fields; render authored markup before binding.

    Keep the canonical typography, field ordering and JSON/SHA-256 encoding used
    by frozen receipts. None shares the existing omitted-field empty string.
    Literal Markdown is data, never an alternate spelling.
    """
    import json
    values = [fields.get(name, '') for name in names]
    values = [_record_field_key('' if value is None else value) for value in values]
    if schema == 'name_description':
        values = [value.replace(' ', '') for value in values]
    return hashlib.sha256(json.dumps(values, ensure_ascii=False, separators=(',', ':')).encode()).hexdigest()


def _anchor_matches(anchor, records, visible, schema, non_record_visible=None,
                    anchor_visible=None, registered_words=(), *, representation=None, contract=None):
    from .metrics import normalize_for_match as normalize
    normalize_for_match = lambda text: normalize(
        text, representation='plain_text' if representation == 'html' else representation)
    if anchor.get('match_scope') == 'non_record_lines':
        if non_record_visible is not None:
            return [value for value in anchor['alternatives']
                    if _stable_literal_match(non_record_visible, value, registered_words, representation=representation)]
        # A known record label cannot stand in for page context. This is a
        # conservative ownership rule, not a claim of recovered DOM ownership.
        # Flattened page output can put headings and records on one physical
        # line, so mask parsed record titles instead of discarding that line.
        titles = [normalize_for_match(r['fields']['title']) for r in records if r.get('valid') and r.get('fields',{}).get('title')]
        folded = normalize_for_match(anchor_visible if anchor_visible is not None else visible)
        for title in sorted(set(titles), key=len, reverse=True):
            folded = folded.replace(title, ' ')
        return [value for value in anchor["alternatives"]
                if _stable_literal_match(folded, value, registered_words, representation=representation)]
    owned = []
    if schema == 'table_row' and contract:
        from .structured_tables import _canonical, _table_header_at, parse_table_records
        matched = []
        for value in anchor['alternatives']:
            from .reference_quality import is_html_document_or_fragment
            reference_representation = representation
            if is_html_document_or_fragment(value):
                reference_representation = 'html'
            elif representation == 'html':
                lines = value.splitlines()
                reference_representation = ('markdown' if any(
                    _table_header_at(lines, index) for index in range(len(lines))) else 'plain_text')
            expected = parse_table_records(
                value, _visible(value, schema, representation=reference_representation), contract,
                representation=reference_representation)
            if len(expected) == 1 and expected[0]['valid']:
                passed = any(record['valid'] and all(
                    _canonical(record['fields'][column['name']], column)
                    == _canonical(expected[0]['fields'][column['name']], column)
                    for column in contract['table_columns']) for record in records)
            else:
                passed = _stable_literal_match(anchor_visible if anchor_visible is not None else visible,
                                               value, registered_words, representation=representation)
            if passed:
                matched.append(value)
        return matched
    if schema != "named_link" and SCHEMA_FIELDS[schema]:
        for value in anchor["alternatives"]:
            from .reference_quality import is_html_document_or_fragment
            reference_html = is_html_document_or_fragment(value)
            reference_literal = representation in {'plain_text', 'html'} or reference_html
            reference_representation = 'html' if reference_html else representation
            reference_visible = _visible(value, schema, representation=reference_representation)
            native_tables = (_html_record_tables(value, schema)
                             if reference_html and schema != 'linked_product' else None)
            expected = (native_tables[0] + _parse_records(
                _visible(native_tables[1], schema, representation='html'), schema, literal=True)
                if native_tables is not None else
                _parse_records(reference_visible, schema, literal=reference_literal))
            if reference_literal and not expected:
                # A declared table alternative has its own column mapping,
                # independent of the answer's literal line representation.
                table_visible = _visible(value, schema, representation='markdown')
                expected, _ = _table_records(table_visible.splitlines(), schema)
            if len(expected) == 1 and expected[0]["valid"]:
                owned.append((value, expected[0]["fields"]))
    if owned:
        # Structured anchors compare complete owned fields, not a flattened
        # document substring. Table columns and Markdown links are equivalent.
        def field_matches(record, fields, field):
            actual = _record_field_key(record["fields"][field])
            expected = _record_field_key(fields[field])
            # Name/description card ownership provides a structural boundary
            # even if an inline DOM separator is omitted. Quote tags are a
            # set-like field, so their separators must remain observable.
            compact = schema == "name_description"
            return (actual.replace(" ", "") == expected.replace(" ", "")
                    if compact else actual == expected)
        matched = {value for value, fields in owned if any(record["valid"] and all(
            field_matches(record, fields, field)
            for field in SCHEMA_FIELDS[schema]) for record in records)}
        structured_values = {value for value, _ in owned}
        matched.update(value for value in anchor['alternatives'] if value not in structured_values
                       and _stable_literal_match(anchor_visible if anchor_visible is not None else visible,
                                                 value, registered_words, representation=representation))
        return [value for value in anchor['alternatives'] if value in matched]
    return [value for value in anchor["alternatives"]
            if _stable_literal_match(anchor_visible if anchor_visible is not None else visible,
                                     value, registered_words, representation=representation,
                                     rendered=schema in {'body_text', 'page_identity'})]


def _escaped_prose_surface(content, schema):
    """Render an answer whose backslash escapes emit literal punctuation, without its code.

    A reference is observed page text, so its Markdown delimiters are characters.
    Code keeps its own meaning and never supplies this surface, and an answer with
    any tag-like text is left to the visible HTML projection. The mask is a caseless
    word letter, so it adds no word or case boundary the answer does not have.
    """
    escape = re.compile(r'\\[!-/:-@\[-`{-~]')
    if not escape.search(content) or re.search(r'<[A-Za-z/!?]', content):
        return ''
    from .reference_quality import markdown_literal_spans
    prose = content
    for start, end in reversed(markdown_literal_spans(content)):
        prose = prose[:start] + '\u01bb' + prose[end:]
    return _visible(prose, schema) if escape.search(prose) else ''


def _native_link_label_surface(content, exclusions):
    """Keep visible native link labels separate without splitting ordinary words."""
    from .dynamic_lists import scoped_html, visible_html_text
    from .reference_quality import is_html_document_or_fragment
    if not is_html_document_or_fragment(content):
        return None
    soup = scoped_html(content)
    for selector in exclusions.get('html_selectors', []):
        for node in soup.select(selector):
            node.decompose()
    labels = '\n'.join(visible_html_text(str(node)) for node in soup.select('a[href]'))
    for pattern in exclusions.get('text_patterns', []):
        labels = re.sub(pattern, ' ', labels, flags=re.I | re.M)
    return labels


def _exclude_html_record_text(content, patterns):
    """Mask rendered text matches while retaining the DOM record owners."""
    from .dynamic_lists import scoped_html, visible_html_text
    from bs4 import NavigableString

    soup = scoped_html(content)
    nodes = list(soup.strings)
    prefix = '\ue028anchor-text'
    while prefix in content or prefix in unescape(content):
        prefix += '-'
    originals = [str(node) for node in nodes]
    source_spans = [[(offset, offset + 1) for offset in range(len(str(node)))]
                    if node.find_parent(['pre', 'textarea']) else
                    [(match.start(), match.end()) for match in re.finditer(r'\s+|\S', str(node))]
                    for node in nodes]
    values = [str(node) if node.find_parent(['pre', 'textarea']) else re.sub(r'\s+', ' ', str(node))
              for node in nodes]
    tokens = {prefix + str(index) + '\ue029': index for index in range(len(nodes))}
    token_nodes = []
    for node, token in zip(nodes, tokens):
        replacement = NavigableString(token)
        node.replace_with(replacement)
        token_nodes.append(replacement)
    layout = visible_html_text(str(soup))
    parts, positions, cursor = [], [], 0
    for match in re.finditer(re.escape(prefix) + r'\d+\ue029', layout):
        parts.append(layout[cursor:match.start()])
        positions.extend([None] * (match.start() - cursor))
        index = tokens[match[0]]
        parts.append(values[index])
        positions.extend((index, offset) for offset in range(len(values[index])))
        cursor = match.end()
    parts.append(layout[cursor:])
    positions.extend([None] * (len(layout) - cursor))
    expanded = ''.join(parts)
    # Apply the existing visible-text owner's final line trimming after tokens
    # have been expanded, keeping the mapping back to actual text nodes.
    pieces, mapped = [], []
    for line in re.finditer(r'[^\n]+', expanded):
        start = line.start() + len(line[0]) - len(line[0].lstrip())
        end = line.start() + len(line[0].rstrip())
        if start < end:
            if pieces:
                pieces.append('\n')
                mapped.append(None)
            pieces.append(expanded[start:end])
            mapped.extend(positions[start:end])
    surface = ''.join(pieces)
    removed = set()
    for pattern in patterns:
        for match in reversed(list(re.finditer(pattern, surface, re.I | re.M))):
            removed.update(position for position in mapped[match.start():match.end()] if position is not None)
            surface = surface[:match.start()] + ' ' + surface[match.end():]
            mapped[match.start():match.end()] = [None]
    if not removed:
        return content
    for index, node in enumerate(token_nodes):
        offsets = {offset for position, (start, end) in enumerate(source_spans[index])
                   if (index, position) in removed for offset in range(start, end)}
        node.replace_with(''.join(' ' if offset in offsets else char
                                  for offset, char in enumerate(originals[index])))
    return str(soup)


def _structured_record_view(content, contract, *, representation=None,
                            allow_closed_title_cards=False, visible=None, text_patterns=()):
    """Use the same format and record owners for count and anchor scopes."""
    from .reference_quality import is_html_document_or_fragment

    schema = contract['schema']
    native = representation == 'html' or (representation != 'plain_text' and is_html_document_or_fragment(content))
    if native and text_patterns:
        content = _exclude_html_record_text(content, text_patterns)
        visible = None
    native_tables = (_html_record_tables(content, schema,
                     allow_closed_title_cards=allow_closed_title_cards)
                     if native and schema not in {'body_text', 'page_identity', 'table_row'} else None)
    record_content = native_tables[1] if native_tables is not None else content
    if visible is None or native_tables is not None:
        visible = _visible(record_content, schema, representation='html' if native else representation)
    if not native:
        for pattern in text_patterns:
            visible = re.sub(pattern, _CARD_BOUNDARY, visible, flags=re.I | re.M)
    if schema == 'table_row':
        from .structured_tables import parse_table_records
        records = parse_table_records(visible if text_patterns and not native else content, visible,
                                      contract, representation=representation)
    elif representation != 'plain_text' and schema == 'name_description' and (html_records := _html_list_descriptions(
            record_content, allow_closed_title_cards=allow_closed_title_cards)) is not None:
        records = html_records
    else:
        records = ([] if schema in {'body_text', 'page_identity'} else _parse_records(
            visible, schema, allow_closed_title_cards=allow_closed_title_cards,
            literal=native or representation == 'plain_text'))
    return (native_tables[0] if native_tables is not None else []) + records


def _validate_visible_control(control):
    """Validate the narrowly opted-in ordinary control attribute surface."""
    if (not isinstance(control, dict) or set(control) != {"selector", "attribute"}
            or not isinstance(control["selector"], str) or not control["selector"].strip()
            or not isinstance(control["attribute"], str)
            or control["attribute"] not in {"placeholder", "value"}):
        raise ValueError("Invalid visible control declaration")
    try:
        soupsieve.compile(control["selector"])
    except (soupsieve.SelectorSyntaxError, NotImplementedError) as error:
        raise ValueError("Invalid visible control selector") from error


def _ordinary_control_text(soup, control):
    """Read only declared rendered-label attributes, never accessibility metadata.

    Explicit hidden nodes are removed by scoped_html. External stylesheet
    visibility must be established by the independent source review.
    """
    parts = []
    for node in soup.select(control["selector"]):
        kind = str(node.get("type", "text")).lower()
        attribute = control["attribute"]
        # A populated control does not display its placeholder, even when the
        # current value contains only whitespace.
        if attribute == "placeholder" and (
                node.get("value", "") if node.name == "input" else node.get_text()):
            continue
        permitted = (
            attribute == "placeholder" and (
                node.name == "textarea" or (node.name == "input" and kind in {
                    "text", "search", "email", "url", "tel", "number"}))) or (
            attribute == "value" and node.name == "input"
            and kind in {"button", "submit", "reset"})
        if permitted and isinstance(node.get(attribute), str):
            parts.append(node[attribute])
    return "\n".join(parts)


def _compile_stable_regex(pattern, flags=0):
    try:
        return re.compile(pattern, flags)
    except re.error as error:
        raise ValueError("Invalid stable acceptance regex") from error


def _compile_stable_selector(selector):
    try:
        return soupsieve.compile(selector)
    except (soupsieve.SelectorSyntaxError, NotImplementedError) as error:
        raise ValueError("Invalid stable acceptance selector") from error


def validate_acceptance_contract(contract: dict) -> None:
    """Fail closed before a caller chooses stable scoring instead of snapshots."""
    if (not isinstance(contract, dict) or type(contract.get("version")) is not int
            or contract.get("version") != 1 or not isinstance(contract.get("schema"), str)
            or contract.get("schema") not in SCHEMA_FIELDS):
        raise ValueError("Unsupported stable acceptance contract")
    if set(contract) - ACCEPTANCE_CONTRACT_FIELDS:
        raise ValueError("Unknown stable acceptance contract fields")
    if (not isinstance(contract.get("stable_text_role", "required"), str)
            or contract.get("stable_text_role", "required") not in {"required", "diagnostic"}):
        raise ValueError("Invalid stable text role")
    if (not isinstance(contract.get('content_scope', 'legacy'), str)
            or contract.get('content_scope', 'legacy') not in {'legacy', 'target_content', 'full_page_stable_coverage'}):
        raise ValueError('Invalid stable content scope')
    exclusions = contract.get('text_exclusions', {})
    if (not isinstance(exclusions, dict)
            or set(exclusions) - {'html_selectors', 'text_patterns'}):
        raise ValueError('Invalid stable text exclusions')
    for key, values in exclusions.items():
        if (not isinstance(values, list)
                or any(not isinstance(value, str) or not value.strip() for value in values)):
            raise ValueError('Invalid stable text exclusions')
        for value in values:
            if key == 'text_patterns':
                _compile_stable_regex(value)
            else:
                _compile_stable_selector(value)
    if contract['schema'] == 'table_row':
        from .structured_tables import validate_table
        validate_table(contract)
    elif any(key in contract for key in ('table_columns', 'record_identity_fields', 'table_scope')):
        raise ValueError('Table declarations apply only to table-row contracts')
    field_counts = contract.get('field_minimum_records', {})
    required_schema_fields = set(SCHEMA_FIELDS[contract['schema']])
    schema_fields = required_schema_fields.copy()
    if contract['schema'] == 'table_row':
        schema_fields = {column['name'] for column in contract['table_columns']}
    elif contract['schema'] == 'product':
        schema_fields.update({'financing', 'shipping'})
    if (not isinstance(field_counts, dict)
            or any(type(value) is not int or value < 1 for value in field_counts.values())
            or contract['schema'] not in {'body_text', 'page_identity'}
            and not set(field_counts) <= schema_fields):
        raise ValueError('Invalid required field counts')
    minimum, maximum = contract.get("minimum_records"), contract.get("maximum_records")
    if contract["schema"] == "body_text":
        if (type(minimum) is not int or minimum != 0
                or maximum is not None and (type(maximum) is not int or maximum != 0) or field_counts):
            raise ValueError("Stable body-text contracts do not declare records")
    elif contract["schema"] == "page_identity":
        if (type(minimum) is not int or minimum != 0
                or maximum is not None and (type(maximum) is not int or maximum != 0) or field_counts):
            raise ValueError("Page identity contracts do not declare records")
    elif type(minimum) is not int or minimum < 1 or maximum is not None and (type(maximum) is not int or maximum < minimum):
        raise ValueError("Stable contract requires positive bounded record counts")
    if 'duplicate_policy' in contract and (contract['schema'] != 'table_row'
            or contract['duplicate_policy'] != 'deduplicate_identical'):
        raise ValueError('Invalid duplicate policy')
    count_policy = contract.get("record_count_policy")
    if count_policy is not None:
        allowed_policy_fields = {
            "list_type", "source_observed_count", "required_ratio",
            "minimum_records", "source_observation_counts",
            "source_evidence_sha256", "reason",
        }
        required_policy_fields = allowed_policy_fields - {"minimum_records"}
        if (contract["schema"] in {"body_text", "page_identity"}
                or not isinstance(count_policy, dict)
                or not required_policy_fields <= set(count_policy) <= allowed_policy_fields
                or not isinstance(count_policy.get("list_type"), str)
                or count_policy.get("list_type") not in {"fixed", "variable"}
                or type(count_policy.get("source_observed_count")) is not int
                or count_policy["source_observed_count"] < 1
                or type(count_policy.get("required_ratio")) not in {int, float}
                or not 0 < count_policy["required_ratio"] <= 1
                or not isinstance(count_policy.get("source_observation_counts"), list)
                or len(count_policy["source_observation_counts"]) < 2
                or any(type(value) is not int or value < 1
                       for value in count_policy["source_observation_counts"])
                or not isinstance(count_policy.get("source_evidence_sha256"), str)
                or not re.fullmatch(r"[0-9a-f]{64}", count_policy["source_evidence_sha256"])
                or not isinstance(count_policy.get("reason"), str)
                or not count_policy["reason"].strip()):
            raise ValueError("Invalid record count policy")
        expected_ratio = 1.0 if count_policy["list_type"] == "fixed" else 0.8
        expected_minimum = math.ceil(
            count_policy["source_observed_count"] * count_policy["required_ratio"])
        if (count_policy["required_ratio"] != expected_ratio
                or minimum != expected_minimum
                or count_policy.get("minimum_records", minimum) != minimum):
            raise ValueError("Record minimum does not match the frozen source policy")
    membership = contract.get('source_membership')
    if 'source_membership' in contract:
        if (not isinstance(membership, dict) or not isinstance(membership.get('kind'), str)
                or membership.get('kind') not in {'fixed', 'changing'}):
            raise ValueError('Invalid source membership policy')
        if membership['kind'] == 'changing':
            if set(membership) != {'kind'}:
                raise ValueError('Changing membership cannot bind historical records')
        else:
            fields, hashes = membership.get('fields'), membership.get('record_hashes')
            if (set(membership) != {'kind', 'fields', 'record_hashes'} or not required_schema_fields
                    or not isinstance(fields, list) or not all(isinstance(f, str) for f in fields)
                    or fields != sorted(set(fields))
                    or not required_schema_fields <= set(fields) <= schema_fields
                    or not isinstance(hashes, list) or not all(isinstance(h, str) for h in hashes)
                    or hashes != sorted(set(hashes)) or len(hashes) < minimum
                    or any(not re.fullmatch(r'[0-9a-f]{64}', h) for h in hashes)):
                raise ValueError('Invalid fixed source record bindings')
    patterns = contract.get("page_patterns", [])
    if not isinstance(patterns, list) or any(not isinstance(p, str) or not p for p in patterns):
        raise ValueError("Invalid stable page patterns")
    anchors = contract.get("stable_text_anchors", [])
    if not isinstance(anchors, list):
        raise ValueError("Stable text anchors must be explicitly declared groups")
    ids = set()
    for anchor in anchors:
        if (not isinstance(anchor, dict) or not {"id", "alternatives"} <= set(anchor)
                or set(anchor) - {
                    "id", "alternatives", "match_scope", "source_location",
                    "observation_status", "source_evidence_sha256",
                    "visible_control",
                }
                or ('match_scope' in anchor and (anchor['match_scope'] != 'non_record_lines' or contract['schema'] != 'named_link'))
                or ('source_location' in anchor and (
                    not isinstance(anchor['source_location'], str)
                    or not anchor['source_location'].strip()))
                or ('observation_status' in anchor and (
                    not isinstance(anchor['observation_status'], str)
                    or not anchor['observation_status'].strip()))
                or ('source_evidence_sha256' in anchor and (
                    not isinstance(anchor['source_evidence_sha256'], str)
                    or not re.fullmatch(r"[0-9a-f]{64}", anchor['source_evidence_sha256'])))
                or not isinstance(anchor.get("id"), str) or not anchor["id"].strip() or anchor["id"] in ids
                or not isinstance(anchor.get("alternatives"), list) or not anchor["alternatives"]
                or any(not isinstance(value, str) or not _visible(value).strip() for value in anchor["alternatives"])):
            raise ValueError("Invalid or duplicate stable text anchor group")
        if "visible_control" in anchor:
            _validate_visible_control(anchor["visible_control"])
        ids.add(anchor["id"])
    if contract.get('content_scope') == 'target_content' and any(
            re.search(r'(^|[ >])(nav|footer)([ >.#:]|$)', anchor.get('source_location', ''), re.I)
            for anchor in anchors):
        raise ValueError('Excluded navigation/footer cannot supply target-content anchors')
    if contract["schema"] == "body_text" and not anchors and not patterns:
        raise ValueError("Stable body-text contracts require text anchors or page patterns")
    if contract["schema"] in {"named_link", "linked_product"}:
        hosts = contract.get("hosts")
        if (not isinstance(hosts, list) or not hosts
                or any(not isinstance(host, str) or not re.fullmatch(r"[a-z0-9]+(?:[.-][a-z0-9]+)*", host) for host in hosts)
                or not isinstance(contract.get("entry_path"), str) or not contract["entry_path"]):
            raise ValueError("Linked records require allowed hosts and detail path pattern")
        if (not isinstance(contract.get("name_validation", "descriptive"), str)
                or contract.get("name_validation", "descriptive") not in {"descriptive", "nonempty"}):
            raise ValueError("Invalid record name validation policy")
        scope = contract.get("entry_scope")
        if scope is not None:
            if (not isinstance(scope, dict) or not scope.get("html_selector")
                    or set(scope) - {"html_selector", "start_pattern", "end_pattern", "required_end", "title_selector", "link_selector", "price_selector", "auxiliary_links_end_pattern", "markdown_title_identity_pattern", "markdown_detail_identity_pattern", "markdown_label_pattern", "markdown_card_title_pattern", "markdown_title_column", "markdown_link_column"}
                    or any(not isinstance(v, str) or not v.strip()
                           for k, v in scope.items() if k != "required_end")
                    or ("required_end" in scope and type(scope["required_end"]) is not bool)
                    or (scope.get("required_end") and not scope.get("end_pattern"))
                    or (("title_selector" in scope) != ("link_selector" in scope))):
                raise ValueError("Invalid linked record section scope")
            _compile_stable_selector(scope["html_selector"])
            for key in ("title_selector", "link_selector", "price_selector"):
                if key in scope:
                    _compile_stable_selector(scope[key])
            for key in ("start_pattern", "end_pattern", "auxiliary_links_end_pattern"):
                if key in scope:
                    compiled = _compile_stable_regex(scope[key], re.M)
                    if compiled.search(""):
                        raise ValueError("Record section boundary cannot match empty text")
            column_keys = ("markdown_title_column", "markdown_link_column")
            if any(key in scope for key in column_keys):
                if (not all(key in scope for key in column_keys)
                        or scope[column_keys[0]] == scope[column_keys[1]]
                        or contract["schema"] != "named_link"):
                    raise ValueError("Named table records require distinct title and link columns")
            pair_keys = ("markdown_title_identity_pattern", "markdown_detail_identity_pattern")
            if 'markdown_label_pattern' in scope:
                pattern = _compile_stable_regex(scope['markdown_label_pattern'])
                if 'label' not in pattern.groupindex or pattern.search(''):
                    raise ValueError('Markdown card label requires a nonempty named label capture')
            if 'markdown_card_title_pattern' in scope:
                pattern = _compile_stable_regex(scope['markdown_card_title_pattern'])
                if ('title' not in pattern.groupindex or pattern.search('')
                        or not scope.get('start_pattern') or not scope.get('end_pattern')):
                    raise ValueError(
                        'Markdown heading cards require bounded scope and a nonempty title capture'
                    )
            if any(key in scope for key in pair_keys):
                if not all(key in scope for key in pair_keys):
                    raise ValueError("Markdown row identity patterns must be paired")
                for key in pair_keys:
                    pattern = _compile_stable_regex(scope[key])
                    if pattern.groups != 1 or pattern.search(""):
                        raise ValueError("Markdown row identity requires one nonempty capture")
        pair_hashes = contract.get("source_record_pair_hashes")
        pair_set_sha256 = contract.get("source_record_pairs_sha256")
        if pair_hashes is not None or pair_set_sha256 is not None:
            if (not isinstance(pair_hashes, list) or not pair_hashes
                    or any(not isinstance(value, str)
                           or not re.fullmatch(r"[0-9a-f]{64}", value)
                           for value in pair_hashes)
                    or pair_hashes != sorted(set(pair_hashes))
                    or not isinstance(pair_set_sha256, str)
                    or not re.fullmatch(r"[0-9a-f]{64}", pair_set_sha256)
                    or hashlib.sha256("\n".join(pair_hashes).encode()).hexdigest()
                    != pair_set_sha256):
                raise ValueError("Invalid frozen source name-link pairs")
        if contract.get('entry_context', 'list_or_card') != 'list_or_card':
            raise ValueError('Named-link entry context must be list_or_card')
        patterns = [*patterns, contract["entry_path"]]
        identity_pattern = contract.get('entry_identity_pattern')
        if identity_pattern is not None:
            if not isinstance(identity_pattern, str) or not identity_pattern:
                raise ValueError('Invalid linked-record identity pattern')
            patterns.append(identity_pattern)
    if contract["schema"] == "named_link":
        exclusions = contract.get('record_name_exclusions', [])
        if (not isinstance(exclusions, list)
                or any(not isinstance(value, str) or not _key(value) for value in exclusions)
                or len({_key(value) for value in exclusions}) != len(exclusions)):
            raise ValueError('Invalid or duplicate record-name exclusions')
        title_exclusions = contract.get('title_exclusion_patterns', [])
        if (not isinstance(title_exclusions, list)
                or any(not isinstance(pattern, str) or not pattern
                       for pattern in title_exclusions)):
            raise ValueError('Invalid named-link title exclusion patterns')
        patterns.extend(title_exclusions)
        if "link_base_url" in contract:
            value = contract["link_base_url"]
            try:
                base = urlsplit(value) if isinstance(value, str) else None
                valid = (base and base.scheme in {"http", "https"}
                         and base.hostname in hosts and not base.username
                         and not base.password and not base.query
                         and not base.fragment
                         and not any(char.isspace() for char in value))
                if base:
                    base.port
            except ValueError:
                valid = False
            if not valid:
                raise ValueError("Link base must be an observed HTTP(S) URL on an allowed host")
    elif contract["schema"] != "linked_product" and any(key in contract for key in (
            'hosts', 'entry_path', 'entry_scope', 'entry_identity_pattern', 'entry_context',
            'name_validation', 'identity_query_keys', 'identity_query_paths',
            'record_name_exclusions', 'title_exclusion_patterns', 'link_base_url',
            'source_record_pair_hashes', 'source_record_pairs_sha256')):
        raise ValueError("Host/path contracts apply only to linked records")
    query_keys = contract.get("identity_query_keys", [])
    if (not isinstance(query_keys, list)
            or any(not isinstance(key, str) or not re.fullmatch(r"[A-Za-z0-9_.~-]+", key)
                   for key in query_keys)):
        raise ValueError("Invalid linked-record identity query keys")
    if query_keys and contract["schema"] not in {"named_link", "linked_product"}:
        raise ValueError("Identity query keys apply only to linked records")
    if "identity_query_paths" in contract:
        query_paths = contract["identity_query_paths"]
        if (not isinstance(query_paths, list) or not query_paths or not query_keys
                or contract.get("entry_identity_pattern")
                or contract["schema"] not in {"named_link", "linked_product"}
                or any(not isinstance(path, str) or not path.strip() for path in query_paths)):
            raise ValueError("Invalid linked-record identity query paths")
        patterns.extend(query_paths)
    if contract["schema"] == "linked_product":
        price_patterns = contract.get("price_patterns")
        price_exclusions = contract.get("price_exclusion_patterns", [])
        if (not isinstance(price_patterns, list) or not price_patterns
                or any(not isinstance(pattern, str) or not pattern for pattern in price_patterns)
                or not isinstance(price_exclusions, list)
                or any(not isinstance(pattern, str) or not pattern for pattern in price_exclusions)):
            raise ValueError("Linked products require explicit price and exclusion patterns")
        unavailable = contract.get("price_unavailable_patterns", [])
        if (not isinstance(unavailable, list)
                or any(not isinstance(p, str) or not p.strip() or _compile_stable_regex(p).search("")
                       for p in unavailable)):
            raise ValueError("Invalid explicit price-unavailability patterns")
        patterns = [*patterns, *price_patterns, *price_exclusions, *unavailable]
    elif any(k in contract for k in ("price_patterns", "price_exclusion_patterns", "price_unavailable_patterns")):
        raise ValueError("Price patterns apply only to linked products")
    if contract["schema"] == "page_identity" and not anchors:
        raise ValueError("Page identity contracts require stable text anchors")
    for pattern in patterns:
        compiled = _compile_stable_regex(pattern)
        if pattern == contract.get('entry_identity_pattern') and compiled.groups != 1:
            raise ValueError('Named-link identity pattern must have one capture group')


def _markdown_code_payloads(text, *, preserve_delimiters=False):
    """Read displayed code payloads once, retaining their source ownership."""
    from .reference_quality import markdown_fenced_code_blocks, markdown_indented_code_contents, markdown_literal_spans

    fenced_blocks, _ = markdown_fenced_code_blocks(text)
    fenced = {(start, end): payload for start, end, payload in fenced_blocks}
    indented = markdown_indented_code_contents(text)
    spans = []
    for start, end in markdown_literal_spans(text):
        literal = text[start:end]
        if preserve_delimiters and ((start, end) in fenced or (start, end) in indented):
            payload = literal
        elif (start, end) in fenced:
            payload = fenced[start, end]
        elif (start, end) in indented:
            payload = indented[start, end]
            if literal.endswith(('\n', '\r')) and not payload.endswith('\n'):
                payload += '\n'
        else:
            width = len(re.match(r'`+', literal)[0])
            payload = literal[width:-width].replace('\n', ' ')
            if preserve_delimiters:
                payload = literal[:width] + payload + literal[-width:]
            elif payload.startswith(' ') and payload.endswith(' ') and payload.strip(' '):
                payload = payload[1:-1]
        spans.append((start, end, payload))
    return spans


def _pattern_text_surface(text, *, representation=None):
    """Render code delimiters only for a source view, not completed visible text."""
    if representation in {'plain_text', 'html'}:
        return text
    for start, end, payload in reversed(_markdown_code_payloads(text)):
        text = text[:start] + payload + text[end:]
    return text


def evaluate_stable_task(content: str, contract: dict, base_url: str = "", eligible: bool = True, links=None, *, anchor_text: str | None = None, anchors_required: bool = True, allow_closed_title_cards: bool = False, representation=None, returned_records=None, page_information=None) -> dict:
    """Validate a version-1 schema/count contract, without fetching or snapshots.

    Required keys: version, schema, minimum_records. Optional maximum_records
    imposes a finite exact scope when equal to the minimum. page_patterns is a
    list of required regexes for stable page/section identity (not current items).
    stable_text_anchors contains explicit {id, alternatives} groups; every group
    must match unless the caller supplies a separate body or composite contract
    and selects diagnostic anchors. Record and page-pattern gates remain required.
    named_link additionally requires hosts and entry_path, using the existing
    same-site detail-link parser. Raw URL identity is not factual verification.
    anchor_text may reuse the caller's normalized visible HTML prose;
    document-title metadata cannot supply page text or structural records.
    An anchor may opt into visible_control {selector, attribute} to recognize
    independently reviewed ordinary input placeholders or button values in HTML.
    This surface belongs only to that anchor, not page identity or body coverage.
    """
    validate_acceptance_contract(contract)
    from .response_input import page_information_texts, unscoped_page_information
    information_texts = page_information_texts(page_information)
    if representation is None:
        from .reference_quality import normalize_native_markdown_layout
        content = normalize_native_markdown_layout(content)
    from .reference_quality import is_html_document_or_fragment as native_html_surface
    native_html = representation == 'html' or (representation != 'plain_text' and native_html_surface(content))
    match_representation = 'html' if native_html else representation
    schema = contract["schema"]
    count_policy = contract.get("record_count_policy")
    minimum, maximum = contract["minimum_records"], contract.get("maximum_records")
    patterns = contract.get("page_patterns", [])
    visible = _visible(content, schema, representation=representation)
    if schema in {'body_text', 'page_identity'} and not native_html and anchor_text is not None:
        from .reference_quality import markdown_literal_spans
        prose_markup = content
        for start, end in reversed(markdown_literal_spans(content)):
            prose_markup = prose_markup[:start] + ' ' + prose_markup[end:]
        if not re.search(r'(?<!\\)</?[A-Za-z][\w:-]*(?:\s|/?>)', prose_markup):
            # An entity-only projection is not a rendered HTML view. Decode
            # the original Markdown once so emitted punctuation stays literal.
            anchor_text = None
    # HTML normalization can also decode entities in Markdown. Preserve its
    # explicit link boundaries before matching atomic page labels.
    anchor_visible = _visible(
        anchor_text, schema, normalized_html=True,
        representation='plain_text' if native_html else representation,
    ) if anchor_text is not None else None
    if anchor_visible is None and representation != 'plain_text':
        from .reference_quality import is_html_document_or_fragment
        html_candidate = content.lstrip()
        first_line, newline, remainder = html_candidate.partition('\n')
        # Legacy callers combine a metadata title and HTML. Recognize the
        # following document, not a guessed title-length limit. Keep metadata
        # as text rather than interpreting angle brackets in a title as HTML.
        if (newline and is_html_document_or_fragment(remainder)
                and not is_html_document_or_fragment(first_line)):
            html_candidate = remainder.lstrip()
        if is_html_document_or_fragment(html_candidate):
            from .dynamic_lists import visible_html_text
            anchor_visible = visible_html_text(html_candidate)
    original_markdown_visible = None
    if anchor_visible is None and representation != 'plain_text':
        from .reference_quality import markdown_heading_text
        heading_text = markdown_heading_text(content)
        if heading_text != content:
            anchor_visible = _visible(heading_text, schema)
            # An unfenced code comment is indistinguishable from an ATX
            # heading. Keep literal evidence on its original visible surface
            # as well; never add this fallback to an HTML projection.
            original_markdown_visible = visible
    non_record_content = content
    exclusions = dict(contract.get('text_exclusions', {}))
    if contract.get('content_scope') == 'target_content':
        exclusions['html_selectors'] = ['nav', 'footer', *exclusions.get('html_selectors', [])]
    if exclusions:
        if representation != 'plain_text' and exclusions.get('html_selectors') and re.search(r'<[a-z][^>]*>', content, re.I):
            from bs4 import BeautifulSoup
            from .dynamic_lists import visible_html_text
            soup = BeautifulSoup(content, 'html.parser')
            for selector in exclusions['html_selectors']:
                for element in soup.select(selector):
                    element.decompose()
            non_record_content = str(soup)
            anchor_visible = visible_html_text(non_record_content)
            original_markdown_visible = None
        else:
            anchor_visible = anchor_visible if anchor_visible is not None else visible
        for pattern in exclusions.get('text_patterns', []):
            anchor_visible = re.sub(pattern, ' ', anchor_visible, flags=re.I | re.M)
            if original_markdown_visible is not None:
                original_markdown_visible = re.sub(pattern, ' ', original_markdown_visible, flags=re.I | re.M)
    text_surface = anchor_visible if anchor_visible is not None else visible
    text_surfaces = [text_surface]
    if original_markdown_visible is not None:
        text_surfaces.append(original_markdown_visible)
    completed_body_view = schema in {'body_text', 'page_identity'}
    pattern_representation = ('plain_text' if completed_body_view or representation == 'html' or native_html_surface(content)
                              else representation)
    pattern_surfaces = [_pattern_text_surface(surface, representation=pattern_representation)
                        for surface in text_surfaces] if patterns else []
    if patterns and unscoped_page_information(contract):
        pattern_surfaces.extend(information_texts)
    page_results = [{"pattern": p, "pass": any(re.search(p, surface, re.I | re.M)
                                               for surface in pattern_surfaces)} for p in patterns]
    content_errors = (["page_identity_mismatch"] if not all(r["pass"] for r in page_results) else [])
    from .reference_quality import blocked_page_hits
    blocked_page = bool(blocked_page_hits(text_surface, representation='plain_text' if completed_body_view else representation))
    errors = []
    if schema in {"named_link", "linked_product"}:
        link_diagnostics = []
        try:
            if schema == "linked_product":
                from .dynamic_lists import extract_linked_products
                entries = extract_linked_products(
                    content, contract, base_url, diagnostics=link_diagnostics, include_excluded=True,
                    representation=representation,
                )
            else:
                entries = extract_named_entries(
                    content, contract, base_url, links, link_diagnostics, include_excluded=True,
                    representation=representation,
                )
        except ValueError:
            # Malformed returned URLs are output validation failures, not
            # malformed contracts and not exceptions that abort a whole run.
            entries = []
            errors.append("invalid_link_output")
        errors.extend(error for error in link_diagnostics if error not in errors)
        excluded_names = {_label_key(value, representation=representation)
                          for value in contract.get("record_name_exclusions", [])}
        excluded_entries = [entry for entry in entries
                            if _label_key(entry["title"], representation=representation) in excluded_names]
        entries = [entry for entry in entries
                   if _label_key(entry["title"], representation=representation) not in excluded_names]
        source_pair_hashes = set(contract.get("source_record_pair_hashes", []))
        source_mismatches = []
        if source_pair_hashes:
            matched_entries = []
            for entry in entries:
                pair_hash = named_entry_source_pair_hash(
                    entry["title"], entry["url"], contract, representation=representation)
                if pair_hash in source_pair_hashes:
                    matched_entries.append(entry)
                else:
                    source_mismatches.append({"title": entry["title"], "url": entry["url"],
                                              "pair_sha256": pair_hash})
            entries = matched_entries
            if source_mismatches:
                errors.append("source_record_mismatch")
        records = []
        for entry in entries:
            parts = urlsplit(entry["url"])
            if (parts.scheme not in {"http", "https"} or not parts.hostname
                    or parts.username or parts.password):
                continue
            records.append({"fields": entry, "valid": True, "errors": []})
    else:
        excluded_entries = []
        source_mismatches = []
        records = _structured_record_view(content, contract, representation=representation,
            allow_closed_title_cards=allow_closed_title_cards, visible=visible)
    # Flat structured rows prove field ownership, but cannot establish a
    # reviewed DOM selector, table region or excluded section boundary.
    if returned_records and SCHEMA_FIELDS[schema] and schema not in {'named_link', 'linked_product'} and not exclusions and not any(
            contract.get(key) for key in ('entry_scope', 'table_scope')):
        if not isinstance(returned_records, list):
            raise ValueError('returned records must be a list')
        names = SCHEMA_FIELDS[schema]
        # Merge only overlap with valid text rows. API paths keep their case,
        # as the uniqueness check does, although frozen hashes fold it.
        def merge_key(fields):
            return (source_record_hash(fields, names, schema),
                    fields.get('path', '').strip() if schema == 'api_operation' else None)

        # An explicit quote row can also be rendered in prose with comma
        # separators between tags. Keep its native field boundaries.
        def quote_identity(fields):
            return (_record_field_key(fields['quote']), _record_field_key(fields['author']),
                    tuple(re.findall(r'[^,\s]+', _record_field_key(fields['tags']))))

        # Each text row absorbs at most one returned row, so further copies stay
        # visible to the duplicate-record check. An equal valid row is kept; an
        # equivalent quote or the same fields parsed as invalid text is replaced.
        text_rows = len(records)
        consumed = set()
        for fields in returned_records:
            if not isinstance(fields, dict) or not all(isinstance(fields.get(key), str) for key in names):
                continue
            selected = {key: fields[key] for key in names}
            key = merge_key(selected)
            record = _record(selected, schema)
            open_rows = [i for i in range(text_rows) if i not in consumed]
            match = next((i for i in open_rows if records[i]['valid']
                          and merge_key(records[i]['fields']) == key), None)
            if match is not None:
                consumed.add(match)
                continue
            if record['valid']:
                match = next((i for i in open_rows if (
                    records[i]['valid'] and schema == 'quote'
                    and quote_identity(records[i]['fields']) == quote_identity(selected))
                    or (not records[i]['valid'] and merge_key(records[i]['fields']) == key)), None)
            if match is None:
                records.append(record)
            else:
                consumed.add(match)
                records[match] = record
    anchor_records = records
    if exclusions and (SCHEMA_FIELDS[schema] or schema == 'table_row') and schema not in {'named_link', 'linked_product'}:
        anchor_records = _structured_record_view(non_record_content, contract, representation=representation,
            allow_closed_title_cards=allow_closed_title_cards,
            text_patterns=exclusions.get('text_patterns', []))
    anchors = []
    control_soup = None
    if representation != 'plain_text' and any("visible_control" in a for a in contract.get("stable_text_anchors", [])):
        from .reference_quality import is_html_document_or_fragment
        if is_html_document_or_fragment(non_record_content):
            from .dynamic_lists import scoped_html
            control_soup = scoped_html(non_record_content)
    non_record_visible = None
    if schema == 'named_link' and any(a.get('match_scope') == 'non_record_lines'
                                      for a in contract.get('stable_text_anchors', [])):
        non_record_visible = _non_record_visible(
            non_record_content, contract, records, contract.get('link_base_url', base_url),
            representation=representation)
        if non_record_visible is not None:
            for pattern in exclusions.get('text_patterns', []):
                non_record_visible = re.sub(pattern, ' ', non_record_visible, flags=re.I | re.M)
    registered_words = tuple(value for anchor in contract.get("stable_text_anchors", [])
                             for value in anchor["alternatives"]
                             if re.fullmatch(r"[A-Za-z]+", value))
    native_labels = None
    escaped_prose = None
    for anchor in contract.get("stable_text_anchors", []):
        selected_surface = anchor_visible
        selected_non_record = non_record_visible
        if control_soup is not None and "visible_control" in anchor:
            control_text = _ordinary_control_text(control_soup, anchor["visible_control"])
            for pattern in exclusions.get('text_patterns', []):
                control_text = re.sub(pattern, ' ', control_text, flags=re.I | re.M)
            selected_surface = (anchor_visible if anchor_visible is not None else visible) + "\n" + control_text
            if non_record_visible is not None:
                selected_non_record = non_record_visible + "\n" + control_text
        matched = _anchor_matches(anchor, anchor_records, visible, schema,
                                  selected_non_record, selected_surface, registered_words,
                                  representation=match_representation, contract=contract)
        if original_markdown_visible is not None:
            literal_matches = _anchor_matches(
                anchor, anchor_records, visible, schema, non_record_visible,
                original_markdown_visible, registered_words, representation=match_representation, contract=contract)
            matched = [value for value in anchor["alternatives"]
                       if value in matched or value in literal_matches]
        if representation != 'plain_text' and not matched and anchor.get('match_scope') != 'non_record_lines':
            if native_labels is None:
                native_labels = _native_link_label_surface(content, exclusions) or ''
            if native_labels:
                matched = _anchor_matches(anchor, anchor_records, visible, schema,
                                          anchor_visible=native_labels, representation=match_representation, contract=contract)
        if (not matched and schema in {'body_text', 'page_identity'} and not native_html
                and representation != 'plain_text' and not exclusions):
            if escaped_prose is None:
                escaped_prose = _escaped_prose_surface(content, schema)
            if escaped_prose:
                matched = [value for value in anchor['alternatives'] if _stable_literal_match(
                    escaped_prose, value, representation=match_representation, rendered=True, literal=True)]
        if (not matched and information_texts and unscoped_page_information(contract)
                and not anchor.get('match_scope') and not anchor.get('visible_control')):
            matched = [value for value in anchor['alternatives'] if any(
                _stable_literal_match(text, value, registered_words, representation='plain_text')
                for text in information_texts)]
        anchors.append({**anchor, "matched": matched, "pass": bool(matched)})
    anchor_errors = ["missing_stable_anchor:" + r["id"] for r in anchors if not r["pass"]]
    if anchors_required:
        content_errors += anchor_errors
    complete = [r for r in records if r["valid"]]
    membership = contract.get('source_membership', {})
    if membership.get('kind') == 'fixed':
        allowed = set(membership['record_hashes'])
        owned = []
        for record in complete:
            record_hash = source_record_hash(record['fields'], membership['fields'], schema)
            if record_hash in allowed:
                owned.append(record)
            else:
                source_mismatches.append({
                    'fields': {name: record['fields'].get(name) for name in membership['fields']},
                    'record_sha256': record_hash})
        if len(owned) != len(complete):
            errors.append('source_record_mismatch')
        complete = owned
    identity_fields = contract["record_identity_fields"] if schema == "table_row" else UNIQUE_FIELDS[schema]
    compact_fields = {c['name'] for c in contract.get('table_columns', [])
                      if c.get('compact_whitespace')}
    def identity_key(record, field):
        if field == 'url' and schema in {'named_link', 'linked_product'}:
            return named_entry_identity(record['fields'][field], contract)
        if schema == 'api_operation' and field == 'path':
            return record['fields'][field]
        value = _key(record['fields'][field])
        return re.sub(r'\s+', '', value) if field in compact_fields else value
    unique = {tuple(identity_key(r, field) for field in identity_fields) for r in complete}
    record_warnings = []
    field_checks = []
    table_columns = {column['name']: column for column in contract.get('table_columns', [])}
    def populated(record, field):
        value = record['fields'].get(field)
        if schema == 'table_row':
            from .structured_tables import _valid
            return isinstance(value, str) and bool(value.strip()) and _valid(value, table_columns[field])
        if schema in {'named_link', 'linked_product'}:
            # Complete linked records already passed their declared extraction rules.
            return isinstance(value, str) and bool(value.strip())
        return _valid_field(field, value)
    for field, count in contract.get('field_minimum_records', {}).items():
        identities = {tuple(identity_key(r, key) for key in identity_fields)
                      for r in complete if populated(r, field)}
        field_checks.append({'field': field, 'count': len(identities), 'minimum': count, 'pass': len(identities) >= count})
        if len(identities) < count:
            errors.append('missing_required_field:' + field)
    if len(complete) != len(records) and maximum is not None:
        errors.append("incomplete_records")
    elif len(complete) != len(records):
        # An at-least-N contract must not fail because unrelated script/UI text
        # happens to resemble the start of a record. Keep the parse noise
        # visible, while requiring N complete unique records independently.
        record_warnings.append("ignored_incomplete_candidates")
    if len(unique) != len(complete):
        grouped = {}
        for record in complete:
            identity = tuple(identity_key(record, field) for field in identity_fields)
            grouped.setdefault(identity, set()).add(tuple(
                (field, identity_key(record, field)) for field in sorted(record['fields'])))
        if (contract.get('duplicate_policy') == 'deduplicate_identical'
                and all(len(values) == 1 for values in grouped.values())):
            record_warnings.append('identical_records_counted_once')
        else:
            errors.append("duplicate_records")
    if len(unique) < minimum:
        errors.append("too_few_complete_records")
    if maximum is not None and len(unique) > maximum:
        errors.append("too_many_records")
    structure_pass = not errors
    errors += content_errors
    if not eligible:
        errors.append("request_ineligible")
    return {"stable_task_pass": not errors, "stable_task_schema": schema,
            "stable_field_results": field_checks,
            "stable_structure_pass": structure_pass, "stable_content_pass": not content_errors,
            "stable_blocked_page": blocked_page,
            "stable_anchors_required": anchors_required,
            "stable_anchor_errors": anchor_errors, "stable_anchor_results": anchors, "stable_page_pattern_results": page_results,
            "stable_anchor_hits": sum(r["pass"] for r in anchors),
            "stable_anchor_total": len(anchors),
            "stable_anchor_coverage": sum(r["pass"] for r in anchors) / len(anchors) if anchors else None,
            "stable_text_scope": "declared_text_groups" if anchors else "not_annotated",
            "stable_record_count": len(records), "stable_complete_record_count": len(complete),
            "stable_unique_record_count": len(unique), "stable_minimum_records": minimum,
            "stable_record_list_type": count_policy.get("list_type") if count_policy else None,
            "stable_source_observed_record_count": (count_policy.get("source_observed_count") if count_policy
                                                    else len(membership['record_hashes'])
                                                    if membership.get('kind') == 'fixed' else None),
            "stable_required_record_ratio": count_policy.get("required_ratio") if count_policy else None,
            "stable_record_coverage": (len(unique) / count_policy["source_observed_count"]
                                       if count_policy else None),
            "stable_maximum_records": maximum, "stable_task_errors": errors, "stable_records": records,
            "stable_excluded_record_names": [entry["title"] for entry in excluded_entries],
            "stable_record_warnings": record_warnings,
            "stable_source_matched_record_count": (len(records)
                                                     if contract.get("source_record_pair_hashes")
                                                     else len(unique) if membership.get('kind') == 'fixed'
                                                     else None),
            "stable_source_mismatched_record_count": len(source_mismatches),
            "stable_source_mismatched_records": source_mismatches[:20],
            "factual_correctness": None,
            "stable_source_membership": membership.get('kind'),
            "stable_verification_scope": (
                "Complete record fields match declared frozen source hashes; semantic truth is not verified"
                if membership.get('kind') == 'fixed' else
                "Mechanical field/group/count checks; source fidelity and factual values are not verified")}