Merge local main changes into origin/main

scripts/dynamic_rectification.py

@@ -0,0 +1,246 @@
+# /// script
+# requires-python = ">=3.11"
+# dependencies = []
+# ///
+# ─── How to run ───
+# .venv/bin/python -m pytest -q tests/test_dynamic_rectification_scoring.py
+"""Public dynamic-rectification packet and deterministic scoring entrypoints."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Literal, TypedDict
+from uuid import NAMESPACE_URL, uuid5
+
+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
+    score: float
+
+
+class WinningSegment(TypedDict):
+    start_time: str
+    end_time: str
+    representative_time: str
+    width_minutes: int
+
+
+def _compute_candidate_model(request: dict) -> dict:
+    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"]
+    )
+    _validated_choice_evidence(request.get("evidence"), candidates)
+    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", []))
+    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": 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": model,
+    }
+
+
+def _minute_value(value: str) -> int:
+    hour, minute = value.split(":", maxsplit=1)
+    return int(hour) * 60 + int(minute)
+
+
+def _winning_segments(rows: Sequence[ChoiceRow], top_score: float) -> list[list[ChoiceRow]]:
+    segments: list[list[ChoiceRow]] = []
+    for row in rows:
+        if row["score"] != top_score:
+            continue
+        follows = bool(segments) and (
+            _minute_value(row["time"]) - _minute_value(segments[-1][-1]["time"])
+        ) % 1_440 == 1
+        if follows:
+            segments[-1].append(row)
+        else:
+            segments.append([row])
+    return segments
+
+
+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 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
+    segments = _winning_segments(ranked, top_score) if ranked else []
+    winning_rows = segments[0] if len(segments) == 1 else []
+    segment: WinningSegment | None = None
+    if winning_rows:
+        segment = {
+            "start_time": winning_rows[0]["time"],
+            "end_time": winning_rows[-1]["time"],
+            "representative_time": winning_rows[(len(winning_rows) - 1) // 2]["time"],
+            "width_minutes": len(winning_rows),
+        }
+    margin = round((top_score - second_score) / max(abs(top_score), 1.0) * 100, 2)
+    blocked = len(segments) != 1 or bool(missing_layers)
+    high = (
+        not blocked and effective_answer_count >= 4 and dimension_count >= 3
+        and segment is not None and segment["width_minutes"] <= 5 and margin >= 20
+    )
+    medium = (
+        not blocked and effective_answer_count >= 3 and dimension_count >= 2
+        and segment is not None and segment["width_minutes"] <= 15 and margin >= 10
+    )
+    confidence: Confidence = "high" if high else "medium" if medium else "low"
+    reasons = []
+    if len(segments) != 1:
+        reasons.append("tied_leader" if segments else "no_candidate_rows")
+    if missing_layers:
+        reasons.append("missing_mandatory_layers")
+    if confidence == "low" and effective_answer_count < 3:
+        reasons.append("insufficient_effective_evidence")
+    fingerprint = request_fingerprint or _canonical_hash(ranked)
+    return {
+        "result_id": str(uuid5(NAMESPACE_URL, f"{ALGORITHM_VERSION}:{fingerprint}")),
+        "confidence": confidence,
+        "can_apply": confidence == "high",
+        "winning_segment": segment,
+        "event_count": effective_answer_count,
+        "domain_count": dimension_count,
+        "top_score": top_score,
+        "second_score": second_score,
+        "margin_percent": margin,
+        "reasons": reasons,
+        "evidence": [],
+        "algorithm_version": ALGORITHM_VERSION,
+        "evidence_mode": "dynamic_choice",
+        "effective_answer_count": effective_answer_count,
+        "dimension_count": dimension_count,
+    }
+
+
+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:
+        raise ValueError("choice evidence may contain at most 10 rows")
+    question_ids: set[str] = set()
+    dimensions: set[str] = set()
+    required = {
+        "question_id", "opportunity_id", "partition_id", "dimension_code",
+        "candidate_scores", "information_gain",
+    }
+    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"
+            )
+            raise ValueError(f"choice evidence contains invalid {field}")
+        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].strip()
+            for key in ("opportunity_id", "partition_id")
+        ):
+            raise ValueError("partition evidence identifier must be a non-empty string")
+        _validate_scores(evidence, candidates)
+        if evidence["dimension_code"] not in SUPPORTED_DIMENSIONS:
+            raise ValueError("choice evidence dimension is unsupported")
+        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"]
+    )
+    evidence_rows, dimensions = _validated_choice_evidence(
+        request.get("choice_evidence"), candidates
+    )
+    totals = {candidate: 0.0 for candidate in candidates}
+    for evidence in evidence_rows:
+        for candidate in candidates:
+            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()
+    ]
+    return adjudicate_choice_rows(
+        rows,
+        effective_answer_count=len(evidence_rows),
+        dimension_count=len(dimensions),
+        missing_layers=[],
+        request_fingerprint=_canonical_hash(request),
+    )

scripts/dynamic_rectification_copy.py

@@ -0,0 +1,59 @@
+"""Localized public copy for dynamic birth-time rectification choices."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from datetime import date
+from typing import Final, Literal, TypedDict, assert_never
+
+DIMENSION_CONTEXT: Final = {
+    "education": "一次明显的升学、转学或学习方向变化",
+    "relocation": "一次明显的搬家、离乡或长期居住地变化",
+    "relationship": "一次明显的关系进入、结束或重要转变",
+    "career": "一次明显的工作、职业方向或身份变化",
+    "health_pressure": "一次持续的健康压力或生活压力变化",
+}
+SUPPORTED_DIMENSIONS: Final = frozenset(DIMENSION_CONTEXT)
+
+
+class DateRange(TypedDict):
+    window_start: str
+    window_end: str
+
+
+RangePrecision = Literal["year", "month", "day"]
+
+
+def _range_label(item: DateRange, precision: RangePrecision) -> str:
+    start = date.fromisoformat(item["window_start"])
+    end = date.fromisoformat(item["window_end"])
+    match precision:
+        case "year":
+            return f"{start.year} 年" if start.year == end.year else f"{start.year}—{end.year} 年"
+        case "month":
+            if start.year == end.year:
+                return f"{start.year} 年 {start.month} 月—{end.month} 月"
+            return f"{start.year} 年 {start.month} 月—{end.year} 年 {end.month} 月"
+        case "day":
+            if start.year == end.year and start.month == end.month:
+                return f"{start.year} 年 {start.month} 月 {start.day} 日—{end.day} 日"
+            return (
+                f"{start.year} 年 {start.month} 月 {start.day} 日—"
+                f"{end.year} 年 {end.month} 月 {end.day} 日"
+            )
+        case unreachable:
+            assert_never(unreachable)
+
+
+def visible_range_labels(items: Sequence[DateRange]) -> list[str]:
+    """Use the least date precision that keeps every visible range distinct."""
+    labels = [_range_label(item, "year") for item in items]
+    for precision in ("month", "day"):
+        duplicates = {label for label in labels if labels.count(label) > 1}
+        if not duplicates:
+            break
+        labels = [
+            _range_label(item, precision) if label in duplicates else label
+            for item, label in zip(items, labels, strict=True)
+        ]
+    return labels

