fix(webfetch): unify source validation and bound browser capture

.agents/skills/benchmark-report/references/evaluation-design.md

@@ -38,7 +38,7 @@ When comparing builds of one implementation, do not infer a code-induced latency
 
 Prefer invariant contracts for changing pages: stable labels, required fields, record shape, types, count bounds, uniqueness, and independently observed explanatory text. Do not freeze ranks, prices, member counts, or current titles unless the benchmark is explicitly a time-bound snapshot.
 
-Ground truth must be independent of the provider under test. Retain source location, bytes or a durable observation, hash, capture method, and scope. Partial annotations support only the requirements they cover. They are not a full-body reference.
+Ground truth must be independent of the provider under test. Retain source location, bytes or a durable observation, hash, capture method, and scope. Partial annotations support only the requirements they cover. They are not a full-body reference. When authoring or expanding a task, inspect its actual reference text against the reader's main content needs. A dashboard's labels do not establish its values; a feed's navigation does not establish its entries. Select applicable fact, body-region, record, pattern, resource and file checks through the shared scorer. Give the source-complete answer and a page shell or incomplete answer opposite outcomes before freezing the new version. Preserve earlier input versions.
 
 ## Machine-readable current profile
 

harness/web_fetch/README.md

@@ -114,6 +114,19 @@ a changed experiment. Checkpoints and final outputs are created exclusively and
 cannot replace an existing observation. Store them under the configured data
 home, outside Git.
 
+The batch isolates each browser worker in a child process. Its wall-clock budget
+covers all declared attempts plus two seconds for startup and cleanup, including
+browser transport calls that do not honor Playwright's page timeout. A stuck or
+crashed worker produces a failed observation and leaves other cases intact.
+
+Explicit `source_verification.reference_mode: live_exact` applies to every track
+and output kind, including stable contracts and mixed text/resource deliveries.
+A required live reference must exist before execution. Without that declaration,
+compiled references remain frozen. Rendered-reference checks apply the declared
+body regions, record contracts, composite fields and patterns as well as facts;
+a fact group is optional when another
+content requirement establishes the source scope.
+
 ## WCXB scoring
 
 `wcxb_scoring.py` owns certified frozen-HTML scoring. The maintained command is
@@ -519,9 +532,10 @@ For each case, independently capture and retain:
    Use scoped visible text/DOM evidence, not script-containing `textContent`.
    Save the rendered DOM or screenshot alongside text when useful for review.
 3. The complete reference for that declared scope, derived independently of
-   provider outputs. Freeze required fact groups before scoring; every group
-   needs at least one nonempty alternative and must occur in both reference and
-   rendered text. Include tail/item-count checks to catch partial lists.
+   provider outputs. Freeze the applicable body, record, field, pattern or fact
+   requirements before scoring and check them against both reference and rendered
+   text. Each declared fact group needs a nonempty alternative. Include tail and
+   item-count checks to catch partial lists.
 
 Use a condition such as “12 product cards visible and loading indicator gone,
 within 10 seconds,” not merely `networkidle` or a fixed sleep. A delayed request
@@ -556,7 +570,9 @@ The result includes exact SHA-256 hashes of initial HTML, rendered text and
 `ground_truth_main_content`, plus group-by-group coverage for initial static
 DOM text, initial payload candidates, rendered text and reference. Matching
 uses the scoring normalizer and numeric boundaries; all groups are mandatory
-regardless of `fact_coverage_threshold`. Empty facts are invalid, not 100%.
+regardless of `fact_coverage_threshold`. Without fact groups, another declared
+content requirement must supply acceptance. Numeric fields require values;
+labels or loading placeholders do not establish a completed dashboard.
 
 Classifications describe the required facts, not the entire page:
 
@@ -566,15 +582,16 @@ Classifications describe the required facts, not the entire page:
   on these facts alone. It includes hidden/template/metadata content.
 - `partial_initial_content`: only some groups found in the initial response.
 - `absent_from_initial_evidence`: no groups found by this mechanical audit.
