feat(web-extract): support source-qualified URL benchmarks and independent reports

examples/web_extract/README.md

@@ -14,7 +14,7 @@ shared cache, works offline after download, and preserves existing hash-bound in
 
 ## Scope and naming
 
-Web Extract measures structured field extraction from archived HTML: the caller
+Web Extract measures structured field extraction from archived pages or live URLs: the caller
 requests fields such as price, stock or author, and the service returns field values.
 Web scraping is the broader term for collecting website data; extraction is the
 part measured here. [Zyte calls it automatic extraction](https://docs.zyte.com/zyte-api/usage/extract/index.html).
@@ -76,7 +76,8 @@ Browser infrastructure requiring a separately connected model is outside this
 comparison. Browserbase JSON uses only the Browserbase API key. The separately
 registered Stagehand route uses its native Model Gateway and needs a project ID;
 it does not read an OpenAI key. Bright Data's site-specific extractor remains discoverable as a candidate, but
-is rejected by the full archived-page plan because it cannot parse arbitrary HTML.
+is outside arbitrary-schema cohorts because it requires a supported site-specific
+extractor. Browser infrastructure alone does not supply the requested field values.
 
 New cohorts give every native API the same public URL for each frozen page.
 Each dataset has its own hosting service. Browser Eval can find and reuse a
@@ -428,6 +429,53 @@ and [versioned preview URLs](https://developers.cloudflare.com/pages/configurati
 
 ## Multi-value field reports
 
+### Source-qualified external cohorts
+
+Use the `records` adapter for another benchmark with explicit field schemas and
+source-validated references. Its definition stays outside Git and binds the source
+versions, sampling rules, page hashes, schemas and reference hashes. References are
+used for local scoring and never sent to an extraction API.
+
+A definition may set `extraction_instructions` to declare its task scope. This
+text is frozen in the dataset profile, plan and request contract. List-query
+benchmarks should request all matching records; the default single-record
+instructions remain available for detail-page benchmarks. Field descriptions
+carry the original question when an API accepts a schema without a separate query.
+
+```sh
+uv run --frozen --extra web-extract-eval python -m harness.web_extract.cli \
+  --home /absolute/data/cohort prepare --benchmark records \
+  --source /absolute/data/cohort-definition.json
+```
+
+The definition declares `input_mode: snapshot` or `input_mode: live`:
+
+| Mode | URL sent to the native API | Source validation |
+| --- | --- | --- |
+| `snapshot` | The immutable hosted page URL selected by `host` | HF revision and the complete HTML/resource inventory before and after collection |
+| `live` | The original website URL in the task | Original reference values checked on the source before and after collection |
+
+Live cohorts reject snapshot bindings. Saved HTML is private source evidence;
+providers still visit the original site. Unavailable pages or references that no
+longer appear stop certification, while completed provider calls remain saved.
+Source checks run at most four origins concurrently, with serial requests and a
+one-second pause within each origin. They do not contribute to provider latency.
+Transient network errors and HTTP 408/429/5xx responses allow up to three source
+checks, with bounded backoff recorded in the receipt. Missing reference values do
+not trigger retries. This policy never retries provider extraction for a better score.
+
+Snapshots can include captured styles, images and other resources. Their hashes
+are part of the whole-snapshot version; changing an asset changes that version
+even when the main HTML is unchanged. Scripts and external resources remain
+blocked by response headers.
+
+The shared scorer reports page-hit macro F1, value-level micro F1 and exact-field
+set rate. Text matching preserves content and units while normalizing whitespace.
+Explicitly typed URL fields resolve relative links against the source URL, using
+the same matching rule in aggregate scores and result details. A cohort using
+these rules must disclose them as an adaptation unless they reproduce the
+upstream evaluator.
+
 The same report owner supports registered record schemas through
 `mode: field-extraction-cohort`. The instance content binds every planned
 participant to its certified source or journal checkpoint, with source-plan

src/harness/web_extract/analysis.py

@@ -169,7 +169,7 @@ def failure_stage(row):
 
 def record_field_results(reference, row, *, text_matching="literal", unavailable_fields=()):
     """Expose exact-set and official page-hit outcomes without changing saved scores."""
-    from harness.web_extract.swde import canonical_text
+    from harness.web_extract.swde import record_value_key
 
     if text_matching not in ("literal", "whitespace"):
         raise ValueError("Unsupported record text matching policy")
@@ -184,16 +184,14 @@ def record_field_results(reference, row, *, text_matching="literal", unavailable
         actual = prediction.get(field, [])
         valid = (
             {
-                canonical_text(value) if text_matching == "whitespace" else value
+                record_value_key(reference, field, value, text_matching)
                 for value in actual
                 if isinstance(value, str)
             }
             if isinstance(actual, list)
             else set()
         )
-        expected = {
-            canonical_text(value) if text_matching == "whitespace" else value for value in gold
-        }
+        expected = {record_value_key(reference, field, value, text_matching) for value in gold}
         if field in unavailable:
             result[field] = {
                 "expected": [],

src/harness/web_extract/certification.py

@@ -4,23 +4,24 @@
 from pathlib import Path
 
 from harness.reporting.templates import metric_contract
-from harness.web_extract.dataset import benchmark_config, task_bindings
+from harness.web_extract.dataset import benchmark_config, input_mode, task_bindings
 from harness.web_extract.pipeline import (
-    RECORD_INSTRUCTIONS,
     available_methods,
     data_home,
     digest,
     instruction_schema,
     load_run,
     probe,
     read_json,
+    record_instructions,
     require_external_output,
     save_json,
     score_run,
     selected_profile,
     validate_dataset,
     validate_fixture_base,
     validate_instruction_schema,
+    validate_live_route,
 )
 
 
@@ -36,6 +37,9 @@ def create_plan(home, methods, output, *, snapshots=None):
     if any(registry[m].get("retired") for m in methods):
         raise ValueError("Retired reference interfaces cannot enter a new cohort")
     validate_dataset(home)
+    live = input_mode(home) == "live"
+    if live and snapshots is not None:
+        raise ValueError("Live cohorts use original URLs, not a snapshot registry")
     truth = read_json(home / "dataset/dataset/ground-truth.json")
     plan = {
         "schema": "archived-product-full-cohort/1",
@@ -74,8 +78,13 @@ def create_plan(home, methods, output, *, snapshots=None):
         plan.update(task_bindings(home))
         for participant in plan["participants"].values():
             participant["maximum_physical_calls"] = len(truth) + 20
-    if snapshots is None and any(
-        registry[m]["input"] == "url" or "input_route" in registry[m]["options"] for m in methods
+    if (
+        not live
+        and snapshots is None
+        and any(
+            registry[m]["input"] == "url" or "input_route" in registry[m]["options"]
+            for m in methods
+        )
     ):
         raise ValueError(
             "URL-capable interfaces require --snapshots; upload-only methods may omit it"
@@ -103,10 +112,19 @@ def create_plan(home, methods, output, *, snapshots=None):
         plan["resource_decision"]["memory"] = (
             "Remote memory unavailable; local client memory does not measure the provider."
         )
+    if live:
+        plan["input_mode"] = "live"
+        for method, participant in plan["participants"].items():
+            participant["input"] = "url"
+            if registry[method]["input"] != "url":
+                if "input_route" not in registry[method].get("options", []):
+                    raise ValueError("A live cohort requires native URL extraction")
+                participant["native_options"] = {"input_route": "url"}
     adapter = benchmark_config(home)["adapter"]
     if adapter is not None:
+        plan["extraction_instructions"] = record_instructions(home)
         for case in truth.values():
-            schema = instruction_schema(adapter.task_schema(home, case), RECORD_INSTRUCTIONS)
+            schema = instruction_schema(adapter.task_schema(home, case), record_instructions(home))
             for method in methods:
                 validate_instruction_schema(method, schema)
     save_json(output, plan, exclusive=True)
@@ -116,6 +134,7 @@ def create_plan(home, methods, output, *, snapshots=None):
 def run(home: Path, profile: Path, method: str, run_dir: Path, fixture_base=None):
     require_external_output(run_dir)
     plan = read_json(profile)
+    live = plan.get("input_mode") == "live"
     if plan.get("schema") == "archived-extraction-repair/1":
         from harness.web_extract.repair import run as repair_run
 
@@ -146,18 +165,28 @@ def run(home: Path, profile: Path, method: str, run_dir: Path, fixture_base=None
     participant = plan["participants"][method]
     if participant.get("input") == "html" and plan.get("public_snapshots"):
         fixture_base = None
-    if available_methods()[method]["input"] == "url" and (
-        not fixture_base or not fixture_base.startswith("https://")
+    if (
+        not live
+        and available_methods()[method]["input"] == "url"
+        and (not fixture_base or not fixture_base.startswith("https://"))
     ):
         raise ValueError("URL participant requires frozen HTTPS fixtures before collection")
     records = benchmark_config(home)["adapter"] is not None
+    if live:
+        if participant.get("input") != "url":
+            raise ValueError("Live participants must declare URL input")
+        validate_live_route(method, participant.get("native_options"))
+    if records and plan.get("extraction_instructions") != record_instructions(home):
+        raise ValueError("Frozen extraction instructions differ from the dataset")
     if records and plan.get("schema") != "archived-extraction-full-cohort/1":
         raise ValueError("Frozen plan schema differs from the selected record benchmark")
     if records or plan.get("task_adapter_sha256"):
         bindings = task_bindings(home)
         if not bindings or any(plan.get(key) != value for key, value in bindings.items()):
             raise ValueError("Frozen task adapter or dataset profile changed")
     validate_dataset(home)
+    if live != (input_mode(home) == "live"):
+        raise ValueError("Frozen input mode differs from the selected dataset")
     truth = read_json(home / "dataset/dataset/ground-truth.json")
     case_hash = digest(("\n".join(sorted(truth)) + "\n").encode())
     if plan.get("dataset_manifest_sha256") != digest((home / "dataset/manifest.json").read_bytes()):

src/harness/web_extract/dataset.py

@@ -11,6 +11,10 @@
 from harness.record_io import bytes_sha256, read_json
 
 BENCHMARKS = {
+    "records": {
+        "profile": None,
+        "module": "harness.web_extract.records",
+    },
     "zyte": {
         "profile": "zyte-product-extraction.json",
         "module": None,
@@ -41,14 +45,33 @@ def benchmark_config(home=None, *, name=None):
     if name not in BENCHMARKS:
         raise ValueError("Unknown extraction benchmark: " + str(name))
     item = BENCHMARKS[name]
+    distribution = item.get("snapshot_distribution")
+    if name == "records" and home is not None:
+        distribution = read_json(Path(home) / "profile.json").get("snapshot_distribution")
     return {
-        "profile": Path(__file__).resolve().parents[1] / "benchmarks/cards" / item["profile"],
+        "profile": (
+            Path(home) / "profile.json"
+            if item["profile"] is None and home is not None
+            else Path(__file__).resolve().parents[1] / "benchmarks/cards" / item["profile"]
+            if item["profile"] is not None
+            else None
+        ),
         "adapter": importlib.import_module(item["module"]) if item["module"] else None,
         "snapshot_project": item.get("snapshot_project"),
-        "snapshot_distribution": item.get("snapshot_distribution"),
+        "snapshot_distribution": distribution,
     }
 
 
+def input_mode(home):
+    """A live source and a hosted immutable page have distinct qualification rules."""
+    from harness.web_extract.pipeline import read_json
+
+    mode = read_json(Path(home) / "dataset/manifest.json").get("input_mode", "snapshot")
+    if mode not in {"snapshot", "live"}:
+        raise ValueError("Unknown extraction input mode")
+    return mode
+
+
 def require_product_reference(home):
     """Keep product-only report and review entries out of other schema families."""
     if benchmark_config(home)["adapter"] is not None:
@@ -68,11 +91,19 @@ def task_bindings(home):
     config = benchmark_config(home)
     if config["adapter"] is None:
         return {}
-    return {
+    result = {
         "task_adapter_sha256": adapter_sha256(home),
         "dataset_profile_sha256": bytes_sha256(config["profile"].read_bytes()),
         "dataset_registry_sha256": registry_sha256(),
     }
+    if hasattr(config["adapter"], "scoring_sources"):
+        result["task_scoring_dependencies_sha256"] = bytes_sha256(
+            b"\n".join(
+                bytes_sha256(path.read_bytes()).encode()
+                for path in config["adapter"].scoring_sources()
+            )
+        )
+    return result
 
 
 def distribution_identity(home: Path, pin, selection):

src/harness/web_extract/messages.json

@@ -295,6 +295,7 @@
     "records_details": "Full result",
     "records_schema": "Frozen request schema",
     "records_preview": "Archived page preview",
+    "records_live_page": "Open live webpage",
     "records_original_url": "Original source URL",
     "records_download": "Frozen HTML",
     "records_error": "Exact runtime response/error",
@@ -718,6 +719,7 @@
     "records_details": "完整结果",
     "records_schema": "冻结请求字段定义",
     "records_preview": "归档页面预览",
+    "records_live_page": "打开实时网页",
     "records_original_url": "原网站地址",
     "records_download": "原始 HTML",
     "records_error": "调用错误",