"""Gate B1 measurement and Discovery fidelity. Does not demand Planner score lifts."""

import hashlib
import json
from pathlib import Path
from types import SimpleNamespace

import pytest

# Imported here, at collection time, with real settings — so its module chain
# (src.core.database creates an engine at import time) never initializes under a
# test's incomplete `get_settings` stub. Tests that monkeypatch reserve_discovery_call
# patch this already-imported module object, not a string import path.
import src.workers.tasks.project_discovery as project_discovery_worker  # noqa: E402
from src.domain.projects.discovery_b1 import (
    classify_reproducibility_pair,
    rate,
    replay_turn,
    score_b1,
    semantic_projection,
)
from src.domain.projects.discovery_catalog import (
    build_discovery_payload,
    model_facing_discovery_payload,
)
from src.domain.projects.discovery_fidelity import (
    apply_fidelity_hygiene,
    invented_constraints,
    normalize_capability_intent,
    question_is_unnecessary,
)
from src.domain.projects.discovery_readiness import finalize_discovery_result
from src.domain.projects.discovery_schemas import (
    Actor,
    BusinessRule,
    DiscoveryQuestion,
    DiscoverySpec,
    DiscoveryTurnResult,
    OpenDecisionOption,
    ProductDefinition,
    Workflow,
)
from src.domain.projects.discovery_schemas import Entity as SpecEntity

ROOT = Path(__file__).resolve().parents[1]
RUN_02 = ROOT / "docs/audits/fixtures/13a_run_02"
WITNESS = json.loads((RUN_02 / "trials/13A-004.json").read_text(encoding="utf-8"))


def _payload(prompt: str) -> dict:
    return {"messages": [{"role": "user", "content": prompt}]}


def _question(text: str, decision_id: str = "access_scope") -> DiscoveryQuestion:
    return DiscoveryQuestion(
        decision_id=decision_id,
        text=text,
        options=[
            OpenDecisionOption(id="admin", label="Admin - full access"),
            OpenDecisionOption(id="editor", label="Editor - can update"),
            OpenDecisionOption(id="viewer", label="Viewer - view only"),
        ],
    )


def _staff_spec(**overrides) -> DiscoverySpec:
    data = {
        "product": ProductDefinition(
            name="Customer App",
            name_status="working",
            purpose="Searchable staff-only customer records",
            first_build_goal="Search and update customer details",
            target_audience=["Staff"],
        ),
        "archetypes": ["internal_tool"],
        "actors": [
            Actor(id="staff", name="Staff", access_scope="varies based on role"),
        ],
        "workflows": [
            Workflow(id="manage", name="Manage customers", description="Search and update")
        ],
        "entities": [SpecEntity(id="customer", name="Customer", description="A person")],
        **overrides,
    }
    return DiscoverySpec(**data)


def test_staff_only_role_question_is_unnecessary():
    prompt = WITNESS["prompt"]
    question = _question("What are the access roles for staff members in the app?")
    assert question_is_unnecessary(question, _staff_spec(), _payload(prompt)) is True


def test_ambiguous_prompt_still_needs_a_real_question():
    question = DiscoveryQuestion(
        decision_id="purpose",
        text="What should this product do for your customers?",
        options=[],
    )
    assert (
        question_is_unnecessary(question, DiscoverySpec(), _payload("something for my business"))
        is False
    )


def test_invented_submission_limit_is_flagged_unique_id_is_not():
    prompt = "Public contact form. Staff review submissions."
    spec = _staff_spec(
        business_rules=[
            BusinessRule(
                id="one",
                rule="Each user can submit one inquiry at a time.",
                impact="medium",
            ),
            BusinessRule(
                id="id",
                rule="Each customer must have a unique identifier.",
                impact="high",
            ),
        ]
    )
    found = invented_constraints(spec, _payload(prompt))
    assert "Each user can submit one inquiry at a time." in found
    assert "Each customer must have a unique identifier." not in found


def test_replay_strips_13a_019_invented_submission_limit():
    witness = json.loads((RUN_02 / "trials/13A-019.json").read_text(encoding="utf-8"))
    result = replay_turn(witness["discovery"]["raw_turn"], _payload(witness["prompt"]))
    rules = [item.rule for item in result.spec.business_rules]
    assert "Each user can submit one inquiry at a time." not in rules
    assert witness["discovery"]["result"]["decision"] == "ready"


def test_hygiene_normalizes_invented_staff_role_split():
    cleaned = apply_fidelity_hygiene(_staff_spec(), _payload(WITNESS["prompt"]))
    assert cleaned.actors[0].access_scope == "all staff records in this workspace"


def test_hygiene_preserves_explicit_roles_and_external_actors():
    spec = _staff_spec(
        actors=[
            Actor(id="admin", name="Admin", access_scope="approve changes"),
            Actor(id="editor", name="Editor", access_scope="own drafts only"),
            Actor(id="customer", name="Customer", access_scope="own requests only"),
        ]
    )
    prompt = "Staff-only editor screen; admin approves changes. Customers submit requests."
    assert apply_fidelity_hygiene(spec, _payload(prompt)).actors == spec.actors
    # Merely mentioning staff must not remove a separately established customer actor.
    assert (
        apply_fidelity_hygiene(spec, _payload("Staff can review customer requests")).actors
        == spec.actors
    )


