diff --git a/.superpowers/sdd/task-2-report.md b/.superpowers/sdd/task-2-report.md index a1951300..ed0c7c07 100644 --- a/.superpowers/sdd/task-2-report.md +++ b/.superpowers/sdd/task-2-report.md @@ -13,8 +13,10 @@ ## Files changed - `scripts/dynamic_rectification.py` +- `scripts/dynamic_rectification_opportunities.py` - `scripts/jyotish_api_server.py` - `tests/test_dynamic_rectification.py` +- `tests/test_dynamic_rectification_scoring.py` - `tests/test_active_rectification_api.py` - `.superpowers/sdd/task-2-report.md` @@ -61,5 +63,33 @@ ## Concerns -- `scripts/dynamic_rectification.py` is above the optional 250 pure-LOC design guideline because the approved Task 2 ownership explicitly requires candidate generation, persisted-model validation, opportunity construction, and versioned scoring in this single module. Splitting it would require expanding the approved file ownership; the code is separated into small pure helpers meanwhile. - Full-file Ruff on `scripts/jyotish_api_server.py` still reports inherited baseline debt (import ordering, legacy f-strings, an existing undefined `swe`, and other unrelated diagnostics). Ruff is clean for the new module and both modified test files; compileall and all focused suites pass. + +## Review fixes + +- Both dynamic routes now fail closed unless `JYOTISH_DYNAMIC_RECTIFICATION_TOKEN` is configured and the request carries its exact bearer value. Comparison uses `secrets.compare_digest`; missing and wrong credentials are rejected before payload validation or scoring. +- Removed both routes from `API_COMMAND_MAP`, `TECHNIQUE_EXAMPLE_ENDPOINTS`, technique-example dispatch, and generated technique summaries. They remain direct authenticated POST routes only. +- Latitude, longitude, and timezone are required. The normalized location triple is persisted inside the candidate model and exact-matched during reuse. +- Adjudication preserves submitted candidate order, so an overnight `23:59` to `00:00` leader is one contiguous segment. +- Question IDs are trimmed opaque nonempty strings. Duplicate detection uses the normalized value and remains enforced. +- Split candidate-model/opportunity work into `dynamic_rectification_opportunities.py` and scoring regressions into `test_dynamic_rectification_scoring.py` without changing public entrypoints. +- Pure LOC after the split: public engine `215`, opportunity module `231`, opportunity tests `168`, scoring/auth tests `213`, active API tests `230`. + +### Review RED + +1. Location/timezone reuse tests initially passed for the wrong reason because the old model rejected the new location field entirely; the original same-location reuse regression also failed until location became part of the canonical model contract. +2. Overnight and opaque-ID regressions failed with UUID validation; the independent reviewer reproduction also showed wall-clock sorting split `23:59` and `00:00` into tied leaders. +3. Missing/wrong bearer regressions reached request validation/scoring instead of raising `Forbidden`; a forged unauthenticated four-row request could therefore reach the scorer. +4. Missing `lat`, `lon`, or `tz` silently normalized to zero. +5. Browser-runnable registration assertions failed because both private endpoints appeared in technique examples, command mapping, dispatch, and summaries. + +### Review GREEN + +1. `/Users/jesse/Downloads/Copse/astrology/yinduzhanxing/.venv/bin/python -m pytest -q tests/test_dynamic_rectification.py tests/test_dynamic_rectification_scoring.py tests/test_active_rectification_api.py tests/test_active_rectification_questions.py tests/test_active_rectification_events.py` + - `45` passed, `0` failed. +2. `/Users/jesse/Downloads/Copse/astrology/yinduzhanxing/.venv/bin/python -m ruff check scripts/dynamic_rectification.py scripts/dynamic_rectification_opportunities.py tests/test_dynamic_rectification.py tests/test_dynamic_rectification_scoring.py tests/test_active_rectification_api.py` + - Passed with no diagnostics after correcting one import-order finding. +3. `/Users/jesse/Downloads/Copse/astrology/yinduzhanxing/.venv/bin/python -m compileall -q scripts/dynamic_rectification.py scripts/dynamic_rectification_opportunities.py scripts/jyotish_api_server.py` + - Passed. +4. `git diff --check` + - Passed after removing one trailing blank line in the API regression file. diff --git a/scripts/dynamic_rectification.py b/scripts/dynamic_rectification.py index 02a4bdc3..759fe248 100644 --- a/scripts/dynamic_rectification.py +++ b/scripts/dynamic_rectification.py @@ -3,26 +3,35 @@ # dependencies = [] # /// # ─── How to run ─── -# .venv/bin/python -m pytest -q tests/test_dynamic_rectification.py -"""Candidate-backed opportunities and deterministic dynamic-choice scoring.""" +# .venv/bin/python -m pytest -q tests/test_dynamic_rectification_scoring.py +"""Public dynamic-rectification packet and deterministic scoring entrypoints.""" from __future__ import annotations -import hashlib -import json -import math -from collections import defaultdict -from collections.abc import Mapping, Sequence -from datetime import date, datetime, time, timedelta -from typing import Final, Literal, TypedDict -from uuid import NAMESPACE_URL, UUID, uuid5 +from collections.abc import Sequence +from typing import Literal, TypedDict +from uuid import NAMESPACE_URL, uuid5 -ALGORITHM_VERSION: Final = "birth-time-choice-scoring-v2" -MIN_INFORMATION_GAIN: Final = 0.15 -SUPPORTED_DIMENSIONS: Final = frozenset( - {"education", "relocation", "relationship", "career", "health_pressure"} +from scripts.dynamic_rectification_opportunities import ( + ALGORITHM_VERSION, + SUPPORTED_DIMENSIONS, + candidate_times, + candidate_window_rows, + canonical_hash, + compute_candidate_model, + experience_windows, + opportunities, + validate_candidate_model, ) +Confidence = Literal["low", "medium", "high"] +_candidate_times = candidate_times +_candidate_window_rows = candidate_window_rows +_canonical_hash = canonical_hash +_experience_windows = experience_windows +_opportunities = opportunities +_validate_candidate_model = validate_candidate_model + class ChoiceRow(TypedDict): time: str @@ -36,228 +45,40 @@ class WinningSegment(TypedDict): width_minutes: int -Confidence = Literal["low", "medium", "high"] - - -def _canonical_hash(value: Mapping | Sequence) -> str: - encoded = json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) - return hashlib.sha256(encoded.encode("utf-8")).hexdigest() - - -def _candidate_times(birth_date: str, start_time: str, end_time: str) -> list[str]: - day = date.fromisoformat(birth_date) - start = datetime.combine(day, time.fromisoformat(start_time)) - end = datetime.combine(day, time.fromisoformat(end_time)) - if end < start: - end += timedelta(days=1) - count = int((end - start).total_seconds() // 60) + 1 - if not 1 <= count <= 1_440: - raise ValueError("candidate range must contain between 1 and 1440 minutes") - return [(start + timedelta(minutes=offset)).strftime("%H:%M") for offset in range(count)] - - -def _experience_windows(birth_date: str, as_of_date: str) -> list[tuple[date, date]]: - born = date.fromisoformat(birth_date) - as_of = date.fromisoformat(as_of_date) - try: - first = born.replace(year=born.year + 12) - except ValueError: - first = born.replace(year=born.year + 12, day=28) - if as_of < first: - return [] - day_count = (as_of - first).days + 1 - window_count = min(4, day_count, max(2, math.ceil(day_count / (6 * 365)))) - boundaries = [first + timedelta(days=day_count * index // window_count) for index in range(window_count)] - return [ - (start, as_of if index == window_count - 1 else boundaries[index + 1] - timedelta(days=1)) - for index, start in enumerate(boundaries) - ] - - -def _candidate_window_rows(request: dict) -> list[dict]: - """Compute each candidate chart once and reuse it for every dimension/window.""" - from scripts.active_rectification_event_engine import ( - DOMAIN_CONFIG, - _candidate_datetimes, - _candidate_row, - ) - - windows = _experience_windows(request["birth_date"], request["as_of_date"]) - if not windows: - return [] - synthetic_events = [] - event_windows: dict[str, tuple[str, date, date]] = {} - for dimension in sorted(SUPPORTED_DIMENSIONS): - for window_start, window_end in windows: - event_id = str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{dimension}:{window_start}:{window_end}")) - midpoint = window_start + (window_end - window_start) / 2 - synthetic_events.append( - {"id": event_id, "domain": dimension, "date": midpoint.isoformat(), "precision": "day"} - ) - event_windows[event_id] = (dimension, window_start, window_end) - calculation_request = { - "birth_date": request["birth_date"], - "start_time": request["start_time"], - "end_time": request["end_time"], - "lat": request["lat"], - "lon": request["lon"], - "tz": request["tz"], - "events": synthetic_events, - } - rows = [_candidate_row(calculation_request, candidate) for candidate in _candidate_datetimes(calculation_request)] - activations = { - event_id: {row["time"]: 0.0 for row in rows} - for event_id in event_windows - } - missing_layers = sorted({layer for row in rows for layer in row["missing_layers"]}) - for row in rows: - for evidence in row["evidence"]: - activations[evidence["event_id"]][row["time"]] = float(evidence["points"]) - return [ - { - "dimension_code": dimension, - "window_start": window_start.isoformat(), - "window_end": window_end.isoformat(), - "activations": activations[event_id], - "missing_layers": [DOMAIN_CONFIG[dimension][0]] - if DOMAIN_CONFIG[dimension][0] in missing_layers else [], - } - for event_id, (dimension, window_start, window_end) in event_windows.items() - ] - - def _compute_candidate_model(request: dict) -> dict: - return { - "version": ALGORITHM_VERSION, - "birth_date": request["birth_date"], - "as_of_date": request["as_of_date"], - "range": {"start_time": request["start_time"], "end_time": request["end_time"]}, - "candidate_times": _candidate_times(request["birth_date"], request["start_time"], request["end_time"]), - "windows": _candidate_window_rows(request), - } - - -def _validate_candidate_model(model: dict, request: dict) -> dict: - expected_keys = {"version", "birth_date", "as_of_date", "range", "candidate_times", "windows"} - candidates = _candidate_times(request["birth_date"], request["start_time"], request["end_time"]) - try: - valid_header = ( - set(model) == expected_keys - and model["version"] == ALGORITHM_VERSION - and model["birth_date"] == request["birth_date"] - and model["as_of_date"] == request["as_of_date"] - and model["range"] == {"start_time": request["start_time"], "end_time": request["end_time"]} - and model["candidate_times"] == candidates - and isinstance(model["windows"], list) - ) - first_window = _experience_windows(request["birth_date"], request["as_of_date"]) - minimum_date = first_window[0][0] if first_window else date.max - maximum_date = date.fromisoformat(request["as_of_date"]) - window_keys = [ - (row.get("dimension_code"), row.get("window_start"), row.get("window_end")) - for row in model["windows"] if isinstance(row, dict) - ] - valid_windows = len(window_keys) == len(set(window_keys)) and all( - isinstance(row, dict) - and set(row) == {"dimension_code", "window_start", "window_end", "activations", "missing_layers"} - and row["dimension_code"] in SUPPORTED_DIMENSIONS - and minimum_date <= date.fromisoformat(row["window_start"]) - <= date.fromisoformat(row["window_end"]) <= maximum_date - and isinstance(row["activations"], dict) - and set(row["activations"]) == set(candidates) - and all( - not isinstance(score, bool) - and isinstance(score, int | float) - and math.isfinite(score) - and score >= 0 - for score in row["activations"].values() - ) - and isinstance(row["missing_layers"], list) - and all(isinstance(layer, str) and layer for layer in row["missing_layers"]) - for row in model["windows"] - ) - except (KeyError, TypeError, ValueError): - valid_header = valid_windows = False - if not valid_header or not valid_windows: - raise ValueError("candidate model does not match the submitted request") - return model - - -def _opportunities(model: dict) -> list[dict]: - grouped: dict[str, list[dict]] = defaultdict(list) - for row in model["windows"]: - if not row["missing_layers"]: - grouped[row["dimension_code"]].append(row) - opportunities = [] - candidates = model["candidate_times"] - for dimension, windows in sorted(grouped.items()): - memberships: dict[int, list[str]] = defaultdict(list) - for candidate in candidates: - winner = max(range(len(windows)), key=lambda index: (windows[index]["activations"][candidate], -index)) - memberships[winner].append(candidate) - populated = [(windows[index], members) for index, members in sorted(memberships.items()) if members] - if not 2 <= len(populated) <= 4: - continue - probabilities = [len(members) / len(candidates) for _, members in populated] - information_gain = -sum(value * math.log(value) for value in probabilities) / math.log(len(populated)) - if information_gain < MIN_INFORMATION_GAIN: - continue - partition_basis = [] - partitions = [] - for window, members in populated: - basis = { - "version": ALGORITHM_VERSION, - "dimension": dimension, - "window_start": window["window_start"], - "window_end": window["window_end"], - "members": sorted(members), - } - partition_basis.append(basis) - partitions.append( - { - "partition_id": _canonical_hash(basis), - "descriptor": f"{window['window_start']}--{window['window_end']}", - "fallback_label": f"{window['window_start'][:4]}—{window['window_end'][:4]}", - "candidate_scores": {candidate: 1.0 if candidate in members else 0.0 for candidate in candidates}, - } - ) - fingerprint = _canonical_hash({"version": ALGORITHM_VERSION, "partitions": partition_basis}) - opportunities.append( - { - "opportunity_id": _canonical_hash({"version": ALGORITHM_VERSION, "dimension": dimension, "partitions": partition_basis}), - "dimension_code": dimension, - "neutral_context": dimension, - "estimated_information_gain": round(information_gain, 6), - "candidate_partition_fingerprint": fingerprint, - "fallback_prompt": f"下面哪个时间段更接近你在 {dimension} 方面的明显变化？", - "partitions": partitions, - } - ) - return sorted(opportunities, key=lambda item: (-item["estimated_information_gain"], item["opportunity_id"])) + return compute_candidate_model(request, _candidate_window_rows) def build_difference_packet(request: dict) -> dict: """Build reusable candidate activations and unused high-gain opportunities.""" - candidates = _candidate_times(request["birth_date"], request["start_time"], request["end_time"]) + candidates = _candidate_times( + request["birth_date"], request["start_time"], request["end_time"] + ) _validated_choice_evidence(request.get("evidence"), candidates) - model = request.get("candidate_model") - candidate_model = _compute_candidate_model(request) if model is None else _validate_candidate_model(model, request) + persisted = request.get("candidate_model") + model = ( + _compute_candidate_model(request) + if persisted is None + else _validate_candidate_model(persisted, request) + ) dismissed = set(request.get("dismissed_opportunity_ids", [])) fingerprints = set(request.get("partition_fingerprints", [])) - opportunities = [ - item for item in _opportunities(candidate_model) + unused = [ + item for item in _opportunities(model) if item["opportunity_id"] not in dismissed and item["candidate_partition_fingerprint"] not in fingerprints ] return { "case_id": request["case_id"], "scoring_version": ALGORITHM_VERSION, - "current_range": {"start_time": request["start_time"], "end_time": request["end_time"]}, - "opportunities": opportunities, + "current_range": { + "start_time": request["start_time"], "end_time": request["end_time"] + }, + "opportunities": unused, "asked_question_fingerprints": list(request.get("question_fingerprints", [])), "candidate_partition_fingerprints": list(request.get("partition_fingerprints", [])), "recent_range_history": list(request.get("recent_ranges", [])), - "candidate_model": candidate_model, + "candidate_model": model, } @@ -271,7 +92,7 @@ def _winning_segments(rows: Sequence[ChoiceRow], top_score: float) -> list[list[ for row in rows: if row["score"] != top_score: continue - follows = segments and ( + follows = bool(segments) and ( _minute_value(row["time"]) - _minute_value(segments[-1][-1]["time"]) ) % 1_440 == 1 if follows: @@ -285,8 +106,8 @@ def adjudicate_choice_rows( rows: Sequence[ChoiceRow], *, effective_answer_count: int, dimension_count: int, missing_layers: Sequence[str], request_fingerprint: str = "", ) -> dict: - """Apply v2 confidence gates to precomputed effective choice evidence.""" - ranked = sorted(rows, key=lambda row: _minute_value(row["time"])) + """Apply v2 confidence gates while preserving submitted candidate chronology.""" + ranked = list(rows) scores = sorted({row["score"] for row in ranked}, reverse=True) top_score = scores[0] if scores else 0.0 second_score = scores[1] if len(scores) > 1 else top_score @@ -318,7 +139,7 @@ def adjudicate_choice_rows( reasons.append("missing_mandatory_layers") if confidence == "low" and effective_answer_count < 3: reasons.append("insufficient_effective_evidence") - fingerprint = request_fingerprint or _canonical_hash(list(ranked)) + fingerprint = request_fingerprint or _canonical_hash(ranked) return { "result_id": str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{fingerprint}")), "confidence": confidence, @@ -338,7 +159,9 @@ def adjudicate_choice_rows( } -def _validated_choice_evidence(evidence_rows: list | None, candidates: Sequence[str]) -> tuple[list[dict], set[str]]: +def _validated_choice_evidence( + evidence_rows: list | None, candidates: Sequence[str], +) -> tuple[list[dict], set[str]]: if not isinstance(evidence_rows, list): raise ValueError("choice evidence must contain partition evidence") if len(evidence_rows) > 10: @@ -351,54 +174,66 @@ def _validated_choice_evidence(evidence_rows: list | None, candidates: Sequence[ } for evidence in evidence_rows: if not isinstance(evidence, dict) or set(evidence) != required: - field = "option_id" if isinstance(evidence, dict) and "option_id" in evidence else "partition evidence" + field = ( + "option_id" + if isinstance(evidence, dict) and "option_id" in evidence + else "partition evidence" + ) raise ValueError(f"choice evidence contains invalid {field}") - try: - UUID(evidence["question_id"]) - except (ValueError, TypeError, AttributeError) as exc: - raise ValueError("partition evidence question_id must be a UUID") from exc - if evidence["question_id"] in question_ids: + question_id = evidence["question_id"].strip() if isinstance(evidence["question_id"], str) else "" + if not question_id: + raise ValueError("partition evidence question identifier must be non-empty") + if question_id in question_ids: raise ValueError("duplicate question evidence is not allowed") if any( - not isinstance(evidence[key], str) or not evidence[key] + not isinstance(evidence[key], str) or not evidence[key].strip() for key in ("opportunity_id", "partition_id") ): raise ValueError("partition evidence identifier must be a non-empty string") - scores = evidence["candidate_scores"] - gain = evidence["information_gain"] - valid_scores = isinstance(scores, dict) and set(scores) == set(candidates) and all( - not isinstance(score, bool) - and isinstance(score, int | float) - and math.isfinite(score) - and score >= 0 - for score in scores.values() - ) - if not valid_scores: - raise ValueError("candidate scores must exactly match the submitted range") + _validate_scores(evidence, candidates) if evidence["dimension_code"] not in SUPPORTED_DIMENSIONS: raise ValueError("choice evidence dimension is unsupported") - if ( - isinstance(gain, bool) or not isinstance(gain, int | float) - or not math.isfinite(gain) or gain < 0 - ): - raise ValueError("choice evidence information gain must be finite") - question_ids.add(evidence["question_id"]) + question_ids.add(question_id) dimensions.add(evidence["dimension_code"]) return evidence_rows, dimensions +def _validate_scores(evidence: dict, candidates: Sequence[str]) -> None: + import math + + scores = evidence["candidate_scores"] + gain = evidence["information_gain"] + valid_scores = isinstance(scores, dict) and set(scores) == set(candidates) and all( + not isinstance(score, bool) + and isinstance(score, int | float) + and math.isfinite(score) + and score >= 0 + for score in scores.values() + ) + if not valid_scores: + raise ValueError("candidate scores must exactly match the submitted range") + if ( + isinstance(gain, bool) or not isinstance(gain, int | float) + or not math.isfinite(gain) or gain < 0 + ): + raise ValueError("choice evidence information gain must be finite") + + def score_choice_evidence(request: dict) -> dict: """Sum only strict server-resolved primary evidence, then adjudicate it.""" - candidates = _candidate_times(request["birth_date"], request["start_time"], request["end_time"]) + candidates = _candidate_times( + request["birth_date"], request["start_time"], request["end_time"] + ) evidence_rows, dimensions = _validated_choice_evidence( request.get("choice_evidence"), candidates ) totals = {candidate: 0.0 for candidate in candidates} for evidence in evidence_rows: - scores = evidence["candidate_scores"] - gain = evidence["information_gain"] for candidate in candidates: - totals[candidate] += float(scores[candidate]) * float(gain) + totals[candidate] += ( + float(evidence["candidate_scores"][candidate]) + * float(evidence["information_gain"]) + ) rows: list[ChoiceRow] = [ {"time": candidate, "score": round(score, 6)} for candidate, score in totals.items() ] diff --git a/scripts/dynamic_rectification_opportunities.py b/scripts/dynamic_rectification_opportunities.py new file mode 100644 index 00000000..2593c7b6 --- /dev/null +++ b/scripts/dynamic_rectification_opportunities.py @@ -0,0 +1,259 @@ +# /// script +# requires-python = ">=3.11" +# dependencies = [] +# /// +# ─── How to run ─── +# .venv/bin/python -m pytest -q tests/test_dynamic_rectification.py +"""Candidate-model construction and opportunity partitioning for rectification.""" + +from __future__ import annotations + +import hashlib +import json +import math +from collections import defaultdict +from collections.abc import Callable, Mapping, Sequence +from datetime import date, datetime, time, timedelta +from typing import Final +from uuid import NAMESPACE_URL, uuid5 + +ALGORITHM_VERSION: Final = "birth-time-choice-scoring-v2" +MIN_INFORMATION_GAIN: Final = 0.15 +SUPPORTED_DIMENSIONS: Final = frozenset( + {"education", "relocation", "relationship", "career", "health_pressure"} +) + + +def canonical_hash(value: Mapping | Sequence) -> str: + encoded = json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(",", ":")) + return hashlib.sha256(encoded.encode("utf-8")).hexdigest() + + +def candidate_times(birth_date: str, start_time: str, end_time: str) -> list[str]: + day = date.fromisoformat(birth_date) + start = datetime.combine(day, time.fromisoformat(start_time)) + end = datetime.combine(day, time.fromisoformat(end_time)) + if end < start: + end += timedelta(days=1) + count = int((end - start).total_seconds() // 60) + 1 + if not 1 <= count <= 1_440: + raise ValueError("candidate range must contain between 1 and 1440 minutes") + return [(start + timedelta(minutes=offset)).strftime("%H:%M") for offset in range(count)] + + +def experience_windows(birth_date: str, as_of_date: str) -> list[tuple[date, date]]: + born = date.fromisoformat(birth_date) + as_of = date.fromisoformat(as_of_date) + try: + first = born.replace(year=born.year + 12) + except ValueError: + first = born.replace(year=born.year + 12, day=28) + if as_of < first: + return [] + day_count = (as_of - first).days + 1 + count = min(4, day_count, max(2, math.ceil(day_count / (6 * 365)))) + boundaries = [first + timedelta(days=day_count * index // count) for index in range(count)] + return [ + (start, as_of if index == count - 1 else boundaries[index + 1] - timedelta(days=1)) + for index, start in enumerate(boundaries) + ] + + +def candidate_window_rows(request: dict) -> list[dict]: + """Compute each candidate chart once and reuse it across every window.""" + from scripts.active_rectification_event_engine import ( + DOMAIN_CONFIG, + _candidate_datetimes, + _candidate_row, + ) + + windows = experience_windows(request["birth_date"], request["as_of_date"]) + if not windows: + return [] + events = [] + event_windows: dict[str, tuple[str, date, date]] = {} + for dimension in sorted(SUPPORTED_DIMENSIONS): + for window_start, window_end in windows: + event_id = str(uuid5( + NAMESPACE_URL, + f"{ALGORITHM_VERSION}:{dimension}:{window_start}:{window_end}", + )) + midpoint = window_start + (window_end - window_start) / 2 + events.append({ + "id": event_id, + "domain": dimension, + "date": midpoint.isoformat(), + "precision": "day", + }) + event_windows[event_id] = (dimension, window_start, window_end) + calculation_request = { + "birth_date": request["birth_date"], + "start_time": request["start_time"], + "end_time": request["end_time"], + "lat": request["lat"], + "lon": request["lon"], + "tz": request["tz"], + "events": events, + } + candidates = _candidate_datetimes(calculation_request) + rows = [_candidate_row(calculation_request, candidate) for candidate in candidates] + activations = { + event_id: {row["time"]: 0.0 for row in rows} for event_id in event_windows + } + missing = {layer for row in rows for layer in row["missing_layers"]} + for row in rows: + for evidence in row["evidence"]: + activations[evidence["event_id"]][row["time"]] = float(evidence["points"]) + return [ + { + "dimension_code": dimension, + "window_start": window_start.isoformat(), + "window_end": window_end.isoformat(), + "activations": activations[event_id], + "missing_layers": [DOMAIN_CONFIG[dimension][0]] + if DOMAIN_CONFIG[dimension][0] in missing else [], + } + for event_id, (dimension, window_start, window_end) in event_windows.items() + ] + + +def compute_candidate_model(request: dict, row_builder: Callable[[dict], list[dict]]) -> dict: + return { + "version": ALGORITHM_VERSION, + "birth_date": request["birth_date"], + "as_of_date": request["as_of_date"], + "range": {"start_time": request["start_time"], "end_time": request["end_time"]}, + "location": { + "lat": request["lat"], + "lon": request["lon"], + "tz": request["tz"], + }, + "candidate_times": candidate_times( + request["birth_date"], request["start_time"], request["end_time"] + ), + "windows": row_builder(request), + } + + +def validate_candidate_model(model: dict, request: dict) -> dict: + expected = { + "version", "birth_date", "as_of_date", "range", "location", + "candidate_times", "windows", + } + candidates = candidate_times(request["birth_date"], request["start_time"], request["end_time"]) + try: + valid_header = ( + set(model) == expected + and model["version"] == ALGORITHM_VERSION + and model["birth_date"] == request["birth_date"] + and model["as_of_date"] == request["as_of_date"] + and model["range"] == { + "start_time": request["start_time"], "end_time": request["end_time"] + } + and model["location"] == { + "lat": request["lat"], "lon": request["lon"], "tz": request["tz"] + } + and model["candidate_times"] == candidates + and isinstance(model["windows"], list) + ) + valid_windows = _validate_windows(model["windows"], request, candidates) + except (KeyError, TypeError, ValueError): + valid_header = valid_windows = False + if not valid_header or not valid_windows: + raise ValueError("candidate model does not match the submitted request") + return model + + +def _validate_windows(windows: list, request: dict, candidates: list[str]) -> bool: + generated = experience_windows(request["birth_date"], request["as_of_date"]) + minimum = generated[0][0] if generated else date.max + maximum = date.fromisoformat(request["as_of_date"]) + keys = [ + (row.get("dimension_code"), row.get("window_start"), row.get("window_end")) + for row in windows if isinstance(row, dict) + ] + return len(keys) == len(set(keys)) and all( + isinstance(row, dict) + and set(row) == { + "dimension_code", "window_start", "window_end", "activations", "missing_layers" + } + and row["dimension_code"] in SUPPORTED_DIMENSIONS + and minimum <= date.fromisoformat(row["window_start"]) + <= date.fromisoformat(row["window_end"]) <= maximum + and isinstance(row["activations"], dict) + and set(row["activations"]) == set(candidates) + and all( + not isinstance(score, bool) + and isinstance(score, int | float) + and math.isfinite(score) + and score >= 0 + for score in row["activations"].values() + ) + and isinstance(row["missing_layers"], list) + and all(isinstance(layer, str) and layer for layer in row["missing_layers"]) + for row in windows + ) + + +def opportunities(model: dict) -> list[dict]: + grouped: dict[str, list[dict]] = defaultdict(list) + for row in model["windows"]: + if not row["missing_layers"]: + grouped[row["dimension_code"]].append(row) + result = [] + for dimension, windows in sorted(grouped.items()): + opportunity = _dimension_opportunity(dimension, windows, model["candidate_times"]) + if opportunity is not None: + result.append(opportunity) + return sorted(result, key=lambda item: (-item["estimated_information_gain"], item["opportunity_id"])) + + +def _dimension_opportunity(dimension: str, windows: list[dict], candidates: list[str]) -> dict | None: + memberships: dict[int, list[str]] = defaultdict(list) + for candidate in candidates: + winner = max( + range(len(windows)), + key=lambda index: (windows[index]["activations"][candidate], -index), + ) + memberships[winner].append(candidate) + populated = [(windows[index], members) for index, members in sorted(memberships.items())] + if not 2 <= len(populated) <= 4: + return None + probabilities = [len(members) / len(candidates) for _, members in populated] + gain = -sum(value * math.log(value) for value in probabilities) / math.log(len(populated)) + if gain < MIN_INFORMATION_GAIN: + return None + basis = [ + { + "version": ALGORITHM_VERSION, + "dimension": dimension, + "window_start": window["window_start"], + "window_end": window["window_end"], + "members": sorted(members), + } + for window, members in populated + ] + partitions = [ + { + "partition_id": canonical_hash(item), + "descriptor": f"{item['window_start']}--{item['window_end']}", + "fallback_label": f"{item['window_start'][:4]}—{item['window_end'][:4]}", + "candidate_scores": { + candidate: 1.0 if candidate in item["members"] else 0.0 + for candidate in candidates + }, + } + for item in basis + ] + fingerprint = canonical_hash({"version": ALGORITHM_VERSION, "partitions": basis}) + return { + "opportunity_id": canonical_hash({ + "version": ALGORITHM_VERSION, "dimension": dimension, "partitions": basis + }), + "dimension_code": dimension, + "neutral_context": dimension, + "estimated_information_gain": round(gain, 6), + "candidate_partition_fingerprint": fingerprint, + "fallback_prompt": f"下面哪个时间段更接近你在 {dimension} 方面的明显变化？", + "partitions": partitions, + } diff --git a/scripts/jyotish_api_server.py b/scripts/jyotish_api_server.py index 996839f8..87d09c6a 100644 --- a/scripts/jyotish_api_server.py +++ b/scripts/jyotish_api_server.py @@ -1286,8 +1286,6 @@ API_COMMAND_MAP = { 'active-rectification-questions': '/api/active_rectification_questions', 'active-rectification-score': '/api/active_rectification_score', 'active-rectification-events': '/api/active_rectification_events', - 'dynamic-rectification-opportunities': '/api/dynamic_rectification_opportunities', - 'dynamic-rectification-score': '/api/dynamic_rectification_score', 'case-validation': '/api/case_validation', 'divisional-yoga': '/api/divisional_yoga', 'deep-varga-avastha': '/api/deep_varga_avastha', @@ -1322,8 +1320,6 @@ TECHNIQUE_EXAMPLE_ENDPOINTS = { '/api/active_rectification_questions', '/api/active_rectification_score', '/api/active_rectification_events', - '/api/dynamic_rectification_opportunities', - '/api/dynamic_rectification_score', '/api/relationship', '/api/remedies', '/api/sade_sati', @@ -1449,6 +1445,13 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): scheme, _, token = authorization.partition(' ') return token.strip() if scheme.lower() == 'bearer' else '' + def _require_dynamic_rectification_token(self): + configured = os.environ.get('JYOTISH_DYNAMIC_RECTIFICATION_TOKEN', '').strip() + supplied = self._job_access_token() + matches = secrets.compare_digest(supplied, configured) + if not configured or not matches: + raise Forbidden('Dynamic rectification server token is missing or invalid') + def _vedastro_status(self): adapter = _load_local_module('vedastro_service_adapter') endpoint = os.environ.get('VEDASTRO_API_ENDPOINT', '').strip() @@ -6909,6 +6912,9 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): raise BadRequest('birth_date must be YYYY-MM-DD') if not isinstance(start_time, str) or not isinstance(end_time, str): raise BadRequest('candidate times must be HH:MM') + for key in ('lat', 'lon', 'tz'): + if key not in body or body[key] in (None, ''): + raise BadRequest(f'{key} is required') try: datetime.strptime(birth_date, '%Y-%m-%d') datetime.strptime(start_time, '%H:%M') @@ -6926,6 +6932,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): } def _compute_dynamic_rectification_opportunities(self, body): + self._require_dynamic_rectification_token() allowed_fields = { 'case_id', 'birth_date', 'as_of_date', 'start_time', 'end_time', 'lat', 'lon', 'tz', 'candidate_model', 'evidence', @@ -6966,6 +6973,7 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): return {'success': True, 'endpoint': 'dynamic_rectification_opportunities', **result} def _compute_dynamic_rectification_score(self, body): + self._require_dynamic_rectification_token() allowed_fields = { 'birth_date', 'start_time', 'end_time', 'lat', 'lon', 'tz', 'choice_evidence', } @@ -7745,8 +7753,6 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): '/api/active_rectification_questions': self._compute_active_rectification_questions, '/api/active_rectification_score': self._compute_active_rectification_score, '/api/active_rectification_events': self._compute_active_rectification_events, - '/api/dynamic_rectification_opportunities': self._compute_dynamic_rectification_opportunities, - '/api/dynamic_rectification_score': self._compute_dynamic_rectification_score, '/api/relationship': self._compute_relationship, '/api/remedies': self._compute_remedies, '/api/sade_sati': self._compute_sade_sati, @@ -7872,8 +7878,6 @@ class JyotishAPIHandler(BaseHTTPRequestHandler): '/api/prashna': 'Compute Prashna chart and answer evidence', '/api/rectification_gate': 'Evaluate birth-time precision gate', '/api/active_rectification_events': 'Score dated life events against actual birth-time candidates', - '/api/dynamic_rectification_opportunities': 'Build candidate-backed dynamic rectification opportunities', - '/api/dynamic_rectification_score': 'Score server-resolved dynamic rectification choices', '/api/relationship': 'Compute relationship and spouse-status evidence', '/api/remedies': 'Generate low-risk remedies from doshas/strength/dasha', '/api/sade_sati': 'Compute Sade Sati status and phase', diff --git a/tests/test_active_rectification_api.py b/tests/test_active_rectification_api.py index 678cdc29..8b3133a9 100644 --- a/tests/test_active_rectification_api.py +++ b/tests/test_active_rectification_api.py @@ -17,6 +17,13 @@ def _handler() -> JyotishAPIHandler: return JyotishAPIHandler.__new__(JyotishAPIHandler) +def _dynamic_handler(monkeypatch) -> JyotishAPIHandler: + monkeypatch.setenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", "server-secret") + handler = _handler() + handler.headers = {"Authorization": "Bearer server-secret"} + return handler + + def test_active_rectification_questions_api_builds_choice_workflow() -> None: result = _handler()._compute_active_rectification_questions( { @@ -178,7 +185,9 @@ def test_dynamic_opportunities_api_accepts_only_server_contract(monkeypatch) -> monkeypatch.setattr(api_server, "_load_local_module", lambda _name: FakeDynamicModule) - result = _handler()._compute_dynamic_rectification_opportunities(_dynamic_base()) + result = _dynamic_handler(monkeypatch)._compute_dynamic_rectification_opportunities( + _dynamic_base() + ) assert result["success"] is True assert result["endpoint"] == "dynamic_rectification_opportunities" @@ -186,31 +195,38 @@ def test_dynamic_opportunities_api_accepts_only_server_contract(monkeypatch) -> assert captured[0]["lat"] == 31.23 -def test_dynamic_opportunities_api_rejects_missing_clock_and_untrusted_fields() -> None: +def test_dynamic_opportunities_api_rejects_missing_clock_and_untrusted_fields(monkeypatch) -> None: + handler = _dynamic_handler(monkeypatch) missing_date = _dynamic_base() del missing_date["as_of_date"] with pytest.raises(BadRequest, match="as_of_date"): - _handler()._compute_dynamic_rectification_opportunities(missing_date) + handler._compute_dynamic_rectification_opportunities(missing_date) with pytest.raises(BadRequest, match="unsupported dynamic rectification opportunity field"): - _handler()._compute_dynamic_rectification_opportunities( + handler._compute_dynamic_rectification_opportunities( {**_dynamic_base(), "confidence": "high"} ) with pytest.raises(BadRequest, match="recent_ranges"): - _handler()._compute_dynamic_rectification_opportunities( + handler._compute_dynamic_rectification_opportunities( {**_dynamic_base(), "recent_ranges": [{"start_time": "05:30", "extra": "05:33"}]} ) with pytest.raises(BadRequest, match="partition evidence"): - _handler()._compute_dynamic_rectification_opportunities( + handler._compute_dynamic_rectification_opportunities( {**_dynamic_base(), "evidence": [{"kind": "unknown"}]} ) + for field in ("lat", "lon", "tz"): + missing_location = _dynamic_base() + del missing_location[field] + with pytest.raises(BadRequest, match=field): + handler._compute_dynamic_rectification_opportunities(missing_location) -def test_dynamic_score_api_rejects_client_option_ids_before_scoring() -> None: + +def test_dynamic_score_api_rejects_client_option_ids_before_scoring(monkeypatch) -> None: with pytest.raises(BadRequest, match="option_id"): - _handler()._compute_dynamic_rectification_score( + _dynamic_handler(monkeypatch)._compute_dynamic_rectification_score( { "birth_date": "1990-01-01", "start_time": "05:30", @@ -247,7 +263,7 @@ def test_dynamic_score_api_returns_versioned_candidate_result(monkeypatch) -> No monkeypatch.setattr(api_server, "_load_local_module", lambda _name: FakeDynamicModule) - result = _handler()._compute_dynamic_rectification_score( + result = _dynamic_handler(monkeypatch)._compute_dynamic_rectification_score( { "birth_date": "1990-01-01", "start_time": "05:30", diff --git a/tests/test_dynamic_rectification.py b/tests/test_dynamic_rectification.py index 6fd88137..553ff9bb 100644 --- a/tests/test_dynamic_rectification.py +++ b/tests/test_dynamic_rectification.py @@ -1,7 +1,6 @@ from __future__ import annotations from datetime import date, datetime -from uuid import uuid4 import pytest @@ -58,32 +57,12 @@ def _fake_model() -> dict: "birth_date": "1990-01-01", "as_of_date": "2026-07-18", "range": {"start_time": "05:30", "end_time": "05:33"}, + "location": {"lat": 31.23, "lon": 121.47, "tz": 8.0}, "candidate_times": ["05:30", "05:31", "05:32", "05:33"], "windows": _fake_rows({}), } -def _score_request() -> dict: - return { - "birth_date": "1990-01-01", - "start_time": "05:30", - "end_time": "05:33", - "lat": 31.23, - "lon": 121.47, - "tz": 8.0, - "choice_evidence": [], - } - - -def _decisive_rows() -> list[dict]: - return [ - {"time": "05:30", "score": 20.0}, - {"time": "05:31", "score": 20.0}, - {"time": "05:32", "score": 20.0}, - {"time": "05:33", "score": 10.0}, - ] - - def test_packet_contains_only_candidate_backed_high_gain_opportunities(monkeypatch) -> None: monkeypatch.setattr(dynamic_rectification, "_candidate_window_rows", _fake_rows) @@ -91,7 +70,7 @@ def test_packet_contains_only_candidate_backed_high_gain_opportunities(monkeypat assert packet["scoring_version"] == "birth-time-choice-scoring-v2" assert packet["current_range"] == {"start_time": "05:30", "end_time": "05:33"} - assert len(packet["opportunities"]) >= 1 + assert packet["opportunities"] for opportunity in packet["opportunities"]: assert opportunity["estimated_information_gain"] >= 0.15 assert 2 <= len(opportunity["partitions"]) <= 4 @@ -99,7 +78,9 @@ def test_packet_contains_only_candidate_backed_high_gain_opportunities(monkeypat opportunity["partitions"] ) for partition in opportunity["partitions"]: - assert set(partition["candidate_scores"]) == {"05:30", "05:31", "05:32", "05:33"} + assert set(partition["candidate_scores"]) == { + "05:30", "05:31", "05:32", "05:33", + } def test_packet_excludes_used_opportunity_and_partition_fingerprints(monkeypatch) -> None: @@ -160,6 +141,14 @@ def test_candidate_model_rejects_out_of_bounds_windows_and_boolean_activations() dynamic_rectification.build_difference_packet({**_base_request(), "candidate_model": model}) +@pytest.mark.parametrize(("field", "changed"), [("lat", 30.0), ("lon", 120.0), ("tz", 7.0)]) +def test_candidate_model_reuse_rejects_location_or_timezone_change(field, changed) -> None: + with pytest.raises(ValueError, match="candidate model"): + dynamic_rectification.build_difference_packet({ + **_base_request(), field: changed, "candidate_model": _fake_model(), + }) + + def test_existing_evidence_summary_must_be_effective_partition_evidence(monkeypatch) -> None: monkeypatch.setattr(dynamic_rectification, "_candidate_window_rows", _fake_rows) @@ -169,18 +158,14 @@ def test_existing_evidence_summary_must_be_effective_partition_evidence(monkeypa ) -def test_candidate_charts_are_computed_once_and_missing_layers_stay_dimension_scoped( +def test_candidate_charts_are_computed_once_and_missing_layers_are_dimension_scoped( monkeypatch, ) -> None: from scripts import active_rectification_event_engine candidates = [datetime(1990, 1, 1, 5, 30), datetime(1990, 1, 1, 5, 31)] calls: list[datetime] = [] - monkeypatch.setattr( - active_rectification_event_engine, - "_candidate_datetimes", - lambda _request: candidates, - ) + monkeypatch.setattr(active_rectification_event_engine, "_candidate_datetimes", lambda _: candidates) def fake_candidate_row(request: dict, candidate: datetime) -> dict: calls.append(candidate) @@ -195,14 +180,12 @@ def test_candidate_charts_are_computed_once_and_missing_layers_stay_dimension_sc "rule_ids": ["fixture"], "points": 1.0, } - for event in request["events"] - if event["domain"] != "career" + for event in request["events"] if event["domain"] != "career" ], "missing_layers": ["D10"], } monkeypatch.setattr(active_rectification_event_engine, "_candidate_row", fake_candidate_row) - rows = dynamic_rectification._candidate_window_rows(_base_request()) assert calls == candidates @@ -210,165 +193,18 @@ def test_candidate_charts_are_computed_once_and_missing_layers_stay_dimension_sc assert {tuple(row["missing_layers"]) for row in rows if row["dimension_code"] != "career"} == {()} -def test_experience_windows_remain_valid_on_the_twelfth_birthday() -> None: - windows = dynamic_rectification._experience_windows("2000-01-01", "2012-01-01") +def test_window_edges_and_under_age_cases_do_not_create_invalid_calculation(monkeypatch) -> None: + assert dynamic_rectification._experience_windows("2000-01-01", "2012-01-01") == [ + (date(2012, 1, 1), date(2012, 1, 1)), + ] - assert windows == [(date(2012, 1, 1), date(2012, 1, 1))] - - -def test_candidate_engine_is_not_called_before_age_twelve(monkeypatch) -> None: from scripts import active_rectification_event_engine monkeypatch.setattr( active_rectification_event_engine, "_candidate_row", - lambda *_args: pytest.fail("candidate chart should not be computed"), + lambda *_: pytest.fail("candidate chart should not be computed"), ) - - rows = dynamic_rectification._candidate_window_rows( - {**_base_request(), "birth_date": "2020-01-01"} - ) - - assert rows == [] - -def test_primary_choice_changes_rankings_and_returns_a_real_range() -> None: - result = dynamic_rectification.score_choice_evidence( - { - **_score_request(), - "choice_evidence": [ - { - "question_id": str(uuid4()), - "opportunity_id": "career-window", - "partition_id": "career-2020-2022", - "dimension_code": "career", - "candidate_scores": {"05:30": 0.0, "05:31": 1.0, "05:32": 1.0, "05:33": 0.0}, - "information_gain": 0.5, - } - ], - } - ) - - assert result["effective_answer_count"] == 1 - assert result["winning_segment"] == { - "start_time": "05:31", - "end_time": "05:32", - "representative_time": "05:31", - "width_minutes": 2, - } - assert result["can_apply"] is False - assert result["evidence"] == [] - - -def test_score_accepts_canonical_candidate_membership_independent_of_json_key_order() -> None: - result = dynamic_rectification.score_choice_evidence( - { - **_score_request(), - "choice_evidence": [ - { - "question_id": str(uuid4()), - "opportunity_id": "career-window", - "partition_id": "career-2020-2022", - "dimension_code": "career", - "candidate_scores": {"05:33": 0.0, "05:32": 1.0, "05:31": 1.0, "05:30": 0.0}, - "information_gain": 0.5, - } - ], - } - ) - - assert result["winning_segment"]["start_time"] == "05:31" - - -def test_unknown_and_unmatched_are_never_choice_evidence() -> None: - with pytest.raises(ValueError, match="partition evidence"): - dynamic_rectification.score_choice_evidence( - {**_score_request(), "choice_evidence": [{"kind": "unknown"}]} - ) - - -def test_high_confidence_requires_versioned_hard_gates() -> None: - result = dynamic_rectification.adjudicate_choice_rows( - _decisive_rows(), - effective_answer_count=4, - dimension_count=3, - missing_layers=[], - ) - - assert result["confidence"] == "high" - assert result["can_apply"] is True - assert result["winning_segment"]["width_minutes"] <= 5 - assert result["margin_percent"] >= 20 - assert result["algorithm_version"] == "birth-time-choice-scoring-v2" - - -def test_medium_and_missing_layers_never_allow_application() -> None: - medium = dynamic_rectification.adjudicate_choice_rows( - _decisive_rows(), - effective_answer_count=3, - dimension_count=2, - missing_layers=[], - ) - blocked = dynamic_rectification.adjudicate_choice_rows( - _decisive_rows(), - effective_answer_count=4, - dimension_count=3, - missing_layers=["D10"], - ) - - assert medium["confidence"] == "medium" - assert medium["can_apply"] is False - assert blocked["confidence"] == "low" - assert blocked["can_apply"] is False - - -def test_score_rejects_client_fields_duplicates_caps_and_invalid_scores() -> None: - evidence = { - "question_id": str(uuid4()), - "opportunity_id": "career-window", - "partition_id": "career-2020-2022", - "dimension_code": "career", - "candidate_scores": {"05:30": 0.0, "05:31": 1.0, "05:32": 1.0, "05:33": 0.0}, - "information_gain": 0.5, - } - with pytest.raises(ValueError, match="option_id"): - dynamic_rectification.score_choice_evidence( - {**_score_request(), "choice_evidence": [{**evidence, "option_id": "client-owned"}]} - ) - with pytest.raises(ValueError, match="duplicate question"): - dynamic_rectification.score_choice_evidence( - {**_score_request(), "choice_evidence": [evidence, evidence]} - ) - with pytest.raises(ValueError, match="at most 10"): - dynamic_rectification.score_choice_evidence( - { - **_score_request(), - "choice_evidence": [ - {**evidence, "question_id": str(uuid4())} for _ in range(11) - ], - } - ) - with pytest.raises(ValueError, match="candidate scores"): - dynamic_rectification.score_choice_evidence( - { - **_score_request(), - "choice_evidence": [ - {**evidence, "candidate_scores": {**evidence["candidate_scores"], "05:34": 1.0}} - ], - } - ) - with pytest.raises(ValueError, match="candidate scores"): - dynamic_rectification.score_choice_evidence( - { - **_score_request(), - "choice_evidence": [ - { - **evidence, - "candidate_scores": {**evidence["candidate_scores"], "05:30": -1.0}, - } - ], - } - ) - with pytest.raises(ValueError, match="identifier"): - dynamic_rectification.score_choice_evidence( - {**_score_request(), "choice_evidence": [{**evidence, "partition_id": ""}]} - ) + assert dynamic_rectification._candidate_window_rows({ + **_base_request(), "birth_date": "2020-01-01", + }) == [] diff --git a/tests/test_dynamic_rectification_scoring.py b/tests/test_dynamic_rectification_scoring.py new file mode 100644 index 00000000..4eb028f1 --- /dev/null +++ b/tests/test_dynamic_rectification_scoring.py @@ -0,0 +1,253 @@ +from __future__ import annotations + +from uuid import uuid4 + +import pytest + +from scripts import dynamic_rectification +from scripts import jyotish_api_server as api_server + + +def _score_request() -> dict: + return { + "birth_date": "1990-01-01", + "start_time": "05:30", + "end_time": "05:33", + "lat": 31.23, + "lon": 121.47, + "tz": 8.0, + "choice_evidence": [], + } + + +def _decisive_rows() -> list[dict]: + return [ + {"time": "05:30", "score": 20.0}, + {"time": "05:31", "score": 20.0}, + {"time": "05:32", "score": 20.0}, + {"time": "05:33", "score": 10.0}, + ] + + +def _evidence(**changes) -> dict: + return { + "question_id": str(uuid4()), + "opportunity_id": "career-window", + "partition_id": "career-2020-2022", + "dimension_code": "career", + "candidate_scores": {"05:30": 0.0, "05:31": 1.0, "05:32": 1.0, "05:33": 0.0}, + "information_gain": 0.5, + **changes, + } + + +def _handler() -> api_server.JyotishAPIHandler: + handler = api_server.JyotishAPIHandler.__new__(api_server.JyotishAPIHandler) + handler.headers = {} + return handler + + +def test_dynamic_routes_fail_closed_and_compare_wrong_bearers_in_constant_time( + monkeypatch, +) -> None: + handler = _handler() + monkeypatch.delenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", raising=False) + with pytest.raises(api_server.Forbidden, match="token"): + handler._compute_dynamic_rectification_score({}) + + calls: list[tuple[str, str]] = [] + original = api_server.secrets.compare_digest + monkeypatch.setattr( + api_server.secrets, + "compare_digest", + lambda supplied, configured: calls.append((supplied, configured)) + or original(supplied, configured), + ) + monkeypatch.setenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", "server-secret") + handler.headers = {"Authorization": "Bearer wrong-secret"} + with pytest.raises(api_server.Forbidden, match="token"): + handler._compute_dynamic_rectification_opportunities({}) + assert calls == [("wrong-secret", "server-secret")] + + +def test_unauthenticated_forged_scores_cannot_obtain_an_applicable_result(monkeypatch) -> None: + monkeypatch.setenv("JYOTISH_DYNAMIC_RECTIFICATION_TOKEN", "server-secret") + scores = {"05:30": 10_000.0, "05:31": 0.0, "05:32": 0.0, "05:33": 0.0} + evidence = [ + _evidence( + question_id=f"question-{index}", + opportunity_id=f"opportunity-{index}", + partition_id=f"partition-{index}", + dimension_code=dimension, + candidate_scores=scores, + information_gain=1.0, + ) + for index, dimension in enumerate( + ["career", "relationship", "education", "career"], start=1 + ) + ] + with pytest.raises(api_server.Forbidden, match="token"): + _handler()._compute_dynamic_rectification_score({ + **_score_request(), "choice_evidence": evidence, + }) + + +def test_dynamic_routes_are_not_browser_runnable_technique_examples() -> None: + endpoints = { + "/api/dynamic_rectification_opportunities", + "/api/dynamic_rectification_score", + } + assert endpoints.isdisjoint(api_server.TECHNIQUE_EXAMPLE_ENDPOINTS) + assert endpoints.isdisjoint(api_server.API_COMMAND_MAP.values()) + for endpoint in endpoints: + with pytest.raises(KeyError): + _handler()._dispatch_technique_endpoint(endpoint, {}) + + +def test_primary_choice_changes_rankings_and_returns_a_real_range() -> None: + result = dynamic_rectification.score_choice_evidence( + {**_score_request(), "choice_evidence": [_evidence()]} + ) + + assert result["effective_answer_count"] == 1 + assert result["winning_segment"] == { + "start_time": "05:31", + "end_time": "05:32", + "representative_time": "05:31", + "width_minutes": 2, + } + assert result["can_apply"] is False + assert result["evidence"] == [] + + +def test_score_accepts_candidate_membership_independent_of_json_key_order() -> None: + scores = {"05:33": 0.0, "05:32": 1.0, "05:31": 1.0, "05:30": 0.0} + result = dynamic_rectification.score_choice_evidence( + {**_score_request(), "choice_evidence": [_evidence(candidate_scores=scores)]} + ) + + assert result["winning_segment"]["start_time"] == "05:31" + + +def test_cross_midnight_leaders_form_one_chronological_segment() -> None: + scores = {"23:58": 0.0, "23:59": 1.0, "00:00": 1.0, "00:01": 0.0} + evidence = [ + _evidence( + question_id=f"question-{index}", + opportunity_id=f"opportunity-{index}", + partition_id=f"partition-{index}", + dimension_code=dimension, + candidate_scores=scores, + information_gain=1.0, + ) + for index, dimension in enumerate( + ["career", "relationship", "education", "career"], start=1 + ) + ] + result = dynamic_rectification.score_choice_evidence({ + **_score_request(), + "start_time": "23:58", + "end_time": "00:01", + "choice_evidence": evidence, + }) + + assert result["confidence"] == "high" + assert result["winning_segment"] == { + "start_time": "23:59", + "end_time": "00:00", + "representative_time": "23:59", + "width_minutes": 2, + } + + +def test_opaque_trimmed_question_ids_are_valid_and_duplicates_remain_rejected() -> None: + evidence = _evidence(question_id=" question-career-window ") + result = dynamic_rectification.score_choice_evidence( + {**_score_request(), "choice_evidence": [evidence]} + ) + assert result["effective_answer_count"] == 1 + + with pytest.raises(ValueError, match="duplicate question"): + dynamic_rectification.score_choice_evidence({ + **_score_request(), + "choice_evidence": [ + evidence, + {**evidence, "question_id": "question-career-window"}, + ], + }) + + +def test_unknown_and_unmatched_are_never_choice_evidence() -> None: + with pytest.raises(ValueError, match="partition evidence"): + dynamic_rectification.score_choice_evidence( + {**_score_request(), "choice_evidence": [{"kind": "unknown"}]} + ) + + +def test_high_confidence_requires_versioned_hard_gates() -> None: + result = dynamic_rectification.adjudicate_choice_rows( + _decisive_rows(), + effective_answer_count=4, + dimension_count=3, + missing_layers=[], + ) + + assert result["confidence"] == "high" + assert result["can_apply"] is True + assert result["winning_segment"]["width_minutes"] <= 5 + assert result["margin_percent"] >= 20 + assert result["algorithm_version"] == "birth-time-choice-scoring-v2" + + +def test_medium_and_missing_layers_never_allow_application() -> None: + medium = dynamic_rectification.adjudicate_choice_rows( + _decisive_rows(), effective_answer_count=3, dimension_count=2, missing_layers=[] + ) + blocked = dynamic_rectification.adjudicate_choice_rows( + _decisive_rows(), + effective_answer_count=4, + dimension_count=3, + missing_layers=["D10"], + ) + + assert medium["confidence"] == "medium" + assert medium["can_apply"] is False + assert blocked["confidence"] == "low" + assert blocked["can_apply"] is False + + +def test_score_rejects_client_fields_duplicates_caps_and_invalid_scores() -> None: + evidence = _evidence() + with pytest.raises(ValueError, match="option_id"): + dynamic_rectification.score_choice_evidence({ + **_score_request(), "choice_evidence": [{**evidence, "option_id": "client"}], + }) + with pytest.raises(ValueError, match="duplicate question"): + dynamic_rectification.score_choice_evidence( + {**_score_request(), "choice_evidence": [evidence, evidence]} + ) + with pytest.raises(ValueError, match="at most 10"): + dynamic_rectification.score_choice_evidence({ + **_score_request(), + "choice_evidence": [ + {**evidence, "question_id": str(uuid4())} for _ in range(11) + ], + }) + with pytest.raises(ValueError, match="candidate scores"): + dynamic_rectification.score_choice_evidence({ + **_score_request(), + "choice_evidence": [_evidence(candidate_scores={ + **evidence["candidate_scores"], "05:34": 1.0, + })], + }) + with pytest.raises(ValueError, match="candidate scores"): + dynamic_rectification.score_choice_evidence({ + **_score_request(), + "choice_evidence": [_evidence(candidate_scores={ + **evidence["candidate_scores"], "05:30": -1.0, + })], + }) + with pytest.raises(ValueError, match="identifier"): + dynamic_rectification.score_choice_evidence( + {**_score_request(), "choice_evidence": [_evidence(partition_id="")]} + )