报告前面加一段模型写的解读（issue #113 R1）

引擎在同一份 packet 上给出与对话同源卡相同的事实（include_reading_facts），
报告任务交付前用一次结构化模型调用写八节解读，校验不过或失败就不出解读、
报告照常交付。解读存在计算快照里，读接口一起返回，页面放在引擎正文之前，
后面标「计算依据」。用量按实际 token 记账（BUG-1460、BUG-1461）。

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01LutgzyJqNhPpZy1nFsJtN3

scripts/professional_report_reference.py

@@ -195,6 +195,18 @@ def _export_args(birth: dict[str, Any]) -> SimpleNamespace:
     return SimpleNamespace(**payload)
 
 
+def _reading_facts(packet: dict, full_reading: dict, minute_confirmed: bool, today: str | None) -> dict[str, Any]:
+    """Facts for the model-written reading (issue #113 R1). Never fails the report."""
+    try:
+        try:
+            from scripts.report_render.reading_facts import build_report_reading_facts
+        except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution path
+            from report_render.reading_facts import build_report_reading_facts
+        return build_report_reading_facts(packet, full_reading, minute_confirmed=minute_confirmed, today=today)
+    except Exception as exc:  # noqa: BLE001 - the reading is optional, the report is not
+        return {"schema": "report-reading-facts-v1", "gaps": ["reading_facts_unavailable"], "error_type": type(exc).__name__}
+
+
 def _load_report_cache():
     try:
         from scripts import consult_full_data_cache as cache
