"""Admission-gated 120-prompt harness: no provider, network, or production writes.""" import asyncio import json import shutil from types import SimpleNamespace import pytest from scripts import measure_discovery_admission_gated_13b as gated def _success(trial_id, prompt, spec, *, decision="ready", question_id=None, caps=None): return { "id": trial_id, "prompt": prompt, "oracle_kind": "admitted", "stratum": "crm", "discovery": { "status": "success", "elapsed_seconds": 1.2, "provider_elapsed_seconds": 1.0, "admission": {"attempted": True, "granted": True, "check_seconds": 0.002}, "raw_turn": { "spec": spec, "requested_capabilities": caps or [], }, "result": { "decision": decision, "question": {"decision_id": question_id} if question_id else {}, "spec": spec, }, }, } def test_oracle_identity_is_the_frozen_120(): prompts = gated.load_oracle() assert len(prompts) == 120 assert prompts[0]["id"] == "13A-001" assert prompts[-1]["id"] == "13A-120" assert {row["id"] for row in prompts} == {f"13A-{i:03d}" for i in range(1, 121)} assert any(row["prompt"] == "Help with bookings." for row in prompts) blocks = next(row for row in prompts if row["id"] == "13A-001")["expected_product_blocks"] assert blocks == ["crm.customer"] def test_harness_does_not_score_planner(): payload = json.loads(gated.ORACLE_PATH.read_text(encoding="utf-8")) assert payload["planner_scored_by_this_harness"] is False empty = {"entities": [], "workflows": [], "archetypes": [], "actors": []} scored = gated.scorecard( [ [_success("13A-001", "a", empty)], [_success("13A-001", "a", empty)], ] ) assert scored["planner_scored"] is False assert scored["raw_planner_control_preserved"] is True assert scored["measurement_kind"] == gated.MEASUREMENT_KIND def test_layer_a_diagnostics_and_option_a_watchlist(): spec_a = { "entities": [{"id": "e1", "name": "Job Request", "description": "jobs"}], "workflows": [{"id": "w1", "name": "Intake", "description": "intake"}], "archetypes": ["internal_tool"], "actors": [{"id": "a1", "name": "Staff"}], } spec_b = { "entities": [{"id": "e1", "name": "quote request", "description": "quotes"}], "workflows": [], "archetypes": ["crm"], "actors": [{"id": "a1", "name": "staff"}, {"id": "a2", "name": "customer"}], } empty = {"entities": [], "workflows": [], "archetypes": [], "actors": []} booking = { "entities": [{"id": "e1", "name": "Booking", "description": "booking"}], "workflows": [{"id": "w1", "name": "Book", "description": "book"}], "archetypes": [], "actors": [], } runs = [ [ _success("13A-025", "Our removal business", spec_a, caps=["auth"]), _success("13A-085", "Help with bookings.", empty), ], [ _success("13A-025", "Our removal business", spec_b, caps=["records"]), _success("13A-085", "Help with bookings.", booking), ], ] result = gated.scorecard(runs) layer = result["layer_a_raw_turn"] assert layer["pair_count"] == 2 assert layer["entity_projection_disagreement"] == 2 assert layer["workflow_projection_disagreement"] == 2 assert layer["archetype_disagreement"] == 1 assert layer["actor_projection_disagreement"] == 1 assert layer["capability_projection_disagreement"] == 1 assert layer["boundary_help_with_bookings_entity_or_workflow"] == 1 assert layer["empty_input_invention"] == 0 assert layer["hello_negative_control_in_corpus"] is False assert layer["watchlist"]["13A-025"][0]["entities"] == ["job request"] def test_protected_outputs_include_historical_denominators(): names = {path.name for path in gated.PROTECTED_OUTPUTS} assert "13a_run_02" in names assert "13b_b1_admission_gated_01" in names assert "13b_b1_extraction_prod_host_01" in names assert "13b_b1_admission_prod_host_01.json" in names def test_refuses_overwrite_and_protected_paths(): existing = gated.ROOT / ".test-tmp-admission-gated-existing" existing.mkdir(exist_ok=True) try: with pytest.raises(ValueError, match="new experiment"): asyncio.run(gated.measure(existing, 2, "production", 1.1)) with pytest.raises(ValueError, match="protected historical fixture"): asyncio.run( gated.measure( gated.ROOT / "docs/audits/fixtures/13b_b1_admission_gated_01", 2, "production", 1.1, ) ) finally: existing.rmdir() def test_preflight_requires_extraction_contract(monkeypatch): monkeypatch.setattr( gated, "DISCOVERY_PROMPT", "UNDERSTAND BEFORE ASKING\nINFERENCE POLICY", ) monkeypatch.setattr( gated, "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="authorized extraction contract"): gated.preflight() def test_verify_requires_production_kind_and_oracle_files(): out = gated.ROOT / ".test-tmp-admission-gated-verify" if out.exists(): shutil.rmtree(out) (out / "run_01" / "trials").mkdir(parents=True) (out / "run_02" / "trials").mkdir(parents=True) try: prompts = gated.load_oracle() (out / "oracle_identity.json").write_text( json.dumps({"oracle_prompts_sha256": gated.ORACLE_PROMPTS_SHA256}), encoding="utf-8", ) (out / "request_contract.json").write_text( json.dumps( { "measurement_kind": gated.MEASUREMENT_KIND, "planner_scored": False, } ), encoding="utf-8", ) (out / "scorecard.json").write_text( json.dumps( { "measurement_kind": gated.MEASUREMENT_KIND, "raw_planner_control_preserved": True, "planner_scored": False, "discovery_completion_rate": {"numerator": 120, "denominator": 120}, "admission": {"admission_policy_exercised": True}, "layer_a_raw_turn": {}, } ), encoding="utf-8", ) (out / "metadata.json").write_text( json.dumps( { "measurement_kind": gated.MEASUREMENT_KIND, "environment": "production", "admission_policy_enabled": True, "planner_scored": False, "repeat_runs": 2, } ), encoding="utf-8", ) for repeat in (1, 2): for case in prompts: (out / f"run_{repeat:02}" / "trials" / f"{case['id']}.json").write_text( "{}", encoding="utf-8", ) result = gated.verify(out) assert result["verification"] == "passed" assert result["oracle_count"] == 120 finally: shutil.rmtree(out)