Fix scoring of equivalent public table answers

harness/web_fetch/README.md

@@ -546,8 +546,11 @@ rankings and catalogs change frequently. Annotate fields, not whole websites:
   `stable_text_anchors` with independently justified equivalent alternatives.
 - Frequently changing names, prices, dates and ranking positions use an
   `acceptance_contract`: required fields, types, record grouping, minimum count,
-  uniqueness and optional stable page identity. Values are not compared to a
-  previous ranking. Duplicate cards cannot satisfy the minimum.
+  uniqueness and stable page text. Values are not compared to a previous
+  ranking. Duplicate cards cannot satisfy the minimum. Final-URL agreement is
+  diagnostic: a missing or mismatched URL does not change acceptance or the
+  evaluated denominator. HTML document-language tags are also diagnostic;
+  content acceptance comes from the declared answer requirements.
 - Full-reference word precision, recall and F1 remain separate diagnostics.
   Historical demonstration snapshots describe their captured sample; they do
   not certify arbitrary later content. Without a comparable full reference,

harness/web_fetch/composite_tasks.py

@@ -13,6 +13,7 @@
 import unicodedata
 from urllib.parse import parse_qsl, unquote, urljoin, urlsplit
 
+from .structured_tables import _pipe_cells, table_text_surface
 from .dynamic_lists import _canonical_entry_path, extract_named_entries
 from .reference_quality import clean_reference_text, markdown_links, visible_markdown_links
 
@@ -541,15 +542,15 @@ def _name_description_group(content: str, group: dict, base_url: str) -> dict:
 
 
 def _table_group(content: str, group: dict) -> dict:
-    lines = [line.strip() for line in clean_reference_text(content).splitlines() if line.strip()]
+    lines = [line.strip() for line in clean_reference_text(table_text_surface(content)).splitlines() if line.strip()]
     required = group["required_columns"]
     records = []
     for index in range(len(lines) - 1):
         delimiter = "|" if "|" in lines[index] else "\t" if "\t" in lines[index] else None
         if not delimiter:
             continue
-        headers = [_key(value) for value in lines[index].strip("|").split(delimiter)]
-        separator = lines[index + 1].strip("|").split(delimiter)
+        headers = [_key(visible_markdown_links(value)) for value in (_pipe_cells(lines[index]) if delimiter == "|" else lines[index].split(delimiter))]
+        separator = (_pipe_cells(lines[index + 1]) if delimiter == "|" else lines[index + 1].split(delimiter))
         if delimiter == "|" and not all(
             re.fullmatch(r"\s*:?-{3,}:?\s*", value) for value in separator
         ):
@@ -577,11 +578,11 @@ def _table_group(content: str, group: dict) -> dict:
                 continue
             row_index = index + 2 if delimiter == "|" else index + 1
             while row_index < len(lines) and delimiter in lines[row_index]:
-                values = [value.strip() for value in lines[row_index].strip("|").split(delimiter)]
+                values = [value.strip() for value in (_pipe_cells(lines[row_index]) if delimiter == "|" else lines[row_index].split(delimiter))]
                 if len(values) == len(headers):
                     row = {}
                     for field, (column, pattern) in mapping.items():