def test_ungrounded_role_ladder_collapses_to_the_generic_base_worker():
    """Positive witness (B1-S): an invented Admin/Manager/Employee ladder with zero textual
    grounding for any tier collapses to the one generic worker actually named. General rule,
    not staff-phrase-gated — the prompt below never says "staff-only" or similar."""
    spec = _staff_spec(
        actors=[
            Actor(id="admin", name="Admin", access_scope="full access"),
            Actor(id="manager", name="Manager", access_scope="team access"),
            Actor(id="employee", name="Employee", access_scope="own tasks only"),
        ]
    )
    prompt = (
        "An internal catalog of products and an ops checklist for our employees. "
        "Business record editing plus task status updates, without inventory automation."
    )
    cleaned = apply_fidelity_hygiene(spec, _payload(prompt))
    assert len(cleaned.actors) == 1
    assert cleaned.actors[0].name == "Employee"
    assert cleaned.actors[0].access_scope == "all employee records in this workspace"


def test_grounded_role_ladder_is_not_collapsed():
    """Negative witness: the same invented-looking ladder is preserved once any tier name is
    actually grounded in the user's own text."""
    spec = _staff_spec(
        actors=[
            Actor(id="admin", name="Admin", access_scope="full access"),
            Actor(id="manager", name="Manager", access_scope="team access"),
            Actor(id="employee", name="Employee", access_scope="own tasks only"),
        ]
    )
    prompt = "Admin approves budget changes for our employees."
    assert apply_fidelity_hygiene(spec, _payload(prompt)).actors == spec.actors


def test_business_specific_actor_names_are_never_collapsed():
    """Boundary witness: a domain-specific actor name (not in the closed generic vocabulary)
    must never be treated as an invented ladder, even alongside a generic tier name."""
    spec = _staff_spec(
        actors=[
            Actor(id="admin", name="Admin", access_scope="full access"),
            Actor(id="purchasing_staff", name="Purchasing Staff", access_scope="edit items"),
        ]
    )
    prompt = "Purchasing staff maintain the item directory; nothing else is stated."
    assert apply_fidelity_hygiene(spec, _payload(prompt)).actors == spec.actors


def test_reconcile_requested_capabilities_restores_omitted_records_and_auth():
    """Positive witness (B1-S): entities imply records, authentication.required implies
    auth — the exact rule DISCOVERY_PROMPT already states to the model — restored only
    when the model's own spec already supports it, never invented from nothing."""
    from src.domain.projects.discovery_fidelity import reconcile_requested_capabilities
    from src.domain.projects.discovery_schemas import AuthenticationSpec

    spec = _staff_spec(authentication=AuthenticationSpec(required=True, methods=["email"]))
    assert reconcile_requested_capabilities(spec, []) == ["auth", "records"]
    # Never removes a capability the model already chose, even an unrelated one.
    assert reconcile_requested_capabilities(spec, ["storage"]) == ["auth", "records", "storage"]


def test_reconcile_requested_capabilities_adds_nothing_without_supporting_facts():
    """Negative witness: no entities, no required auth — nothing is added."""
    from src.domain.projects.discovery_fidelity import reconcile_requested_capabilities

    spec = _staff_spec(entities=[], authentication=None)
    assert reconcile_requested_capabilities(spec, []) == []
    assert reconcile_requested_capabilities(spec, ["ai"]) == ["ai"]


def test_reconcile_requested_capabilities_does_not_add_auth_when_not_required():
    """Boundary witness: authentication present but not required must not imply auth."""
    from src.domain.projects.discovery_fidelity import reconcile_requested_capabilities
    from src.domain.projects.discovery_schemas import AuthenticationSpec

    spec = _staff_spec(authentication=AuthenticationSpec(required=False, methods=[]))
    assert reconcile_requested_capabilities(spec, []) == ["records"]


def test_customer_language_admits_auth_and_signup_without_platform_vocabulary():
    spec = DiscoverySpec(
        product=ProductDefinition(
            purpose="Customer intake website",
            first_build_goal="Let customers create accounts and sign in",
            target_audience=["Customers"],
        )
    )
    normalized = normalize_capability_intent(
        spec,
        _payload("Build a customer intake website where visitors can sign up and sign in."),
    )
    assert normalized.authentication is not None
    assert normalized.authentication.required is True
    assert normalized.authentication.audiences == ["customer"]
    assert normalized.authentication.signup_audiences == ["customer"]
    assert normalized.authentication.audience_methods["customer"] == ["magic_link", "password"]


def test_explicit_auth_method_is_preserved_and_staff_only_language_is_not_promoted():
    spec = DiscoverySpec(
        product=ProductDefinition(
            purpose="Internal operations tool",
            first_build_goal="Let staff review submissions",
        )
    )
    normalized = normalize_capability_intent(
        spec,
        _payload("Staff must log in with a password to review submissions."),
    )
    assert normalized.authentication is None

    normalized = normalize_capability_intent(
        spec,
        _payload("Customers use a magic link to sign in; staff review the submissions."),
    )
    assert normalized.authentication is not None
    assert normalized.authentication.methods == ["magic_link"]
    assert normalized.authentication.audience_methods["customer"] == ["magic_link"]


