fix(web): keep unrectified birth data usable without inventing a minute

Rectification stays optional. Reported minutes can consult and generate reports; date-plus-period uses a declared window instead of a midpoint or 00:00. Updates BUG-341.

Co-authored-by: Cursor <cursoragent@cursor.com>

scripts/domain_calculation_service.py

@@ -337,3 +337,220 @@ def compute_sade_sati(
     )
     result["result_hash"] = _canonical_hash(result)
     return result
+
+
+_CLOCK_PLANETS = (
+    "Sun",
+    "Moon",
+    "Mars",
+    "Mercury",
+    "Jupiter",
+    "Venus",
+    "Saturn",
+    "Rahu",
+    "Ketu",
+)
+_MINUTES_PER_DAY = 24 * 60
+
+
+def _require_hhmm(value: Any, *, field: str) -> str:
+    clock = str(value or "").strip()
+    if len(clock) != 5 or clock[2] != ":":
+        raise CalculationError(f"{field} must be HH:MM")
+    try:
+        hour = int(clock[:2])
+        minute = int(clock[3:])
+    except ValueError as exc:
+        raise CalculationError(f"{field} must be HH:MM") from exc
+    if hour < 0 or hour > 23 or minute < 0 or minute > 59:
+        raise CalculationError(f"{field} must be HH:MM")
+    return f"{hour:02d}:{minute:02d}"
+
+
+def _clock_minutes(clock: str) -> int:
+    return int(clock[:2]) * 60 + int(clock[3:])
+
+
+def _minutes_to_clock(total: int) -> str:
+    normalized = total % _MINUTES_PER_DAY
+    if normalized < 0:
+        normalized += _MINUTES_PER_DAY
+    return f"{normalized // 60:02d}:{normalized % 60:02d}"
+
+
+def declared_window_probe_clocks(range_start: str, range_end: str) -> list[str]:
+    start = _clock_minutes(_require_hhmm(range_start, field="range_start"))
+    end = _clock_minutes(_require_hhmm(range_end, field="range_end"))
+    span = end + _MINUTES_PER_DAY - start if end < start else end - start
+    if span < 1:
+        raise CalculationError("declared window must span at least one minute")
+    clocks: list[str] = []
+    seen: set[str] = set()
+    for numerator in (0, 1, 2, 3):
+        offset = int((span * numerator) / 3 + 0.5)
+        clock = _minutes_to_clock(start + offset)
+        if clock not in seen:
+            seen.add(clock)
+            clocks.append(clock)
+    if len(clocks) < 2:
+        raise CalculationError("declared window must yield at least two distinct probes")
+    return clocks
+
+
+def _probe_role(index: int, count: int) -> str:
+    if index == 0:
+        return "range_start"
+    if index == count - 1:
+        return "range_end"
+    return "interior"
+
+
+def _window_layer_snapshot(chart: dict[str, Any]) -> dict[str, Any]:
+    planets = chart.get("planets") if isinstance(chart.get("planets"), dict) else {}
+    planet_signs: dict[str, str] = {}
+    for name in _CLOCK_PLANETS:
+        planet = planets.get(name)
+        sign = planet.get("sign") if isinstance(planet, dict) else None
+        if isinstance(sign, str) and sign:
+            planet_signs[name] = sign
+    moon = planets.get("Moon") if isinstance(planets.get("Moon"), dict) else {}
+    moon_nakshatra = moon.get("nakshatra") if isinstance(moon.get("nakshatra"), str) else None
+    ascendant = chart.get("ascendant") if isinstance(chart.get("ascendant"), dict) else {}
+    ascendant_sign = ascendant.get("sign") if isinstance(ascendant.get("sign"), str) else None
+    houses_raw = chart.get("houses") if isinstance(chart.get("houses"), dict) else {}
+    house_cusp_signs: dict[str, str] = {}
+    for index in range(1, 13):
+        house = houses_raw.get(f"house_{index}")
+        sign = house.get("cusp_sign") if isinstance(house, dict) else None
+        if isinstance(sign, str) and sign:
+            house_cusp_signs[str(index)] = sign
+    return {
+        "planet_signs": planet_signs,
+        "moon_nakshatra": moon_nakshatra,
+        "ascendant_sign": ascendant_sign,
+        "house_cusp_signs": house_cusp_signs,
+    }
+
+
+def _unique_in_order(values: list[Any]) -> list[Any]:
+    ordered: list[Any] = []
+    for value in values:
+        if value not in ordered:
+            ordered.append(value)
+    return ordered
+
+
+def compute_declared_window_chart(payload: dict[str, Any]) -> dict[str, Any]:
+    if "hour" in payload or "minute" in payload or "second" in payload:
+        raise CalculationError("declared window must not include a single birth minute")
+    range_start = _require_hhmm(payload.get("range_start", payload.get("rangeStart")), field="range_start")
+    range_end = _require_hhmm(payload.get("range_end", payload.get("rangeEnd")), field="range_end")
+    clocks = declared_window_probe_clocks(range_start, range_end)
+    snapshots: list[dict[str, Any]] = []
+    probes: list[dict[str, str]] = []
+    for index, clock in enumerate(clocks):
+        hour = int(clock[:2])
+        minute = int(clock[3:])
+        chart = compute_chart({
+            "year": payload["year"],
+            "month": payload["month"],
+            "day": payload["day"],
+            "hour": hour,
+            "minute": minute,
+            "second": 0,
+            "lat": payload["lat"],
+            "lon": payload["lon"],
+            "tz": payload["tz"],
+            "timezone_id": payload.get("timezone_id", payload.get("timezoneId")),
+            "ayanamsa": payload.get("ayanamsa"),
+            "node_mode": payload.get("node_mode", payload.get("nodeMode", "mean")),
+        })
+        snapshots.append(_window_layer_snapshot(chart))
+        probes.append({"clock": clock, "role": _probe_role(index, len(clocks))})
+
+    stable_planet_signs: dict[str, str] = {}
+    varying_planet_signs: dict[str, list[str]] = {}
+    for name in _CLOCK_PLANETS:
+        signs = _unique_in_order([
+            snapshot["planet_signs"][name]
+            for snapshot in snapshots
+            if name in snapshot["planet_signs"]
+        ])
+        if len(signs) == 1:
+            stable_planet_signs[name] = signs[0]
+        elif len(signs) > 1:
+            varying_planet_signs[name] = signs
+
+    moon_nakshatras = _unique_in_order([
+        snapshot["moon_nakshatra"]
+        for snapshot in snapshots
+        if snapshot["moon_nakshatra"]
+    ])
+    ascendant_signs = _unique_in_order([
+        snapshot["ascendant_sign"]
+        for snapshot in snapshots
+        if snapshot["ascendant_sign"]
+    ])
+    house_variation: dict[str, list[str]] = {}
+    stable_houses: dict[str, str] = {}
+    for house in (str(index) for index in range(1, 13)):
+        signs = _unique_in_order([
+            snapshot["house_cusp_signs"][house]
+            for snapshot in snapshots
+            if house in snapshot["house_cusp_signs"]
+        ])
+        if len(signs) == 1:
+            stable_houses[house] = signs[0]
+        elif len(signs) > 1:
+            house_variation[house] = signs
+
+    stable_layers: dict[str, Any] = {"planet_signs": stable_planet_signs}
+    if len(moon_nakshatras) == 1:
+        stable_layers["moon_nakshatra"] = moon_nakshatras[0]
+    if len(ascendant_signs) == 1:
+        stable_layers["ascendant_sign"] = ascendant_signs[0]
+        if stable_houses:
+            stable_layers["house_cusp_signs"] = stable_houses
+
+    varying_layers: dict[str, Any] = {}
+    if varying_planet_signs:
+        varying_layers["planet_signs"] = varying_planet_signs
+    if len(moon_nakshatras) > 1:
+        varying_layers["moon_nakshatra"] = moon_nakshatras
+    if len(ascendant_signs) > 1:
+        varying_layers["ascendant_signs"] = ascendant_signs
+    if house_variation:
+        varying_layers["house_cusp_signs"] = house_variation
+
+    wraps_midnight = _clock_minutes(range_end) < _clock_minutes(range_start)
+    packet = {
+        "declared_range": {
+            "start": range_start,
+            "end": range_end,
+            "wraps_midnight": wraps_midnight,
+        },
+        "probe_count": len(probes),
+        "probes": probes,
+        "stable_layers": stable_layers,
+        "varying_layers": varying_layers,
+        "blocked_layers": [
+            "vimshottari_boundaries",
+            "narayana_boundaries",
+            "vargas",
+            "personal_transits",
+            *(["lagna", "houses"] if len(ascendant_signs) > 1 else []),
+        ],
+        "answer_policy": {
+            "can_answer_direction": bool(
+                stable_planet_signs
+                or stable_layers.get("moon_nakshatra")
+                or stable_layers.get("ascendant_sign")
+            ),
+            "can_answer_precise_timing": False,
+            "birth_time_confidence": "declared_window",
+            "candidate_is_confirmed": False,
+            "should_lead_with_limitations": True,
+        },
+    }
+    packet["result_hash"] = _canonical_hash(packet)
+    return packet