-- `unclassified`: no required facts were supplied; validation fails.
+- `unclassified`: no fact groups were supplied. Other declared content checks
+  still determine validity.
 
 The static visibility proxy excludes scripts, head, templates, noscript and
 explicitly hidden elements; it cannot compute stylesheets, layout or browser
 HTML repair. Payload detection recognizes source literals and JSON string
 escapes, not arbitrary expressions, compression or external scripts. Therefore
 absence is not proof of JS necessity: a follow-up HTTP/API request may suffice.
 `js_required_proven` is always false. Likewise, `valid` checks declared
-completion and fact/length gates; it does not independently verify the declared
+completion, length and declared content gates; it does not independently verify the declared
 capture outcome or prove full-page completeness. Retain the actual evidence.
 
 Run the synthetic checks without credentials or network:

harness/web_fetch/browser_observations.py

@@ -7,6 +7,9 @@
 import hashlib
 import ipaddress
 import json
+import os
+import signal
+import subprocess
 import socket
 import sys
 import tempfile
@@ -53,6 +56,8 @@
         ? node.textContent : '';
     }
     if (node.nodeType !== Node.ELEMENT_NODE) return '';
+    // Unrendered elements can return textContent from innerText.
+    if (['NOSCRIPT', 'TEMPLATE'].includes(node.tagName)) return '';
     const style = getComputedStyle(node);
     if (style.display === 'none' || Number(style.opacity) === 0) return '';
     if (!affected.has(node)) return node.innerText || '';
@@ -564,6 +569,11 @@ def capture_with_retries(
                     "message": str(error)[:1000],
                 }
             )
+    return _failed_observation(case, observer, errors, attempts, timeout_ms, last_row)
+
+
+def _failed_observation(case, observer, errors, attempts, timeout_ms, last_row=None):
+    last_row = last_row or {}
     return {
         **last_row,
         "case_id": case["id"],
@@ -599,6 +609,68 @@ def capture_source_identity() -> dict:
             "lexbench_webfetch/scoring": SCORING_SOURCE_SHA256}
 
 
+
+_CAPTURE_PROGRAM = (
+    "import json,sys; from harness.web_fetch.browser_observations import capture_with_retries; "
+    "print(json.dumps(capture_with_retries(**json.load(sys.stdin))))"
+)
+
+
+def _capture_in_process(case, timeout_ms, attempts, browser_binary=None, resolved_ip=None,
+                        *, observation_scope="reference", dns_mode="system"):
+    """Bound browser transport and cleanup as well as page operations."""
+    payload = dict(case=case, timeout_ms=timeout_ms, attempts=attempts,
+                   browser_binary=browser_binary, resolved_ip=resolved_ip,
+                   observation_scope=observation_scope, dns_mode=dns_mode)
+    budget = timeout_ms * attempts / 1000 + 2
+    began = time.monotonic()
+    diagnostics = ""
+    with subprocess.Popen(
+        [sys.executable, "-c", _CAPTURE_PROGRAM], stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
+        cwd=Path(__file__).resolve().parents[2], start_new_session=os.name == "posix",
+        env={**os.environ, "DEBUG": "pw:api"},
+    ) as process:
+        try:
+            stdout, stderr = process.communicate(json.dumps(payload), timeout=budget)
+            if process.returncode:
+                raise RuntimeError(stderr[-1000:] or f"Capture worker exited {process.returncode}")
+            result = json.loads(stdout)
+            if not isinstance(result, dict) or result.get('case_id') != case['id']:
+                raise ValueError("Capture worker returned another case")
+            return result
+        except subprocess.TimeoutExpired:
+            # Each POSIX worker owns a new process group, including its driver.
+            # Terminate only that group so a stuck transport cannot hold pipes open.
+            signals = (signal.SIGTERM, signal.SIGKILL) if os.name == "posix" else (None, None)
+            for sig in signals:
+                try:
+                    if os.name == "posix":
+                        os.killpg(process.pid, sig)
+                    else:
+                        process.kill()
+                except ProcessLookupError:
+                    pass
+                try:
+                    _stdout, diagnostics = process.communicate(timeout=1)
+                    break
+                except subprocess.TimeoutExpired:
+                    continue
+            message = f"Browser observation worker exceeded {budget:g}s wall-clock budget"
+            error_type = "TimeoutError"
+        except (RuntimeError, ValueError) as error:
+            message, error_type = str(error), type(error).__name__
+    row = _failed_observation(
+        case, {"browser": "Chromium", "binary": "explicit" if browser_binary else "Playwright default",
+               "dns_mode": dns_mode},
+        [{"attempt": None, "type": error_type, "message": message}], None, timeout_ms)
+    row['completion']['elapsed_ms'] = round((time.monotonic() - began) * 1000, 2)
+    row['worker_wall_budget_ms'] = budget * 1000
+    if diagnostics:
+        row['worker_diagnostics'] = diagnostics[-4000:]
+    return row
+
+
 def capture_batch(
     cases,
     *,
@@ -683,7 +755,7 @@ def observe(case):
         )
         if dns_mode != "system":
             options["dns_mode"] = dns_mode