@pytest.mark.parametrize(
    "prompt,question",
    [
        (
            "Staff-only customer list with editing",
            "What user roles or permissions should staff have?",
        ),
        (
            "Employees sign in and correct phone numbers",
            "What specific details should employees correct?",
        ),
        (
            "Visitors submit messages; staff review after signing in",
            "How will staff access submissions?",
        ),
        ("Visitors submit; employees can view responses", "How will employee access be managed?"),
    ],
)
def test_supplied_staff_and_field_facts_do_not_require_restatement(prompt, question):
    assert question_is_unnecessary(_question(question), _staff_spec(), _payload(prompt))


def test_explicit_record_restrictions_are_not_treated_as_generic_staff_access():
    assert not question_is_unnecessary(
        _question("What roles or permissions should staff have?"),
        _staff_spec(),
        _payload("Staff can edit their own records; access for supervisors is undecided."),
    )


@pytest.mark.parametrize(
    "question",
    [
        "Who can approve changes?",
        "Can staff access other employees' records?",
        "Which payment provider component should handle payment?",
    ],
)
def test_consequential_questions_are_not_conventional_details(question):
    assert not question_is_unnecessary(
        _question(question), _staff_spec(), _payload("Staff-only app")
    )


def test_customer_field_paraphrase_promotes_only_without_real_blockers():
    question = _question(
        "What specific details about each customer do you want to be able to update?"
    )
    turn = DiscoveryTurnResult(
        decision="clarify",
        message="Customer list",
        question=question,
        spec=_staff_spec(),
        project_type="nuxt_app",
    )
    assert finalize_discovery_result(turn, _payload(WITNESS["prompt"])).decision == "ready"
    from src.domain.projects.discovery_schemas import TenancySpec

    turn.spec = _staff_spec(archetypes=["saas"], tenancy=TenancySpec(model="unknown"))
    assert finalize_discovery_result(turn, _payload(WITNESS["prompt"])).decision == "clarify"


def test_finalize_promotes_13a_004_witness_without_planner_input():
    turn = DiscoveryTurnResult.model_validate(WITNESS["discovery"]["raw_turn"])
    evaluated = finalize_discovery_result(turn, _payload(WITNESS["prompt"]))
    assert evaluated.decision == "ready"
    assert evaluated.question is None
    assert evaluated.specification
    assert evaluated.message != turn.message
    assert evaluated.readiness.buildable is True
    assert "conventional implementation details" in evaluated.readiness.reason
    assert "crm.customer" not in evaluated.specification


def test_grounded_blocker_id_replaces_model_free_text_id_across_repeats():
    """Two turns hitting the same Core-detected gap must expose the same decision_id even when
    the model phrases and ids its own question differently each time — this is what let
    equivalent live repeats disagree on `same_prompt_reproducibility` despite agreeing on the
    actual reason (missing_primary_workflow)."""
    spec = _staff_spec(workflows=[])
    prompt = "Staff-only customer list with editing"
    first = DiscoveryTurnResult(
        decision="clarify",
        message="Understood the audience.",
        question=DiscoveryQuestion(
            decision_id="lead_tracking_roles",
            text="What roles or permissions should staff have?",
            options=[],
        ),
        spec=spec,
        project_type="nuxt_app",
    )
    second = DiscoveryTurnResult(
        decision="clarify",
        message="Understood the audience.",
        question=DiscoveryQuestion(
            decision_id="lead_tracking_access",
            text="Who will have access to the lead tracking system?",
            options=[],
        ),
        spec=spec,
        project_type="nuxt_app",
    )
    result_a = finalize_discovery_result(first, _payload(prompt))
    result_b = finalize_discovery_result(second, _payload(prompt))
    assert result_a.decision == result_b.decision == "clarify"
    assert result_a.question.decision_id == "missing_primary_workflow"
    assert result_b.question.decision_id == "missing_primary_workflow"
    # The user-facing question text is untouched; only the tracking id is grounded.
    assert result_a.question.text == "What roles or permissions should staff have?"
    assert result_b.question.text == "Who will have access to the lead tracking system?"


def test_ungrounded_clarification_keeps_the_models_own_decision_id():
    """When no Core-deterministic blocker explains the clarify, the model's own id is kept —
    Core must not invent grounding it cannot justify."""
    turn = DiscoveryTurnResult(
        decision="clarify",
        message="Understood.",
        question=DiscoveryQuestion(
            decision_id="approval_authority",
            text="Who has authority to approve a refund?",
            options=[],
        ),
        spec=_staff_spec(),
        project_type="nuxt_app",
    )
    evaluated = finalize_discovery_result(turn, _payload("Staff-only refund approvals"))
    assert evaluated.decision == "clarify"
    assert evaluated.question.decision_id == "approval_authority"


def test_finalize_reconciles_capabilities_end_to_end():
    """Integration witness: finalize_discovery_result, not just the unit function, must
    apply the capabilities reconciliation — this is what the live evidence showed missing
    (13A-001, 13A-007, 13A-049, 13A-058, 13A-060, 13A-098)."""
    turn = DiscoveryTurnResult(
        decision="clarify",
        message="Understood.",
        question=DiscoveryQuestion(decision_id="scope", text="What else is needed?", options=[]),
        spec=_staff_spec(),
        project_type="nuxt_app",
        requested_capabilities=[],
    )
    evaluated = finalize_discovery_result(turn, _payload("Staff-only customer directory"))
    assert "records" in evaluated.requested_capabilities