@@ -215,6 +227,7 @@ def _report_cache_identity(
     languages: list[str],
     include_fact_tables: bool,
     sha: str,
+    reading: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     return {
         "birth": birth,
@@ -225,6 +238,8 @@ def _report_cache_identity(
         "include_fact_tables": include_fact_tables,
         "report_version": REPORT_VERSION_FULL,
         "github_sha": sha,
+        "include_reading_facts": reading is not None,
+        **({"reading": reading} if reading is not None else {}),
     }
 
 
@@ -237,6 +252,7 @@ def _assemble_professional_report_reference(
     languages: list[str],
     include_fact_tables: bool,
     engine,
+    reading: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     full_reading = handler._compute_full_reading_for_thematic(birth)
     resolved_engine = engine or _load_engine()
@@ -270,6 +286,9 @@ def _assemble_professional_report_reference(
                 "reader_dasha_applicability": clean_reader_appendix_markdown(render_dasha_applicability(packet))}
                if include_fact_tables else {}),
             **(english if edition == FULL_DATA_EDITION and "en" in languages else {}),
+            **({"reading_facts": _reading_facts(
+                packet, full_reading, reading["minute_confirmed"], reading.get("today"),
+            )} if reading is not None and edition == FULL_DATA_EDITION else {}),
         }
     return {
         "format": "json",
@@ -278,6 +297,21 @@ def _assemble_professional_report_reference(
     }
 
 
+def _reading_request(body: dict[str, Any]) -> dict[str, Any] | None:
+    """`include_reading_facts: true` asks for the reading facts; off by default."""
+    if body.get("include_reading_facts") is not True:
+        return None
+    try:
+        from scripts.consult_card_domain_facts import minute_confirmed_from_body
+    except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution path
+        from consult_card_domain_facts import minute_confirmed_from_body
+    today = body.get("today")
+    return {
+        "minute_confirmed": minute_confirmed_from_body(body),
+        "today": today if isinstance(today, str) else None,
+    }
+
+
 def build_professional_report_reference(handler, body: dict[str, Any], *, engine=None) -> dict[str, Any]:
     """Build one JSON or Markdown response from one reused full-reading result."""
     if not isinstance(body, dict):
@@ -291,6 +325,7 @@ def build_professional_report_reference(handler, body: dict[str, Any], *, engine
 
     birth = handler._high_rigor_birth_payload(body)
     include_fact_tables = body.get("include_fact_tables") is True
+    reading = _reading_request(body) if edition == FULL_DATA_EDITION and output_format == "markdown" else None
     sha = _report_cache_sha()
     use_cache = (
         report_reference_cache_enabled()
@@ -300,16 +335,16 @@ def build_professional_report_reference(handler, body: dict[str, Any], *, engine
     )
     if not use_cache:
         return _assemble_professional_report_reference(
-            handler, birth, output_format, edition, packs, languages, include_fact_tables, engine,
+            handler, birth, output_format, edition, packs, languages, include_fact_tables, engine, reading,
         )
     identity = _report_cache_identity(
-        birth, output_format, edition, packs, languages, include_fact_tables, sha,
+        birth, output_format, edition, packs, languages, include_fact_tables, sha, reading,
     )
     started = time.perf_counter()
 
     def builder(_identity: dict[str, Any]) -> dict[str, Any]:
         return _assemble_professional_report_reference(
-            handler, birth, output_format, edition, packs, languages, include_fact_tables, engine,
+            handler, birth, output_format, edition, packs, languages, include_fact_tables, engine, reading,
         )
 
     cached = _load_report_cache().get_or_build(

scripts/report_render/reading_facts.py

@@ -0,0 +1,126 @@
+"""Facts for the model-written reading at the front of the report (issue #113 R1).
+
+The reading is written by a model, but only from what the engine already
+computed. This module copies the consult-card leaves (the same facts the chat
+uses) from the report's own full-data packet, drops paths and gaps, and adds
+the current dasha dates. It computes nothing new.
+"""
+
+from __future__ import annotations
+
+import copy
+from typing import Any
+
+try:
+    from scripts.consult_card_domain_facts import assert_fact_leaves, build_consult_card_facts, dig
+    from scripts.consult_full_data_cache import _attach_graha_drishti
+except ModuleNotFoundError:  # pragma: no cover - direct scripts/ execution path
+    from consult_card_domain_facts import assert_fact_leaves, build_consult_card_facts, dig
+    from consult_full_data_cache import _attach_graha_drishti
+
+READING_FACTS_SCHEMA = "report-reading-facts-v1"
+READING_DOMAINS = ("general", "career", "wealth", "marriage", "health")
+DASHA = "worksheets.timing_and_predictive_systems.dasha.current_dasha"
+UPCOMING_SUB_PERIODS = 2
+
+
+def _compact(node: Any) -> Any:
+    """Leaf values only. Gaps, withheld and blocked leaves are left out."""
+    if isinstance(node, dict):
+        if "status" in node and "check" in node:
+            return node.get("value") if node["status"] == "value" else None
+        out = {}
+        for key, value in node.items():
+            kept = _compact(value)
+            if kept is not None and kept != {} and kept != []:
+                out[key] = kept
+        return out
+    if isinstance(node, list):
+        return [item for item in (_compact(entry) for entry in node) if item is not None]
+    return node
+
+
+def _period(row: Any) -> dict[str, Any] | None:
+    if not isinstance(row, dict) or not isinstance(row.get("lord"), str):
+        return None
+    out = {"lord": row["lord"]}
+    for key in ("start", "end"):
+        if isinstance(row.get(key), str):
+            out[key] = row[key]
+    return out
+
+
+def _dasha_dates(packet: dict) -> dict[str, Any]:
+    current = dig(packet, DASHA)
+    if not isinstance(current, dict):
+        return {}
+    dates: dict[str, Any] = {}
+    md = _period(current)
+    if md:
+        dates["mahadasha"] = md
+    rows = current.get("antardasha_timeline")
+    if isinstance(rows, list):
+        index = next((i for i, row in enumerate(rows) if isinstance(row, dict) and row.get("is_current") is True), None)
+        if index is not None:
+            now = _period(rows[index])
+            if now:
+                dates["antardasha"] = now
+            upcoming = [_period(row) for row in rows[index + 1:index + 1 + UPCOMING_SUB_PERIODS]]
+            dates["next_antardashas"] = [row for row in upcoming if row]
+    return dates
+
+
+def _domain_factors(row: dict) -> list[dict[str, Any]]:
+    factors = []
+    for factor in row.get("factors") or []:
+        if isinstance(factor, dict) and factor.get("kind") == "planet" and isinstance(factor.get("planet"), str):
+            factors.append({"role": factor.get("role"), "planet": factor["planet"]})
+    return factors
+
+
+def build_report_reading_facts(
+    packet: dict,
+    full_reading: dict,
+    *,
+    minute_confirmed: bool,
+    today: str | None = None,
+) -> dict[str, Any]:
+    """Compact, packet-sourced facts for the reading. Raises on any non-packet leaf."""
+    working = _attach_graha_drishti(copy.deepcopy(packet), full_reading)
+    raw = build_consult_card_facts(working, list(READING_DOMAINS), minute_confirmed=minute_confirmed)
+    rows = {row["domain"]: row for row in raw["domains"]}
+    for row in rows.values():
+        assert_fact_leaves(working, row)
+    general = rows["general"]
+    planets = {}
+    for name, record in general["planets"].items():
+        kept = _compact({key: value for key, value in record.items() if key != "domain_varga_house"})
+        if kept.get("present"):
+            kept.pop("present")
+            planets[name] = kept
+    domains = {}
+    for name, row in rows.items():
+        domains[name] = {
+            "varga": row["domain_varga"],
+            "houses": _compact(row["houses"]),
+            "planets_in_varga": {
+                planet: house
+                for planet, record in row["planets"].items()
+                if (house := _compact(record.get("domain_varga_house"))) is not None
+            },
+            "significators": _domain_factors(row),
+        }
+    ascendant = dig(packet, "core_chart.ascendant")
+    return {
+        "schema": READING_FACTS_SCHEMA,
+        "minute_confirmed": minute_confirmed,
+        **({"today": today} if today else {}),
+        "ascendant": _compact({
+            key: {"status": "value", "check": "path", "value": ascendant.get(key)}
+            for key in ("sign", "lord", "nakshatra")
+            if isinstance(ascendant, dict) and ascendant.get(key) is not None
+        }),
+        "planets": planets,
+        "domains": domains,
+        "timing": {**_compact(general["timing"]), **_dasha_dates(packet)},
+    }