scripts/jyotish_api_server.py

@@ -3000,6 +3000,7 @@ API_COMMAND_MAP = {
     'solar-return': '/api/annual',
     'muhurta': '/api/muhurta',
     'panchanga-range': '/api/panchanga_range',
+    'declared-window-chart': '/api/declared_window_chart',
     'bhava-chalit': '/api/bhava_chalit',
     'sudarshana': '/api/sudarshana',
     'nakshatra-full': '/api/nakshatra_full',
@@ -3375,6 +3376,9 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
             elif path == '/api/chart':
                 result = self._compute_chart(body)
                 self._json(result)
+            elif path == '/api/declared_window_chart':
+                result = self._compute_declared_window_chart(body)
+                self._json(result)
             elif path == '/api/daily_guidance':
                 result = _load_local_module('daily_guidance_service').build_daily_guidance(body)
                 self._json(result)
@@ -8533,6 +8537,32 @@ class JyotishAPIHandler(BaseHTTPRequestHandler):
             raise BadRequest(str(e)) from e
         return {'success': True, 'endpoint': 'panchanga_range', 'report': report}
 
+    def _compute_declared_window_chart(self, body):
+        if body.get('hour') is not None or body.get('minute') is not None or body.get('second') is not None:
+            raise BadRequest('declared window must not include a single birth minute')
+        if 'year' not in body or 'month' not in body or 'day' not in body:
+            raise BadRequest('year, month and day are required')
+        if 'lat' not in body or 'lon' not in body or body.get('tz') in {None, ''}:
+            raise BadRequest('lat, lon and tz are required')
+        calculation_service = _load_local_module('domain_calculation_service')
+        try:
+            packet = calculation_service.compute_declared_window_chart({
+                'year': self._get_int(body, 'year', None, 1800, 2400),
+                'month': self._get_int(body, 'month', None, 1, 12),
+                'day': self._get_int(body, 'day', None, 1, 31),
+                'lat': self._get_float(body, 'lat', None, -90, 90),
+                'lon': self._get_float(body, 'lon', None, -180, 180),
+                'tz': self._get_float(body, 'tz', None, -14, 14),
+                'range_start': body.get('range_start', body.get('rangeStart')),
+                'range_end': body.get('range_end', body.get('rangeEnd')),
+                'timezone_id': body.get('timezone_id', body.get('timezoneId')),
+                'ayanamsa': _request_ayanamsa(body),
+                'node_mode': body.get('node_mode', body.get('nodeMode', 'mean')),
+            })
+        except calculation_service.CalculationError as exc:
+            raise BadRequest(str(exc)) from exc
+        return {'success': True, 'endpoint': 'declared_window_chart', 'packet': packet}
+
     def _compute_muhurta_panchanga(self, body):
         reference_date = body.get('reference_date') or body.get('transit_date') or body.get('today') or datetime.now().strftime('%Y-%m-%d')
         if not isinstance(reference_date, str):
@@ -10958,6 +10988,7 @@ def start_server(port=5200, host='127.0.0.1', allowed_origins=None, allowed_host
     print(f'  POST /api/annual — 年运/Tajika')
     print(f'  POST /api/muhurta — 择日')
     print(f'  POST /api/panchanga_range — Panchanga日期范围')
+    print(f'  POST /api/declared_window_chart — 声明出生窗口探针比较')
     print(f'  POST /api/bhava_chalit — Bhava Chalit')
     print(f'  POST /api/sudarshana — Sudarshana Chakra')
     print(f'  POST /api/nakshatra_full — Nakshatra深层报告')

scripts/run_quality_gate.py

@@ -55,6 +55,7 @@ CORE_PYTEST_TARGETS = [
     # These pin the answer-truth contract every product consultation is built on, and their failure
     # mode is silent widening — nothing errors when they regress (BUG-267, BUG-270).
     "tests/test_consultation_consumer_context.py",
+    "tests/test_declared_window_chart.py",
 ]
 
 RUNTIME_TRUTH_PYTEST_TARGETS = [