def test_saas_tenancy_gap_is_not_auto_promoted():
    from src.domain.projects.discovery_schemas import TenancySpec

    turn = DiscoveryTurnResult(
        decision="clarify",
        message="Need the workspace model",
        question=DiscoveryQuestion(
            decision_id="tenancy",
            text="How should companies be separated?",
            options=[],
        ),
        spec=_staff_spec(
            archetypes=["saas"],
            tenancy=TenancySpec(model="unknown"),
        ),
        project_type="nuxt_app",
    )
    evaluated = finalize_discovery_result(turn, _payload("Build a SaaS for agencies"))
    assert evaluated.decision == "clarify"
    assert evaluated.question is not None
    assert evaluated.specification == ""


def test_model_payload_drops_worker_aliases_and_is_smaller():
    payload = build_discovery_payload(
        "nuxt_app",
        "generated",
        {"messages": [{"role": "user", "content": "Staff-only customer list"}]},
        [
            SimpleNamespace(
                ref="demo",
                name="Demo",
                description="d" * 400,
                framework="nuxt_app",
                tags=["demo"],
            )
        ],
    )
    model = model_facing_discovery_payload(payload)
    assert "messages" in payload
    assert "templates" in payload
    assert "messages" not in model
    assert "templates" not in model
    assert "platform_context" in model
    assert len(json.dumps(model)) < len(json.dumps(payload))


def test_timeout_increase_alone_is_not_a_b1_pass():
    scored = score_b1([], timeout_policy_changed=True)
    assert scored["timeout_policy_changed"] is True
    assert scored["discovery_completion_rate"]["rate"] is None
    assert scored["first_turn_ready_rate"]["rate"] is None
    assert rate(0, 0, "policy only")["rate"] is None


def test_reproducibility_does_not_count_matching_timeouts_or_missing_trials():
    failed = {
        "id": "one",
        "prompt": "same",
        "discovery": {"status": "error", "error_type": "TimeoutError"},
    }
    scored = score_b1([], repeats=[[failed], [failed]])
    assert scored["same_prompt_reproducibility"]["rate"] == 0
    assert scored["same_prompt_reproducibility"]["completed_in_all_runs"] == 0
    assert score_b1([], repeats=[[failed], []])["same_prompt_reproducibility"]["denominator"] == 1


def test_reproducibility_checks_every_run_and_rejects_changed_prompts():
    ready = {
        "id": "one",
        "prompt": "same",
        "discovery": {"status": "success", "result": {"decision": "ready"}},
    }
    clarify = {**ready, "discovery": {"status": "success", "result": {"decision": "clarify"}}}
    assert score_b1([], repeats=[[ready], [ready]])["same_prompt_reproducibility"]["rate"] == 1
    assert (
        score_b1([], repeats=[[ready], [ready], [clarify]])["same_prompt_reproducibility"]["rate"]
        == 0
    )
    with pytest.raises(ValueError, match="same prompt"):
        score_b1([], repeats=[[ready], [{**ready, "prompt": "different"}]])


@pytest.mark.asyncio
async def test_live_measure_rejects_timeout_change_before_provider_calls(monkeypatch):
    from scripts.measure_discovery_readiness_13b import live_measure

    monkeypatch.setattr(
        "src.config.get_settings", lambda: SimpleNamespace(project_intake_timeout_seconds=30)
    )
    with pytest.raises(ValueError, match="frozen 8s"):
        await live_measure([], ROOT, {}, 2)


def test_harness_protects_frozen_parent_directory():
    from scripts.measure_discovery_readiness_13b import measure

    with pytest.raises(ValueError, match="Refusing to write"):
        measure(SimpleNamespace(out=str(RUN_02.parent), allow_existing=True))


@pytest.mark.asyncio
async def test_live_harness_preserves_errors_and_uses_fresh_first_turns(monkeypatch):
    from scripts import measure_discovery_readiness_13b as harness

    saved, calls = {}, []
    monkeypatch.setattr(
        "src.config.get_settings",
        lambda: SimpleNamespace(
            project_intake_timeout_seconds=8,
            project_intake_model="gpt-4o-mini",
            openai_api_key="test-only",
            project_intake_calls_per_minute=60,
        ),
    )

    async def context(_baseline):
        return (
            {"type": "nuxt_app", "source_kind": "generated"},
            [
                SimpleNamespace(
                    ref=str(i), name="Demo", description="Demo", framework="nuxt_app", tags=[]
                )
                for i in range(24)
            ],
            ["145"],
        )

    async def provider(payload, settings, *, trace):
        calls.append(payload)
        if len(calls) == 1:
            raise TimeoutError()
        return DiscoveryTurnResult.model_validate(WITNESS["discovery"]["raw_turn"])

    async def no_sleep(_delay):
        pass

    monkeypatch.setattr(harness, "live_context", context)
    monkeypatch.setattr(harness, "write_json", lambda path, value: saved.update({str(path): value}))
    monkeypatch.setattr(harness.asyncio, "sleep", no_sleep)
    monkeypatch.setattr("src.integrations.project_intake_client.discover_project", provider)
    score = await harness.live_measure([WITNESS], ROOT / "unused-live-test", {}, 2)
    assert score["discovery_timeout_rate"]["numerator"] == 1
    assert score["per_run"][1]["discovery_completion_rate"]["numerator"] == 1
    assert score["same_prompt_reproducibility"]["rate"] == 0
    assert len(calls) == 2
    assert calls[0] == calls[1]
    assert calls[0]["discovery_state"]["spec"] is None
    assert "expected_product_blocks" not in json.dumps(calls)
    assert any("run_01" in path and "trials" in path for path in saved)


