root/Jyotisha
2
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

fix: harden dynamic rectification boundary

Browse Source
This commit is contained in:
Jesse_Chen
2026-07-19 01:06:23 +08:00
parent df0adcdfc8
commit 49ab717752
7 changed files with 691 additions and 458 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.superpowers/sdd/task-2-report.md
+31 -1
View File
@@ -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.
scripts/dynamic_rectification.py
+87 -252
View File
@@ -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()
]
scripts/dynamic_rectification_opportunities.py
+259
View File
@@ -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,
}
scripts/jyotish_api_server.py
+12 -8
View File
@@ -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',
tests/test_active_rectification_api.py
+25 -9
View File
@@ -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",
tests/test_dynamic_rectification.py
+24 -188
View File
@@ -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",
}) == []
tests/test_dynamic_rectification_scoring.py
+253
View File
@@ -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="")]}
)