"""Gate B1 scorecard: Discovery execution and readiness columns. Planner matching is an independent denominator and is not scored here. Timeout policy is a separate column and is never treated as a reliability win. """ from __future__ import annotations from collections import Counter from src.domain.projects.discovery_fidelity import ( invented_constraints, question_is_unnecessary, ) from src.domain.projects.discovery_schemas import ( DiscoveryQuestion, DiscoverySpec, DiscoveryTurnResult, ) # Same Core-computed blocker ids as validate_discovery_readiness. Kept here so # measurement can classify pairs without requiring a readiness export on main. DETERMINISTIC_BLOCKER_IDS = frozenset( { "missing_product_purpose", "missing_first_build_goal", "missing_primary_workflow", "missing_actor_model", "unresolved_tenancy", "unresolved_access_boundary", "unresolved_billing_owner", "unresolved_marketplace_relationship", "unresolved_required_integration", } ) TIMEOUT_TYPES = {"TimeoutError", "APITimeoutError"} B1_COLUMNS = ( "discovery_completion_rate", "discovery_latency_p50_p95", "discovery_timeout_rate", "first_turn_ready_rate", "unnecessary_clarification_rate", "invented_assumption_rate", "same_prompt_reproducibility", "timeout_policy_changed", ) def rate(numerator: int, denominator: int, reason: str | None = None) -> dict: return { "numerator": numerator, "denominator": denominator, "rate": numerator / denominator if denominator else None, "unmeasured_reason": reason if not denominator else None, } def latency_distribution(elapsed: list[float]) -> dict: if not elapsed: return { "p50": None, "p95": None, "n": 0, "unmeasured_reason": "No elapsed samples.", } ordered = sorted(elapsed) return { "p50": round(_percentile(ordered, 50), 3), "p95": round(_percentile(ordered, 95), 3), "n": len(ordered), "unmeasured_reason": None, } def _percentile(ordered: list[float], percent: int) -> float: if len(ordered) == 1: return ordered[0] rank = (percent / 100) * (len(ordered) - 1) low = int(rank) high = min(low + 1, len(ordered) - 1) return ordered[low] + (ordered[high] - ordered[low]) * (rank - low) def _question_from(raw: dict | None) -> DiscoveryQuestion | None: if not raw: return None return DiscoveryQuestion.model_validate(raw) def _spec_from(raw: dict | None) -> DiscoverySpec: return DiscoverySpec.model_validate(raw or {}) def trial_payload(trial: dict) -> dict: return {"messages": [{"role": "user", "content": trial["prompt"]}]} def capacity_trace_fields(trace: dict | None) -> dict: """Copy non-secret capacity diagnostics from a provider trace onto a trial row.""" trace = trace or {} return { "provider_capacity": dict(trace.get("provider_capacity") or {}), "provider_execution_ms": trace.get("provider_execution_ms"), "estimated_request_tokens": trace.get("estimated_request_tokens"), "actual_provider_usage": trace.get("actual_provider_usage"), "reservation_difference": trace.get("reservation_difference"), "provider_429_initial": bool(trace.get("provider_429_initial")), "provider_429_retry_success": bool(trace.get("provider_429_retry_success")), "provider_429_final": bool(trace.get("provider_429_final")), "rate_limit_retries": list(trace.get("rate_limit_retries") or []), } def evaluate_trial_fidelity(trial: dict) -> dict: discovery = trial.get("discovery") or {} if discovery.get("status") != "success": return { "completed": False, "timeout": discovery.get("error_type") in TIMEOUT_TYPES, "ready": False, "clarify": False, "unnecessary_clarification": False, "invented_assumption": False, "elapsed_seconds": discovery.get("elapsed_seconds"), "capacity_unavailable": discovery.get("error_type") == "ProviderCapacityUnavailableError", "provider_429": discovery.get("error_type") == "RateLimitError", } result = discovery.get("result") or {} raw_turn = discovery.get("raw_turn") or {} spec = _spec_from(result.get("spec") or raw_turn.get("spec")) question = _question_from(result.get("question") or raw_turn.get("question")) payload = trial_payload(trial) decision = result.get("decision") or (discovery.get("raw_turn") or {}).get("decision") return { "completed": True, "timeout": False, "ready": decision == "ready", "clarify": decision == "clarify", "unnecessary_clarification": decision == "clarify" and question_is_unnecessary(question, spec, payload), "invented_assumption": bool(invented_constraints(spec, payload)), "elapsed_seconds": discovery.get("elapsed_seconds"), "capacity_unavailable": discovery.get("error_type") == "ProviderCapacityUnavailableError", "provider_429": discovery.get("error_type") == "RateLimitError", } def rate_limit_retry_summary(trials: list[dict]) -> dict: """Informational only; not part of B1_COLUMNS and never substitutes for completion, timeout, or reproducibility columns. Counts client-side retries against a provider 429, introduced in `src/integrations/project_intake_client.py`.""" retried = 0 attempts = 0 successes = 0 exhausted = 0 for trial in trials: discovery = trial.get("discovery") or {} retries = discovery.get("rate_limit_retries") or [] if not retries: continue retried += 1 attempts += len(retries) if discovery.get("status") == "success": successes += 1 elif discovery.get("error_type") == "RateLimitError": exhausted += 1 return { "requests_retried": retried, "retry_attempts": attempts, "retry_successes": successes, "retry_exhausted": exhausted, } def admission_summary(trials: list[dict]) -> dict: """Informational only; not part of B1_COLUMNS. Reports whether this run went through production's real admission gate (reserve_discovery_call) and, if so, how it behaved. `reserve_discovery_call` rejects immediately rather than queueing, so there is no meaningful "admission wait" distinct from the Redis round trip (`check_seconds`).""" attempts = 0 granted = 0 denied = 0 denied_policy = 0 denied_infrastructure = 0 check_seconds: list[float] = [] for trial in trials: admission = (trial.get("discovery") or {}).get("admission") if not admission or not admission.get("attempted"): continue attempts += 1 if admission.get("granted"): granted += 1 else: denied += 1 if admission.get("denial_kind") == "infrastructure": denied_infrastructure += 1 else: denied_policy += 1 if admission.get("check_seconds") is not None: check_seconds.append(float(admission["check_seconds"])) return { "admission_policy_exercised": attempts > 0, "admission_attempts": attempts, "admission_granted": granted, "admission_denied": denied, # Distinguishes a real budget_exhausted policy denial from a Redis connection/ # timeout failure during the admission check itself — conflating the two would # misrepresent infrastructure flakiness as production capacity behavior. "admission_denied_by_policy": denied_policy, "admission_denied_by_infrastructure_error": denied_infrastructure, "admission_check_latency_p50_p95": latency_distribution(check_seconds), } def capacity_summary(trials: list[dict]) -> dict: """Informational only; not part of B1_COLUMNS. Distinguishes capacity queueing, capacity deadline, provider 429, and provider timeout. Never a semantic score.""" queued = 0 granted = 0 immediate = 0 deadline = 0 unavailable = 0 waits: list[float] = [] executions: list[float] = [] initial_429 = 0 retry_attempts = 0 retry_success = 0 retry_exhausted = 0 final_429 = 0 for trial in trials: discovery = trial.get("discovery") or {} capacity = discovery.get("provider_capacity") or {} retries = discovery.get("rate_limit_retries") or [] if capacity.get("capacity_queued"): queued += 1 if capacity.get("capacity_granted"): granted += 1 if capacity.get("capacity_immediate_grant") or ( capacity.get("capacity_granted") and not capacity.get("capacity_queued") ): immediate += 1 if capacity.get("capacity_deadline_exceeded"): deadline += 1 if discovery.get("error_type") == "ProviderCapacityUnavailableError": unavailable += 1 if capacity.get("capacity_wait_ms") is not None: waits.append(float(capacity["capacity_wait_ms"])) if capacity.get("provider_execution_ms") is not None: executions.append(float(capacity["provider_execution_ms"])) elif discovery.get("provider_execution_ms") is not None: executions.append(float(discovery["provider_execution_ms"])) if discovery.get("provider_429_initial") or discovery.get("error_type") == "RateLimitError": initial_429 += 1 retry_attempts += len(retries) if discovery.get("provider_429_retry_success"): retry_success += 1 if discovery.get("provider_429_final") or ( discovery.get("error_type") == "RateLimitError" and not discovery.get("provider_429_retry_success") ): final_429 += 1 if retries: retry_exhausted += 1 wait_dist = _ms_distribution(waits) return { "capacity_queued": queued, "capacity_granted": granted, "capacity_immediate_grants": immediate, "capacity_deadline_exceeded": deadline, "provider_capacity_unavailable": unavailable, "capacity_wait_ms_p50_p95": wait_dist, "capacity_wait_ms_max": round(max(waits), 1) if waits else None, "provider_execution_ms_p50_p95": _ms_distribution(executions), "provider_429_initial": initial_429, "retry_attempt_count": retry_attempts, "provider_429_retry_success": retry_success, "retry_exhausted_count": retry_exhausted, "provider_429_final": final_429, } def _ms_distribution(samples: list[float]) -> dict: if not samples: return {"p50": None, "p95": None, "n": 0} ordered = sorted(samples) return { "p50": round(_percentile(ordered, 50), 1), "p95": round(_percentile(ordered, 95), 1), "n": len(ordered), } def score_b1( trials: list[dict], *, timeout_policy_changed: bool = False, repeats: list[list[dict]] | None = None, ) -> dict: observations = [evaluate_trial_fidelity(trial) for trial in trials] admitted = [trial for trial in trials if trial.get("oracle_kind") == "admitted"] admitted_obs = [evaluate_trial_fidelity(trial) for trial in admitted] completed = [row for row in observations if row["completed"]] timeouts = [row for row in observations if row["timeout"]] ready = [row for row in admitted_obs if row["ready"]] clarify = [row for row in completed if row["clarify"]] unnecessary = [row for row in clarify if row["unnecessary_clarification"]] invented = [row for row in completed if row["invented_assumption"]] elapsed = [ float(row["elapsed_seconds"]) for row in observations if row.get("elapsed_seconds") is not None ] provider_elapsed = [ float((trial.get("discovery") or {}).get("provider_elapsed_seconds")) for trial in trials if (trial.get("discovery") or {}).get("provider_elapsed_seconds") is not None ] return { "phase": "13B", "gate": "B1", "corpus_size": len(trials), "capability_scoring_eligible": len(admitted), "discovery_completion_rate": rate(len(completed), len(trials) or 0), # End-to-end: includes the admission check when admission_gated=true, otherwise # identical to the provider-only figure below. "discovery_latency_p50_p95": latency_distribution(elapsed), # Provider-call-only latency; only distinct from the above under admission gating. "discovery_provider_latency_p50_p95": latency_distribution(provider_elapsed), "discovery_timeout_rate": rate(len(timeouts), len(trials) or 0), "discovery_error_types": dict( Counter( (trial.get("discovery") or {}).get("error_type", "unknown") for trial in trials if (trial.get("discovery") or {}).get("status") == "error" ) ), "first_turn_ready_rate": rate(len(ready), len(admitted) or 0), "unnecessary_clarification_rate": rate(len(unnecessary), len(clarify)), "invented_assumption_rate": rate(len(invented), len(completed)), "same_prompt_reproducibility": _reproducibility(repeats), "semantic_reproducibility": _semantic_reproducibility(repeats), "timeout_policy_changed": timeout_policy_changed, "raw_planner_control_preserved": True, "required_columns": list(B1_COLUMNS), "rate_limit_retries": rate_limit_retry_summary(trials), "admission": admission_summary(trials), "provider_capacity": capacity_summary(trials), "fidelity_scoring": ( "Keyword diagnostics only; zero findings do not certify semantic fidelity." ), } def _reproducibility(repeats: list[list[dict]] | None) -> dict: if not repeats or len(repeats) < 2: return rate(0, 0, "Same-prompt repeats were not collected in this measurement.") indexed = [{row["id"]: row for row in run} for run in repeats] ids = set().union(*(set(run) for run in indexed)) agree = 0 completed = 0 for trial_id in ids: rows = [run.get(trial_id) for run in indexed] if not all(row and row.get("discovery", {}).get("status") == "success" for row in rows): continue if len({row["prompt"] for row in rows}) != 1: raise ValueError("Repeat IDs must refer to the same prompt") completed += 1 if len({_decision_key(row) for row in rows}) == 1: agree += 1 return { **rate(agree, len(ids)), "completed_in_all_runs": completed, "runs": len(repeats), "definition": ( "Same decision and question decision_id across all runs; " "missing/error runs fail agreement." ), } def _decision_key(trial: dict) -> tuple: discovery = trial.get("discovery") or {} if discovery.get("status") != "success": return (discovery.get("status"), discovery.get("error_type")) result = discovery.get("result") or {} question = (result.get("question") or {}).get("decision_id") return (result.get("decision"), question) SEMANTIC_REPRODUCIBILITY_DEFINITION = ( "Planner-relevant projection agreement across every run for the same prompt: " "archetypes, actor-name set, entity-name set, workflow count, requested_capabilities, " "tenancy model, billing.required, and explicit deferred_scope — built only from fields " "the model already outputs and normalizes today. It answers whether repeated Discovery " "runs preserved what P2HS believes the user wants to build, not whether the surface " "text or decision_id matched. It does not consult Planner or any gold label, and it " "does not replace same_prompt_reproducibility, which remains the strict diagnostic." ) def semantic_projection(spec: DiscoverySpec, requested_capabilities: list[str] | None) -> dict: """The Planner-relevant meaning of one turn, independent of surface wording. Deliberately restricted to fields that are already structured or enumerated by the Discovery schema (archetypes, tenancy.model, billing.required, requested_capabilities) or are simple identifier sets/counts (actor and entity names, workflow count). It does not parse free-text access_scope or business_rules — doing so would mean inventing a second interpretive layer on top of Discovery, which this measurement must not do. """ return { "archetypes": frozenset(item.strip().lower() for item in spec.archetypes), "actor_count": len(spec.actors), "actor_names": frozenset(actor.name.strip().lower() for actor in spec.actors), "entities": frozenset(entity.name.strip().lower() for entity in spec.entities), "workflow_count": len(spec.workflows), "requested_capabilities": frozenset(requested_capabilities or []), "tenancy_model": spec.tenancy.model if spec.tenancy else None, "billing_required": bool(spec.billing.required) if spec.billing else False, "explicit_exclusions": frozenset(item.strip().lower() for item in spec.deferred_scope), } def _projection_from_result(result: dict) -> dict: spec = DiscoverySpec.model_validate(result.get("spec") or {}) return semantic_projection(spec, result.get("requested_capabilities")) def classify_reproducibility_pair(row_a: dict, row_b: dict) -> str: """Classify one same-prompt repeat pair. Uses only fields finalize_discovery_result and validate_discovery_readiness already compute — no new NLP/interpretive layer, no Planner, no gold label. Categories, most to least severe: execution_error, contradictory_product_interpretation, planner_relevant_spec_variance, ready_vs_clarify_same_product_spec, different_valid_primary_question, same_valid_blocker_different_question, decision_id_only, wording_only. """ disc_a = row_a.get("discovery") or {} disc_b = row_b.get("discovery") or {} if disc_a.get("status") != "success" or disc_b.get("status") != "success": return "execution_error" result_a = disc_a.get("result") or {} result_b = disc_b.get("result") or {} decision_a, decision_b = result_a.get("decision"), result_b.get("decision") same_projection = _projection_from_result(result_a) == _projection_from_result(result_b) if decision_a != decision_b: return ( "ready_vs_clarify_same_product_spec" if same_projection else "contradictory_product_interpretation" ) if decision_a != "clarify": # Both ready/blocked/out_of_scope: same_prompt_reproducibility already agrees here # by definition (no question to disagree about), so this path is not reachable as # a disagreement in practice; kept for completeness of the taxonomy. return "wording_only" if same_projection else "planner_relevant_spec_variance" question_a = result_a.get("question") or {} question_b = result_b.get("question") or {} decision_id_a, decision_id_b = question_a.get("decision_id"), question_b.get("decision_id") text_a, text_b = question_a.get("text"), question_b.get("text") if decision_id_a == decision_id_b: return "wording_only" if text_a == text_b: return "decision_id_only" if not same_projection: return "planner_relevant_spec_variance" grounded_a = decision_id_a in DETERMINISTIC_BLOCKER_IDS grounded_b = decision_id_b in DETERMINISTIC_BLOCKER_IDS if grounded_a and grounded_b: return "same_valid_blocker_different_question" return "different_valid_primary_question" def _semantic_reproducibility(repeats: list[list[dict]] | None) -> dict: if not repeats or len(repeats) < 2: return { **rate(0, 0, "Same-prompt repeats were not collected in this measurement."), "classification": {}, "definition": SEMANTIC_REPRODUCIBILITY_DEFINITION, } indexed = [{row["id"]: row for row in run} for run in repeats] ids = set().union(*(set(run) for run in indexed)) agree = 0 completed = 0 classification: Counter = Counter() for trial_id in sorted(ids): rows = [run.get(trial_id) for run in indexed] if not all(row and row.get("discovery", {}).get("status") == "success" for row in rows): classification["execution_error"] += 1 continue completed += 1 projections = [ _projection_from_result((row.get("discovery") or {}).get("result") or {}) for row in rows ] if all(projection == projections[0] for projection in projections): agree += 1 if len(rows) == 2: classification[classify_reproducibility_pair(rows[0], rows[1])] += 1 return { **rate(agree, len(ids)), "completed_in_all_runs": completed, "runs": len(repeats), "classification": dict(classification), "classification_scope": ( "Pairwise (row-by-row) for exactly 2 runs; not computed per-pair for >2 runs." if len(repeats) != 2 else "All disagreement/agreement pairs classified." ), "definition": SEMANTIC_REPRODUCIBILITY_DEFINITION, } def replay_turn(raw_turn: dict, payload: dict): from src.domain.projects.discovery_readiness import finalize_discovery_result turn = DiscoveryTurnResult.model_validate(raw_turn) return finalize_discovery_result(turn, payload)