@pytest.mark.asyncio
async def test_admission_gated_denial_never_calls_the_provider(monkeypatch):
    """M1: --admission-gated must route through the real production gate
    (reserve_discovery_call) and, when it denies, must not fall back to calling the
    provider directly — that would silently misrepresent AdmissionDenied as completion."""
    from scripts import measure_discovery_readiness_13b as harness

    saved: dict = {}
    monkeypatch.setattr(
        "src.config.get_settings",
        lambda: SimpleNamespace(
            project_intake_timeout_seconds=8,
            project_intake_model="gpt-4o-mini",
            openai_api_key="test-only",
            project_intake_calls_per_minute=60,
        ),
    )

    async def context(_baseline):
        return (
            {"type": "nuxt_app", "source_kind": "generated"},
            [
                SimpleNamespace(
                    ref=str(i), name="Demo", description="Demo", framework="nuxt_app", tags=[]
                )
                for i in range(24)
            ],
            ["145"],
        )

    provider_calls = []

    async def provider(payload, settings, *, trace):
        provider_calls.append(payload)
        return DiscoveryTurnResult.model_validate(WITNESS["discovery"]["raw_turn"])

    async def deny(_settings):
        raise ValueError("budget_exhausted")

    async def no_sleep(_delay):
        pass

    monkeypatch.setattr(harness, "live_context", context)
    monkeypatch.setattr(harness, "write_json", lambda path, value: saved.update({str(path): value}))
    monkeypatch.setattr(harness.asyncio, "sleep", no_sleep)
    monkeypatch.setattr("src.integrations.project_intake_client.discover_project", provider)
    monkeypatch.setattr(project_discovery_worker, "reserve_discovery_call", deny)
    score = await harness.live_measure(
        [WITNESS], ROOT / "unused-admission-test", {}, 2, admission_gated=True
    )
    assert provider_calls == []
    # The top-level scorecard is scored from run 1 only (existing score_b1 convention,
    # unchanged here); both runs are checked individually via per_run below.
    assert score["discovery_completion_rate"]["numerator"] == 0
    assert score["discovery_error_types"] == {"AdmissionDenied": 1}
    assert score["admission"]["admission_policy_exercised"] is True
    assert score["admission"]["admission_attempts"] == 1
    assert score["admission"]["admission_granted"] == 0
    assert score["admission"]["admission_denied"] == 1
    for run_score in score["per_run"]:
        assert run_score["admission"]["admission_attempts"] == 1
        assert run_score["admission"]["admission_denied"] == 1
    metadata_writes = [v for k, v in saved.items() if k.endswith("metadata.json")]
    assert any(m.get("measurement_kind") == "production_admission_gated" for m in metadata_writes)
    assert any(m.get("admission_policy_enabled") is True for m in metadata_writes)


@pytest.mark.asyncio
async def test_admission_gated_grant_still_calls_the_provider_and_records_latency(monkeypatch):
    from scripts import measure_discovery_readiness_13b as harness

    saved: dict = {}
    monkeypatch.setattr(
        "src.config.get_settings",
        lambda: SimpleNamespace(
            project_intake_timeout_seconds=8,
            project_intake_model="gpt-4o-mini",
            openai_api_key="test-only",
            project_intake_calls_per_minute=60,
        ),
    )

    async def context(_baseline):
        return (
            {"type": "nuxt_app", "source_kind": "generated"},
            [
                SimpleNamespace(
                    ref=str(i), name="Demo", description="Demo", framework="nuxt_app", tags=[]
                )
                for i in range(24)
            ],
            ["145"],
        )

    async def provider(payload, settings, *, trace):
        return DiscoveryTurnResult.model_validate(WITNESS["discovery"]["raw_turn"])

    async def grant(_settings):
        return None

    async def no_sleep(_delay):
        pass

    monkeypatch.setattr(harness, "live_context", context)
    monkeypatch.setattr(harness, "write_json", lambda path, value: saved.update({str(path): value}))
    monkeypatch.setattr(harness.asyncio, "sleep", no_sleep)
    monkeypatch.setattr("src.integrations.project_intake_client.discover_project", provider)
    monkeypatch.setattr(project_discovery_worker, "reserve_discovery_call", grant)
    score = await harness.live_measure(
        [WITNESS], ROOT / "unused-admission-test-2", {}, 2, admission_gated=True
    )
    # score_b1's top-level scorecard is run 1 only; check both runs via per_run.
    assert score["discovery_completion_rate"]["numerator"] == 1
    assert score["admission"]["admission_granted"] == 1
    assert score["admission"]["admission_denied"] == 0
    assert score["admission"]["admission_check_latency_p50_p95"]["n"] == 1
    assert score["discovery_provider_latency_p50_p95"]["n"] == 1
    for run_score in score["per_run"]:
        assert run_score["discovery_completion_rate"]["numerator"] == 1