scripts/dynamic_rectification_opportunities.py

@@ -0,0 +1,302 @@
+# /// 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
+
+from scripts.dynamic_rectification_copy import (
+    DIMENSION_CONTEXT,
+    SUPPORTED_DIMENSIONS,
+    visible_range_labels,
+)
+
+ALGORITHM_VERSION: Final = "birth-time-choice-scoring-v2"
+OPPORTUNITY_MODEL_VERSION: Final = "birth-time-opportunity-model-v2"
+MIN_INFORMATION_GAIN: Final = 0.15
+
+
+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_window_sets(
+    birth_date: str, as_of_date: str,
+) -> list[tuple[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
+    counts = [1] if day_count == 1 else list(range(2, min(4, day_count) + 1))
+    return [
+        (
+            f"periods-{count}",
+            [
+                (
+                    first + timedelta(days=day_count * index // count),
+                    as_of if index == count - 1 else (
+                        first + timedelta(days=day_count * (index + 1) // count - 1)
+                    ),
+                )
+                for index in range(count)
+            ],
+        )
+        for count in counts
+    ]
+
+
+def experience_windows(birth_date: str, as_of_date: str) -> list[tuple[date, date]]:
+    sets = experience_window_sets(birth_date, as_of_date)
+    return sets[-1][1] if sets else []
+
+
+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,
+    )
+
+    window_sets = experience_window_sets(request["birth_date"], request["as_of_date"])
+    if not window_sets:
+        return []
+    events = []
+    event_windows: dict[str, tuple[str, str, date, date]] = {}
+    for dimension in sorted(SUPPORTED_DIMENSIONS):
+        for window_group, windows in window_sets:
+            for window_start, window_end in windows:
+                event_id = str(uuid5(
+                    NAMESPACE_URL,
+                    f"{ALGORITHM_VERSION}:{window_group}:{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] = (
+                    window_group, 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 [
+        {
+            "window_group": window_group,
+            "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, (window_group, 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,
+        "opportunity_model_version": OPPORTUNITY_MODEL_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", "opportunity_model_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["opportunity_model_version"] == OPPORTUNITY_MODEL_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_window_sets(request["birth_date"], request["as_of_date"])
+    minimum = generated[0][1][0][0] if generated else date.max
+    maximum = date.fromisoformat(request["as_of_date"])
+    groups = {name for name, _windows in generated}
+    keys = [
+        (row.get("window_group"), 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) == {
+            "window_group", "dimension_code", "window_start", "window_end", "activations", "missing_layers"
+        }
+        and row["window_group"] in groups
+        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[tuple[str, str], list[dict]] = defaultdict(list)
+    for row in model["windows"]:
+        if not row["missing_layers"]:
+            grouped[(row["dimension_code"], row["window_group"])].append(row)
+    variants: dict[str, list[dict]] = defaultdict(list)
+    for (dimension, window_group), windows in sorted(grouped.items()):
+        opportunity = _dimension_opportunity(
+            dimension, window_group, windows, model["candidate_times"],
+        )
+        if opportunity is not None:
+            variants[dimension].append(opportunity)
+    result = [
+        sorted(items, key=lambda item: (-item["estimated_information_gain"], item["opportunity_id"]))[0]
+        for items in variants.values()
+    ]
+    return sorted(result, key=lambda item: (-item["estimated_information_gain"], item["opportunity_id"]))
+
+
+def _dimension_opportunity(
+    dimension: str, window_group: str, windows: list[dict], candidates: list[str],
+) -> dict | None:
+    neutral_context = DIMENSION_CONTEXT[dimension]
+    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,
+            "window_group": window_group,
+            "dimension": dimension,
+            "window_start": window["window_start"],
+            "window_end": window["window_end"],
+            "members": sorted(members),
+        }
+        for window, members in populated
+    ]
+    labels = visible_range_labels([
+        {"window_start": item["window_start"], "window_end": item["window_end"]}
+        for item in basis
+    ])
+    partitions = [
+        {
+            "partition_id": canonical_hash(item),
+            "descriptor": f"{item['window_start']}--{item['window_end']}",
+            "fallback_label": label,
+            "candidate_scores": {
+                candidate: 1.0 if candidate in item["members"] else 0.0
+                for candidate in candidates
+            },
+        }
+        for item, label in zip(basis, labels, strict=True)
+    ]
+    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": neutral_context,
+        "estimated_information_gain": round(gain, 6),
+        "candidate_partition_fingerprint": fingerprint,
+        "fallback_prompt": f"哪一个时间段更接近{neutral_context}？",
+        "partitions": partitions,
+    }

scripts/jyotish_api_server.py

@@ -1445,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()
@@ -1748,6 +1755,12 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
             elif path == '/api/active_rectification_events':
                 result = self._compute_active_rectification_events(body)
                 self._json(result)
+            elif path == '/api/dynamic_rectification_opportunities':
+                result = self._compute_dynamic_rectification_opportunities(body)
+                self._json(result)
+            elif path == '/api/dynamic_rectification_score':
+                result = self._compute_dynamic_rectification_score(body)
+                self._json(result)
             elif path == '/api/case_validation':
                 result = self._compute_case_validation(body)
                 self._json(result)
@@ -6891,6 +6904,95 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
             **result,
         }
 
+    def _dynamic_rectification_base(self, body, allowed_fields, field_label):
+        unsupported_fields = sorted(set(body) - allowed_fields)
+        if unsupported_fields:
+            raise BadRequest(f'unsupported dynamic rectification {field_label} field: {unsupported_fields[0]}')
+        birth_date = body.get('birth_date')
+        start_time = body.get('start_time')
+        end_time = body.get('end_time')
+        if not isinstance(birth_date, str):
+            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')
+            datetime.strptime(end_time, '%H:%M')
+        except ValueError as exc:
+            raise BadRequest('birth date or candidate time has invalid format') from exc
+        return {
+            **body,
+            'birth_date': birth_date,
+            'start_time': start_time,
+            'end_time': end_time,
+            'lat': self._get_float(body, 'lat', 0, -90, 90),
+            'lon': self._get_float(body, 'lon', 0, -180, 180),
+            'tz': self._get_float(body, 'tz', 0, -14, 14),
+        }
+
+    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',
+            'dismissed_opportunity_ids', 'question_fingerprints',
+            'partition_fingerprints', 'recent_ranges',
+        }
+        normalized = self._dynamic_rectification_base(body, allowed_fields, 'opportunity')
+        case_id = normalized.get('case_id')
+        as_of_date = normalized.get('as_of_date')
+        if not isinstance(case_id, str) or not case_id.strip():
+            raise BadRequest('case_id must be a non-empty string')
+        if not isinstance(as_of_date, str):
+            raise BadRequest('as_of_date must be YYYY-MM-DD')
+        try:
+            datetime.strptime(as_of_date, '%Y-%m-%d')
+        except ValueError as exc:
+            raise BadRequest('as_of_date must be YYYY-MM-DD') from exc
+        for key in ('evidence', 'dismissed_opportunity_ids', 'question_fingerprints', 'partition_fingerprints', 'recent_ranges'):
+            if not isinstance(normalized.get(key), list):
+                raise BadRequest(f'{key} must be an array')
+        if any(not isinstance(item, str) or not item for key in ('dismissed_opportunity_ids', 'question_fingerprints', 'partition_fingerprints') for item in normalized[key]):
+            raise BadRequest('dynamic rectification fingerprints and IDs must be non-empty strings')
+        for item in normalized['recent_ranges']:
+            if not isinstance(item, dict) or set(item) != {'start_time', 'end_time'}:
+                raise BadRequest('recent_ranges must contain only start_time and end_time')
+            try:
+                datetime.strptime(item['start_time'], '%H:%M')
+                datetime.strptime(item['end_time'], '%H:%M')
+            except (KeyError, TypeError, ValueError) as exc:
+                raise BadRequest('recent_ranges must contain valid HH:MM times') from exc
+        if normalized.get('candidate_model') is not None and not isinstance(normalized['candidate_model'], dict):
+            raise BadRequest('candidate_model must be an object')
+        module = _load_local_module('dynamic_rectification')
+        try:
+            result = module.build_difference_packet(normalized)
+        except (KeyError, TypeError, ValueError) as exc:
+            raise BadRequest(str(exc)) from exc
+        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',
+        }
+        normalized = self._dynamic_rectification_base(body, allowed_fields, 'score')
+        evidence = normalized.get('choice_evidence')
+        if not isinstance(evidence, list):
+            raise BadRequest('choice_evidence must be an array')
+        if any(isinstance(item, dict) and 'option_id' in item for item in evidence):
+            raise BadRequest('option_id is client-owned and cannot be scored')
+        module = _load_local_module('dynamic_rectification')
+        try:
+            result = module.score_choice_evidence(normalized)
+        except (KeyError, TypeError, ValueError) as exc:
+            raise BadRequest(str(exc)) from exc
+        return {'success': True, 'endpoint': 'dynamic_rectification_score', **result}
+
     def _compute_case_validation(self, body):
         planets, _, _ = self._normalized_planets_from_body(body)
         current_md = body.get('current_md', body.get('dasha_lord', ''))