-                        match = pattern.fullmatch(visible_markdown_links(values[column])) if pattern else None
+                        match = pattern.fullmatch(visible_markdown_links(values[column]).replace("**", "").strip()) if pattern else None
                         row[field] = (match[field] if match else '') if pattern else values[column]
                     if all(
                         _value_matches_spec(row[spec["id"]], spec, whole_cell=True)
@@ -601,15 +602,15 @@ def _table_group(content: str, group: dict) -> dict:
 
 
 def _linked_table_group(content: str, group: dict, base_url: str) -> dict:
-    """Read named links and all required fields from the same Markdown row."""
-    lines = [line.strip() for line in clean_reference_text(content).splitlines() if line.strip()]
+    """Read named links and required fields from the same owned table row."""
+    lines = [line.strip() for line in clean_reference_text(table_text_surface(content)).splitlines() if line.strip()]
     required = group["required_columns"]
     records = {}
     for index in range(len(lines) - 1):
         if "|" not in lines[index]:
             continue
-        headers = [_key(value) for value in lines[index].strip("|").split("|")]
-        separator = lines[index + 1].strip("|").split("|")
+        headers = [_key(visible_markdown_links(value)) for value in _pipe_cells(lines[index])]
+        separator = _pipe_cells(lines[index + 1])
         if not all(re.fullmatch(r"\s*:?-{3,}:?\s*", value) for value in separator):
             continue
         mapping = {}
@@ -625,7 +626,7 @@ def _linked_table_group(content: str, group: dict, base_url: str) -> dict:
         else:
             row_index = index + 2
             while row_index < len(lines) and "|" in lines[row_index]:
-                values = [value.strip() for value in lines[row_index].strip("|").split("|")]
+                values = [value.strip() for value in _pipe_cells(lines[row_index])]
                 if len(values) == len(headers):
                     row = {field: values[column] for field, column in mapping.items()}
                     cell = row[group["link_column"]]
@@ -679,12 +680,15 @@ def _field_report_group(content: str, group: dict) -> dict:
 
 
 def evaluate_composite_acceptance(
-    content: str, contract: dict, *, base_url: str, links: list | None = None, eligible: bool = True
+    content: str, contract: dict, *, base_url: str, links: list | None = None, eligible: bool = True, table_content: str | None = None
 ) -> dict:
     validate_composite_acceptance(contract)
     results = []
     for group in contract["groups"]:
-        scoped = _scope(content, group)
+        source = table_content if table_content is not None and group["kind"] in {"table", "linked_table"} else content
+        if group["kind"] in {"table", "linked_table"}:
+            source = table_text_surface(source)
+        scoped = _scope(source, group)
         if group["kind"] == "named_link":
             result = _named_link_group(
                 scoped, group, base_url, links if scoped == content else None

harness/web_fetch/metrics.py

@@ -325,9 +325,6 @@ def task_passed(ok: bool, metrics: dict, task_kind: str | None = None) -> bool:
     # Scored records carry the case kind so dataset-free summaries use the same
     # predicate as the report. Explicit case context still takes precedence.
     task_kind = task_kind if task_kind is not None else metrics.get('task_kind', 'body')
-    if (metrics.get('page_identity_required')
-            and metrics.get('page_identity_status') != 'verified'):
-        return False
     if (not task_measured(metrics, task_kind) or metrics.get('deadline_overrun')
             or metrics.get('blocked_page')):
         return False
@@ -347,10 +344,6 @@ def task_passed(ok: bool, metrics: dict, task_kind: str | None = None) -> bool:
 def task_measured(metrics: dict, task_kind: str | None = None) -> bool:
     if metrics.get('rule_review_status') == 'unresolved':
         return False
-    if (metrics.get('page_identity_required')
-            and metrics.get('page_identity_status') == 'unavailable'
-            and metrics.get('request_ok') is not False):
-        return False
     kind = task_kind or metrics.get('task_kind', 'body')
     return kind != 'file' or isinstance(metrics.get('raw_file_pass'), bool)
 
@@ -643,6 +636,7 @@ def evaluate_content(
         from .composite_tasks import evaluate_composite_acceptance
         composite = evaluate_composite_acceptance(
             visible_content, composite_acceptance, base_url=base_url, links=named_links,
+            table_content=content,
         )
         result.update(composite)
         result["content_pass"] = bool(result["content_pass"] and composite["composite_acceptance_pass"])

harness/web_fetch/scoring.py

@@ -51,6 +51,7 @@ def load_list_reference(case: EvalCase | dict[str, Any], source: Path) -> dict |
     "structure_pass", "named_entry_count", "minimum_entries", "snapshot_f1",
     "dynamic_titles", "verification_note", "confirmed_entries", "checked_entries",
     "page_identity_required", "page_identity_status", "final_url_consistent",
+    "document_language_required",
     "language_consistent", "observed_document_language",
     "acceptable_final_urls",
     "ordered_tree", "ordered_tree_similarity",
@@ -187,7 +188,7 @@ def score_attempt(
             else "unavailable"
         )
         result.update(
-            page_identity_required=True,
+            page_identity_required=False,
             page_identity_status=identity_status,
             final_url_consistent=final_consistent,
             language_consistent=language_consistent,
@@ -205,7 +206,7 @@ def score_attempt(
             and any(_same_page_url(final_url, expected) for expected in acceptable_urls)
         )
         result.update(
-            page_identity_required=True,
+            page_identity_required=False,
             page_identity_status=(
                 "verified" if final_consistent else "mismatch" if final_url else "unavailable"
             ),
@@ -255,13 +256,6 @@ def score_attempt(
             body_requirements, sort_keys=True, ensure_ascii=False).encode()).hexdigest()
     if not result["content_score_applicable"] or not ok:
         result["content_pass"] = False
-    if result.get("page_identity_required") and result.get("page_identity_status") != "verified":
-        result["content_pass"] = False
-        if contract:
-            result["stable_task_pass"] = False
-            result["stable_content_pass"] = False
-            if "page_identity_not_verified" not in result["stable_task_errors"]:
-                result["stable_task_errors"].append("page_identity_not_verified")
     if not ok:
         if contract:
             result["stable_task_pass"] = False

harness/web_fetch/stable_tasks.py

@@ -143,7 +143,7 @@ def _visible(text, schema=None):
     text = _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'\\([!"#$%&\'()*+,\-./:;<=>?@\[\]\\^_`{|}~])', r'\1', text)
+    text = re.sub(r'\\([!"#$%&\'()*+,\-./:;<=>?@\[\]\\^_`{|}~])', lambda m: m[0] if schema == 'table_row' and m[1] in {'|', '\\'} else m[1], 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

harness/web_fetch/structured_tables.py

@@ -174,6 +174,70 @@ def _without_example_fences(text):
     return ''.join(lines)
 
 
+def table_header_label(value):
+    """The first visual line names a column; later lines are controls/help."""
+    return re.split(r'<br\s*/?>', value, maxsplit=1, flags=re.I)[0].strip()
+
+
+def table_text_surface(content):
+    """Preserve owned HTML cells/links and adjacent split Markdown bodies."""
+    from bs4 import BeautifulSoup
+    content = _without_example_fences(content)
+    if re.search(r"<table\b", content, re.I):
+        from .dynamic_lists import _scoped_html
+        content = str(_scoped_html(content))
+        content = re.sub(
+            r"<h([1-6])\b[^>]*>[\s\S]*?</h\1\s*>",
+            lambda m: "\n" + BeautifulSoup(m[0], "html.parser").get_text(" ", strip=True) + "\n",
+            content, flags=re.I,
+        )
+
+    def convert(match):
+        soup = BeautifulSoup(match[0], 'html.parser')
+        table = soup.find('table')
+        lines = []
+        for row in table.find_all('tr'):
+            if row.find_parent('table') is not table:
+                continue
+            cells = row.find_all(['th', 'td'], recursive=False)
+            values = []
+            for cell in cells:
+                if cell.get('colspan', '1') != '1' or cell.get('rowspan', '1') != '1':
+                    break
+                for link in cell.find_all('a', href=True):
+                    link.replace_with('[' + link.get_text(' ', strip=True) + '](' + link['href'] + ')')
+                values.append(cell.get_text(' ', strip=True).replace('\\', r'\\').replace('|', r'\|'))
+            else:
+                if values:
+                    lines.append('| ' + ' | '.join(values) + ' |')
+                    if all(cell.name == 'th' for cell in cells):
+                        lines.append('| ' + ' | '.join('---' for _ in cells) + ' |')
+        return '\n' + '\n'.join(lines) + '\n'
+    content = re.sub(r'<table\b[^>]*>[\s\S]*?</table\s*>', convert, content, flags=re.I)
+    lines = content.splitlines()
+    output = []
+    index = 0
+    def separator(line):
+        return '|' in line and all(
+            re.fullmatch(r'\s*:?-{3,}:?\s*', c) for c in _pipe_cells(line))
+    while index < len(lines):
+        prior = [v for v in output if v.strip()]
+        cells = _pipe_cells(lines[index])
+        if ('|' in lines[index] and all(not c.strip() for c in cells)
+                and index + 1 < len(lines) and separator(lines[index + 1])
+                and len(_pipe_cells(lines[index + 1])) == len(cells)
+                and len(prior) >= 2 and separator(prior[-1])
+                and len(_pipe_cells(prior[-2])) == len(cells)
+                and any(c.strip() for c in _pipe_cells(prior[-2]))):
+            while output and not output[-1].strip():
+                output.pop()
+            index += 2
+            continue
+        output.append(lines[index])
+        index += 1
+    return '\n'.join(output)
+
+
 def parse_table_records(content, visible, contract):
     columns, scope = contract['table_columns'], contract['table_scope']
     names = [c['name'] for c in columns]
@@ -227,7 +291,13 @@ def parse_table_records(content, visible, contract):
     if root is None:
         # Inline layout breaks do not create new Markdown rows/cells. Preserve
         # a space so split numeric tokens cannot become a different valid value.
-        visible = re.sub(r'<br\s*/?>', ' ', visible, flags=re.I)
+        lines = table_text_surface(visible).splitlines()
+        for i in range(len(lines) - 1):
+            cells = _pipe_cells(lines[i])
+            if ('|' in lines[i] and len(cells) == len(_pipe_cells(lines[i + 1]))
+                    and all(re.fullmatch(r'\s*:?-{3,}:?\s*', c) for c in _pipe_cells(lines[i + 1]))):
+                lines[i] = '| ' + ' | '.join(table_header_label(c) for c in cells) + ' |'
+        visible = re.sub(r'<br\s*/?>', ' ', '\n'.join(lines), flags=re.I)
         # Only source-verified, case-declared auxiliary text is removed.
         # This also normalizes headers before locating the target table.
         for pattern in scope.get('plain_text_ignored_patterns', []):
@@ -309,5 +379,23 @@ def capture(match):
 
 
 def _pipe_cells(line):
-    # Remove only the table's outer delimiters; adjacent pipes are empty cells.
-    return line.strip().removeprefix('|').removesuffix('|').split('|')
+    """Split real delimiters, preserving escaped pipes and empty owned cells."""
+    line = line.strip()
+    cells = ['']
+    index = 0
+    while index < len(line):
+        char = line[index]
+        if char == '\\' and index + 1 < len(line) and line[index + 1] in {'|', '\\'}:
+            cells[-1] += line[index + 1]
+            index += 2
+            continue
+        if char == '|':
+            cells.append('')
+        else:
+            cells[-1] += char
+        index += 1
+    if line.startswith('|'):
+        cells.pop(0)
+    if cells and cells[-1] == '' and line.endswith('|'):
+        cells.pop()
+    return cells

tests/test_scoring.py

@@ -63,7 +63,7 @@ def test_untagged_source_requires_explicitly_observed_empty_language(self):
                                         final_url="https://example.test/page")
         self.assertEqual(observed_tagged["page_identity_status"], "mismatch")
 
-    def test_risk_control_requires_declared_page_identity(self):
+    def test_risk_control_final_url_is_diagnostic(self):
         navigation = {
             "risk_control": {
                 "navigation_contract": {
@@ -85,14 +85,24 @@ def test_risk_control_requires_declared_page_identity(self):
             source_benchmark=navigation,
         )
         wrong = score_attempt(c, b"Expected", ok=True, final_url="https://other.test/")
-        self.assertTrue(wrong["page_identity_required"])
+        self.assertFalse(wrong["page_identity_required"])
         self.assertEqual(wrong["page_identity_status"], "mismatch")
         self.assertTrue(task_measured(wrong))
-        self.assertFalse(task_passed(True, wrong))
+        self.assertTrue(task_passed(True, wrong))
         missing = score_attempt(c, b"Expected", ok=True)
         self.assertEqual(missing["page_identity_status"], "unavailable")
-        self.assertFalse(task_measured(missing))
-        self.assertFalse(task_passed(True, missing))
+        self.assertTrue(task_measured(missing))
+        self.assertTrue(task_passed(True, missing))
+        for diagnostic in (wrong, missing):
+            stats = summarize([{
+                "provider": "fixture", "provider_type": "fixture",
+                "case_id": c.id, "case_name": c.name, "category": c.category,
+                "ok": True, "elapsed_ms": 1,
+                "metrics": diagnostic,
+            }])["providers"]["fixture"]
+            self.assertEqual(stats["task_evaluated_attempts"], 1)
+            self.assertEqual(stats["task_unmeasured_attempts"], 0)
+            self.assertEqual(stats["task_pass_rate"], 1.0)
         failed = score_attempt(c, b"", ok=False)
         self.assertEqual(failed["page_identity_status"], "unavailable")
         self.assertTrue(task_measured(failed))

tests/test_scoring_regressions.py

@@ -73,6 +73,15 @@ def test_every_registered_false_verdict_control_runs_through_shared_scorer(self)
                 self.assertEqual(verdict, row["expected_task_pass"])
                 for key, expected in row.get("expected_metrics", {}).items():
                     self.assertEqual(metrics.get(key), expected, key)
+                if (row["root_cause"] == "table_escaped_pipe_boundaries"
+                        and row["expected_task_pass"]):
+                    if row["case"].get("composite_acceptance"):
+                        record = metrics["composite_group_results"][0]["records"][0]
+                        fields = record.get("fields", record)
+                        self.assertEqual(fields["owner"], "Alice | Bob")
+                    else:
+                        self.assertEqual(metrics["stable_records"][0]["fields"]["name"],
+                                         "Alpha | Beta")
                 root_modes.setdefault(row["root_cause"], set()).add(row["failure_mode"])
         paired = {"answer_surface_omission", "body_scope_calibration",
                   "declared_scope_exclusions", "declared_truncation",
@@ -95,6 +104,11 @@ def test_every_registered_false_verdict_control_runs_through_shared_scorer(self)
         paired.add("declared_full_page_stable_text_scope")
         paired.add("inline_emphasis_in_structured_fields")
         paired.add("markdown_list_prefix_ownership")
+        paired.add("final_url_diagnostic_not_acceptance")
+        paired.add("html_language_metadata_diagnostic_not_acceptance")
+        paired.update({"table_visible_section_boundaries", "table_escaped_pipe_boundaries"})
+        paired.update({"table_header_visual_line", "table_adjacent_split_body",
+                       "composite_html_table_ownership"})
         for root_cause in paired:
             self.assertEqual(root_modes.get(root_cause), {"false-positive", "false-negative"})