@pytest.mark.asyncio
async def test_admission_gated_redis_failure_is_recorded_per_trial_not_fatal(monkeypatch):
    """A Redis connection failure during the admission check must not crash the whole
    asyncio.gather() batch (observed live: a bare redis.exceptions.TimeoutError from
    reserve_discovery_call propagated past a narrower `except ValueError` and aborted the
    entire run). It must be recorded per-trial and distinguished from a real policy denial."""
    from scripts import measure_discovery_readiness_13b as harness

    saved: dict = {}
    monkeypatch.setattr(
        "src.config.get_settings",
        lambda: SimpleNamespace(
            project_intake_timeout_seconds=8,
            project_intake_model="gpt-4o-mini",
            openai_api_key="test-only",
            project_intake_calls_per_minute=60,
        ),
    )

    async def context(_baseline):
        return (
            {"type": "nuxt_app", "source_kind": "generated"},
            [
                SimpleNamespace(
                    ref=str(i), name="Demo", description="Demo", framework="nuxt_app", tags=[]
                )
                for i in range(24)
            ],
            ["145"],
        )

    provider_calls = []

    async def provider(payload, settings, *, trace):
        provider_calls.append(payload)
        return DiscoveryTurnResult.model_validate(WITNESS["discovery"]["raw_turn"])

    async def redis_down(_settings):
        raise TimeoutError("Timeout connecting to server")

    async def no_sleep(_delay):
        pass

    monkeypatch.setattr(harness, "live_context", context)
    monkeypatch.setattr(harness, "write_json", lambda path, value: saved.update({str(path): value}))
    monkeypatch.setattr(harness.asyncio, "sleep", no_sleep)
    monkeypatch.setattr("src.integrations.project_intake_client.discover_project", provider)
    monkeypatch.setattr(project_discovery_worker, "reserve_discovery_call", redis_down)
    score = await harness.live_measure(
        [WITNESS], ROOT / "unused-admission-test-3", {}, 2, admission_gated=True
    )
    assert provider_calls == []
    assert score["discovery_error_types"] == {"AdmissionCheckFailed:TimeoutError": 1}
    assert score["admission"]["admission_denied"] == 1
    assert score["admission"]["admission_denied_by_infrastructure_error"] == 1
    assert score["admission"]["admission_denied_by_policy"] == 0


def _semantic_row(decision, *, spec=None, question=None, capabilities=None, status="success"):
    return {
        "discovery": {
            "status": status,
            "result": {
                "decision": decision,
                "spec": spec or {},
                "question": question,
                "requested_capabilities": capabilities or [],
            },
        }
    }


_STAFF_SPEC = {
    "archetypes": ["internal_tool"],
    "actors": [{"id": "staff", "name": "Staff"}],
    "entities": [{"id": "customer", "name": "Customer", "description": "x"}],
}
_SAAS_SPEC = {
    "archetypes": ["saas"],
    "actors": [{"id": "owner", "name": "Owner"}, {"id": "member", "name": "Member"}],
    "entities": [{"id": "org", "name": "Organization", "description": "x"}],
}


def test_semantic_projection_ignores_free_text_access_scope_wording():
    a = semantic_projection(
        DiscoverySpec.model_validate(
            {**_STAFF_SPEC, "actors": [{"id": "staff", "name": "Staff", "access_scope": "full"}]}
        ),
        [],
    )
    b = semantic_projection(
        DiscoverySpec.model_validate(
            {
                **_STAFF_SPEC,
                "actors": [{"id": "staff", "name": "Staff", "access_scope": "everything"}],
            }
        ),
        [],
    )
    assert a == b


def test_classify_execution_error():
    a = _semantic_row("clarify", status="error")
    b = _semantic_row("clarify", spec=_STAFF_SPEC)
    assert classify_reproducibility_pair(a, b) == "execution_error"


def test_classify_ready_vs_clarify_same_spec():
    a = _semantic_row("ready", spec=_STAFF_SPEC)
    b = _semantic_row(
        "clarify",
        spec=_STAFF_SPEC,
        question={"decision_id": "access_scope", "text": "Who has access?"},
    )
    assert classify_reproducibility_pair(a, b) == "ready_vs_clarify_same_product_spec"


def test_classify_contradictory_product_interpretation():
    a = _semantic_row("ready", spec=_STAFF_SPEC)
    b = _semantic_row(
        "clarify",
        spec=_SAAS_SPEC,
        question={"decision_id": "tenancy", "text": "How are orgs separated?"},
    )
    assert classify_reproducibility_pair(a, b) == "contradictory_product_interpretation"


def test_classify_wording_only_same_decision_id():
    a = _semantic_row(
        "clarify", spec=_STAFF_SPEC, question={"decision_id": "access", "text": "Who can see it?"}
    )
    b = _semantic_row(
        "clarify",
        spec=_STAFF_SPEC,
        question={"decision_id": "access", "text": "Which staff can access this?"},
    )
    assert classify_reproducibility_pair(a, b) == "wording_only"


def test_classify_decision_id_only_same_text_different_id():
    question_text = "Who can access customer records?"
    a = _semantic_row(
        "clarify", spec=_STAFF_SPEC, question={"decision_id": "access_a", "text": question_text}
    )
    b = _semantic_row(
        "clarify", spec=_STAFF_SPEC, question={"decision_id": "access_b", "text": question_text}
    )
    assert classify_reproducibility_pair(a, b) == "decision_id_only"


