"""B1-X focused extraction harness: no provider, network, or production writes.""" import asyncio import json import shutil from types import SimpleNamespace import pytest from scripts import measure_discovery_extraction_13b as extraction def _row(witness_id, cls, spec, *, expected_names=None, expect_empty=False, error=False): row = { "id": witness_id, "class": cls, "prompt": witness_id, "discovery": ( {"status": "error"} if error else { "raw_turn": { "spec": spec, "requested_capabilities": spec.pop("_caps", []), } } ), } if expected_names is not None: row["expected_names"] = expected_names if expect_empty: row["expect_empty"] = True return row def test_extraction_discipline_is_in_prompt_and_schema(): from src.domain.projects.discovery_schemas import DiscoverySpec, Entity from src.integrations.project_intake_client import DISCOVERY_PROMPT assert "EXTRACTION DISCIPLINE" in DISCOVERY_PROMPT assert "INTENT FIDELITY" in DISCOVERY_PROMPT assert "grammatical number" in (Entity.model_fields["name"].description or "") assert "including when decision is clarify" in ( DiscoverySpec.model_fields["workflows"].description or "" ) def test_witness_set_covers_b1s_variance_and_controls(): cases = {item["id"]: item for item in extraction.witnesses()} assert set(cases) >= { "13A-018", "13A-025", "13A-042", "13A-051", "13A-060", "13A-063", "13A-075", "13A-085", "13A-103", "X-nouns", "X-plural", "X-singular", "X-none", } assert cases["13A-042"]["class"] == "archetype_variance_booking" assert cases["X-none"]["expect_empty"] is True assert cases["X-plural"]["expected_names"] == ["inspection reports"] assert "staff-only appointment booking" in cases["13A-042"]["prompt"] def test_projection_and_diagnostics_cover_required_fields(): spec_a = { "entities": [{"name": "Job Request"}], "workflows": [{}], "archetypes": ["internal_tool"], "actors": [{"name": "Staff"}], "_caps": ["auth", "records"], } spec_b = { "entities": [{"name": "quote request"}], "workflows": [], "archetypes": ["booking"], "actors": [{"name": "staff"}, {"name": "customer"}], "_caps": ["records"], } empty = { "entities": [], "workflows": [], "archetypes": [], "actors": [], "_caps": [], } invented = { "entities": [{"name": "hello"}], "workflows": [{}], "archetypes": [], "actors": [], "_caps": [], } runs = [ [ _row("13A-025", "implicit_entity_explicit_workflow", spec_a), _row("X-none", "no_meaningful_workflow", empty, expect_empty=True), ], [ _row("13A-025", "implicit_entity_explicit_workflow", spec_b), _row("X-none", "no_meaningful_workflow", invented, expect_empty=True), ], ] result = extraction.diagnostics(runs) assert result["pair_count"] == 2 assert result["entity_projection_disagreement"] == 2 assert result["workflow_projection_disagreement"] == 2 assert result["archetype_disagreement"] == 1 assert result["actor_projection_disagreement"] == 1 assert result["capability_projection_disagreement"] == 1 assert result["explicit_workflow_omissions"] == 1 assert result["empty_input_invention"] == 1 assert result["per_witness"]["13A-025"]["entity_stable"] is False assert result["per_witness"]["13A-025"]["workflow_count_stable"] is False def test_protected_outputs_include_historical_b1_and_admission(): names = {path.name for path in extraction.PROTECTED_OUTPUTS} assert "extraction_before" in names assert "13b_b1_admission_prod_host_01.operator_summary.json" in names assert "13b_b1_admission_gated_01" in names def test_refuses_overwrite_and_protected_paths(): existing = extraction.ROOT / ".test-tmp-extraction-13b-existing" existing.mkdir(exist_ok=True) try: with pytest.raises(ValueError, match="new experiment"): asyncio.run(extraction.measure(existing, 4, "production")) with pytest.raises(ValueError, match="protected historical fixture"): asyncio.run( extraction.measure( extraction.ROOT / "docs/audits/fixtures/13b_b1_completion/extraction_before", 4, "production", ) ) finally: existing.rmdir() def test_preflight_requires_extraction_contract(monkeypatch): monkeypatch.setattr( extraction, "DISCOVERY_PROMPT", "UNDERSTAND BEFORE ASKING\nINFERENCE POLICY", ) monkeypatch.setattr( extraction, "get_settings", lambda: SimpleNamespace( project_intake_timeout_seconds=8, project_intake_model="gpt-4o-mini", openai_api_key="x", redis_url="redis://test", project_intake_calls_per_minute=60, ), ) with pytest.raises(ValueError, match="Extraction-discipline"): extraction.preflight() def test_verify_requires_production_kind_and_witness_files(): out = extraction.ROOT / ".test-tmp-extraction-13b-verify" if out.exists(): shutil.rmtree(out) (out / "run_01").mkdir(parents=True) (out / "run_02").mkdir() try: cases = [{"id": "13A-042"}] (out / "witnesses.json").write_text(json.dumps(cases), encoding="utf-8") (out / "scorecard.json").write_text( json.dumps( { "measurement_kind": "focused_live_extraction", "diagnostics": { "entity_projection_disagreement": 0, "workflow_projection_disagreement": 0, "archetype_disagreement": 0, "actor_projection_disagreement": 0, "capability_projection_disagreement": 0, "explicit_workflow_omissions": 0, "empty_input_invention": 0, "per_witness": {"13A-042": {}}, }, } ), encoding="utf-8", ) (out / "request_contract.json").write_text( json.dumps( { "measurement_kind": "focused_live_extraction", "prompt": "EXTRACTION DISCIPLINE\n", "repeats": 2, } ), encoding="utf-8", ) (out / "metadata.json").write_text( json.dumps( { "environment": "production", "admission_policy_enabled": True, "repeats": 2, } ), encoding="utf-8", ) (out / "run_01" / "13A-042.json").write_text("{}", encoding="utf-8") with pytest.raises(AssertionError, match="missing"): extraction.verify(out) (out / "run_02" / "13A-042.json").write_text("{}", encoding="utf-8") result = extraction.verify(out) assert result["verification"] == "passed" assert result["measurement_kind"] == "focused_live_extraction" finally: shutil.rmtree(out, ignore_errors=True)