-        row = capture_with_retries(
+        row = _capture_in_process(
             case, timeout_ms, attempts, browser_binary, resolved_ip, **options
         )
         guard()

harness/web_fetch/reference_lifecycle.py

@@ -143,16 +143,11 @@ def requires_fresh_reference(case) -> bool:
     if contract:
         from lexbench_webfetch.stable_tasks import validate_acceptance_contract
         validate_acceptance_contract(contract)
-        return False
     verification = _get(case, "source_verification", {}) or {}
     mode = verification.get("reference_mode", "frozen")
     if mode not in {"frozen", "live_exact"}:
         raise ReferenceLifecycleError("Unknown reference mode; use frozen or live_exact")
-    return (mode == "live_exact"
-            and _get(case, "track", "web") in {"web", "dynamic", "challenge"}
-            and _get(case, "task_kind", "body") == "body"
-            and bool((_get(case, "ground_truth_main_content", "") or "").strip())
-            and bool(verification))
+    return mode == "live_exact"
 
 
 def expected_providers(dataset, case):
@@ -593,6 +588,8 @@ def _binding(case, dataset_path: Path, now: datetime) -> dict[str, Any]:
     identifier = safe_identifier(_get(case, "id"))
     verification = _get(case, "source_verification")
     reference = _get(case, "ground_truth_main_content")
+    if not isinstance(reference, str) or not reference.strip():
+        raise ReferenceLifecycleError(f"{identifier}: required live reference is missing")
     if not isinstance(verification, dict) or verification.get("state") != "valid":
         raise ReferenceLifecycleError(f"{identifier}: independent source verification is not valid")
     if not isinstance(verification.get("method"), str) or not verification["method"].strip():

harness/web_fetch/render_evidence.py

@@ -117,7 +117,12 @@ def validate_render_reference(
             errors.append(f"duplicate_fact_id:{identifier}")
         seen.add(identifier)
         facts.append({"id": identifier, "alternatives": alternatives})
-    if not facts:
+    body_requirements = get("body_requirements", {}) or {}
+    required_patterns = get("required_patterns", []) or []
+    acceptance_contract = get("acceptance_contract", {}) or {}
+    composite_acceptance = get("composite_acceptance", {}) or {}
+    if not facts and not (body_requirements or required_patterns or acceptance_contract
+                          or composite_acceptance):
         errors.append("required_facts_empty")
     texts = {"initial_visible": _initial_visible(html),
              "initial_payload": _payload_candidates(html),
@@ -147,6 +152,22 @@ def validate_render_reference(
         for group in coverage[key]["groups"]:
             if not group["pass"]:
                 errors.append(f"{key}_missing_fact:{group['id']}")
+    if body_requirements or required_patterns or acceptance_contract or composite_acceptance:
+        from lexbench_webfetch.metrics import evaluate_content
+        for key in ("reference", "rendered"):
+            checks = evaluate_content(texts[key], [], body_requirements=body_requirements,
+                                      required_patterns=required_patterns,
+                                      acceptance_contract=acceptance_contract,
+                                      composite_acceptance=composite_acceptance,
+                                      base_url=get("url", ""))
+            if not checks["body_coverage_pass"]:
+                errors.append(f"{key}_body_requirements_not_met")
+            if not all(row['pass'] for row in checks['pattern_results']):
+                errors.append(f"{key}_pattern_requirements_not_met")
+            for field, label in (("stable_task_pass", "record"),
+                                 ("composite_acceptance_pass", "composite")):
+                if checks.get(field) is False:
+                    errors.append(f"{key}_{label}_requirements_not_met")
     evidence = (get("source_benchmark", {}) or {}).get("render_evidence", {})
     completion = evidence.get("completion", {})
     condition = completion.get("condition")

tests/package_smoke.py

@@ -1040,21 +1040,21 @@ def verify_browser_observation_capture(root, source_root):
     binary = root / 'observation-browser'
     binary.write_bytes(b'synthetic installed browser identity')
     calls = []
-    original = observations.capture_with_retries
+    original = observations._capture_in_process
 
     def observe(case, *args, **kwargs):
         calls.append(case['id'])
         return {'case_id': case['id'], 'status': 'success', 'content_text': 'Installed capture'}
 
-    observations.capture_with_retries = observe
+    observations._capture_in_process = observe
     try:
         options = dict(timeout_ms=1000, attempts=1, browser_binary=str(binary),
                        checkpoints=root / 'observation-checkpoints')
         first = observations.capture_batch([task], **options)
         assert observations.capture_batch([task], **options) == first
         assert calls == ['installed-observation']
     finally:
-        observations.capture_with_retries = original
+        observations._capture_in_process = original
 
 
 def verify_reference_promotion(root, source_root):

tests/test_browser_observation_contracts.py

@@ -36,7 +36,7 @@ def test_invalid_policy_rejected_before_browser_calls(rule, monkeypatch):
     def unexpected(*args, **kwargs):
         pytest.fail("Invalid policy reached browser capture")
 
-    monkeypatch.setattr(capture, "capture_with_retries", unexpected)
+    monkeypatch.setattr(capture, "_capture_in_process", unexpected)
     with pytest.raises(ValueError, match="policy"):
         capture.capture_batch([case_with(rule)], timeout_ms=1000, attempts=1)
 
@@ -93,7 +93,7 @@ def observe(case, *args, **kwargs):
         calls.append(case["id"])
         return {"case_id": case["id"], "status": "success", "content_text": "Article"}
 
-    monkeypatch.setattr(capture, "capture_with_retries", observe)
+    monkeypatch.setattr(capture, "_capture_in_process", observe)
     options = dict(
         timeout_ms=1000,
         attempts=1,
@@ -133,7 +133,7 @@ def observe(case, *args, **kwargs):
         binary.write_bytes(b"changed during observation")
         return {"case_id": case["id"]}
 
-    monkeypatch.setattr(capture, "capture_with_retries", observe)
+    monkeypatch.setattr(capture, "_capture_in_process", observe)
     with pytest.raises(ValueError, match="implementation changed"):
         capture.capture_batch([case_with()], **options)
     assert not (options["checkpoints"] / "new-task.json").exists()
@@ -207,6 +207,8 @@ def do_GET(self):
             body = b"""<html><body><nav>Navigation only</nav><main>
             <div style="display:none">Hidden panel</div>
             <div style="opacity:0"><span>Transparent menu</span></div>
+            <noscript>Fallback code is not visible with JavaScript enabled</noscript>
+            <template>Template payload is not visible</template>
             <span>Visible duplicate</span><div style="opacity:0">Visible duplicate</div>
             <p style="opacity:0.5">Translucent text</p>
             <p>Inline <strong>formatting</strong> preserved</p>
@@ -256,6 +258,8 @@ def test_live_multielement_completion_preserves_declared_content_scope(rendered_
     if mode == "visible":
         assert "Inline formatting preserved" in row["content_text"]
         assert row["content_text"].count("Visible duplicate") == 1
+        assert "Fallback code" not in row["content_text"]
+        assert "Template payload" not in row["content_text"]
     assert ("Hidden panel" in row["content_text"]) == (mode != "visible")
     assert ("Excluded panel" in row["content_text"]) == (mode == "dom")
 
@@ -268,3 +272,24 @@ def test_live_ambiguous_content_scope_is_rejected_instead_of_truncated(rendered_
     task["url"] = rendered_page
     with pytest.raises(ValueError, match="exactly one element"):
         capture.capture(task, 10000, os.environ["BROWSER_EVAL_TEST_BROWSER_BINARY"])
+
+
+def test_stuck_capture_worker_is_bounded_and_remains_an_unqualified_observation(monkeypatch):
+    import time
+    monkeypatch.setattr(capture, '_CAPTURE_PROGRAM', 'import time; time.sleep(30)')
+    began = time.monotonic()
+    row = capture._capture_in_process(case_with(), 20, 1)
+    assert time.monotonic() - began < 8
+    assert row['status'] == 'failure'
+    assert row['completion']['satisfied'] is False
+    assert row['attempt_count'] is None
+    assert row['attempt_errors'][0]['type'] == 'TimeoutError'
+    assert 'wall-clock budget' in row['attempt_errors'][0]['message']
+
+
+def test_crashed_capture_worker_does_not_erase_other_case_results(monkeypatch):
+    monkeypatch.setattr(capture, '_CAPTURE_PROGRAM', "raise RuntimeError('controlled worker failure')")
+    row = capture._capture_in_process(case_with(), 1000, 1)
+    assert row['case_id'] == 'new-task'
+    assert row['status'] == 'failure'
+    assert 'controlled worker failure' in row['attempt_errors'][0]['message']

tests/test_browser_reference_scope.py

@@ -85,7 +85,7 @@ def observe(case, *args, **kwargs):
         calls.append(case['id'])
         return {'case_id':case['id'],'status':'failure' if case['id']=='blocked' else 'success',
                 'content_text':'Verify you are human' if case['id']=='blocked' else 'Article body'}
-    monkeypatch.setattr(browser_capture,'capture_with_retries',observe)
+    monkeypatch.setattr(browser_capture,'_capture_in_process',observe)
     cases=[{'id':k,'url':'https://example.com/'+k} for k in ('article','blocked')]
     binary = tmp_path / 'browser'
     binary.write_bytes(b'synthetic browser identity')
@@ -109,7 +109,7 @@ def observe(case, *args, **kwargs):
 def test_checkpoint_cannot_cross_dns_modes(tmp_path, monkeypatch):
     binary = tmp_path / 'browser'
     binary.write_bytes(b'synthetic browser identity')
-    monkeypatch.setattr(browser_capture, 'capture_with_retries', lambda case, *args, **kwargs: {
+    monkeypatch.setattr(browser_capture, '_capture_in_process', lambda case, *args, **kwargs: {
         'case_id': case['id'], 'status': 'success', 'completion': {'satisfied': True},
     })
     options = dict(timeout_ms=1000, attempts=1, checkpoints=tmp_path / 'checkpoints',
@@ -403,7 +403,7 @@ def fake_capture(case, timeout_ms, attempts, browser_binary, resolved_ip):
             'completion': {'satisfied': succeeded},
         }
 
-    monkeypatch.setattr(browser_capture, 'capture_with_retries', fake_capture)
+    monkeypatch.setattr(browser_capture, '_capture_in_process', fake_capture)
     monkeypatch.setattr('sys.argv', [
         'capture_headless_browser_observations.py',
         '--dataset', str(dataset), '--output', str(output),