def test_classify_different_valid_primary_question_same_spec():
    a = _semantic_row(
        "clarify",
        spec=_STAFF_SPEC,
        question={"decision_id": "data_retention", "text": "Should messages be kept?"},
    )
    b = _semantic_row(
        "clarify",
        spec=_STAFF_SPEC,
        question={"decision_id": "field_scope", "text": "What fields matter?"},
    )
    assert classify_reproducibility_pair(a, b) == "different_valid_primary_question"


def test_classify_same_valid_blocker_different_question():
    a = _semantic_row(
        "clarify",
        spec=_STAFF_SPEC,
        question={"decision_id": "missing_primary_workflow", "text": "What is the workflow?"},
    )
    b = _semantic_row(
        "clarify",
        spec=_STAFF_SPEC,
        question={"decision_id": "unresolved_access_boundary", "text": "Who can access this?"},
    )
    assert classify_reproducibility_pair(a, b) == "same_valid_blocker_different_question"


def test_classify_planner_relevant_spec_variance():
    a = _semantic_row(
        "clarify",
        spec=_STAFF_SPEC,
        question={"decision_id": "data_retention", "text": "Should messages be kept?"},
    )
    b = _semantic_row(
        "clarify",
        spec=_SAAS_SPEC,
        question={"decision_id": "field_scope", "text": "What fields matter?"},
    )
    assert classify_reproducibility_pair(a, b) == "planner_relevant_spec_variance"


def test_semantic_reproducibility_side_by_side_with_strict_metric():
    run1 = [
        {
            "id": "one",
            "prompt": "same",
            "discovery": _semantic_row(
                "clarify",
                spec=_STAFF_SPEC,
                question={"decision_id": "data_retention", "text": "Should messages be kept?"},
            )["discovery"],
        }
    ]
    run2 = [
        {
            "id": "one",
            "prompt": "same",
            "discovery": _semantic_row(
                "clarify",
                spec=_STAFF_SPEC,
                question={"decision_id": "field_scope", "text": "What fields matter?"},
            )["discovery"],
        }
    ]
    scored = score_b1([], repeats=[run1, run2])
    # Strict metric disagrees (different decision_id); semantic metric agrees (same spec).
    assert scored["same_prompt_reproducibility"]["rate"] == 0
    assert scored["semantic_reproducibility"]["rate"] == 1
    assert scored["semantic_reproducibility"]["classification"] == {
        "different_valid_primary_question": 1
    }


@pytest.mark.asyncio
async def test_live_replay_rejects_changed_payload_before_writing(monkeypatch):
    from scripts import measure_discovery_readiness_13b as harness

    async def context(_metadata):
        return {"type": "nuxt_app", "source_kind": "generated"}, [], ["145"]

    monkeypatch.setattr(harness, "live_context", context)
    monkeypatch.setattr(
        harness,
        "load_trials",
        lambda _source: [{**WITNESS, "payload_sha256": "different-catalog-or-payload"}],
    )
    original_read = Path.read_text

    def read(path, *args, **kwargs):
        if path.name == "metadata.json":
            return json.dumps({"mode": "live", "status": "complete", "repeat_runs": 2})
        return original_read(path, *args, **kwargs)

    monkeypatch.setattr(Path, "read_text", read)

    def no_write(*args):
        pytest.fail("Changed payload must not produce evidence")

    monkeypatch.setattr(harness, "write_json", no_write)
    with pytest.raises(ValueError, match="non-equivalent replay"):
        await harness.replay_live(ROOT / "fake-source", ROOT / "unused-live-replay-test")


def test_replay_uses_frozen_raw_turn_not_live_provider():
    result = replay_turn(WITNESS["discovery"]["raw_turn"], _payload(WITNESS["prompt"]))
    assert result.decision == "ready"
    assert WITNESS["discovery"]["result"]["decision"] == "clarify"


def test_harness_refuses_to_write_frozen_13a_evidence():
    from scripts.measure_discovery_readiness_13b import measure

    args = SimpleNamespace(
        out=str(ROOT / "docs/audits/fixtures/13a_run_02"),
        source=str(RUN_02),
        mode="before",
        allow_existing=True,
    )
    with pytest.raises(ValueError, match="Refusing to write"):
        measure(args)


