""" Assemble held-out corpus per DESIGN_v4.md. 15 documents target (5 A + 5 B + 5 C). Same content-validation gates as DESIGN_v3. Halt at n < 12. Writes corpus_heldout/ + corpus_heldout/manifest.json + assembly_v4_log.json. """ from __future__ import annotations import hashlib import json import os import re import subprocess import sys import urllib.error import urllib.request from pathlib import Path REPO_ROOT = Path(__file__).resolve().parents[2] STUDY_DIR = REPO_ROOT / "fvs_eval" / "validation_study" HELDOUT_DIR = STUDY_DIR / "corpus_heldout" HELDOUT_DIR.mkdir(exist_ok=True) # ============================================================================ # Source lists (frozen per DESIGN_v4 ยง2) # ============================================================================ STRATUM_A = [ { "slug": "ah01_marginal_revolution", "source_type": "blog_generic", "url": "https://marginalrevolution.com/marginalrevolution/2024/05/the-rise-of-factories-in-china-the-united-states-and-japan.html", "title": "Marginal Revolution: The Rise of Factories in China, the United States, and Japan", }, { "slug": "ah02_dan_luu_productivity", "source_type": "blog_generic", "url": "https://danluu.com/productivity-scale/", "title": "Dan Luu: Productivity at scale", }, { "slug": "ah03_ben_kuhn_outliers", "source_type": "blog_generic", "url": "https://www.benkuhn.net/outliers/", "title": "Ben Kuhn: Outliers", }, { "slug": "ah04_arxiv_qwq32b", "source_type": "arxiv", "url": "https://export.arxiv.org/api/query?id_list=2403.03186", "title": "arxiv 2403.03186", }, { "slug": "ah05_arxiv_dbrx", "source_type": "arxiv", "url": "https://export.arxiv.org/api/query?id_list=2404.07143", "title": "arxiv 2404.07143", }, ] STRATUM_B_PROMPTS = [ { "slug": "bh01_supply_chain", "topic": "global supply chain resilience", "prompt": "Analyze the state of global supply chain resilience in 2026. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "bh02_nonprofit_impact", "topic": "nonprofit impact measurement", "prompt": "Describe how a non-profit organization should evaluate impact measurement. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "bh03_gene_editing", "topic": "gene editing ethics", "prompt": "Discuss the ethical implications of gene editing technology for human embryos. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "bh04_retail_shifts", "topic": "retail industry", "prompt": "Analyze recent shifts in the retail industry and what they mean for brick-and-mortar stores. Write a considered written analysis, approximately 400-500 words.", }, { "slug": "bh05_autonomous_vehicles", "topic": "autonomous vehicles", "prompt": "Describe the current state and near-term outlook of autonomous vehicle deployment. Write a considered written analysis, approximately 400-500 words.", }, ] STRATUM_C_TITLES = [ {"slug": "ch01_photosynthesis", "title": "Photosynthesis"}, {"slug": "ch02_nuclear_power", "title": "Nuclear_power"}, {"slug": "ch03_cryptocurrency", "title": "Cryptocurrency"}, {"slug": "ch04_vaccine_hesitancy", "title": "Vaccine_hesitancy"}, {"slug": "ch05_renaissance", "title": "Renaissance"}, ] # ============================================================================ # Extractors (reuse from v3 where possible) # ============================================================================ def _curl(url: str, timeout: int = 30) -> str: result = subprocess.run( ["curl", "-sSL", "--max-time", str(timeout), "-A", "FrameCheck-ValidationStudy/1.0 (research)", url], capture_output=True, text=True, timeout=timeout + 15, ) if result.returncode != 0 or not result.stdout: raise RuntimeError(f"curl failed: rc={result.returncode}") return result.stdout def _replace_html_entities(text: str) -> str: for old, new in [ ("&", "&"), ("<", "<"), (">", ">"), (""", '"'), ("'", "'"), ("'", "'"), (" ", " "), ("'", "'"), ("—", "-"), ("–", "-"), ("’", "'"), ("‘", "'"), ("“", '"'), ("”", '"'), ("…", "..."), ("—", "-"), ]: text = text.replace(old, new) return text def extract_blog_generic(html: str) -> str: """Generic blog extractor: tries container patterns in priority order. Covers WordPress (entry-content), simple static sites (article or main), and raw body. """ html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) html = re.sub(r"]+/?>", "", html, flags=re.IGNORECASE) html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) html = re.sub(r"]*>.*?", "", html, flags=re.DOTALL | re.IGNORECASE) container_patterns = [ r']*>(.*?)', r']*class=["\'][^"\']*entry-content[^"\']*["\'][^>]*>(.*?)\s*(?:<(?:/div|footer|aside)|\Z)', r']*class=["\'][^"\']*post-content[^"\']*["\'][^>]*>(.*?)\s*(?:<(?:/div|footer|aside)|\Z)', r']*class=["\'][^"\']*post-body[^"\']*["\'][^>]*>(.*?)', r']*id=["\']content["\'][^>]*>(.*?)\s*(?:<(?:/div|footer)|\Z)', r']*>(.*?)', r']*>(.*?)', ] body = None for pat in container_patterns: m = re.search(pat, html, flags=re.DOTALL | re.IGNORECASE) if m: body = m.group(1) break if body is None: raise RuntimeError("blog_generic extractor: no container found") body = re.sub(r"", "\n\n", body, flags=re.IGNORECASE) body = re.sub(r"<(br|hr)[^>]*/?>", "\n", body, flags=re.IGNORECASE) text = re.sub(r"<[^>]+>", "", body) text = _replace_html_entities(text) text = re.sub(r"\n{2,}", "\n\n", text) paragraphs = text.split("\n\n") cleaned = [] for p in paragraphs: p = re.sub(r"\n+", " ", p).strip() p = re.sub(r" +", " ", p) if p: cleaned.append(p) return "\n\n".join(cleaned) def extract_arxiv(xml: str) -> str: m = re.search(r"]*>(.*?)", xml, flags=re.DOTALL) if not m: raise RuntimeError("arxiv extractor: no ") abstract = m.group(1).strip() abstract = _replace_html_entities(abstract) abstract = re.sub(r"(?= threshold, f"G1 {'pass' if wc>=threshold else 'FAIL'}: {wc} words (threshold {threshold})") def gate_G2(text, source_type): paras = [p.strip() for p in re.split(r"\n\s*\n", text) if p.strip()] if source_type == "arxiv": qualifying = [p for p in paras if len(p.split()) >= 100] return (len(qualifying) >= 1, f"G2 {'pass' if len(qualifying)>=1 else 'FAIL'} (arxiv): {len(qualifying)} para >= 100w") qualifying = [p for p in paras if len(p.split()) >= 40] return (len(qualifying) >= 3, f"G2 {'pass' if len(qualifying)>=3 else 'FAIL'}: {len(qualifying)} paragraphs >= 40w (total {len(paras)})") def gate_G3(text, source_type): if source_type == "arxiv": return (True, "G3 skipped (arxiv)") lines = [ln.strip() for ln in text.split("\n") if ln.strip()] if not lines: return (False, "G3 FAIL: empty") short = sum(1 for ln in lines if len(ln.split()) <= 5) ratio = short / len(lines) ok = ratio <= 0.30 return (ok, f"G3 {'pass' if ok else 'FAIL'}: {short}/{len(lines)} short lines ({ratio:.0%})") def validate(text, source_type): g1 = gate_G1(text, source_type) g2 = gate_G2(text, source_type) g3 = gate_G3(text, source_type) return (g1[0] and g2[0] and g3[0]), [g1[1], g2[1], g3[1]] # ============================================================================ # Anthropic API (reused) # ============================================================================ def _call_anthropic(prompt: str, max_tokens: int = 1200) -> dict: api_key = os.environ.get("ANTHROPIC_API_KEY") if not api_key: raise RuntimeError("ANTHROPIC_API_KEY not set") system_prompt = ( "You are a helpful assistant. Respond to the user's analytical " "prompt with a considered written analysis." ) candidates = ["claude-opus-4-7", "claude-sonnet-4-6", "claude-haiku-4-5-20251001"] last_err = None for cand in candidates: payload = { "model": cand, "max_tokens": max_tokens, "system": system_prompt, "temperature": 0.7, "messages": [{"role": "user", "content": prompt}], } req = urllib.request.Request( "https://api.anthropic.com/v1/messages", method="POST", data=json.dumps(payload).encode("utf-8"), headers={"x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json"}, ) try: with urllib.request.urlopen(req, timeout=90) as resp: data = json.loads(resp.read()) text = "".join(b.get("text", "") for b in data.get("content", []) if b.get("type") == "text") return {"text": text, "model": data.get("model"), "requested_model": cand, "stop_reason": data.get("stop_reason"), "input_tokens": data.get("usage", {}).get("input_tokens"), "output_tokens": data.get("usage", {}).get("output_tokens")} except urllib.error.HTTPError as e: body = e.read().decode("utf-8", errors="replace") if e.fp else str(e) last_err = f"{e.code} {e.reason}: {body[:300]}" continue except Exception as e: last_err = str(e) continue raise RuntimeError(f"All candidates failed; last: {last_err}") def _fetch_wikipedia_extract(title: str, max_chars: int = 6000) -> str: import urllib.parse api = ("https://en.wikipedia.org/w/api.php?" "action=query&prop=extracts&explaintext=1&redirects=1&format=json&titles=" + urllib.parse.quote(title)) req = urllib.request.Request(api, headers={"User-Agent": "FrameCheck-ValidationStudy/1.0"}) with urllib.request.urlopen(req, timeout=30) as resp: data = json.loads(resp.read()) pages = data.get("query", {}).get("pages", {}) for pid, page in pages.items(): if page.get("missing") is not None: raise RuntimeError(f"wikipedia not found: {title}") extract = page.get("extract", "") if extract: if len(extract) > max_chars: cut = extract[:max_chars] last_para = cut.rfind("\n\n") if last_para > max_chars // 2: extract = cut[:last_para] else: extract = cut return extract.strip() raise RuntimeError(f"no extract: {title}") # ============================================================================ # Assembly # ============================================================================ def _sha(s): return hashlib.sha256(s.encode("utf-8")).hexdigest() def _wc(s): return len(s.split()) def _now(): return subprocess.check_output(["date", "-u", "+%Y-%m-%dT%H:%M:%SZ"]).decode().strip() def main(): accepted = [] log = [] print("=" * 70) print("Held-out corpus assembly (DESIGN_v4)") print("=" * 70) # Stratum A print("\n== Stratum A (human-authored, non-tuning sources) ==") for spec in STRATUM_A: entry = {"slug": spec["slug"], "url": spec["url"], "source_type": spec["source_type"]} print(f"\n {spec['slug']}") try: raw = _curl(spec["url"]) entry["fetch"] = f"OK ({len(raw)} bytes)" text = EXTRACTORS[spec["source_type"]](raw) entry["extract"] = f"OK ({_wc(text)} words)" passed, gates = validate(text, spec["source_type"]) entry["gates"] = gates for g in gates: print(f" {g}") if passed: (HELDOUT_DIR / f"{spec['slug']}.txt").write_text(text, encoding="utf-8") accepted.append({ "doc_id": spec["slug"], "stratum": "A", "title": spec["title"], "source": spec["url"], "source_type": spec["source_type"], "fetched_at": _now(), "word_count": _wc(text), "sha256": _sha(text), "validation_gates": gates, }) entry["accepted"] = True print(f" ACCEPTED ({_wc(text)} words)") else: entry["accepted"] = False print(f" EXCLUDED") except Exception as e: entry["error"] = str(e)[:150] entry["accepted"] = False print(f" FAILED: {str(e)[:120]}") log.append(entry) # Stratum B print("\n== Stratum B (fresh AI-generated) ==") for spec in STRATUM_B_PROMPTS: entry = {"slug": spec["slug"], "prompt": spec["prompt"]} print(f"\n {spec['slug']} ({spec['topic']})") try: result = _call_anthropic(spec["prompt"]) text = result["text"] entry["model"] = result["model"] passed, gates = validate(text, "blog_generic") entry["gates"] = gates for g in gates: print(f" {g}") if passed: (HELDOUT_DIR / f"{spec['slug']}.txt").write_text(text, encoding="utf-8") accepted.append({ "doc_id": spec["slug"], "stratum": "B", "title": spec["topic"], "source": f"anthropic-api::{result['model']}", "fetched_at": _now(), "word_count": _wc(text), "sha256": _sha(text), "validation_gates": gates, "notes": f"Generated; requested={result['requested_model']}, served={result['model']}. Prompt: {spec['prompt']}", }) entry["accepted"] = True print(f" ACCEPTED ({_wc(text)} words, model={result['model']})") else: entry["accepted"] = False print(f" EXCLUDED") except Exception as e: entry["error"] = str(e)[:150] entry["accepted"] = False print(f" FAILED: {str(e)[:120]}") log.append(entry) # Stratum C print("\n== Stratum C (Wikipedia, non-tuning titles) ==") for spec in STRATUM_C_TITLES: entry = {"slug": spec["slug"], "title": spec["title"]} print(f"\n {spec['slug']} ({spec['title']})") try: text = _fetch_wikipedia_extract(spec["title"]) passed, gates = validate(text, "wikipedia") entry["gates"] = gates for g in gates: print(f" {g}") if passed: (HELDOUT_DIR / f"{spec['slug']}.txt").write_text(text, encoding="utf-8") accepted.append({ "doc_id": spec["slug"], "stratum": "C", "title": spec["title"], "source": f"https://en.wikipedia.org/wiki/{spec['title']}", "fetched_at": _now(), "word_count": _wc(text), "sha256": _sha(text), "validation_gates": gates, }) entry["accepted"] = True print(f" ACCEPTED ({_wc(text)} words)") else: entry["accepted"] = False print(f" EXCLUDED") except Exception as e: entry["error"] = str(e)[:150] entry["accepted"] = False print(f" FAILED: {str(e)[:120]}") log.append(entry) # Write manifest strata = {"A": 0, "B": 0, "C": 0} for d in accepted: strata[d["stratum"]] += 1 manifest = { "version": 1, "study": "heldout_v4", "assembled_at": _now(), "total_documents": len(accepted), "target": 15, "halt_threshold": 12, "per_stratum": strata, "documents": accepted, } (HELDOUT_DIR / "manifest.json").write_text(json.dumps(manifest, indent=2)) log_path = STUDY_DIR / "assembly_v4_log.json" log_path.write_text(json.dumps({"version": 1, "executed_at": _now(), "log": log}, indent=2)) print("\n" + "=" * 70) print("Summary") print("=" * 70) print(f" Total held-out: {len(accepted)} (target 15)") print(f" Per stratum: A={strata['A']}, B={strata['B']}, C={strata['C']}") print(f" Halt threshold: 12") if len(accepted) >= 12: print(f" Status: PROCEED (n >= 12)") else: print(f" Status: HALT (n < 12)") if __name__ == "__main__": main()