@pytest.mark.asyncio
async def test_structured_call_retries_once_after_rate_limit_then_succeeds(monkeypatch):
    """A 429 must not be treated as a timeout or a completion failure when the provider
    recovers within the same request budget — this is what let a transient rate-limit
    collapse second-pass completion during equal-load validation (79/120, 26 429s)."""
    import httpx
    import openai
    from src.domain.projects.discovery_schemas import DiscoveryTurnResult
    from src.integrations import project_intake_client as client_module

    request = httpx.Request("POST", "https://api.openai.com/v1/chat/completions")
    rate_limited = openai.RateLimitError(
        "rate limited",
        response=httpx.Response(429, request=request, headers={"retry-after": "0.01"}),
        body=None,
    )
    parsed = WITNESS["discovery"]["raw_turn"]
    success_message = SimpleNamespace(
        refusal=None, parsed=DiscoveryTurnResult.model_validate(parsed)
    )
    success_response = SimpleNamespace(
        choices=[SimpleNamespace(message=success_message)],
        model_dump=lambda **_kw: {"ok": True},
    )
    calls = []

    async def parse(**kwargs):
        calls.append(kwargs)
        if len(calls) == 1:
            raise rate_limited
        return success_response

    class FakeClient:
        def __init__(self, **_kwargs):
            self.chat = SimpleNamespace(completions=SimpleNamespace(parse=parse))

        async def __aenter__(self):
            return self

        async def __aexit__(self, *_exc):
            return False

    monkeypatch.setattr(openai, "AsyncOpenAI", FakeClient)
    settings = SimpleNamespace(openai_api_key="test", project_intake_model="gpt-4o-mini")
    trace: dict = {}
    result = await client_module._project_structured_call(
        {"messages": []}, settings, "prompt", DiscoveryTurnResult, 8, trace=trace
    )
    assert result == success_message.parsed
    assert len(calls) == 2
    assert trace["rate_limit_retries"] == [{"attempt": 1, "retry_after": 0.01}]


@pytest.mark.asyncio
async def test_structured_call_still_raises_after_repeated_rate_limits(monkeypatch):
    import httpx
    import openai
    from src.domain.projects.discovery_schemas import DiscoveryTurnResult
    from src.integrations import project_intake_client as client_module

    request = httpx.Request("POST", "https://api.openai.com/v1/chat/completions")

    async def parse(**_kwargs):
        raise openai.RateLimitError(
            "rate limited",
            response=httpx.Response(429, request=request, headers={}),
            body=None,
        )

    class FakeClient:
        def __init__(self, **_kwargs):
            self.chat = SimpleNamespace(completions=SimpleNamespace(parse=parse))

        async def __aenter__(self):
            return self

        async def __aexit__(self, *_exc):
            return False

    monkeypatch.setattr(openai, "AsyncOpenAI", FakeClient)
    settings = SimpleNamespace(openai_api_key="test", project_intake_model="gpt-4o-mini")
    with pytest.raises(openai.RateLimitError):
        await client_module._project_structured_call(
            {"messages": []}, settings, "prompt", DiscoveryTurnResult, 8
        )


def test_b1_comparison_keeps_b2_closed_and_timeout_unchanged():
    card = json.loads((ROOT / "docs/audits/fixtures/13b_b1_comparison.json").read_text())
    assert card["phase_13b_authorized"] is True
    assert card["phase_13b_gate"] == "B1"
    assert card["b2_authorized"] is False
    assert card["b1_exit_satisfied"] is False
    assert card["timeout_policy_changed"] is False
    assert card["planner_changed"] is False
    assert card["13a_evidence_overwritten"] is False
    assert card["before"]["discovery_completion_rate"] == "36/120"
    assert card["after"]["first_turn_ready_rate"] == "6/108"
    assert card["after"]["unnecessary_clarification_rate"] == "0/30"
    assert card["after"]["invented_assumption_rate"] == "0/36"


def test_frozen_13a_evidence_is_not_rewritten_by_b1_authorization():
    freeze = json.loads((ROOT / "docs/audits/fixtures/13a_run_02_freeze.json").read_text())
    run_01 = ROOT / "docs/audits/fixtures/13a_run_01/scorecard.json"
    run_02 = RUN_02 / "scorecard.json"
    assert freeze["phase_13b_authorized"] is False
    assert freeze["phase_13b_gate"] == "none"
    assert freeze["implementation_started"] is False
    assert freeze["discovery_completion"] == "36/120"
    assert freeze["first_turn_ready"] == "4/108"
    assert run_01.is_file()
    assert run_02.is_file()
    assert hashlib.sha256(run_02.read_bytes()).hexdigest()
    brief = (ROOT / "docs/audits/PHASE_13B_IMPLEMENTATION_BRIEF.md").read_text(encoding="utf-8")
    assert "AUTHORIZE PHASE 13B." in brief
    assert "Do not begin Gate B2 until B1" in brief
    assert "timeout_increase_is_not_success=yes" in brief


def _template_turn(project_type: str, template_ref: str) -> DiscoveryTurnResult:
    return DiscoveryTurnResult(
        decision="clarify",
        message="Understood.",
        question=DiscoveryQuestion(decision_id="scope", text="What else is needed?", options=[]),
        spec=_staff_spec(),
        project_type=project_type,
        requested_capabilities=[],
        template_ref=template_ref,
        template_fit="strong",
    )


_TEMPLATE_PAYLOAD = {
    "messages": [{"role": "user", "content": "Landing page with a lead form"}],
    "templates": [
        {"ref": "revio-landing-page-react", "framework": "react_app", "auto_selectable": True},
        {"ref": "revio-landing-page-nuxt", "framework": "nuxt_app", "auto_selectable": True},
    ],
}


def test_template_from_another_framework_is_dropped():
    """Regression: a react template paired with an html/nuxt project type made the worker
    reject the whole result as invalid_template_selection. Finalize must drop it instead."""
    evaluated = finalize_discovery_result(
        _template_turn("nuxt_app", "revio-landing-page-react"), _TEMPLATE_PAYLOAD
    )
    assert evaluated.template_ref is None