merge origin/staging into marriage match p1 (只碰文档, issue #79)
Independent Staging Quality Gate / validate (push) Successful in 14m3s
Independent Staging Quality Gate / publish (push) Successful in 4m33s

This commit is contained in:
Jesse_Chen
2026-10-18 21:44:38 +08:00
19 changed files with 2646 additions and 3 deletions
@@ -0,0 +1,7 @@
# 上游 `yinduzhanxing` main `7a773d13` 相对 `bde5d615` 的合盘相关差异(2026-10-18)
只放与 `bde5d615` 相比改动过的文件;未列出的文件以 `docs/research/upstream_full_synastry_bde5d615/` 为准。`scripts/marriage_match.py` 两版逐字相同。
- `skill/`:上游 `skills/jyotish-full-synastry-reading/` 中改动的 6 个文件(单语言交付、D4 键名改为 `D4_Turyamsa`、Niryaana Shoola 对恋爱婚姻应期标为不适用)。
- `tests/`:对应测试。
- `niryaana_shoola_dasha.py.txt`:Niryaana Shoola 改为 PVR 口径(`pvr_niryaana_shoola_v1`:二宫 / 八宫取强、奇顺偶逆、动 / 固 / 双 7 / 8 / 9 年);旧 `goel_nirayana_shoola_v1`(上升 / 七宫取强、每宫 9 年)判定为其实是 Shoola Dasha 的规则,已禁止运行。
@@ -0,0 +1,167 @@
#!/usr/bin/env python3
"""Source-bounded Niryaana Shoola Dasha Mahadasha producer.
Niryaana Shoola and Shoola are adjacent, but distinct, Jaimini systems. This
module implements the former only. The latter retains its Lagna/seventh,
fixed-nine-year implementation in ``scripts/shoola_dasha.py``.
"""
from __future__ import annotations
from datetime import datetime
from typing import Any
from scripts.dasha_profile_arbitration import attach_profile_arbitration
from scripts.rashi_dasha_explicit_contract import add_calendar_years
PROFILE_ID = "pvr_niryaana_shoola_v1"
LEGACY_MISCLASSIFIED_PROFILE_ID = "goel_nirayana_shoola_v1"
PYJHORA_PROFILE_ID = "pyjhora_2nd_8th_observation_v1"
TOTAL_CYCLE = 96
SIGNS = (
"Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", "Libra",
"Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces",
)
SIGN_LORDS = {
0: "Mars", 1: "Venus", 2: "Mercury", 3: "Moon", 4: "Sun", 5: "Mercury",
6: "Venus", 7: "Mars", 8: "Jupiter", 9: "Saturn", 10: "Saturn", 11: "Jupiter",
}
def _sign_index(longitude: float) -> int:
return int(float(longitude) % 360.0 // 30.0)
def _forward_order(start: int) -> list[int]:
return [(int(start) + offset) % 12 for offset in range(12)]
def _reverse_order(start: int) -> list[int]:
return [(int(start) - offset) % 12 for offset in range(12)]
def resolve_lagna_seventh_tie_break(
*,
lagna_sign_index: int,
planets: dict[str, dict[str, Any]] | None = None,
planet_strengths: dict[str, float] | None = None,
) -> dict[str, Any]:
"""Explicit Shoola-only tie-break retained for its separate producer."""
planets = planets if isinstance(planets, dict) else {}
strengths = planet_strengths if isinstance(planet_strengths, dict) else {}
rows = []
for label, sign in (("lagna", int(lagna_sign_index) % 12), ("seventh", (int(lagna_sign_index) + 6) % 12)):
lord = SIGN_LORDS[sign]
lord_data = planets.get(lord) if isinstance(planets.get(lord), dict) else {}
degree = lord_data.get("degree_in_sign", lord_data.get("degree"))
if not isinstance(degree, (int, float)) and isinstance(lord_data.get("lon"), (int, float)):
degree = float(lord_data["lon"]) % 30.0
occupants = [
name for name, data in planets.items()
if isinstance(data, dict) and int(data.get("sign_idx", -1)) % 12 == sign
]
rows.append({
"label": label, "sign_index": sign, "sign": SIGNS[sign],
"occupants": occupants, "occupant_count": len(occupants), "lord": lord,
"lord_strength": float(strengths[lord]) if isinstance(strengths.get(lord), (int, float)) else None,
"lord_degree_in_sign": float(degree) if isinstance(degree, (int, float)) else None,
})
keys = [
(row["occupant_count"], row["lord_strength"] if row["lord_strength"] is not None else float("-inf"), row["lord_degree_in_sign"] if row["lord_degree_in_sign"] is not None else float("-inf"))
for row in rows
]
if keys[0] == keys[1]:
return {"status": "blocked", "reason": "Tie-break evidence remains equal after occupants, lord strength, and lord degree.", "rows": rows}
selected = rows[0] if keys[0] > keys[1] else rows[1]
return {"status": "selected", "selected": selected, "rows": rows, "rule": "occupants_then_sign_lord_strength_then_sign_lord_degree"}
def _sign_years(sign_index: int) -> int:
"""PVR Niryaana Shoola: movable/fixed/dual = 7/8/9 years."""
if int(sign_index) % 12 in {1, 4, 7, 10}:
return 8
if int(sign_index) % 12 in {2, 5, 8, 11}:
return 9
return 7
def niryaana_shoola_applicability(
*, ascendant_longitude: float, second_strength: float, eighth_strength: float,
) -> dict[str, Any]:
"""Select the stronger second/eighth sign; the source supplies no tie-break."""
lagna = _sign_index(ascendant_longitude)
second, eighth = (lagna + 1) % 12, (lagna + 7) % 12
second_strength, eighth_strength = float(second_strength), float(eighth_strength)
if second_strength == eighth_strength:
return {
"applicable": False, "execution_status": "blocked", "ascendant_sign_index": lagna,
"second_sign_index": second, "eighth_sign_index": eighth,
"second_strength": second_strength, "eighth_strength": eighth_strength,
"rule_family": "pvr_niryaana_shoola_2nd_8th_modality",
"reason": "The cited Niryaana Shoola rule does not provide a tie-break for equal second/eighth strength.",
}
start = second if second_strength > eighth_strength else eighth
return {
"applicable": True, "ascendant_sign_index": lagna, "second_sign_index": second,
"eighth_sign_index": eighth, "second_strength": second_strength,
"eighth_strength": eighth_strength, "starting_sign_index": start,
"starting_sign": SIGNS[start], "direction": "forward" if start % 2 == 0 else "reverse",
"rule_family": "pvr_niryaana_shoola_2nd_8th_modality",
"reason": "Niryaana Shoola starts from the stronger second/eighth sign; odd seeds run forward and even seeds reverse.",
}
def calculate_niryaana_shoola_dasha(birth_info: dict[str, Any]) -> dict[str, Any]:
"""Calculate source-bounded Niryaana Shoola Mahadashas only.
PVR's chapter proposes borrowing Narayana rules for children rather than
specifying its own AD/PD algorithm, so child periods remain deliberately
blocked. This system is for longevity analysis and is not relationship
timing evidence.
"""
profile_id = birth_info.get("profile_id") or PROFILE_ID
if profile_id == LEGACY_MISCLASSIFIED_PROFILE_ID:
return {
"applicable": False, "execution_status": "blocked", "confidence_status": "blocked",
"profile_id": profile_id, "major": [], "antardasha": [], "pratyantardasha": [],
"reason": "goel_nirayana_shoola_v1 conflates Niryaana Shoola with Shoola Dasha; choose pvr_niryaana_shoola_v1 or the separate Shoola producer.",
}
if profile_id not in {PROFILE_ID, PYJHORA_PROFILE_ID}:
return {"applicable": False, "execution_status": "blocked", "confidence_status": "blocked", "profile_id": profile_id, "major": [], "reason": "Unsupported Niryaana Shoola profile_id: " + str(profile_id)}
required = ("birth_datetime", "ascendant_longitude", "second_strength", "eighth_strength")
missing = [field for field in required if birth_info.get(field) is None]
if missing:
return {"applicable": False, "execution_status": "blocked", "confidence_status": "blocked", "profile_id": profile_id, "major": [], "reason": "Missing required input: " + ", ".join(missing)}
birth = birth_info["birth_datetime"]
if isinstance(birth, str):
birth = datetime.fromisoformat(birth)
applicability = niryaana_shoola_applicability(
ascendant_longitude=float(birth_info["ascendant_longitude"]),
second_strength=float(birth_info["second_strength"]),
eighth_strength=float(birth_info["eighth_strength"]),
)
if not applicability["applicable"]:
return {**applicability, "profile_id": profile_id, "total_cycle": TOTAL_CYCLE, "major": [], "antardasha": [], "pratyantardasha": [], "confidence_status": "blocked"}
order = _forward_order(applicability["starting_sign_index"]) if applicability["direction"] == "forward" else _reverse_order(applicability["starting_sign_index"])
try:
rows, cursor = [], birth
for sign_index in order:
years = _sign_years(sign_index)
end = add_calendar_years(cursor, years)
rows.append({"level": "mahadasha", "sign_index": sign_index, "sign": SIGNS[sign_index], "years": years, "start_date": cursor.isoformat(timespec="seconds"), "end_date": end.isoformat(timespec="seconds")})
cursor = end
except ValueError:
return {**applicability, "profile_id": profile_id, "execution_status": "blocked", "confidence_status": "blocked", "major": [], "reason": "Calendar-year anniversary cannot be represented for this birth date."}
payload = {
**applicability, "profile_id": profile_id, "execution_status": "executed",
"confidence_status": "source_bounded", "verification_status": "pvr_rule_replay",
"period_depth_status": "mahadasha_executed_ad_pd_blocked_source_rule",
"total_cycle": TOTAL_CYCLE, "duration_basis": "calendar_years", "major": rows,
"antardasha": [], "pratyantardasha": [],
"relationship_applicability": "not_applicable_longevity_specific",
"reason": "PVR Niryaana Shoola MD replay. Its child-period rule is not independently specified, and longevity-specific periods are not relationship timing evidence.",
"source_locator": "PVR, An Integrated Approach to Vedic Astrology, ch. 22.2, source pages 285-292.",
}
attach_profile_arbitration(payload, "niryaana_shoola", profile_id)
return payload
@@ -0,0 +1,275 @@
---
name: jyotish-full-synastry-reading
description: Produce a full Jyotish love and marriage synastry reading from two birth records or two mature personal Jyotish reports. Use when a user wants a professional relationship interpretation, interaction patterns, shared timing, and relationship-relevant life-period analysis rather than a score-only matching result.
---
# Full Jyotish Synastry Reading
Use this skill to deliver a relationship reading, not a matchmaking scorecard.
Accept either two birth profiles or two mature personal Jyotish reports. A
birth profile contains local birth date, local birth time, and full birthplace
for each person. Do not require a relationship goal, sex/gender label,
questionnaire, caste, or a choice between relationship categories. Infer the
reading emphasis from the two charts.
## Input Contract
1. Resolve each birthplace to a sourced latitude, longitude, and timezone.
Display the normalized places and coordinate precision before calculation.
Ask a follow-up only for an ambiguous place, missing local time, or a time
range. Never invent a hospital/street coordinate.
2. Use one visible calculation profile for both people. If neither person
specifies one, use `lahiri` with `mean` nodes and state that default. Do not
silently substitute Raman or mix ayanamsas, node modes, timezones, or house
systems between the two charts.
3. Keep birth data in the active session or user-approved private runtime
storage. Never place it in a repository file, test, Skill, prompt example,
Git commit, or public report metadata.
4. If either time is not sufficiently precise for D9, UL, DK, houses, or timing,
calculate the stable Moon-based layer but label all time-sensitive layers
`blocked` or `parameter_sensitive`. Do not replace them with a score.
## Full-Report Input
When the user supplies two formal personal reports, treat them as rich evidence
packets, not as unverified prose to summarize.
1. Identify each report's birth profile, calculation profile, report generation
date, current/transit reference date, and available raw modules. Prefer an
accompanying JSON packet, delivery manifest, or structured appendix over
extracting values from narrative sentences.
2. Verify that both reports use the same ayanamsa, node mode, zodiac and house
profile before comparing signs, houses, Vargas, strength values, or Dashas.
If profiles differ, retain two separate lanes and do not make a direct
placement comparison until the differing chart is recomputed.
3. Reuse stable natal material: D1, Vargas, house lords, D9, UL, DK, Yoga,
Shadbala, Ashtakavarga, Sahams, natal Dasha roots, and book citations.
Recalculate all date-sensitive material from the preserved birth inputs for
the current day: active Dasha, transits, annual charts, current-year Sahams,
and timing windows. If the report has no usable birth inputs, retain old
dynamic material as a dated historical snapshot rather than calling it now.
4. Build a source map before interpreting: `report field -> pair comparison ->
user-facing section -> assertion label`. Do not add a second prose-only
interpretation layer that loses the reports' structured evidence.
## Execution Order
Run this sequence for every complete reading. Reuse repository producers; do
not hand-calculate degrees, vargas, dashas, or house overlays in prose.
1. Run the repository pre-work gate and capability route:
```bash
python3 scripts/pre_work_check.py --remote-timeout 8 --command-timeout 45
python3 scripts/jyotish_engine.py audit-capabilities --mode table --route relationship_timing_strict
```
2. With birth profiles, run `full-reading` separately for person A and person
B using the same `--ayanamsa lahiri --node-mode mean`, their normalized
coordinates, and the current date for `--today`/`--transit-date` unless the
user explicitly names a historical date. With report packets, perform the
Full-Report Input checks first, then recompute only date-sensitive layers
from the verified birth inputs. Preserve both raw result packets.
For the normal birth-record workflow, use the controlled private runner
rather than manually assembling two engine commands. It runs both readings
concurrently, blocks mixed calculation profiles, and writes only outside
the repository:
```bash
python3 skills/jyotish-full-synastry-reading/scripts/prepare_full_synastry_packets.py \
--person-a /private/runtime/person-a-birth.json \
--person-b /private/runtime/person-b-birth.json \
--output-dir /private/runtime/synastry-run
```
Each profile contains `date`, `time`, full `place`, `latitude`, `longitude`,
`timezone`, `ayanamsa`, and `node_mode`. Omit `--today` for the actual
current date; use it only for an explicitly requested historical reading.
The runner performs the repository preflight before consuming either profile
and emits `preflight.json`, `person_a.json`, `person_b.json`,
`execution_receipt.json`, and `dossier.json`, all private.
3. After both packets are complete, run the execution receipt. It is a gate,
not a cosmetic appendix:
```bash
python3 skills/jyotish-full-synastry-reading/scripts/validate_full_synastry_packets.py \
--person-a /private/runtime/person-a-full-reading.json \
--person-b /private/runtime/person-b-full-reading.json
```
Do not write birth records, packets, or the resulting receipt into Git.
The receipt must show the calculation-profile gate, every individual
technique row, and the status plus source inputs for every pair task before
writing conclusions. `ready_for_comparison` means only that both source
packets contain usable inputs; it is not an interpretation or a positive
relationship finding. Read
`references/full-technique-execution-matrix.md` in full for the required
pair comparisons and event-class-specific timing rules.
4. Assemble the private two-person evidence dossier before narrative writing:
```bash
python3 skills/jyotish-full-synastry-reading/scripts/build_full_synastry_dossier.py \
--person-a /private/runtime/person-a-full-reading.json \
--person-b /private/runtime/person-b-full-reading.json \
--output /private/runtime/full-synastry-dossier.json
```
It inventories every source module from both packets and produces a private
`module_narrative_plan` showing the reader question or non-assertive
disposition for each; derives directional pair overlays, angular
contacts and Jaimini rashi contacts; builds a separate reciprocal-care
brief for what helps each recipient feel cherished, safe or pressured; and
provides a `reader_section_brief` that binds every customer section to its
relevant source modules, pair lens and separate D1/D9 recipient evidence;
then attaches the neutral
`marriage-match` dossier. The contacts are raw facts, not a compatibility
verdict. A report may only claim a full pair reading when its
`delivery_readiness` is `ready_for_narrative`; otherwise retain the raw data
and state exactly which primary comparison is blocked.
5. Build the pair layer from those two computed charts. Call the new
`marriage-match` product for its narrow structural evidence only. It must
receive the two calculated snapshots, never raw birth data. Treat its
35-point Ashtakoota core, Kuja, and Papasamya as separate evidence rows.
6. Execute and compare all relationship-relevant layers available from the two
full readings:
| Layer | Required comparison |
|---|---|
| D1 and functional roles | 1st/7th axes, 5th/7th/8th/12th lords, Venus, Jupiter, Moon, mutual house overlays, dignity, aspects, yogas, Shadbala and Ashtakavarga where relevant |
| D9 | each D9 Lagna and 7th axis, D9 7th lord, Venus/Jupiter, DK, vargottama, Pushkara, D1-D9 agreement and conflict |
| Jaimini | seven- and eight-karaka variants, DK, UL, UL second house, A7/Darapada, Arudha and Argala; retain rule-profile differences |
| Periods and transits | each person's Vimshottari plus Narayana; relation-related MD/AD/PD, Dasha sandhi, double-transit and LL/7L triggers; keep active periods separate. Calendar overlap is not couple resonance. A relationship-event conclusion requires an explicit event class, both charts' event-relevant promise and activation, pair interaction, transit/annual corroboration and supplied relationship status/history. |
| Annual and specialized layers | Tajika/Varshaphala, Vivah Saham, relevant Sahams, KP, Nakshatra, Bhrigu Pada, Chara and other route-required producers; show `executed`, `blocked`, or `not_applicable` exactly as returned |
| Matching evidence | international 35-point Ashtakoota core, exception review, Mars-only Kuja from Lagna/Moon/Venus, and five-planet Papasamya kept distinct |
Evidence priority is fixed: each person's relationship promise first, the
two-chart relationship structure second, mutually activated timing third,
and matching/pressure indicators last. Kuja is a narrow conflict-pressure
indicator. It cannot lead the report, overturn a sound relationship
structure by itself, or become a pass/fail conclusion.
7. Search the local book corpus with `scripts/search_jyotish_books.py` for
each material interpretive rule. Record book, page/line, and exact snippet.
For claims that rely on a web-only rule, complete MEVG research, record the
source tier and conflict, and add required intake records before treating it
as more than `reference_only`.
8. Before writing Chinese prose, read
`references/chinese-synastry-writing-contract.md`. Preserve its protected
terms, degrees, dates, assertion labels and source paths. Apply the local
Chinese editing pass only after the evidence map is frozen; it may improve
sentence rhythm and remove boilerplate, but may not change an astrological
fact, a term mapping, a source boundary or a time window.
9. Run global-web evidence and real-case calibration according to
`references/mandatory-verification-gate-protocol.md`. Absence of external
closure lowers assertion strength; it does not erase a runnable native
calculation or its raw data.
## User-Facing Report
Write a professional Chinese relationship report, not Western-style
relationship coaching. Read `references/customer-report-shape.md` before
writing. Follow its judgment order exactly: individual promise ->
spouse/marriage carrying capacity -> lived pair interaction -> relationship
progression -> shared gains and costs -> long-term durability -> event-specific
timing -> bounded auxiliary evidence. The internal writer must use the actual
Jyotish structures. Explain primary Jyotish terms when they make the judgment
clear, but do not turn the report into a formula table. Never expose the name,
formula, score or lineage of a supplementary cultural-star, Western or
question-chart technique; it may only add a convergent experiential detail.
Open the Chinese and English reader reports with an evidence-backed experience
of affection, being valued or drawing closer, not an analyst-led contrast.
Where the reciprocal-care brief is executed, the **Attraction and Closeness**
section must write both directions: what helps A feel warmly connected, what
helps B feel warmly connected, what pressure makes either withdraw, and one
mutual practice tied to the actual pair evidence. This is reciprocal-care
guidance, never a promise of love or a tactic for extracting consent.
The private dossier, not the customer report, must retain every row from the
execution receipt, the two-person comparison recorded for that row, whether it
informed a conclusion, and why it did not if it was not used. It retains raw
placements, degrees, period boundaries, source-book citations and
MEVG/case-calibration state. Never commit that dossier or append it to the
customer copy.
Never use the total Ashtakoota value, a percentage, or a pass/fail label in the
headline. If a traditional matching value is helpful, describe it only in the
appendix as one limited Moon/nakshatra reference layer.
When mature personal reports are available, add a short section before
relationship weather called **What each person is carrying into the
relationship**. Explain each person's relationship-relevant life season:
available capacity, career or relocation pressure, family/home obligations,
inner change, and current relationship activation. Only include non-relationship
domains when their chart evidence shows a material effect on relationship
availability or decision-making. Do not duplicate two full personal reports.
Use one evidence label for every substantive conclusion:
`multi_system_consensus`, `single_system_inference`,
`parameter_sensitive`, `unclosed_divisional_chart`,
`user_history_verification_required`, or `blocked`.
When bilingual delivery is requested, produce **two independently written
Markdown reports** from the same frozen dossier. Write Chinese under
`references/chinese-synastry-writing-contract.md`, then start a separate
English composition pass under
`references/english-synastry-writing-contract.md`. English must use the
structured dossier, original placements and original citations; never translate
the Chinese draft. Run the private delivery gate after both files exist:
```bash
python3 skills/jyotish-full-synastry-reading/scripts/validate_bilingual_synastry_delivery.py \
--dossier /private/runtime/full-synastry-dossier.json \
--chinese /private/runtime/Chinese-Full-Synastry-Report.md \
--english /private/runtime/English-Full-Synastry-Report.md \
--output /private/runtime/bilingual-delivery-receipt.json
```
The receipt checks shared evidence coverage, direct-English composition
metadata, terminology, section-by-section narrative depth, reciprocal-care
coverage and report shape. It does not justify copying a Chinese paragraph into
English.
## Boundaries
- Do not output an approve/reject verdict, relationship ranking, percentage,
“soulmate” claim, or automatic marriage/divorce prediction.
- Do not call 35 points a 36-point score. Varna/caste is excluded from the
default product, and the 35-point core is only a small traditional evidence
layer.
- Do not merge Papasamya into Kuja, apply sex-asymmetric tolerance, or claim
that both people having Kuja automatically cancels it.
- Do not use a present relationship problem to retrofit chart calculations.
- Do not call a full-reading module "used" merely because its key exists. The
execution receipt must show an actual non-blocked result for both people, and
pair comparisons must be recorded separately from individual calculations.
- Do not hide a conflict between D1, D9, Jaimini, Dasha, or transit evidence.
Explain the disagreement and reduce the conclusion accordingly.
- Never make health, fertility, legal-marriage, financial, or safety claims
solely from a compatibility layer.
## Runtime References
- `references/strict-workflow-router.md`: relationship timing route and audit table.
- `references/mandatory-verification-gate-protocol.md`: web evidence and case calibration.
- `references/navamsa-marriage-deep-analysis.md`: D9 relationship microscope.
- `references/marriage-timing-comprehensive-techniques.md`: timing source map.
- `docs/marriage_match_product_contract.md`: narrow pair-evidence contract.
- `references/customer-report-shape.md`: user-readable report and full-report-input shape.
- `references/full-technique-execution-matrix.md`: mandatory all-technique and
pair-comparison execution matrix.
- `references/interaction-progression-evidence-map.md`: mandatory comparison
map for lived interaction, progression, shared gains and shared costs.
- `references/customer-topic-priorities.md`: reader-first ordering for the
customer report; presentation source only, never a Jyotish rule source.
- `references/chinese-synastry-writing-contract.md`: Chinese terminology,
evidence-preserving editing and final-language gate.
- `references/english-synastry-writing-contract.md`: direct English report
composition and terminology gate.
@@ -0,0 +1,4 @@
interface:
display_name: "Full Jyotish Synastry Reading"
short_description: "Narrative love and marriage chart interpretation"
default_prompt: "Use $jyotish-full-synastry-reading for a full love/marriage synastry reading. Ask only for each person's local birth date, local birth time, and full birthplace when missing; resolve coordinates, calculate both charts under one visible profile, run the private full-synastry dossier, and write independently composed Chinese and English reports. Lead with lived interaction, each person's distinct experience, progression, shared gains/pressures, and event-class-specific timing. Do not turn Ashtakoota, Kuja, or coincident Dasha dates into a score, verdict, or shared relationship event."
@@ -0,0 +1,227 @@
#!/usr/bin/env python3
"""Compile two independently written private synastry drafts for delivery.
This is a report compiler, not an astrology interpreter or a translation
service. It accepts separately authored Chinese and English section payloads,
binds them to the full-synastry dossier, and writes only after the existing
bilingual delivery gate passes.
"""
from __future__ import annotations
import argparse
import importlib.util
import json
import shutil
import tempfile
from pathlib import Path
from typing import Any, Mapping
ROOT = Path(__file__).resolve().parents[3]
VALIDATOR_PATH = Path(__file__).with_name("validate_bilingual_synastry_delivery.py")
EXPECTED_COMPOSITION = "independent_from_structured_dossier"
SECTION_KEYS = (
"relationship_thesis",
"what_each_person_brings",
"emotional_safety",
"attraction_and_closeness",
"communication_conflict_and_repair",
"relationship_progression",
"shared_life_gains_and_pressures",
"can_it_stay_steady",
"relationship_weather",
)
def _load_module(name: str, path: Path) -> Any:
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module
_VALIDATOR = _load_module("full_synastry_bilingual_delivery", VALIDATOR_PATH)
ZH_HEADINGS = dict(zip(SECTION_KEYS, _VALIDATOR.ZH_HEADINGS, strict=True))
EN_HEADINGS = dict(zip(SECTION_KEYS, _VALIDATOR.EN_HEADINGS, strict=True))
def _private_output_directory(value: str) -> Path:
path = Path(value).expanduser().resolve()
try:
path.relative_to(ROOT)
except ValueError:
return path
raise ValueError("--output-dir must be outside the repository because it can contain private relationship information")
def _payload(value: Mapping[str, Any], *, language: str) -> dict[str, Any]:
if not isinstance(value, Mapping):
raise ValueError(f"{language} section payload must be a JSON object")
if value.get("language") != language:
raise ValueError(f"{language} section payload must declare language={language!r}")
if value.get("composition") != EXPECTED_COMPOSITION:
raise ValueError(f"{language} section payload must declare independent composition from the structured dossier")
title = value.get("title", "")
if not isinstance(title, str):
raise ValueError(f"{language} title must be text")
sections = value.get("sections")
if not isinstance(sections, Mapping):
raise ValueError(f"{language} payload requires a sections object")
missing = [key for key in SECTION_KEYS if not isinstance(sections.get(key), str) or not sections[key].strip()]
unknown = sorted(set(sections) - set(SECTION_KEYS))
if missing or unknown:
detail = []
if missing:
detail.append("missing=" + ", ".join(missing))
if unknown:
detail.append("unknown=" + ", ".join(unknown))
raise ValueError(f"{language} section payload keys are invalid: {'; '.join(detail)}")
return {"title": title.strip(), "sections": {key: sections[key].strip() for key in SECTION_KEYS}}
def _dossier_ready(dossier: Mapping[str, Any]) -> None:
if not isinstance(dossier, Mapping):
raise ValueError("dossier must be a JSON object")
if dossier.get("schema") != "jyotish.full_synastry_dossier.v1":
raise ValueError("dossier schema is not jyotish.full_synastry_dossier.v1")
if dossier.get("delivery_readiness") != "ready_for_narrative":
raise ValueError("dossier is not ready_for_narrative")
brief = dossier.get("reader_section_brief")
if not isinstance(brief, list):
raise ValueError("dossier has no reader_section_brief")
actual = {row.get("section") for row in brief if isinstance(row, Mapping)}
if actual != set(SECTION_KEYS):
raise ValueError("dossier reader_section_brief does not cover every customer section")
def _render_markdown(payload: Mapping[str, Any], headings: Mapping[str, str]) -> str:
blocks = []
if payload["title"]:
blocks.append(f"# {payload['title']}")
for key in SECTION_KEYS:
blocks.append(f"## {headings[key]}\n\n{payload['sections'][key]}")
return "\n\n".join(blocks).strip() + "\n"
def compile_delivery(
dossier: Mapping[str, Any], chinese_payload: Mapping[str, Any], english_payload: Mapping[str, Any],
) -> dict[str, Any]:
"""Compile independently authored reports and reject a failed delivery in memory."""
_dossier_ready(dossier)
chinese = _render_markdown(_payload(chinese_payload, language="zh"), ZH_HEADINGS)
english = _render_markdown(_payload(english_payload, language="en"), EN_HEADINGS)
receipt = _VALIDATOR.build_delivery_receipt(dossier, chinese, english)
if receipt.get("status") != "passed":
raise ValueError("bilingual synastry delivery gate failed; inspect the private receipt before writing any report")
return {"chinese_markdown": chinese, "english_markdown": english, "delivery_receipt": receipt}
def compile_single_language_delivery(
dossier: Mapping[str, Any], payload: Mapping[str, Any], language: str,
) -> dict[str, Any]:
"""Compile one independently authored language from the frozen dossier."""
if language not in {"zh", "en"}:
raise ValueError("language must be 'zh' or 'en'")
_dossier_ready(dossier)
headings = ZH_HEADINGS if language == "zh" else EN_HEADINGS
markdown = _render_markdown(_payload(payload, language=language), headings)
receipt = _VALIDATOR.build_single_language_delivery_receipt(dossier, markdown, language)
if receipt.get("status") != "passed":
raise ValueError("single-language synastry delivery gate failed; inspect the private receipt before writing any report")
return {"language": language, "markdown": markdown, "delivery_receipt": receipt}
def _publish_delivery_bundle(output_dir: Path, compiled: Mapping[str, Any]) -> None:
"""Publish all three files together; never overwrite an existing delivery."""
if output_dir.exists():
raise ValueError("--output-dir already exists; choose a new private release directory")
output_dir.parent.mkdir(parents=True, exist_ok=True)
staging = Path(tempfile.mkdtemp(prefix=".synastry-delivery-", dir=output_dir.parent))
try:
(staging / "Chinese-Full-Synastry-Report.md").write_text(compiled["chinese_markdown"], encoding="utf-8")
(staging / "English-Full-Synastry-Report.md").write_text(compiled["english_markdown"], encoding="utf-8")
(staging / "bilingual-delivery-receipt.json").write_text(
json.dumps(compiled["delivery_receipt"], ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
staging.replace(output_dir)
except BaseException:
shutil.rmtree(staging, ignore_errors=True)
raise
def _publish_single_language_delivery(output_dir: Path, compiled: Mapping[str, Any]) -> None:
"""Atomically publish one language without creating a fictional peer report."""
if output_dir.exists():
raise ValueError("--output-dir already exists; choose a new private release directory")
output_dir.parent.mkdir(parents=True, exist_ok=True)
staging = Path(tempfile.mkdtemp(prefix=".synastry-delivery-", dir=output_dir.parent))
language = str(compiled["language"])
report_name = "Chinese-Full-Synastry-Report.md" if language == "zh" else "English-Full-Synastry-Report.md"
receipt_name = f"{language}-delivery-receipt.json"
try:
(staging / report_name).write_text(compiled["markdown"], encoding="utf-8")
(staging / receipt_name).write_text(
json.dumps(compiled["delivery_receipt"], ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
staging.replace(output_dir)
except BaseException:
shutil.rmtree(staging, ignore_errors=True)
raise
def main() -> int:
parser = argparse.ArgumentParser(description="Compile private, independently written full-synastry Markdown delivery.")
parser.add_argument("--dossier", required=True, help="Private full-synastry dossier JSON")
parser.add_argument("--chinese-sections", help="Private independent Chinese section payload JSON")
parser.add_argument("--english-sections", help="Private independent English section payload JSON")
parser.add_argument("--language", choices=("zh", "en"), help="Compile one independently authored language")
parser.add_argument("--sections", help="Private section payload JSON used with --language")
parser.add_argument("--output-dir", required=True, help="Private directory outside the repository")
parser.add_argument("--dry-run", action="store_true", help="Validate and print the intended delivery without writing files")
args = parser.parse_args()
output_dir = _private_output_directory(args.output_dir)
dossier = json.loads(Path(args.dossier).read_text(encoding="utf-8"))
if args.language:
if not args.sections or args.chinese_sections or args.english_sections:
parser.error("--language requires --sections and cannot be combined with bilingual section arguments")
compiled = compile_single_language_delivery(
dossier, json.loads(Path(args.sections).read_text(encoding="utf-8")), args.language,
)
files = [
"Chinese-Full-Synastry-Report.md" if args.language == "zh" else "English-Full-Synastry-Report.md",
f"{args.language}-delivery-receipt.json",
]
else:
if not args.chinese_sections or not args.english_sections:
parser.error("bilingual delivery requires --chinese-sections and --english-sections")
chinese_payload = json.loads(Path(args.chinese_sections).read_text(encoding="utf-8"))
english_payload = json.loads(Path(args.english_sections).read_text(encoding="utf-8"))
compiled = compile_delivery(dossier, chinese_payload, english_payload)
files = ["Chinese-Full-Synastry-Report.md", "English-Full-Synastry-Report.md", "bilingual-delivery-receipt.json"]
if args.dry_run:
print(json.dumps({
"status": "would-deliver",
"output_dir": str(output_dir),
"files": files,
"receipt_status": compiled["delivery_receipt"]["status"],
}, ensure_ascii=False))
return 0
if args.language:
_publish_single_language_delivery(output_dir, compiled)
else:
_publish_delivery_bundle(output_dir, compiled)
print(json.dumps({
"status": "delivered",
"output_dir": str(output_dir),
"files": files,
}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,188 @@
#!/usr/bin/env python3
"""Prepare two private full-reading packets and a synastry dossier.
This is the operational entry point for the full synastry Skill. It accepts
two private birth profiles, runs the native full-reading engine separately,
then builds the receipt and dossier consumed by the bilingual writing pass.
It deliberately does not write customer prose: that requires a chart reader
to distinguish the two people's evidence and live conditions.
"""
from __future__ import annotations
import argparse
import importlib.util
import json
import subprocess
import sys
from concurrent.futures import ThreadPoolExecutor
from datetime import date, datetime
from pathlib import Path
from typing import Any, Mapping
ROOT = Path(__file__).resolve().parents[3]
ENGINE = ROOT / "scripts" / "jyotish_engine.py"
PRE_WORK_CHECK = ROOT / "scripts" / "pre_work_check.py"
DOSSIER_PATH = Path(__file__).with_name("build_full_synastry_dossier.py")
RECEIPT_PATH = Path(__file__).with_name("validate_full_synastry_packets.py")
def _load_module(name: str, path: Path) -> Any:
spec = importlib.util.spec_from_file_location(name, path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module
_DOSSIER = _load_module("full_synastry_dossier", DOSSIER_PATH)
_RECEIPT = _load_module("full_synastry_receipt", RECEIPT_PATH)
def _private_output_directory(value: str) -> Path:
path = Path(value).expanduser().resolve()
try:
path.relative_to(ROOT)
except ValueError:
return path
raise ValueError("--output-dir must be outside the repository because it contains private birth data")
def _number(profile: Mapping[str, Any], key: str, *, minimum: float, maximum: float) -> float:
value = profile.get(key)
if not isinstance(value, (int, float)) or isinstance(value, bool) or not minimum <= float(value) <= maximum:
raise ValueError(f"{key} must be a number between {minimum} and {maximum}")
return float(value)
def normalize_profile(profile: Mapping[str, Any]) -> dict[str, Any]:
"""Validate a private birth profile without retaining identity fields."""
if not isinstance(profile, Mapping):
raise ValueError("birth profile must be a JSON object")
birth_date = profile.get("date")
birth_time = profile.get("time")
if not isinstance(birth_date, str):
raise ValueError("date is required as YYYY-MM-DD")
if not isinstance(birth_time, str):
raise ValueError("time is required as HH:MM or HH:MM:SS")
try:
day = date.fromisoformat(birth_date)
try:
parsed_time = datetime.strptime(birth_time, "%H:%M:%S").time()
except ValueError:
parsed_time = datetime.strptime(birth_time, "%H:%M").time()
except ValueError as exc:
raise ValueError("date/time must use YYYY-MM-DD and HH:MM or HH:MM:SS") from exc
place = profile.get("place")
if not isinstance(place, str) or not place.strip():
raise ValueError("place is required; keep the full birth-place text in the private profile")
ayanamsa = str(profile.get("ayanamsa", "lahiri")).lower()
node_mode = str(profile.get("node_mode", "mean")).lower()
if node_mode not in {"mean", "true"}:
raise ValueError("node_mode must be mean or true")
return {
"year": day.year, "month": day.month, "day": day.day,
"hour": parsed_time.hour, "minute": parsed_time.minute, "second": parsed_time.second,
"lat": _number(profile, "latitude", minimum=-90, maximum=90),
"lon": _number(profile, "longitude", minimum=-180, maximum=180),
"tz": _number(profile, "timezone", minimum=-14, maximum=14),
"ayanamsa": ayanamsa, "node_mode": node_mode,
"gender": profile.get("gender") if profile.get("gender") in {"male", "female"} else None,
}
def full_reading_command(profile: Mapping[str, Any], *, today: str | None = None) -> list[str]:
"""Build the native command from validated input; never hand-calculate charts."""
command = [
sys.executable, str(ENGINE), "full-reading",
"--year", str(profile["year"]), "--month", str(profile["month"]), "--day", str(profile["day"]),
"--hour", str(profile["hour"]), "--minute", str(profile["minute"]), "--second", str(profile["second"]),
"--lat", str(profile["lat"]), "--lon", str(profile["lon"]), "--tz", str(profile["tz"]),
"--ayanamsa", str(profile["ayanamsa"]), "--node-mode", str(profile["node_mode"]),
]
if profile.get("gender"):
command.extend(("--gender", str(profile["gender"])))
if today:
command.extend(("--today", today, "--transit-date", today))
return command
def preflight_command() -> list[str]:
"""Use the repository gate before native chart calculation."""
return [
sys.executable, str(PRE_WORK_CHECK),
"--remote-timeout", "8", "--command-timeout", "45",
]
def _run_preflight() -> dict[str, Any]:
completed = subprocess.run(preflight_command(), check=False, capture_output=True, text=True)
if completed.returncode:
detail = completed.stderr.strip().splitlines()[-1:] or completed.stdout.strip().splitlines()[-1:]
raise RuntimeError("synastry preflight failed: " + (detail[0] if detail else "unknown failure"))
return {
"status": "passed",
"command": ["scripts/pre_work_check.py", "--remote-timeout", "8", "--command-timeout", "45"],
}
def _run_full_reading(profile: Mapping[str, Any], today: str | None) -> dict[str, Any]:
completed = subprocess.run(full_reading_command(profile, today=today), check=False, capture_output=True, text=True)
if completed.returncode:
message = completed.stderr.strip().splitlines()[-1:] or ["native full-reading failed"]
raise RuntimeError("full-reading failed: " + message[0])
try:
result = json.loads(completed.stdout)
except json.JSONDecodeError as exc:
raise RuntimeError("full-reading did not return JSON") from exc
if not isinstance(result, dict):
raise RuntimeError("full-reading returned a non-object result")
return result
def prepare_synastry_packets(person_a: Mapping[str, Any], person_b: Mapping[str, Any], *, today: str | None = None) -> dict[str, Any]:
"""Run both native charts concurrently and assemble their private evidence."""
a, b = normalize_profile(person_a), normalize_profile(person_b)
if (a["ayanamsa"], a["node_mode"]) != (b["ayanamsa"], b["node_mode"]):
raise ValueError("both profiles must use the same ayanamsa and node_mode before direct comparison")
preflight = _run_preflight()
with ThreadPoolExecutor(max_workers=2) as pool:
future_a = pool.submit(_run_full_reading, a, today)
future_b = pool.submit(_run_full_reading, b, today)
packet_a, packet_b = future_a.result(), future_b.result()
return {
"preflight": preflight,
"person_a": packet_a,
"person_b": packet_b,
"execution_receipt": _RECEIPT.build_execution_receipt(packet_a, packet_b),
"dossier": _DOSSIER.build_full_synastry_dossier(packet_a, packet_b),
}
def main() -> int:
parser = argparse.ArgumentParser(description="Prepare private full-synastry evidence from two birth profiles.")
parser.add_argument("--person-a", required=True, help="Private JSON profile: date, time, place, latitude, longitude, timezone, ayanamsa, node_mode")
parser.add_argument("--person-b", required=True, help="Private JSON profile: date, time, place, latitude, longitude, timezone, ayanamsa, node_mode")
parser.add_argument("--output-dir", required=True, help="Private directory outside the repository")
parser.add_argument("--today", help="Optional historical reference date YYYY-MM-DD; omit for the actual current date")
args = parser.parse_args()
if args.today:
date.fromisoformat(args.today)
output_dir = _private_output_directory(args.output_dir)
person_a = json.loads(Path(args.person_a).read_text(encoding="utf-8"))
person_b = json.loads(Path(args.person_b).read_text(encoding="utf-8"))
prepared = prepare_synastry_packets(person_a, person_b, today=args.today)
output_dir.mkdir(parents=True, exist_ok=True)
for name, value in prepared.items():
(output_dir / f"{name}.json").write_text(json.dumps(value, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
print(json.dumps({
"status": "prepared", "output_dir": str(output_dir),
"dossier_status": prepared["dossier"].get("delivery_readiness"),
"next_step": "Write Chinese and English reports independently from dossier.json, then run validate_bilingual_synastry_delivery.py.",
}, ensure_ascii=False))
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,348 @@
#!/usr/bin/env python3
"""Validate independently written Chinese and English synastry reports.
The validator checks shared evidence coverage against a private dossier. It
does not translate prose and never treats one language as the other's source.
"""
from __future__ import annotations
import argparse
import json
import re
from pathlib import Path
from typing import Any, Mapping
SCHEMA = "jyotish.full_synastry_bilingual_delivery_receipt.v1"
SINGLE_LANGUAGE_SCHEMA = "jyotish.full_synastry_single_language_delivery_receipt.v1"
ZH_HEADINGS = ("结论总纲", "各自带入关系的需求", "情绪安全与相处感受", "吸引与亲密", "沟通、冲突与修复", "关系如何推进", "共同生活的收益与压力", "能不能稳定走下去", "关系应期")
EN_HEADINGS = ("Relationship Thesis", "What Each Person Brings", "Emotional Safety and Lived Experience", "Attraction and Closeness", "Communication, Conflict and Repair", "How the Relationship Progresses", "Shared Life: Gains and Pressures", "Can It Stay Steady?", "Relationship Weather")
EVIDENCE_LABELS = (
"multi_system_consensus", "single_system_inference", "parameter_sensitive",
"unclosed_divisional_chart", "user_history_verification_required", "blocked",
)
SUPPLEMENTARY_METHOD_TERMS = ("Shukuyo", "Western synastry", "Composite chart", "Davison", "Prashna", "宿曜", "星宿关系", "西占合盘", "组合盘", "问卦")
PRIVATE_EVIDENCE_TERMS = (
"execution_receipt", "source_map", "delivery_readiness", "module_inventory", "modules.",
"pair_interaction_evidence", "mutual_affection_activation", "reader_section_brief",
"recipient_experience_evidence", "d1_recipient_evidence", "d9_recipient_evidence",
"parallel_personal_timing", "event_resonance_gate",
"原始模块索引", "执行回执", "证据包路径", "章节证据简报", "接收方证据",
)
ZH_READER_MIN_CHARACTERS = 2000
EN_READER_MIN_WORDS = 1200
READER_BANNED_TERMS = ("灵魂伴侣", "宇宙安排", "百分百", "一定会", "soulmate", "the universe", "guaranteed", "100%")
ZH_D9_TERMS = ("D9", "九分盘", "婚姻分盘", "Navamsa", "Navamsha")
EN_D9_TERMS = ("D9", "Navamsa", "Navamsha")
ZH_ANALYST_LED_OPENERS = ("这段关系的核心", "这段关系不是", "不是有没有吸引", "不是“", "不是\"")
EN_ANALYST_LED_OPENERS = ("the core of this relationship", "this relationship is not")
ZH_AFFECTION_OPENING_TERMS = ("珍惜", "靠近", "在意", "被理解", "被重视", "安心", "温暖", "好感", "心动", "亲近", "回应", "投入")
EN_AFFECTION_OPENING_TERMS = ("cherish", "close", "care", "understood", "valued", "safe", "warm", "drawn", "affection", "respond", "invest")
ZH_SECTION_MIN_CHARACTERS = 100
EN_SECTION_MIN_WORDS = 70
ZH_AFFECTION_SECTION_WARMTH_TERMS = ("珍惜", "靠近", "在意", "被理解", "被重视", "安心", "温暖", "好感", "心动", "亲近")
ZH_AFFECTION_SECTION_BOUNDARY_TERMS = ("空间", "压力", "边界", "催促", "忽略", "退缩")
EN_AFFECTION_SECTION_WARMTH_TERMS = ("cherish", "close", "care", "understood", "valued", "safe", "warm", "drawn", "affection")
EN_AFFECTION_SECTION_BOUNDARY_TERMS = ("space", "pressure", "boundary", "rush", "ignored", "withdraw")
def _heading_status(text: str, headings: tuple[str, ...]) -> dict[str, bool]:
return {heading: heading in text for heading in headings}
def _opening_after_heading(text: str, heading: str, headings: tuple[str, ...]) -> str:
"""Return the reader-visible opening of one section, not the whole report."""
return _section_bodies(text, headings).get(heading, "")[:500]
def _section_bodies(text: str, headings: tuple[str, ...]) -> dict[str, str]:
"""Extract heading bodies without mistaking prose mentions for headings."""
found: list[tuple[int, int, str]] = []
for heading in headings:
pattern = re.compile(
rf"^[ \t]*(?:#{{1,6}}[ \t]*)?(?:(?:[一二三四五六七八九十]+|[0-9]+|[IVXLCDM]+)[、..)][ \t]*)?{re.escape(heading)}[ \t#]*$",
re.MULTILINE,
)
match = pattern.search(text)
if match:
found.append((match.start(), match.end(), heading))
found.sort()
bodies = {heading: "" for heading in headings}
for index, (_start, end, heading) in enumerate(found):
next_start = found[index + 1][0] if index + 1 < len(found) else len(text)
bodies[heading] = text[end:next_start].strip()
return bodies
def _opening_quality(chinese: str, english: str) -> dict[str, Any]:
zh_opening = _opening_after_heading(chinese, "结论总纲", ZH_HEADINGS)
en_opening = _opening_after_heading(english, "Relationship Thesis", EN_HEADINGS)
zh_hits = [phrase for phrase in ZH_ANALYST_LED_OPENERS if phrase in zh_opening]
en_lower = en_opening.lower()
en_hits = [phrase for phrase in EN_ANALYST_LED_OPENERS if phrase in en_lower]
return {
"chinese_analyst_led_openers": zh_hits,
"english_analyst_led_openers": en_hits,
"status": "passed" if not zh_hits and not en_hits else "review_required",
"boundary": "The thesis must begin with the reader's lived relationship question and answer, not an analyst-led contrast such as 'not X but Y'.",
}
def _dossier_status(dossier: Mapping[str, Any]) -> dict[str, str]:
rows = dossier.get("source_map", []) if isinstance(dossier.get("source_map"), list) else []
return {
str(row.get("label")): str(row.get("status"))
for row in rows if isinstance(row, Mapping) and row.get("label")
}
def _d9_pair_evidence_status(dossier: Mapping[str, Any]) -> str:
interaction = dossier.get("pair_interaction_evidence")
if not isinstance(interaction, Mapping):
return "not_available"
d9 = interaction.get("d9_interaction_evidence")
return str(d9.get("status")) if isinstance(d9, Mapping) else "not_available"
def _mutual_affection_status(dossier: Mapping[str, Any]) -> str:
brief = dossier.get("mutual_affection_activation")
return str(brief.get("status")) if isinstance(brief, Mapping) else "not_available"
def _opening_affection_coverage(
dossier: Mapping[str, Any], chinese: str, english: str,
) -> dict[str, Any]:
"""Require a warm, evidence-backed opening only when the brief exists."""
status = _mutual_affection_status(dossier)
required = status == "executed"
zh_opening = _opening_after_heading(chinese, "结论总纲", ZH_HEADINGS)
en_opening = _opening_after_heading(english, "Relationship Thesis", EN_HEADINGS).lower()
chinese_terms = [term for term in ZH_AFFECTION_OPENING_TERMS if term in zh_opening]
english_terms = [term for term in EN_AFFECTION_OPENING_TERMS if term in en_opening]
return {
"available_status": status,
"required_in_opening": required,
"chinese_terms": chinese_terms,
"english_terms": english_terms,
"status": "passed" if not required or (chinese_terms and english_terms) else "review_required",
"boundary": (
"A warm opening describes evidence-backed conditions for reciprocal care. "
"It must not promise love, consent, fidelity, or an outcome."
),
}
def _section_depth_coverage(chinese: str, english: str) -> dict[str, Any]:
chinese_bodies = _section_bodies(chinese, ZH_HEADINGS)
english_bodies = _section_bodies(english, EN_HEADINGS)
chinese_rows = {
heading: len(re.findall(r"[\u4e00-\u9fff]", body))
for heading, body in chinese_bodies.items()
}
english_rows = {
heading: len(re.findall(r"\b[A-Za-z][A-Za-z'-]*\b", body))
for heading, body in english_bodies.items()
}
return {
"chinese_characters": chinese_rows,
"chinese_minimum_per_section": ZH_SECTION_MIN_CHARACTERS,
"english_words": english_rows,
"english_minimum_per_section": EN_SECTION_MIN_WORDS,
"status": "passed" if (
all(count >= ZH_SECTION_MIN_CHARACTERS for count in chinese_rows.values())
and all(count >= EN_SECTION_MIN_WORDS for count in english_rows.values())
) else "review_required",
"boundary": "This guards against heading-only sections and does not certify astrological correctness or narrative quality.",
}
def _attraction_reciprocity_coverage(dossier: Mapping[str, Any], chinese: str, english: str) -> dict[str, Any]:
"""Require warmth and boundary coverage when reciprocal evidence exists."""
required = _mutual_affection_status(dossier) == "executed"
chinese_section = _section_bodies(chinese, ZH_HEADINGS)["吸引与亲密"]
english_section = _section_bodies(english, EN_HEADINGS)["Attraction and Closeness"].lower()
chinese_warmth = [term for term in ZH_AFFECTION_SECTION_WARMTH_TERMS if term in chinese_section]
chinese_boundary = [term for term in ZH_AFFECTION_SECTION_BOUNDARY_TERMS if term in chinese_section]
english_warmth = [term for term in EN_AFFECTION_SECTION_WARMTH_TERMS if term in english_section]
english_boundary = [term for term in EN_AFFECTION_SECTION_BOUNDARY_TERMS if term in english_section]
passed = not required or (
chinese_warmth and chinese_boundary and english_warmth and english_boundary
)
return {
"required": required,
"chinese_warmth_terms": chinese_warmth,
"chinese_boundary_terms": chinese_boundary,
"english_warmth_terms": english_warmth,
"english_boundary_terms": english_boundary,
"status": "passed" if passed else "review_required",
"boundary": (
"This requires reciprocal-care and space/pressure language when evidence is executed. "
"It does not prove consent, attraction, fidelity or any relationship outcome."
),
}
def build_delivery_receipt(dossier: Mapping[str, Any], chinese: str, english: str) -> dict[str, Any]:
source_status = _dossier_status(dossier)
cited_labels = {label: {"zh": label in chinese, "en": label in english} for label in EVIDENCE_LABELS}
primary_pair_tasks = [
row for row in (dossier.get("execution_receipt", {}) or {}).get("pair_execution_tasks", [])
if isinstance(row, Mapping) and row.get("status") in {"ready_for_comparison", "parameter_sensitive", "blocked"}
]
english_cjk_ratio = len(re.findall(r"[\u4e00-\u9fff]", english)) / max(len(english), 1)
chinese_characters = len(re.findall(r"[\u4e00-\u9fff]", chinese))
english_words = len(re.findall(r"\b[A-Za-z][A-Za-z'-]*\b", english))
zh_headings = _heading_status(chinese, ZH_HEADINGS)
en_headings = _heading_status(english, EN_HEADINGS)
d9_status = _d9_pair_evidence_status(dossier)
d9_required = d9_status == "executed"
d9_coverage = {
"available_status": d9_status,
"required_in_narrative": d9_required,
"chinese": any(term in chinese for term in ZH_D9_TERMS),
"english": any(term.lower() in english.lower() for term in EN_D9_TERMS),
}
opening_quality = _opening_quality(chinese, english)
opening_affection = _opening_affection_coverage(dossier, chinese, english)
section_depth = _section_depth_coverage(chinese, english)
attraction_reciprocity = _attraction_reciprocity_coverage(dossier, chinese, english)
return {
"schema": SCHEMA,
"dossier_schema": dossier.get("schema"),
"composition": {"chinese": "direct_from_structured_dossier", "english": "direct_from_structured_dossier_not_translated_from_chinese"},
"headings": {"chinese": zh_headings, "english": en_headings},
"evidence_labels": cited_labels,
"source_map_status": source_status,
"pair_tasks": [{"task": row.get("task"), "status": row.get("status")} for row in primary_pair_tasks],
"d9_narrative_coverage": d9_coverage,
"opening_quality": opening_quality,
"opening_affection_coverage": opening_affection,
"section_depth_coverage": section_depth,
"attraction_reciprocity_coverage": attraction_reciprocity,
"supplementary_method_terms": {"chinese": {term: term in chinese for term in SUPPLEMENTARY_METHOD_TERMS}, "english": {term: term in english for term in SUPPLEMENTARY_METHOD_TERMS}},
"private_evidence_terms": {"chinese": {term: term in chinese for term in PRIVATE_EVIDENCE_TERMS}, "english": {term: term in english for term in PRIVATE_EVIDENCE_TERMS}},
"english_cjk_ratio": round(english_cjk_ratio, 5),
"reader_depth": {"chinese_characters": chinese_characters, "chinese_minimum": ZH_READER_MIN_CHARACTERS, "english_words": english_words, "english_minimum": EN_READER_MIN_WORDS, "status": "passed" if chinese_characters >= ZH_READER_MIN_CHARACTERS and english_words >= EN_READER_MIN_WORDS else "review_required"},
"reader_language_terms": {"chinese": {term: term in chinese for term in READER_BANNED_TERMS}, "english": {term: term in english.lower() for term in READER_BANNED_TERMS}},
"status": "passed" if (
dossier.get("delivery_readiness") == "ready_for_narrative" and all(zh_headings.values()) and all(en_headings.values())
and opening_quality["status"] == "passed" and opening_affection["status"] == "passed"
and section_depth["status"] == "passed" and attraction_reciprocity["status"] == "passed"
and (not d9_required or (d9_coverage["chinese"] and d9_coverage["english"]))
and not any(term in chinese for term in SUPPLEMENTARY_METHOD_TERMS) and not any(term in english for term in SUPPLEMENTARY_METHOD_TERMS)
and not any(term in chinese for term in PRIVATE_EVIDENCE_TERMS) and not any(term in english for term in PRIVATE_EVIDENCE_TERMS)
and chinese_characters >= ZH_READER_MIN_CHARACTERS and english_words >= EN_READER_MIN_WORDS
and not any(term in chinese for term in READER_BANNED_TERMS) and not any(term in english.lower() for term in READER_BANNED_TERMS)
and english_cjk_ratio < 0.05
) else "review_required",
"boundary": "This validates delivery structure and shared evidence coverage. It does not prove a translation is absent or replace technical review.",
}
def build_single_language_delivery_receipt(
dossier: Mapping[str, Any], report: str, language: str,
) -> dict[str, Any]:
"""Validate one independently composed customer language without a placeholder peer."""
if language not in {"zh", "en"}:
raise ValueError("language must be 'zh' or 'en'")
headings = ZH_HEADINGS if language == "zh" else EN_HEADINGS
text = report if language == "zh" else report.lower()
d9_status = _d9_pair_evidence_status(dossier)
affection_status = _mutual_affection_status(dossier)
section_bodies = _section_bodies(report, headings)
section_sizes = {
heading: (
len(re.findall(r"[\u4e00-\u9fff]", body)) if language == "zh"
else len(re.findall(r"\b[A-Za-z][A-Za-z'-]*\b", body))
)
for heading, body in section_bodies.items()
}
min_size = ZH_SECTION_MIN_CHARACTERS if language == "zh" else EN_SECTION_MIN_WORDS
opening_heading = "结论总纲" if language == "zh" else "Relationship Thesis"
attraction_heading = "吸引与亲密" if language == "zh" else "Attraction and Closeness"
opening = _opening_after_heading(report, opening_heading, headings)
analyst_terms = ZH_ANALYST_LED_OPENERS if language == "zh" else EN_ANALYST_LED_OPENERS
analyst_hits = [term for term in analyst_terms if term in (opening if language == "zh" else opening.lower())]
warmth_terms = ZH_AFFECTION_OPENING_TERMS if language == "zh" else EN_AFFECTION_OPENING_TERMS
warmth_hits = [term for term in warmth_terms if term in (opening if language == "zh" else opening.lower())]
attraction = section_bodies[attraction_heading] if language == "zh" else section_bodies[attraction_heading].lower()
attraction_warmth = ZH_AFFECTION_SECTION_WARMTH_TERMS if language == "zh" else EN_AFFECTION_SECTION_WARMTH_TERMS
attraction_boundary = ZH_AFFECTION_SECTION_BOUNDARY_TERMS if language == "zh" else EN_AFFECTION_SECTION_BOUNDARY_TERMS
attraction_warmth_hits = [term for term in attraction_warmth if term in attraction]
attraction_boundary_hits = [term for term in attraction_boundary if term in attraction]
narrative_d9_terms = ZH_D9_TERMS if language == "zh" else EN_D9_TERMS
d9_present = any(term.lower() in text for term in narrative_d9_terms)
reader_depth = len(re.findall(r"[\u4e00-\u9fff]", report)) if language == "zh" else len(re.findall(r"\b[A-Za-z][A-Za-z'-]*\b", report))
reader_min = ZH_READER_MIN_CHARACTERS if language == "zh" else EN_READER_MIN_WORDS
forbidden = READER_BANNED_TERMS
forbidden_hits = [term for term in forbidden if term in (report if language == "zh" else report.lower())]
supplementary_hits = [term for term in SUPPLEMENTARY_METHOD_TERMS if term in report]
private_hits = [term for term in PRIVATE_EVIDENCE_TERMS if term in report]
checks = {
"dossier_ready": dossier.get("delivery_readiness") == "ready_for_narrative",
"all_headings": all(_heading_status(report, headings).values()),
"opening_is_reader_led": not analyst_hits,
"opening_affection": affection_status != "executed" or bool(warmth_hits),
"section_depth": all(size >= min_size for size in section_sizes.values()),
"attraction_reciprocity": affection_status != "executed" or bool(attraction_warmth_hits and attraction_boundary_hits),
"d9_coverage": d9_status != "executed" or d9_present,
"reader_depth": reader_depth >= reader_min,
"no_forbidden_language": not forbidden_hits,
"no_supplementary_method_names": not supplementary_hits,
"no_private_evidence_terms": not private_hits,
"english_is_not_mostly_cjk": language != "en" or len(re.findall(r"[\u4e00-\u9fff]", report)) / max(len(report), 1) < 0.05,
}
return {
"schema": SINGLE_LANGUAGE_SCHEMA,
"dossier_schema": dossier.get("schema"),
"language": language,
"composition": "direct_from_structured_dossier" if language == "zh" else "direct_from_structured_dossier_not_translated_from_chinese",
"headings": _heading_status(report, headings),
"checks": checks,
"source_map_status": _dossier_status(dossier),
"d9_narrative_coverage": {"available_status": d9_status, "required_in_narrative": d9_status == "executed", "present": d9_present},
"opening_quality": {"analyst_led_openers": analyst_hits, "status": "passed" if not analyst_hits else "review_required"},
"opening_affection_coverage": {"available_status": affection_status, "terms": warmth_hits, "status": "passed" if checks["opening_affection"] else "review_required"},
"section_depth_coverage": {"sizes": section_sizes, "minimum": min_size, "status": "passed" if checks["section_depth"] else "review_required"},
"attraction_reciprocity_coverage": {"warmth_terms": attraction_warmth_hits, "boundary_terms": attraction_boundary_hits, "status": "passed" if checks["attraction_reciprocity"] else "review_required"},
"reader_depth": {"value": reader_depth, "minimum": reader_min, "status": "passed" if checks["reader_depth"] else "review_required"},
"status": "passed" if all(checks.values()) else "review_required",
"boundary": "This validates one independently authored language. It does not certify a future second-language report or treat one language as a translation source.",
}
def main() -> int:
parser = argparse.ArgumentParser(description="Validate independent Chinese and English full synastry reports.")
parser.add_argument("--dossier", required=True, help="Private full-synastry dossier JSON")
parser.add_argument("--chinese", help="Chinese Markdown report")
parser.add_argument("--english", help="English Markdown report written directly from the dossier")
parser.add_argument("--language", choices=("zh", "en"), help="Validate one independently authored language")
parser.add_argument("--report", help="Single-language Markdown report used with --language")
parser.add_argument("--output", help="Private receipt output path")
args = parser.parse_args()
dossier = json.loads(Path(args.dossier).read_text(encoding="utf-8"))
if args.language:
if not args.report or args.chinese or args.english:
parser.error("--language requires --report and cannot be combined with --chinese or --english")
receipt = build_single_language_delivery_receipt(
dossier, Path(args.report).read_text(encoding="utf-8"), args.language,
)
else:
if not args.chinese or not args.english:
parser.error("bilingual validation requires --chinese and --english")
receipt = build_delivery_receipt(
dossier,
Path(args.chinese).read_text(encoding="utf-8"),
Path(args.english).read_text(encoding="utf-8"),
)
encoded = json.dumps(receipt, ensure_ascii=False, indent=2)
if args.output:
Path(args.output).write_text(encoded + "\n", encoding="utf-8")
else:
print(encoded)
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,311 @@
#!/usr/bin/env python3
"""Create a privacy-safe execution receipt for two full-reading packets.
This is deliberately an audit adapter, not an interpreter. It establishes
whether both people were calculated under comparable settings and whether the
relationship route's individual evidence modules are actually present before a
synastry narrative is written.
"""
from __future__ import annotations
import argparse
import json
from pathlib import Path
from typing import Any, Mapping
SCHEMA = "jyotish.full_synastry_execution_receipt.v1"
# The paths are the native full-reading outputs. Several technique labels
# intentionally share a producer because one calculation supports more than
# one relationship question; the report writer must still account for each.
REQUIRED_TECHNIQUES: tuple[tuple[str, tuple[str, ...], str], ...] = (
("D1/Rashi", ("chart",), "natal promise and house framework"),
("Functional Benefic/Malefic", ("functional_benefic_malefic",), "Lagna-specific roles"),
("Bhava Chalit", ("bhava_chalit",), "house-boundary confirmation"),
("Planetary Aspects", ("aspects",), "actual contact and friction channels"),
("Planetary Friendship", ("planetary_friendship",), "planetary cooperation context"),
("D9/Navamsa", ("varga_full.D9_Navamsa", "d9_navamsa_expanded"), "long-term relationship structure"),
("Vimshottari", ("dasha",), "primary planet-period timing"),
("Dasha Sandhi", ("dasha_sandhi",), "period-transition pressure"),
("Narayana Dasha", ("narayana_dasha",), "rashi-period cross-check"),
("Jaimini/Chara", ("jaimini",), "karakas, arudhas and rashi timing"),
("Darakaraka (DK)", ("jaimini",), "partner significator"),
("Upapada Lagna (UL)", ("special_lagnas", "yogas_doshas"), "marriage formation and sustenance"),
("Argala", ("argala",), "support and obstruction"),
("Shadbala", ("shadbala",), "relative planetary strength"),
("Ashtakavarga", ("ashtakavarga",), "transit capacity"),
("Vimsopaka", ("vimsopaka",), "cross-varga dignity weighting"),
("Vargottama", ("vargottama",), "D1-D9 reinforcement"),
("Pushkara", ("pushkara",), "supportive Navamsa condition"),
("Dispositor Chain", ("dispositor_chains", "final_dispositors"), "energy-flow resolution"),
("Inter-chart Linkage", ("inter_chart_linkage",), "D1-D9 linkage within each chart"),
("Avasthas", ("avasthas",), "condition and capacity of relationship significators"),
("Bhava Bala", ("bhava_bala",), "relative capacity of relationship and shared-life houses"),
("Sensitive Points", ("sensitive_points",), "derived point context; never a standalone verdict"),
("Lagna Vargottamamsa", ("lagna_vargottamamsa",), "D1-D9 ascendant reinforcement"),
("Extended Vargas", ("varga_extended",), "D5/D6/D8/D11 context; use only by documented topic"),
("Tajika Yogas", ("tajika_yogas",), "annual activation context"),
("Vivah and other Sahams", ("sahams",), "event-sensitive points"),
("Mangal Dosha", ("yogas_doshas",), "Mars-only marriage-pressure evidence"),
("Punarphoo", ("punarphoo",), "observation-only Moon-Saturn layer"),
("Arudha Lagna", ("special_lagnas", "yogas_doshas"), "public relationship frame"),
("Tithi Lord", ("tithi_lord",), "lunar relationship context"),
("Rashi Tulya Navamsa", ("rashi_tulya_navamsa",), "D1-D9 comparison"),
("Marriage Counting", ("marriage_counting",), "supplemental relationship-pattern layer"),
("D30 Trimshamsa", ("trimshamsa_d30",), "crisis-pressure context only"),
("Muntha / Solar Return", ("muntha", "solar_return"), "annual relationship weather"),
("Bhrigu Pada Dasha", ("bhrigu_pada_dasha",), "parameter-sensitive comparison layer"),
("Nakshatra Advanced", ("nakshatra_dasha", "nakshatra_advanced"), "nakshatra timing/context"),
("Double Transit / LL7L", ("double_transit_pac", "transit_ll7l"), "reference-date activation"),
("Transit Multi-reference", ("transit_positions", "transit_multi_reference"), "cross-reference present activation"),
("KP 7th-house layer", ("kp",), "formalisation question support"),
("Sudarshana Chakra", ("sudarshana",), "triple-lagna confirmation"),
("Yogini Dasha", ("yogini_dasha",), "secondary timing only; retain its own birth-balance boundary"),
("Ashtottari Dasha", ("ashtottari_dasha",), "conditional secondary timing; verify applicability"),
("Kalachakra Dasha", ("kalachakra_dasha",), "conditional secondary timing; verify method profile"),
("Navamsha Dasha", ("navamsha_dasha",), "D9-oriented secondary timing; parameter-sensitive"),
("Niryaana Shoola Dasha", ("niryaana_shoola_dasha",), "longevity-specific system; not applicable to relationship timing"),
("Shastihayani Dasha", ("shastihayani_dasha",), "conditional secondary timing; applicability required"),
("Dasha Convergence", ("dasa_convergence",), "cross-dasha agreement within each chart"),
("Relationship Strict Evidence", ("relationship_strict_evidence",), "relationship-route rule and book-evidence payload"),
("Western comparison layer", ("unified_western_reading",), "independent supporting lens"),
)
PAIR_TASKS: tuple[tuple[str, tuple[str, ...], str], ...] = (
(
"D1/D9 mutual overlays and relationship axes",
("chart", "varga_full.D9_Navamsa"),
"primary Jyotish pair structure",
),
(
"lived interaction: emotional safety, attraction, communication and conflict pattern",
("chart", "aspects", "planetary_friendship", "varga_full.D9_Navamsa"),
"compare mutual Lagna/Moon/Venus/Mars/Mercury contacts and D9 confirmation; write observable interaction, not personality labels",
),
(
"shared gain and cost: resources, home, work and private-space axes",
("chart", "varga_full.D2_Hora", "varga_full.D4_Turyamsa", "varga_full.D10_Dasamsa", "bhava_bala", "ashtakavarga"),
"compare 2/4/6/10/11/12-house capacity and relevant varga context; no financial prediction from compatibility alone",
),
(
"long-term intimacy and repair: 5/7/8/12 axes, UL/DK/A7, Argala and pressure layers",
("chart", "jaimini", "special_lagnas", "argala", "yogas_doshas", "trimshamsa_d30"),
"separate attraction, commitment, deep binding, recovery and pressure; D30 is bounded context only",
),
(
"DK, UL, UL-second and A7/Darapada cross-comparison",
("jaimini", "special_lagnas"),
"primary Jaimini marriage structure",
),
(
"two-person Dasha/Narayana/Chara context comparison",
("dasha", "narayana_dasha", "jaimini"),
"primary timing convergence",
),
(
"secondary-dasha convergence and disagreement review",
("nakshatra_dasha", "yogini_dasha", "ashtottari_dasha", "kalachakra_dasha", "navamsha_dasha", "dasa_convergence"),
"secondary systems qualify or challenge a primary timing window; never invent a date when profiles conflict",
),
(
"shared transit, Tajika, Saham and annual-window comparison",
("tajika_yogas", "sahams", "muntha"),
"annual and transit corroboration",
),
(
"international Ashtakoota core, additional Kutas, Mars-only Kuja and separate Papasamya",
("chart",),
"limited Moon/nakshatra and pressure context; build through marriage-match",
),
(
"Nakshatra relationship context",
("nakshatra_dasha",),
"Jyotish nakshatra context; never a standalone marriage verdict",
),
(
"optional 27-nakshatra Shukuyo narrative layer",
(),
"separate cultural narrative only; never merge into Jyotish evidence",
),
(
"optional Western synastry/composite/Davison layer",
("unified_western_reading",),
"separate supporting perspective; a natal Western packet is not a composite chart",
),
(
"Prashna context when a real question chart was supplied",
(),
"guarded context only; never replace the two natal charts",
),
)
def _packet(value: Mapping[str, Any]) -> Mapping[str, Any]:
nested = value.get("raw_full_reading") if isinstance(value, Mapping) else None
return nested if isinstance(nested, Mapping) else value
def _get_path(value: Mapping[str, Any], dotted: str) -> Any:
cursor: Any = value
for key in dotted.split("."):
if not isinstance(cursor, Mapping) or key not in cursor:
return None
cursor = cursor[key]
return cursor
def _status(value: Any) -> str:
if value is None:
return "blocked"
if isinstance(value, Mapping):
state = str(value.get("execution_status") or value.get("status", "")).lower()
if state in {"blocked", "not_available", "unavailable", "skipped"}:
return "blocked"
if state in {
"partial", "partial_verified", "parameter_sensitive", "observation_only",
"guarded", "reference_only", "computed_pending_validation",
}:
return "parameter_sensitive"
if value.get("skipped") is True or value.get("available") is False:
return "blocked"
return "executed"
def _profile(packet: Mapping[str, Any]) -> dict[str, Any]:
birth = packet.get("birth_info", {}) if isinstance(packet.get("birth_info"), Mapping) else {}
profile = packet.get("calculation_profile", {}) if isinstance(packet.get("calculation_profile"), Mapping) else {}
settings = profile.get("effective_settings", {}) if isinstance(profile.get("effective_settings"), Mapping) else {}
return {
"ayanamsa": birth.get("ayanamsa_name") or birth.get("ayanamsa") or profile.get("ayanamsa") or settings.get("ayanamsa"),
"node_mode": birth.get("node_mode") or profile.get("node_mode") or settings.get("node_mode"),
"position_mode": profile.get("position_mode") or settings.get("position_mode"),
"house_system": profile.get("house_system") or settings.get("house_system"),
}
def _technique_row(packet: Mapping[str, Any], definition: tuple[str, tuple[str, ...], str]) -> dict[str, Any]:
label, candidates, purpose = definition
if label == "Niryaana Shoola Dasha":
return {
"technique": label,
"status": "not_applicable",
"source_path": None,
"purpose": purpose,
}
modules = packet.get("modules", {}) if isinstance(packet.get("modules"), Mapping) else {}
found_path = next((path for path in candidates if _get_path(modules, path) is not None), None)
value = _get_path(modules, found_path) if found_path else None
return {
"technique": label,
"status": _status(value),
"source_path": f"modules.{found_path}" if found_path else None,
"purpose": purpose,
}
def _pair_task_row(
person_a: Mapping[str, Any],
person_b: Mapping[str, Any],
definition: tuple[str, tuple[str, ...], str],
) -> dict[str, Any]:
"""State whether both packets can support a comparison, not its conclusion."""
task, required_paths, boundary = definition
if not required_paths:
return {
"task": task,
"status": "not_applicable",
"required_inputs": [],
"boundary": boundary,
}
modules_a = person_a.get("modules", {}) if isinstance(person_a.get("modules"), Mapping) else {}
modules_b = person_b.get("modules", {}) if isinstance(person_b.get("modules"), Mapping) else {}
inputs = []
for path in required_paths:
a_value = _get_path(modules_a, path)
b_value = _get_path(modules_b, path)
a_status, b_status = _status(a_value), _status(b_value)
if a_status == b_status == "executed":
status = "ready_for_comparison"
elif "blocked" in {a_status, b_status}:
status = "blocked"
else:
status = "parameter_sensitive"
inputs.append({
"source_path": f"modules.{path}",
"status": status,
"person_a_status": a_status,
"person_b_status": b_status,
})
overall = "ready_for_comparison"
if any(row["status"] == "blocked" for row in inputs):
overall = "blocked"
elif any(row["status"] == "parameter_sensitive" for row in inputs):
overall = "parameter_sensitive"
return {
"task": task,
"status": overall,
"required_inputs": inputs,
"boundary": boundary,
}
def build_execution_receipt(person_a: Mapping[str, Any], person_b: Mapping[str, Any]) -> dict[str, Any]:
"""Return a non-identifying audit receipt for a pair of full readings."""
a, b = _packet(person_a), _packet(person_b)
profile_a, profile_b = _profile(a), _profile(b)
profile_complete = all(profile_a.values()) and all(profile_b.values())
profile_match = profile_complete and all(profile_a[key] == profile_b[key] for key in profile_a)
rows = []
for definition in REQUIRED_TECHNIQUES:
a_row, b_row = _technique_row(a, definition), _technique_row(b, definition)
individual_statuses = {a_row["status"], b_row["status"]}
combined = "executed" if individual_statuses == {"executed"} else "blocked"
if individual_statuses == {"not_applicable"}:
combined = "not_applicable"
elif combined == "blocked" and individual_statuses <= {"executed", "partial", "parameter_sensitive", "observation_only", "guarded", "not_applicable"}:
combined = "parameter_sensitive"
rows.append({
"technique": definition[0],
"status": combined,
"person_a": a_row,
"person_b": b_row,
})
return {
"schema": SCHEMA,
"calculation_profile": {"person_a": profile_a, "person_b": profile_b, "compatible": profile_match},
"profile_gate": "passed" if profile_match else "blocked",
"profile_gate_reason": None if profile_match else (
"calculation_profile_incomplete" if not profile_complete else "calculation_profile_mismatch"
),
"individual_technique_audit": rows,
"pair_execution_tasks": [_pair_task_row(a, b, task) for task in PAIR_TASKS],
"boundary": (
"This receipt proves packet presence and calculation-profile comparability only. "
"It does not turn a module into a conclusion; pair tasks, book evidence, MEVG and case calibration remain required."
),
}
def main() -> int:
parser = argparse.ArgumentParser(description="Validate two full-reading packets before a full synastry interpretation.")
parser.add_argument("--person-a", required=True, help="JSON full-reading or export packet for person A")
parser.add_argument("--person-b", required=True, help="JSON full-reading or export packet for person B")
parser.add_argument("--output", help="Optional receipt path; do not place customer data in a repository path.")
args = parser.parse_args()
with Path(args.person_a).open(encoding="utf-8") as handle:
person_a = json.load(handle)
with Path(args.person_b).open(encoding="utf-8") as handle:
person_b = json.load(handle)
receipt = build_execution_receipt(person_a, person_b)
encoded = json.dumps(receipt, ensure_ascii=False, indent=2)
if args.output:
Path(args.output).write_text(encoded + "\n", encoding="utf-8")
else:
print(encoded)
return 0
if __name__ == "__main__":
raise SystemExit(main())
@@ -0,0 +1,210 @@
from __future__ import annotations
import importlib.util
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
MODULE_PATH = ROOT / "skills" / "jyotish-full-synastry-reading" / "scripts" / "validate_bilingual_synastry_delivery.py"
spec = importlib.util.spec_from_file_location("bilingual_synastry_delivery", MODULE_PATH)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
def _full_chinese_report() -> str:
headings = ["结论总纲", "各自带入关系的需求", "情绪安全与相处感受", "吸引与亲密", "沟通、冲突与修复", "关系如何推进", "共同生活的收益与压力", "能不能稳定走下去", "关系应期"]
prose = "双方需要清楚说明节奏、边界、现实安排与实际互动;任何判断都需由两人的现实经验验证,不能由单一配置替代。"
sections = []
for heading in headings:
suffix = "D9 婚姻分盘用于检验长期承诺的结构,不能代替本命互动。" if heading == "能不能稳定走下去" else ""
sections.append(f"{heading}\n{(prose * 5) + suffix}")
return "\n".join(sections)
def _full_english_report() -> str:
headings = [
"Relationship Thesis", "What Each Person Brings", "Emotional Safety and Lived Experience",
"Attraction and Closeness", "Communication, Conflict and Repair",
"How the Relationship Progresses", "Shared Life: Gains and Pressures",
"Can It Stay Steady?", "Relationship Weather",
]
prose = "Each conclusion remains conditional on both charts, actual interaction, practical circumstances, and repeated real-world observation by both people. "
sections = []
for heading in headings:
suffix = "Navamsa (D9) qualifies long-term commitment and does not replace lived natal interaction. " if heading == "Can It Stay Steady?" else ""
sections.append(f"{heading}\n{(prose * 8) + suffix}")
return "\n".join(sections)
def test_receipt_requires_independent_full_report_shapes_without_forced_term_dumping():
dossier = {
"schema": "jyotish.full_synastry_dossier.v1",
"delivery_readiness": "ready_for_narrative",
"source_map": [],
"execution_receipt": {"pair_execution_tasks": []},
}
chinese = _full_chinese_report()
english = _full_english_report()
receipt = module.build_delivery_receipt(dossier, chinese, english)
assert receipt["status"] == "passed"
assert receipt["composition"]["english"] == "direct_from_structured_dossier_not_translated_from_chinese"
assert not any(receipt["supplementary_method_terms"]["english"].values())
def test_receipt_rejects_incomplete_or_chinese_english_delivery():
receipt = module.build_delivery_receipt({"delivery_readiness": "blocked"}, "结论总纲", "结论总纲")
assert receipt["status"] == "review_required"
def test_receipt_allows_primary_jyotish_terms_but_hides_supplementary_method_names():
dossier = {
"schema": "jyotish.full_synastry_dossier.v1",
"delivery_readiness": "ready_for_narrative",
"source_map": [],
"execution_receipt": {"pair_execution_tasks": []},
}
chinese = _full_chinese_report() + "\nD9 与 UL 的结论在此一致"
english = _full_english_report() + "\nNavamsa Upapada Darakaraka Vimshottari Narayana"
assert module.build_delivery_receipt(dossier, chinese, english)["status"] == "passed"
assert module.build_delivery_receipt(dossier, chinese, english + "\nWestern synastry")["status"] == "review_required"
assert module.build_delivery_receipt(dossier, chinese, english + "\nexecution_receipt")["status"] == "review_required"
assert module.build_delivery_receipt(dossier, chinese, english + "\nreader_section_brief")["status"] == "review_required"
def test_receipt_rejects_heading_only_or_fated_language():
dossier = {"schema": "jyotish.full_synastry_dossier.v1", "delivery_readiness": "ready_for_narrative", "source_map": [], "execution_receipt": {"pair_execution_tasks": []}}
assert module.build_delivery_receipt(dossier, "结论总纲", "Relationship Thesis")["reader_depth"]["status"] == "review_required"
assert module.build_delivery_receipt(dossier, _full_chinese_report() + "灵魂伴侣", _full_english_report())["status"] == "review_required"
def test_receipt_rejects_analyst_led_not_x_but_y_opening():
dossier = {"schema": "jyotish.full_synastry_dossier.v1", "delivery_readiness": "ready_for_narrative", "source_map": [], "execution_receipt": {"pair_execution_tasks": []}}
chinese = _full_chinese_report().replace("结论总纲", "结论总纲\n这段关系的核心,不是有没有吸引,而是能否落实承诺。", 1)
receipt = module.build_delivery_receipt(dossier, chinese, _full_english_report())
assert receipt["opening_quality"]["status"] == "review_required"
assert receipt["status"] == "review_required"
def test_receipt_requires_warm_reciprocal_opening_when_affection_evidence_is_executed():
dossier = {
"schema": "jyotish.full_synastry_dossier.v1",
"delivery_readiness": "ready_for_narrative",
"source_map": [],
"execution_receipt": {"pair_execution_tasks": []},
"mutual_affection_activation": {"status": "executed"},
}
cold_chinese = _full_chinese_report()
cold_english = _full_english_report()
cold = module.build_delivery_receipt(dossier, cold_chinese, cold_english)
assert cold["opening_affection_coverage"]["required_in_opening"] is True
assert cold["status"] == "review_required"
chinese = cold_chinese.replace(
"结论总纲\n", "结论总纲\nA 会在 B 的稳定回应里感到被重视与安心,B 也会在 A 的投入中感到被珍惜。\n", 1,
)
english = cold_english.replace(
"Relationship Thesis\n", "Relationship Thesis\nA feels valued and safe in B's steady response, while B feels cherished by A's investment.\n", 1,
)
receipt = module.build_delivery_receipt(dossier, chinese, english)
assert receipt["opening_affection_coverage"]["chinese_terms"]
assert receipt["opening_affection_coverage"]["english_terms"]
assert receipt["attraction_reciprocity_coverage"]["status"] == "review_required"
def test_receipt_rejects_long_report_with_empty_reader_sections():
dossier = {
"schema": "jyotish.full_synastry_dossier.v1",
"delivery_readiness": "ready_for_narrative",
"source_map": [],
"execution_receipt": {"pair_execution_tasks": []},
}
headings_zh = "\n".join(["结论总纲", "各自带入关系的需求", "情绪安全与相处感受", "吸引与亲密", "沟通、冲突与修复", "关系如何推进", "共同生活的收益与压力", "能不能稳定走下去", "关系应期"])
headings_en = "\n".join(["Relationship Thesis", "What Each Person Brings", "Emotional Safety and Lived Experience", "Attraction and Closeness", "Communication, Conflict and Repair", "How the Relationship Progresses", "Shared Life: Gains and Pressures", "Can It Stay Steady?", "Relationship Weather"])
receipt = module.build_delivery_receipt(
dossier,
headings_zh + "\n" + "双方需要核对实际互动与现实安排。" * 140,
headings_en + "\n" + "Both people need to verify practical interaction and real circumstances before drawing conclusions. " * 180,
)
assert receipt["reader_depth"]["status"] == "passed"
assert receipt["section_depth_coverage"]["status"] == "review_required"
assert receipt["status"] == "review_required"
def test_section_depth_recognizes_markdown_and_numbered_headings():
chinese = "## 一、结论总纲\n" + "双方需要把现实互动说清。" * 20
english = "## I. Relationship Thesis\n" + "Both people need clear real-world interaction before conclusions. " * 20
assert module._section_bodies(chinese, ("结论总纲",))["结论总纲"]
assert module._section_bodies(english, ("Relationship Thesis",))["Relationship Thesis"]
def test_receipt_requires_warmth_and_space_in_attraction_section_when_evidence_exists():
dossier = {
"schema": "jyotish.full_synastry_dossier.v1",
"delivery_readiness": "ready_for_narrative",
"source_map": [],
"execution_receipt": {"pair_execution_tasks": []},
"mutual_affection_activation": {"status": "executed"},
}
chinese = _full_chinese_report().replace(
"结论总纲\n", "结论总纲\nA 会在 B 的稳定回应里感到被重视与安心,B 也会在 A 的投入中感到被珍惜。\n", 1,
)
english = _full_english_report().replace(
"Relationship Thesis\n", "Relationship Thesis\nA feels valued and safe in B's steady response, while B feels cherished by A's investment.\n", 1,
)
cold_attraction = module.build_delivery_receipt(dossier, chinese, english)
assert cold_attraction["attraction_reciprocity_coverage"]["status"] == "review_required"
chinese = chinese.replace("吸引与亲密\n", "吸引与亲密\n双方在被珍惜时更愿意靠近,也要为彼此保留空间,避免压力累积。\n", 1)
english = english.replace("Attraction and Closeness\n", "Attraction and Closeness\nBoth feel warmer and closer when valued, while each needs space so pressure does not accumulate.\n", 1)
receipt = module.build_delivery_receipt(dossier, chinese, english)
assert receipt["attraction_reciprocity_coverage"]["status"] == "passed"
assert receipt["status"] == "passed"
def test_receipt_requires_d9_narrative_when_native_d9_pair_evidence_executed():
dossier = {
"schema": "jyotish.full_synastry_dossier.v1",
"delivery_readiness": "ready_for_narrative",
"source_map": [],
"execution_receipt": {"pair_execution_tasks": []},
"pair_interaction_evidence": {"d9_interaction_evidence": {"status": "executed"}},
}
receipt = module.build_delivery_receipt(dossier, _full_chinese_report(), _full_english_report())
assert receipt["status"] == "passed"
assert receipt["d9_narrative_coverage"]["required_in_narrative"] is True
without_d9 = _full_chinese_report().replace("D9 婚姻分盘用于检验长期承诺的结构,不能代替本命互动。", "")
assert module.build_delivery_receipt(dossier, without_d9, _full_english_report())["status"] == "review_required"
def test_single_language_receipt_accepts_an_independent_chinese_delivery():
dossier = {
"schema": "jyotish.full_synastry_dossier.v1",
"delivery_readiness": "ready_for_narrative",
"source_map": [],
"execution_receipt": {"pair_execution_tasks": []},
}
receipt = module.build_single_language_delivery_receipt(
dossier, _full_chinese_report(), "zh",
)
assert receipt["status"] == "passed"
assert receipt["schema"] == module.SINGLE_LANGUAGE_SCHEMA
assert receipt["composition"] == "direct_from_structured_dossier"
def test_single_language_receipt_keeps_affection_and_section_gates():
dossier = {
"schema": "jyotish.full_synastry_dossier.v1",
"delivery_readiness": "ready_for_narrative",
"source_map": [],
"execution_receipt": {"pair_execution_tasks": []},
"mutual_affection_activation": {"status": "executed"},
}
receipt = module.build_single_language_delivery_receipt(dossier, _full_chinese_report(), "zh")
assert receipt["status"] == "review_required"
warm = _full_chinese_report().replace(
"结论总纲\n", "结论总纲\n双方在稳定回应中感到被重视与珍惜,也能安心靠近。\n", 1,
).replace(
"吸引与亲密\n", "吸引与亲密\n彼此被珍惜时更愿意靠近,也需保留空间和边界,避免压力累积。\n", 1,
)
assert module.build_single_language_delivery_receipt(dossier, warm, "zh")["status"] == "passed"
@@ -0,0 +1,166 @@
from __future__ import annotations
import importlib.util
import json
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
MODULE_PATH = ROOT / "skills" / "jyotish-full-synastry-reading" / "scripts" / "compose_full_synastry_delivery.py"
spec = importlib.util.spec_from_file_location("compose_full_synastry_delivery", MODULE_PATH)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
def _dossier(*, affection="not_available"):
return {
"schema": "jyotish.full_synastry_dossier.v1",
"delivery_readiness": "ready_for_narrative",
"reader_section_brief": [{"section": key} for key in module.SECTION_KEYS],
"source_map": [],
"execution_receipt": {"pair_execution_tasks": []},
"mutual_affection_activation": {"status": affection},
}
def _payload(language: str):
if language == "zh":
prose = "双方需要清楚说明节奏、边界、现实安排与实际互动;任何判断都需由两人的现实经验验证,不能由单一配置替代。"
sections = {key: prose * 5 for key in module.SECTION_KEYS}
return {
"language": "zh",
"composition": module.EXPECTED_COMPOSITION,
"title": "双方的恋爱与婚姻合盘",
"sections": sections,
}
prose = "Each conclusion remains conditional on both charts, actual interaction, practical circumstances, and repeated real-world observation by both people. "
sections = {key: prose * 8 for key in module.SECTION_KEYS}
return {
"language": "en",
"composition": module.EXPECTED_COMPOSITION,
"title": "Full Love and Marriage Synastry Reading",
"sections": sections,
}
def test_compiler_requires_ready_dossier_and_independent_complete_section_payloads():
compiled = module.compile_delivery(_dossier(), _payload("zh"), _payload("en"))
assert compiled["delivery_receipt"]["status"] == "passed"
assert "## 结论总纲" in compiled["chinese_markdown"]
assert "## Relationship Thesis" in compiled["english_markdown"]
assert "reader_section_brief" not in compiled["english_markdown"]
incomplete = _payload("zh")
del incomplete["sections"]["relationship_weather"]
with pytest.raises(ValueError, match="missing=relationship_weather"):
module.compile_delivery(_dossier(), incomplete, _payload("en"))
with pytest.raises(ValueError, match="not ready_for_narrative"):
module.compile_delivery({**_dossier(), "delivery_readiness": "blocked"}, _payload("zh"), _payload("en"))
def test_compiler_requires_private_output_directory(tmp_path):
assert module._private_output_directory(str(tmp_path)) == tmp_path.resolve()
with pytest.raises(ValueError, match="outside the repository"):
module._private_output_directory(str(ROOT / "outputs"))
def test_cli_writes_only_a_passed_private_delivery(tmp_path, monkeypatch):
dossier_path = tmp_path / "dossier.json"
chinese_path = tmp_path / "chinese.json"
english_path = tmp_path / "english.json"
output_dir = tmp_path / "delivery"
dossier_path.write_text(json.dumps(_dossier()), encoding="utf-8")
chinese_path.write_text(json.dumps(_payload("zh"), ensure_ascii=False), encoding="utf-8")
english_path.write_text(json.dumps(_payload("en")), encoding="utf-8")
monkeypatch.setattr(sys, "argv", [
"compose_full_synastry_delivery.py",
"--dossier", str(dossier_path),
"--chinese-sections", str(chinese_path),
"--english-sections", str(english_path),
"--output-dir", str(output_dir),
])
assert module.main() == 0
assert (output_dir / "Chinese-Full-Synastry-Report.md").is_file()
assert (output_dir / "English-Full-Synastry-Report.md").is_file()
receipt = json.loads((output_dir / "bilingual-delivery-receipt.json").read_text(encoding="utf-8"))
assert receipt["status"] == "passed"
def test_cli_does_not_create_output_directory_when_the_gate_fails(tmp_path, monkeypatch):
dossier_path = tmp_path / "dossier.json"
chinese_path = tmp_path / "chinese.json"
english_path = tmp_path / "english.json"
output_dir = tmp_path / "failed-delivery"
dossier_path.write_text(json.dumps({**_dossier(), "delivery_readiness": "blocked"}), encoding="utf-8")
chinese_path.write_text(json.dumps(_payload("zh"), ensure_ascii=False), encoding="utf-8")
english_path.write_text(json.dumps(_payload("en")), encoding="utf-8")
monkeypatch.setattr(sys, "argv", [
"compose_full_synastry_delivery.py",
"--dossier", str(dossier_path),
"--chinese-sections", str(chinese_path),
"--english-sections", str(english_path),
"--output-dir", str(output_dir),
])
with pytest.raises(ValueError, match="not ready_for_narrative"):
module.main()
assert not output_dir.exists()
def test_cli_dry_run_validates_without_creating_a_delivery(tmp_path, monkeypatch, capsys):
dossier_path = tmp_path / "dossier.json"
chinese_path = tmp_path / "chinese.json"
english_path = tmp_path / "english.json"
output_dir = tmp_path / "dry-run-delivery"
dossier_path.write_text(json.dumps(_dossier()), encoding="utf-8")
chinese_path.write_text(json.dumps(_payload("zh"), ensure_ascii=False), encoding="utf-8")
english_path.write_text(json.dumps(_payload("en")), encoding="utf-8")
monkeypatch.setattr(sys, "argv", [
"compose_full_synastry_delivery.py",
"--dossier", str(dossier_path),
"--chinese-sections", str(chinese_path),
"--english-sections", str(english_path),
"--output-dir", str(output_dir),
"--dry-run",
])
assert module.main() == 0
assert json.loads(capsys.readouterr().out)["status"] == "would-deliver"
assert not output_dir.exists()
def test_publisher_refuses_to_overwrite_an_existing_private_delivery(tmp_path):
output_dir = tmp_path / "existing-delivery"
output_dir.mkdir()
marker = output_dir / "do-not-overwrite.txt"
marker.write_text("existing", encoding="utf-8")
compiled = module.compile_delivery(_dossier(), _payload("zh"), _payload("en"))
with pytest.raises(ValueError, match="already exists"):
module._publish_delivery_bundle(output_dir, compiled)
assert marker.read_text(encoding="utf-8") == "existing"
def test_single_language_compiler_and_cli_publish_without_a_placeholder_peer(tmp_path, monkeypatch):
compiled = module.compile_single_language_delivery(_dossier(), _payload("zh"), "zh")
assert compiled["delivery_receipt"]["status"] == "passed"
assert "## 结论总纲" in compiled["markdown"]
dossier_path = tmp_path / "dossier.json"
sections_path = tmp_path / "zh.json"
output_dir = tmp_path / "zh-delivery"
dossier_path.write_text(json.dumps(_dossier()), encoding="utf-8")
sections_path.write_text(json.dumps(_payload("zh"), ensure_ascii=False), encoding="utf-8")
monkeypatch.setattr(sys, "argv", [
"compose_full_synastry_delivery.py", "--dossier", str(dossier_path),
"--language", "zh", "--sections", str(sections_path), "--output-dir", str(output_dir),
])
assert module.main() == 0
assert (output_dir / "Chinese-Full-Synastry-Report.md").is_file()
assert (output_dir / "zh-delivery-receipt.json").is_file()
assert not (output_dir / "English-Full-Synastry-Report.md").exists()
@@ -0,0 +1,253 @@
from __future__ import annotations
import importlib.util
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
MODULE_PATH = ROOT / "skills" / "jyotish-full-synastry-reading" / "scripts" / "build_full_synastry_dossier.py"
spec = importlib.util.spec_from_file_location("full_synastry_dossier", MODULE_PATH)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
def _packet(*, ayanamsa="lahiri"):
modules = {"chart": {
"ascendant": {"lon": 0.0},
"planets": {
"Moon": {"degree_raw": 0.0}, "Venus": {"degree_raw": 20.0},
"Mars": {"degree_raw": 180.0}, "Saturn": {"degree_raw": 210.0},
"Sun": {"degree_raw": 120.0}, "Rahu": {"degree_raw": 300.0}, "Ketu": {"degree_raw": 120.0},
},
}, "functional_benefic_malefic": {}}
for _, candidates, _ in module._RECEIPT.REQUIRED_TECHNIQUES:
root, *rest = candidates[0].split(".")
modules.setdefault(root, {})
cursor = modules[root]
for key in rest[:-1]:
cursor = cursor.setdefault(key, {})
if rest:
cursor[rest[-1]] = {}
modules.setdefault("varga_full", {}).update({
"D2_Hora": {}, "D4_Turyamsa": {}, "D10_Dasamsa": {}, "D9_Navamsa": {},
})
return {
"birth_info": {"ayanamsa_name": ayanamsa, "node_mode": "mean"},
"calculation_profile": {"zodiac": "sidereal", "position_mode": "legacy", "house_system": "whole_sign"},
"modules": modules,
}
def _d9_packet(*, ayanamsa="lahiri"):
packet = _packet(ayanamsa=ayanamsa)
d9 = {
"Ascendant": {"sign_idx": 0, "degree_in_sign": 5.0},
"Moon": {"sign_idx": 3, "degree_in_sign": 10.0},
"Venus": {"sign_idx": 6, "degree_in_sign": 15.0},
"Mars": {"sign_idx": 9, "degree_in_sign": 20.0},
"Saturn": {"sign_idx": 10, "degree_in_sign": 2.0},
}
packet["modules"]["varga_full"]["D9_Navamsa"] = d9
return packet
def test_dossier_preserves_full_module_inventory_and_uses_neutral_match_product():
dossier = module.build_full_synastry_dossier(_packet(), _packet())
assert dossier["delivery_readiness"] == "ready_for_narrative"
assert dossier["marriage_match"]["product"] == "marriage_match_product_v1"
assert dossier["marriage_match"]["kuja_mangal_dosha"]["gender_asymmetry_used"] is False
assert {row["person"] for row in dossier["module_inventory"]} == {"person_a", "person_b"}
assert all("source_path" in row for row in dossier["source_map"] if row["kind"] == "individual_technique")
narrative_plan = dossier["module_narrative_plan"]
assert {row["person"] for row in narrative_plan} == {"person_a", "person_b"}
d9_rows = [row for row in narrative_plan if row["module_id"] == "varga_full"]
assert all("can_it_stay_steady" in row["reader_sections"] for row in d9_rows)
assert {row["lane"] for row in dossier["relationship_lenses"]} == {
"relationship_state", "emotional_safety", "attraction_and_closeness",
"communication_conflict_and_repair", "relationship_progression",
"shared_life_gains_and_pressures", "stability_and_long_term_repair",
"near_term_relationship_weather",
}
assert all(row["status"] == "ready_for_comparison" for row in dossier["relationship_lenses"])
interactions = dossier["pair_interaction_evidence"]
assert interactions["status"] == "executed"
assert len(interactions["person_a_to_person_b_overlays"]) == 7
assert interactions["angular_contacts"]
assert interactions["jaimini_rashi_contacts"]
assert interactions["recipient_experience_evidence"]["person_a"]
assert interactions["recipient_experience_evidence"]["person_b"]
affection = dossier["mutual_affection_activation"]
assert affection["status"] == "executed"
assert affection["recipients"]["person_a"]["d1_evidence"]
assert len(affection["recipients"]["person_b"]["writer_obligations"]) == 4
section_brief = {row["section"]: row for row in dossier["reader_section_brief"]}
assert set(section_brief) == {
"relationship_thesis", "what_each_person_brings", "emotional_safety",
"attraction_and_closeness", "communication_conflict_and_repair",
"relationship_progression", "shared_life_gains_and_pressures",
"can_it_stay_steady", "relationship_weather",
}
attraction = section_brief["attraction_and_closeness"]
assert attraction["d1_recipient_evidence"]["person_a"]
assert attraction["d1_recipient_evidence"]["person_b"]
assert "d9_recipient_evidence" in attraction
assert attraction["reciprocal_care"]["status"] == "executed"
assert section_brief["relationship_weather"]["event_resonance_gate"]["boundary"]
def test_dossier_blocks_mixed_calculation_profiles_without_dropping_inventory():
dossier = module.build_full_synastry_dossier(_packet(), _packet(ayanamsa="raman"))
assert dossier["delivery_readiness"] == "blocked"
assert dossier["profile_gate"] == "blocked"
assert dossier["module_inventory"]
def test_pair_interaction_evidence_keeps_direction_and_blocks_without_both_ascendants():
a = _packet()
b = _packet()
b["modules"]["chart"]["ascendant"] = {}
interactions = module.build_pair_interaction_evidence(a, b)
assert interactions["status"] == "blocked"
assert not interactions["person_a_to_person_b_overlays"]
assert interactions["angular_contacts"]
assert {row["direction"] for row in interactions["angular_contacts"]} <= {
"person_a_to_person_b", "person_b_to_person_a", "mutual_angle", "undetermined_direction",
}
def test_same_named_special_aspect_uses_directed_angle_not_planet_label():
a = _packet()
b = _packet()
a["modules"]["chart"]["planets"]["Mars"]["degree_raw"] = 0.0
b["modules"]["chart"]["planets"]["Mars"]["degree_raw"] = 270.0
contacts = module.build_pair_interaction_evidence(a, b)["angular_contacts"]
mars_contact = next(
row for row in contacts
if row["person_a_planet"] == row["person_b_planet"] == "Mars"
and row["aspect_type"] == "Mars_aspect_4"
)
assert mars_contact["direction"] == "person_b_to_person_a"
def test_recipient_experience_evidence_keeps_each_persons_received_contacts_separate():
a, b = _packet(), _packet()
a["modules"]["chart"]["planets"]["Mars"]["degree_raw"] = 180.0
b["modules"]["chart"]["ascendant"]["lon"] = 0.0
interactions = module.build_pair_interaction_evidence(a, b)
received_by_b = interactions["recipient_experience_evidence"]["person_b"]
mars_overlay = next(item for item in received_by_b if item["evidence_type"] == "house_overlay" and item["source_planet"] == "Mars")
assert mars_overlay["source_direction"] == "person_a_to_person_b"
assert mars_overlay["recipient_house"] == 7
assert "attraction_and_closeness" in mars_overlay["reader_topics"]
assert "relationship_progression_and_commitment" in mars_overlay["reader_topics"]
def test_pair_interaction_evidence_keeps_native_d9_contacts_separate_from_d1():
a, b = _d9_packet(), _d9_packet()
b["modules"]["varga_full"]["D9_Navamsa"]["Ascendant"] = {"sign_idx": 6, "degree_in_sign": 0.0}
interactions = module.build_pair_interaction_evidence(a, b)
d9 = interactions["d9_interaction_evidence"]
assert interactions["division"] == "D1"
assert d9["status"] == "executed"
assert d9["division"] == "D9"
assert d9["source_paths"] == ["modules.varga_full.D9_Navamsa"]
moon_overlay = next(row for row in d9["person_a_to_person_b_overlays"] if row["planet"] == "Moon")
assert moon_overlay["recipient_house"] == 10
assert "Do not average" in interactions["division_boundary"]
def test_parallel_timing_context_keeps_native_periods_separate_without_rebuilding_dasha():
a, b = _packet(), _packet()
a["modules"]["dasha"] = {
"reference_date": "2026-10-17",
"current_dasha": {
"lord": "Jupiter", "start_datetime": "2024-01-01T00:00:00", "end_datetime": "2040-01-01T00:00:00",
"antardasha": {
"lord": "Venus", "start_datetime": "2026-01-01T00:00:00", "end_datetime": "2028-01-01T00:00:00",
"pratyantar_dasha_timeline": [
{"lord": "Moon", "start_datetime": "2026-10-01T00:00:00", "end_datetime": "2026-12-01T00:00:00", "is_current": True},
],
},
},
}
b["modules"]["dasha"] = {
"reference_date": "2026-10-17",
"current_dasha": {
"lord": "Saturn", "start_datetime": "2020-06-01T00:00:00", "end_datetime": "2039-06-01T00:00:00",
"antardasha": {
"lord": "Mercury", "start_datetime": "2026-06-01T00:00:00", "end_datetime": "2027-06-01T00:00:00",
"pratyantar_dasha_timeline": [
{"lord": "Mars", "start_datetime": "2026-10-15T00:00:00", "end_datetime": "2026-11-15T00:00:00", "is_current": True},
],
},
},
}
timing = module.build_full_synastry_dossier(a, b)["parallel_timing_context"]
assert timing["status"] == "executed"
assert timing["reference_alignment"] == "matched"
assert {(item["level"], item["label"]) for item in timing["person_a_active_periods"]} == {
("mahadasha", "Jupiter"), ("antardasha", "Venus"), ("pratyantar", "Moon"),
}
assert {(item["level"], item["label"]) for item in timing["person_b_active_periods"]} == {
("mahadasha", "Saturn"), ("antardasha", "Mercury"), ("pratyantar", "Mars"),
}
assert timing["event_resonance_requirement"]["status"] == "not_established_by_parallel_periods"
assert "Never calculate" in timing["boundary"]
def test_event_resonance_gate_rejects_generic_window_scores_and_requires_named_event():
a, b = _packet(), _packet()
strict = {
"event_judgement": {"event_class": "relationship_formation", "verdict": "high_probability_window", "score": 89},
"prediction_boundary_contract": {
"promise": {"status": "present"}, "activation": {"status": "present"},
"manifestation": {"status": "present"}, "label": {"status": "present"},
},
}
a["modules"]["relationship_strict_evidence"] = strict
b["modules"]["relationship_strict_evidence"] = strict
gate = module.build_full_synastry_dossier(a, b)["event_resonance_gate"]
formation = next(row for row in gate["event_classes"] if row["event_class"] == "relationship_formation")
assert formation["status"] == "candidate_alignment_requires_event_history"
assert gate["person_a"]["ignored_non_event_fields"]["score"] == 89
assert "never asserts" in gate["boundary"]
def test_event_resonance_gate_never_promotes_a_generic_verdict_without_event_label():
a, b = _packet(), _packet()
strict = {
"event_judgement": {"verdict": "high_probability_window", "score": 83},
"prediction_boundary_contract": {
"promise": {"status": "present"}, "activation": {"status": "weak"},
"manifestation": {"status": "weak"}, "label": {"status": "missing"},
},
}
a["modules"]["relationship_strict_evidence"] = strict
b["modules"]["relationship_strict_evidence"] = strict
gate = module.build_full_synastry_dossier(a, b)["event_resonance_gate"]
assert gate["person_a"]["event_class"] is None
assert gate["person_a"]["status"] == "event_class_not_established"
assert all(row["status"] == "not_established" for row in gate["event_classes"])
def test_event_resonance_gate_keeps_native_event_candidate_observation_only():
a, b = _packet(), _packet()
for packet in (a, b):
packet["modules"]["chart"]["ascendant"]["sign"] = "Leo"
packet["modules"]["dasha"] = {
"current_dasha": {"mahadasha": "Saturn", "antardasha": {"lord": "Mercury"}},
}
packet["modules"]["relationship_strict_evidence"] = {
"event_judgement": {"verdict": "high_probability_window", "score": 83},
"prediction_boundary_contract": {
"promise": {"status": "present"}, "activation": {"status": "weak"},
"manifestation": {"status": "weak"}, "label": {"status": "missing"},
},
}
gate = module.build_full_synastry_dossier(a, b)["event_resonance_gate"]
formation = next(row for row in gate["event_classes"] if row["event_class"] == "relationship_formation")
assert gate["person_a"]["event_class"] == "relationship_formation"
assert gate["person_a"]["event_class_source"] == "native_observation_only"
assert gate["person_a"]["status"] == "event_class_observation_only"
assert formation["status"] == "parallel_observation_only"
@@ -0,0 +1,142 @@
from __future__ import annotations
import importlib.util
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
MODULE_PATH = ROOT / "skills" / "jyotish-full-synastry-reading" / "scripts" / "validate_full_synastry_packets.py"
spec = importlib.util.spec_from_file_location("full_synastry_execution", MODULE_PATH)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
def _packet(*, ayanamsa="lahiri", blocked=None, parameter_sensitive=None):
modules = {"chart": {"houses": {}}, "functional_benefic_malefic": {}}
for _, candidates, _ in module.REQUIRED_TECHNIQUES:
root, *rest = candidates[0].split(".")
root_status = (
"blocked" if candidates[0] == blocked
else "observation_only" if candidates[0] == parameter_sensitive
else None
)
modules.setdefault(root, {"status": root_status} if root_status else {})
cursor = modules[root]
for key in rest[:-1]:
cursor = cursor.setdefault(key, {})
if rest:
status = (
"blocked" if candidates[0] == blocked
else "observation_only" if candidates[0] == parameter_sensitive
else None
)
cursor[rest[-1]] = {"status": status} if status else {}
return {
"birth_info": {"ayanamsa_name": ayanamsa, "node_mode": "mean"},
"calculation_profile": {
"zodiac": "sidereal",
"position_mode": "legacy",
"house_system": "whole_sign",
},
"modules": modules,
}
def test_full_synastry_receipt_requires_every_registered_technique_for_both_people():
receipt = module.build_execution_receipt(_packet(), _packet())
assert receipt["profile_gate"] == "passed"
assert len(receipt["individual_technique_audit"]) == len(module.REQUIRED_TECHNIQUES)
by_technique = {row["technique"]: row["status"] for row in receipt["individual_technique_audit"]}
assert by_technique["Niryaana Shoola Dasha"] == "not_applicable"
assert {status for technique, status in by_technique.items() if technique != "Niryaana Shoola Dasha"} == {"executed"}
statuses = {task["task"]: task["status"] for task in receipt["pair_execution_tasks"]}
assert statuses["D1/D9 mutual overlays and relationship axes"] == "ready_for_comparison"
assert statuses["DK, UL, UL-second and A7/Darapada cross-comparison"] == "ready_for_comparison"
assert statuses["two-person Dasha/Narayana/Chara context comparison"] == "ready_for_comparison"
assert statuses["shared transit, Tajika, Saham and annual-window comparison"] == "ready_for_comparison"
assert statuses["Nakshatra relationship context"] == "ready_for_comparison"
assert statuses["optional 27-nakshatra Shukuyo narrative layer"] == "not_applicable"
assert statuses["Prashna context when a real question chart was supplied"] == "not_applicable"
def test_shared_life_pair_task_uses_the_native_d4_turyamsa_key():
a, b = _packet(), _packet()
for packet in (a, b):
packet["modules"].setdefault("varga_full", {}).update({
"D2_Hora": {}, "D4_Turyamsa": {}, "D10_Dasamsa": {},
})
packet["modules"]["bhava_bala"] = {}
packet["modules"]["ashtakavarga"] = {}
receipt = module.build_execution_receipt(a, b)
shared_life = next(
row for row in receipt["pair_execution_tasks"]
if row["task"].startswith("shared gain and cost")
)
assert shared_life["status"] == "ready_for_comparison"
assert "modules.varga_full.D4_Turyamsa" in {
row["source_path"] for row in shared_life["required_inputs"]
}
def test_full_synastry_receipt_preserves_blocked_module_and_profile_mismatch():
receipt = module.build_execution_receipt(
_packet(ayanamsa="lahiri"),
_packet(ayanamsa="raman", blocked="dasha"),
)
dasha = next(row for row in receipt["individual_technique_audit"] if row["technique"] == "Vimshottari")
assert receipt["profile_gate"] == "blocked"
assert dasha["status"] == "blocked"
assert dasha["person_b"]["status"] == "blocked"
def test_full_synastry_receipt_keeps_observation_only_technique_out_of_completed_rows():
receipt = module.build_execution_receipt(
_packet(parameter_sensitive="punarphoo"),
_packet(parameter_sensitive="punarphoo"),
)
punarphoo = next(row for row in receipt["individual_technique_audit"] if row["technique"] == "Punarphoo")
assert punarphoo["status"] == "parameter_sensitive"
assert punarphoo["person_a"]["status"] == "parameter_sensitive"
def test_synastry_surfaces_lead_with_evidence_not_a_match_score():
main_js = (ROOT / "jyotish-app" / "main.js").read_text(encoding="utf-8")
export_js = (ROOT / "jyotish-app" / "export.js").read_text(encoding="utf-8")
assert "<h3>恋爱 / 婚姻合盘</h3>" in main_js
assert "展开传统 Ashtakoota 技术参考" in main_js
assert "关系结构、互动模式与双人应期优先;传统匹配数值仅作技术参考。" in export_js
assert "传统 Ashtakoota 技术参考" in export_js
assert "computeMarriageMatchProduct" in main_js
assert "不能回退到旧式顺序型合盘" in main_js
assert "computeSynastry?.({ male_moon" not in main_js
def test_full_synastry_skill_uses_jyotish_judgement_order_and_full_receipt_appendix():
shape = (
ROOT / "skills" / "jyotish-full-synastry-reading" / "references" / "customer-report-shape.md"
).read_text(encoding="utf-8")
skill = (ROOT / "skills" / "jyotish-full-synastry-reading" / "SKILL.md").read_text(encoding="utf-8")
for heading in (
"命局承诺",
"配偶与婚姻承载",
"两盘交感",
"能否久处",
"关系应期:四类事件,四套门槛",
"私有证据包",
"情绪安全与相处感受",
"共同生活的收益与压力",
):
assert heading in shape
assert "individual promise ->\nspouse/marriage carrying" in skill
assert "every row" in skill
def test_synastry_skill_default_prompt_routes_to_the_full_non_score_product():
agent = (ROOT / "skills" / "jyotish-full-synastry-reading" / "agents" / "openai.yaml").read_text(encoding="utf-8")
assert "local birth date" in agent
assert "private full-synastry dossier" in agent
assert "independently composed Chinese and English reports" in agent
assert "coincident Dasha dates" in agent
@@ -0,0 +1,146 @@
from __future__ import annotations
import os
import sys
import pytest
SCRIPTS = os.path.join(os.path.dirname(__file__), "..", "scripts")
if SCRIPTS not in sys.path:
sys.path.insert(0, SCRIPTS)
from marriage_match import build_marriage_match_dossier
def chart(asc, moon, venus, mars, *, saturn=210, sun=120, rahu=300, ketu=120):
return {
"ascendant": {"longitude": asc},
"planets": {
"Moon": {"longitude": moon}, "Venus": {"longitude": venus},
"Mars": {"longitude": mars}, "Saturn": {"longitude": saturn},
"Sun": {"longitude": sun}, "Rahu": {"longitude": rahu}, "Ketu": {"longitude": ketu},
},
}
def test_default_profile_separates_mars_kuja_from_papasamya():
a = chart(0, 30, 60, 180, saturn=180)
b = chart(90, 120, 150, 270, saturn=270)
report = build_marriage_match_dossier(a, b)
assert report["rule_profile"]["id"] == "practitioner_consensus_v1"
assert report["kuja_mangal_dosha"]["person_a"]["scope"] == "Mars only"
pressure = report["papasamya"]["person_a"]
assert {row["planet"] for row in pressure["planet_rows"]} == {"Mars", "Saturn", "Sun", "Rahu", "Ketu"}
assert "score" not in report["kuja_mangal_dosha"]
def test_kuja_comparison_is_gender_neutral_and_not_numeric_tolerance():
a = chart(0, 30, 60, 180)
b = chart(90, 120, 150, 270)
forward = build_marriage_match_dossier(a, b)["kuja_mangal_dosha"]
reverse = build_marriage_match_dossier(b, a)["kuja_mangal_dosha"]
assert forward["gender_asymmetry_used"] is False
assert forward["outcome"] == reverse["outcome"]
assert forward["comparison_method"] == "neutral_reference_count"
def test_raman_variant_is_visible_and_first_house_difference_is_explicit():
# Mars in the first house from each reference frame: default sees it;
# Raman's selected lineage does not.
same_sign = chart(0, 0, 0, 0)
default = build_marriage_match_dossier(same_sign, same_sign)
raman = build_marriage_match_dossier(same_sign, same_sign, profile_id="raman_muhurtha_v1")
assert default["kuja_mangal_dosha"]["person_a"]["active_reference_count"] == 3
assert raman["kuja_mangal_dosha"]["person_a"]["active_reference_count"] == 0
assert raman["rule_profile"]["id"] == "raman_muhurtha_v1"
def test_bhakoot_and_nadi_exceptions_are_exposed_without_rewriting_raw_score():
# Moon longitudes both in Ashwini but different padas: same Nadi raw zero,
# candidate cancellation stays a separately auditable flag.
a = chart(0, 1, 20, 180)
b = chart(90, 4, 120, 270)
report = build_marriage_match_dossier(a, b)
nadi = report["ashtakoota_36"]["exception_review"]["nadi"]
assert nadi["raw_score"] == 0
assert "same_nakshatra_different_pada" in nadi["candidate_cancellations"]
def test_international_core_is_order_independent_and_excludes_caste_point():
a = chart(0, 30, 60, 180)
b = chart(90, 120, 150, 270)
forward = build_marriage_match_dossier(a, b)["ashtakoota_36"]["international_core"]
reverse = build_marriage_match_dossier(b, a)["ashtakoota_36"]["international_core"]
assert forward["total_score"] == reverse["total_score"]
assert forward["scores"] == reverse["scores"]
assert forward["max_score"] == 35.0
assert "ValuesOrientation" not in forward["scores"]
assert "Varna" not in forward["scores"]
dialogue = build_marriage_match_dossier(a, b)["ashtakoota_36"]["shared_life_dialogue"]
assert dialogue["status"] == "not_scored"
def test_reported_directional_reproduction_is_order_independent_without_varna():
# The historical directional calculation yields 10 vs 9 when 45 and 200
# degrees are swapped because it includes Varna. The product core omits
# Varna and averages every included Kuta, so input order is immaterial.
a = chart(0, 45, 80, 120)
b = chart(90, 200, 170, 210)
forward = build_marriage_match_dossier(a, b)["ashtakoota_36"]["international_core"]
reverse = build_marriage_match_dossier(b, a)["ashtakoota_36"]["international_core"]
assert forward["total_score"] == reverse["total_score"] == 9.0
assert forward["scores"] == reverse["scores"]
assert forward["directionality"]["status"] == "symmetric"
def test_zero_degree_moon_is_a_valid_ashtakoota_input():
result = build_marriage_match_dossier(chart(0, 0, 20, 180), chart(90, 120, 150, 270))
assert result["ashtakoota_36"]["status"] == "executed"
assert result["ashtakoota_36"]["international_core"]["person_a_details"]["moon_sign"] == "Aries"
def test_aries_named_ascendant_is_a_valid_lagna_reference():
# Aries is sign index 0 and must not be discarded by a boolean fallback.
person_a = {"ascendant": "Aries", "planets": {"Mars": {"longitude": 120.0}}}
person_b = {"ascendant": "Taurus", "planets": {"Mars": {"longitude": 240.0}}}
kuja = build_marriage_match_dossier(person_a, person_b)["kuja_mangal_dosha"]
lagna_frame = kuja["person_a"]["frames"][0]
assert lagna_frame["reference"] == "lagna"
assert lagna_frame["reference_sign"] == "Aries"
assert lagna_frame["mars_house"] == 5
assert lagna_frame["status"] == "not_triggered"
def test_moon_only_quick_reference_stays_neutral_and_marks_missing_pressure_layers():
result = build_marriage_match_dossier(
{"planets": {"Moon": {"longitude": 0.0}}},
{"planets": {"Moon": {"longitude": 120.0}}},
)
assert result["ashtakoota_36"]["status"] == "executed"
assert result["ashtakoota_36"]["international_core"]["total_score"] == 18.5
assert result["kuja_mangal_dosha"]["person_a"]["status"] == "not_available"
assert result["kuja_mangal_dosha"]["person_b"]["status"] == "not_available"
def test_shared_life_dialogue_is_not_a_score_or_questionnaire_store():
dialogue = build_marriage_match_dossier(chart(0, 30, 60, 180), chart(90, 120, 150, 270))["ashtakoota_36"]["shared_life_dialogue"]
assert dialogue["status"] == "not_scored"
assert dialogue["storage"] == "no_answers_collected"
assert dialogue["prompts"]
def test_non_romantic_scope_is_rejected():
with pytest.raises(ValueError, match="romantic/marriage"):
build_marriage_match_dossier(chart(0, 1, 20, 180), chart(90, 4, 120, 270), relationship_type="business")
def test_api_handler_uses_the_new_product_contract_without_legacy_kuja_score():
from jyotish_api_server import JyotishAPIHandler
payload = {
"person_a": chart(0, 1, 20, 180),
"person_b": chart(90, 4, 120, 270),
}
result = JyotishAPIHandler._compute_marriage_match(None, payload)
assert result["product"] == "marriage_match_product_v1"
assert result["kuja_mangal_dosha"]["gender_asymmetry_used"] is False
@@ -0,0 +1,74 @@
"""Regression tests for the distinct Niryaana Shoola rule family."""
from __future__ import annotations
from datetime import datetime
from scripts.niryaana_shoola_dasha import (
PROFILE_ID,
calculate_niryaana_shoola_dasha,
niryaana_shoola_applicability,
)
def test_niryaana_shoola_selects_stronger_second_or_eighth() -> None:
second = niryaana_shoola_applicability(
ascendant_longitude=0.0, second_strength=2.0, eighth_strength=1.0,
)
eighth = niryaana_shoola_applicability(
ascendant_longitude=0.0, second_strength=1.0, eighth_strength=2.0,
)
assert second["starting_sign"] == "Taurus"
assert second["direction"] == "reverse"
assert eighth["starting_sign"] == "Scorpio"
assert eighth["direction"] == "reverse"
def test_niryaana_shoola_blocks_equal_second_eighth_without_invented_tie_break() -> None:
result = niryaana_shoola_applicability(
ascendant_longitude=0.0, second_strength=1.0, eighth_strength=1.0,
)
assert result["execution_status"] == "blocked"
assert "does not provide a tie-break" in result["reason"]
def test_niryaana_shoola_uses_modality_years_and_parity_direction() -> None:
result = calculate_niryaana_shoola_dasha({
"profile_id": PROFILE_ID,
"birth_datetime": datetime(2000, 1, 1, 12, 0),
"ascendant_longitude": 15.0,
"second_strength": 2.0,
"eighth_strength": 1.0,
})
assert result["execution_status"] == "executed"
assert result["total_cycle"] == 96
assert [row["sign"] for row in result["major"][:3]] == ["Taurus", "Aries", "Pisces"]
assert [row["years"] for row in result["major"][:3]] == [8, 7, 9]
assert result["period_depth_status"] == "mahadasha_executed_ad_pd_blocked_source_rule"
assert result["antardasha"] == []
assert result["relationship_applicability"] == "not_applicable_longevity_specific"
def test_legacy_goel_profile_is_blocked_as_misclassified() -> None:
result = calculate_niryaana_shoola_dasha({
"profile_id": "goel_nirayana_shoola_v1",
"birth_datetime": datetime(2000, 1, 1, 12, 0),
"ascendant_longitude": 0.0,
"lagna_strength": 2.0,
"seventh_strength": 1.0,
})
assert result["execution_status"] == "blocked"
assert "conflates Niryaana Shoola with Shoola Dasha" in result["reason"]
def test_pyjhora_observation_profile_replays_same_native_rule_shape() -> None:
result = calculate_niryaana_shoola_dasha({
"profile_id": "pyjhora_2nd_8th_observation_v1",
"birth_datetime": "2000-01-01T00:00:00",
"ascendant_longitude": 15.0,
"second_strength": 2.0,
"eighth_strength": 1.0,
})
assert result["execution_status"] == "executed"
assert result["starting_sign"] == "Taurus"
assert result["profile_arbitration"]["selection_status"] == "explicit_profile_selected"
+1
View File
@@ -505,5 +505,6 @@
| `TASK-marriage-match-p1-fix-20261018.md` | #79 | 只存两个 profile 的配对结果;旧历史列表不读新行;并发重复写入改为重读 | 待验收 | `codex/marriage-match-p1-20261017` |
| `TASK-marriage-match-p1-fix-20261018-2.md` | #79 | 胶水挪进子包,根目录文件数回到 446;调用清单改指向 marriage_match.py | 待验收 | `codex/marriage-match-p1-20261017` |
| `TASK-full-reading-module-parity-20261018.md` | #81 | full-reading 补齐上游 5 个模块(西占层文件没搬致永远 blocked、Navamsha / Niryaana 用我们自己的实现填键、Lagna Vargottamamsa / Pushkara / Punarphoo)(BUG-1375 起) | 待执行(可与 #78 并行) | `codex/full-reading-module-parity-20261018` |
| `TASK-niryaana-shoola-pvr-20261018.md` | #84 | Niryaana Shoola 改 PVR 口径(我们的表与 Shoola 逐行相同;推翻 BUG-1283 所选 profile;占星师 7a773d13) | 排队(等 #81) | `codex/niryaana-shoola-pvr-20261018` |
| `TASK-synastry-full-report-20261018.md` | #82 | 婚恋合盘:选人页 + 完整合盘报告(关系画像九节:结论+折叠的分析逻辑+关系时间线+技术依据摘要;英文按需;计费键 synastry.full) | 排队(等 #78、#79、#81) | `codex/synastry-full-report-20261018`;动表 |
| `TASK-synastry-report-chat-20261018.md` | #83 | 从合盘报告某一节进入对话问依据(绑定报告与节、对话里可用术语、只依据已存计算结果并逐条核对、计费照普通对话) | 排队(等 #82) | `codex/synastry-report-chat-20261018`;动表 |
@@ -34,6 +34,7 @@ Claude 10-18 用两位公开名人(布拉德·皮特、安吉丽娜·朱莉,
| --- | --- |
| F1 | 比上游少的模块都补上,让我们的 full-reading 是上游的超集 |
| F2 | `navamsha_dasha`、`niryaana_shoola_dasha` 两个键**用我们自己的 `rashi_dashas_native` 结果填**,不搬上游那两份文件(BUG-1283 的红线不变)。键名与上游一致,形状按上游合盘校验器能识别的方式包一层(见 T3) |
| F2b | **10-18 更正(占星师 `7a773d13`)**:我们 `rashi_dashas_native` 里的 Niryaana Shoola(`goel_nirayana_shoola_v1`)其实是 Shoola 的规则,正确口径另立单 `TASK-niryaana-shoola-pvr-20261018.md`。所以本单 **`niryaana_shoola_dasha` 键不再用它填**,只写占位 `{"execution_status": "blocked", "reason": "口径更正中,见 TASK-niryaana-shoola-pvr-20261018"}`;合盘校验器对这一行本来就固定判「不适用」,不影响合盘。`navamsha_dasha` 照原计划 |
| F3 | 西占层(`unified_western_reading`)按上游搬入。它在报告里的用途与上游一致:只作补充,合盘报告不得出现方法名(合盘任务书红线 3) |
| F4 | 个人报告里如果因为西占层从 `blocked` 变成有结果而多出内容,这是预期的修复,不是回归;改动的断言写三栏说明 |
@@ -0,0 +1,60 @@
# TASK · full-reading 补齐上游模块 · 修改单(issue #81)
- 日期:2026-10-18
- 被看提交:`900581068e06df14ab1f3de6bf5e7bbfe2ad1dee`(分支 `codex/full-reading-module-parity-20261018`)
- 原任务书:`docs/tasks/TASK-full-reading-module-parity-20261018.md`
- 执行方:Grok,同一分支追加提交;`origin/staging` 前进时用 `git merge origin/staging`,不 rebase
- 性质:**不是打回**。验收中发现一条与既有合同冲突的规则,产品已拍板;本单写清怎么按决定改,连同 issue 里的 F2b 补充一起做
- BUG 编号:沿用 BUG-1375、BUG-1376,不新开号
## 决策记录(产品负责人 2026-10-18,issue #81「我选 A」)
`tests/test_python_hygiene_budget.py` 的 `ROOT_PY_BUDGET = 446` 规定 `scripts/` 根目录 `.py` 文件数只许减不许增(`TASK-python-hygiene-20261013`)。本分支根目录 449 个。
产品选 **A**:**上游原样文件**可以放在 `scripts/` 根目录(与上游同路径,方便以后同步),预算为这几个文件相应调高;**我们自己写的文件仍不许增加**。
## 已核对(Claude)
- `scripts/classical_astrology.py`、`scripts/unified_western_reading.py` 与 `docs/research/upstream_full_reading_parity_bde5d615/` 副本逐字一致(只差 CRLF / BOM)。
- `scripts/western_chart_engine.py`、`scripts/western_real_case_calibration.py` 只有新增函数,没有整份覆盖。
## 要改的
### M1 · 删掉 `scripts/_script_root.py`
这是本地新建的文件,不是上游原样文件(上游副本里没有它),按决定不许占根目录名额。两个上游模块对它的引用都包在 `try / except ModuleNotFoundError: pass` 里,是可选的。
- 删掉该文件(它还带了 UTF-8 BOM)。
- 若删掉后两个上游模块有 import 失败,**不要**改上游文件;改为在调用方(`jyotish_engine.py` full-reading 组装处或测试)保证仓库根在 `sys.path` 上,或把辅助放进子目录,在进度里写明。
- 验收:`tests/test_full_reading_module_parity.py` 全绿,且其中「模块能导入」那条在没有 `_script_root.py` 时仍成立。
### M2 · 预算按决定调高,并写明例外
`tests/test_python_hygiene_budget.py`:
- `ROOT_PY_BUDGET` 由 446 改为 **446 + 本单新增的上游原样文件数**(预期 448),常量旁写三栏:
- 原值:446
- 新值:448
- 原因:产品 2026-10-18(issue #81 选 A):上游原样文件按上游路径放在 `scripts/` 根目录,`classical_astrology.py`、`unified_western_reading.py` 两个;本地新写的文件仍不得增加。
- 合并时若 `origin/staging` 的根目录计数已因其他单变化(例如 #79 合入),按合并后的实际数重算,只为本单的上游原样文件加名额。
### M3 · F2b:`niryaana_shoola_dasha` 改为占位
照 issue #81 2026-10-18 20:40 的补充:本单的 `niryaana_shoola_dasha` 键不再用 `rashi_dashas_native` 的 Niryaana Shoola 结果填,只写
```json
{"execution_status": "blocked", "reason": "口径更正中,见 TASK-niryaana-shoola-pvr-20261018"}
```
`navamsha_dasha` 照原计划。相应测试与进度同步改;原任务书 T3 对 Niryaana 那一行的验收改为「校验器判为不适用,合盘任务不因它 blocked」。
### M4 · 交回前必跑
- 全量 `pytest tests` 与开工时 `origin/staging` 逐名比对(预算这条正是全量才抓得到的)。
- `python3 -m pytest tests/test_python_hygiene_budget.py tests/test_rashi_dashas_native.py tests/test_full_reading_module_parity.py -q`。
- 屏蔽 PyJHora 后再跑新增测试(原任务书已要求)。
## 不要做
- 不改两个上游原样文件;不整份覆盖 `western_*`。
- 不把预算调到超过上游原样文件所需的数。
@@ -0,0 +1,63 @@
# 任务书:Niryaana Shoola 改为 PVR 口径(2026-10-18)
## 基线
- `origin/staging` 以开工时最新为准(`git fetch origin --prune`)。
- 分支:`codex/niryaana-shoola-pvr-20261018`;工作树:`.worktrees/niryaana-shoola-pvr-20261018`。
- 顺序:与 #81(full-reading 补齐模块)都碰 `niryaana_shoola_dasha` 这个键。#81 先合入;本单在它之后开分支,把 #81 留下的「暂不填」占位换成真结果。
- 上游参照:占星师 `yinduzhanxing` main `7a773d13`(10-18)「correct niryaana shoola」。副本:`docs/research/upstream_full_synastry_7a773d13/niryaana_shoola_dasha.py.txt` 及同目录 `tests/test_niryaana_shoola_dasha.py.txt`。
## 事故实证
| # | 事实 | 位置 |
| --- | --- | --- |
| 1 | 我们完整数据版报告里的「Niryaana Shoola」与「Shoola」**是同一套算法**:两者都调 `_nine_year`,从上升与第七宫中取强、每宫 9 年、顺行,只是 profile 名不同(`shoola_forward_md_direct_ad_v1` / `goel_nirayana_shoola_v1`)。报告里这两张大运表逐行相同 | `scripts/rashi_dashas_native.py` 的 `calculate_rashi_dashas`、`_nine_year`;报告节表 `scripts/rashi_dasha_report.py` |
| 2 | 占星师 10-18 裁定:「上升 / 七宫取强、固定 9 年」属于 Shoola Dasha;Niryaana Shoola 应按 PVR:从**第二宫与第八宫**取强起运,奇数星座顺行、偶数星座逆行,动 / 固 / 双星座分别 7 / 8 / 9 年。上游已把 `goel_nirayana_shoola_v1` 标为误归类并禁止运行,新 profile 为 `pvr_niryaana_shoola_v1` | 上游 `scripts/niryaana_shoola_dasha.py`(`PROFILE_ID`、`LEGACY_MISCLASSIFIED_PROFILE_ID`) |
| 3 | 上游同时把 Niryaana Shoola 从恋爱婚姻合盘的应期里排除(寿元类大运,不用于关系时间) | 上游 `validate_full_synastry_packets.py`:该行固定 `not_applicable` |
| 4 | 只在 staging:`origin/main` 还没有 `rashi_dashas_native.py`,生产不受影响 | — |
另:上游 `cba6a031`(10-18 晚)把 full-reading 接上了,二宫 / 八宫力量取 `bhava_bala.tri_component` 的 `total`。Claude 用两位公开名人复测:两张盘都能算出来,奇顺偶逆与 7 / 8 / 9 年都对;**但用我们的 Jaimini 星座力量(`stronger_sign`)比,两张盘选出的起运星座都和上游相反**(一张上游选双子、我们选射手;另一张上游选狮子、我们选水瓶)。上游所用的那套宫位力量在它自己的代码里标着「替代口径、待验证」。用哪种力量来比二宫 / 八宫,已请占星师确认(见 H4)。
## 根因
BUG-1283(10-09)把旧 `niryaana_shoola_dasha.py` 换成 `goel_nirayana_shoola_v1` 时,选的规则其实是 Shoola 的,而且没有和同文件里的 Shoola 对比输出,两张表一样没被发现。
## 决策记录(产品负责人 2026-10-18)
| # | 决定 |
| --- | --- |
| H1 | Niryaana Shoola 采用占星师 10-18 的 PVR 口径(`pvr_niryaana_shoola_v1`)。**推翻 BUG-1283 选定的 `goel_nirayana_shoola_v1`**;BUG-1283 的另一半(旧 `niryaana_shoola_dasha.py` 不得被新路径调用)保持不变 |
| H2 | Shoola 保持现状(上升 / 七宫取强、每宫 9 年)——那本来就是 Shoola 的规则 |
| H3 | 合盘不用 Niryaana Shoola 做关系应期(与上游一致);本单只影响个人完整数据版报告和 full-reading 的 `niryaana_shoola_dasha` 键 |
| H4 | **力量比较先用 Jaimini 星座力量**(本文件已有的 `rashi_strengths` / `stronger_sign`,与 Chara、Narayana 等星座大运同一套),不用上游的 Bhava Bala。理由:Niryaana Shoola 是 Jaimini 星座大运,上游所用 Bhava Bala 替代口径在其代码中标为待验证。**占星师若确认应该用 Bhava Bala,再改**;开工前 Claude 会在评论里同步她的答复 |
| H5 | **只出主运,不出子运**(与上游一致:上游说明 PVR 没有独立给出子运规则,子运标为 blocked)。报告里该表只列主运并写一句「子运规则未定,暂不列出」 |
## 硬红线
1. 在 `scripts/rashi_dashas_native.py` 内改,不 import 旧的 `scripts/niryaana_shoola_dasha.py`(BUG-1283 语法树断言保持绿);不整份搬上游文件(上游版本依赖我们没有的 `dasha_profile_arbitration` 等)。
2. 力量比较用本文件已有的 `rashi_strengths` / `stronger_sign`,平手时按现有 tie-break 规则走,仍分不出就「本轮不上表」,不得默认挑一个。
3. 全量 pytest 与基线逐条比对;改既有断言写三栏。
## 任务分解
| # | 内容 | 验收标准 |
| --- | --- | --- |
| T1 | `niryaana_shoola` 改为 PVR:第二宫与第八宫取强起运;起运星座奇数顺行、偶数逆行;动 / 固 / 双星座 7 / 8 / 9 年(12 宫一轮 96 年);**只出主运**(H5);profile 名 `pvr_niryaana_shoola_v1`,出处写 PVR《Vedic Astrology: An Integrated Approach》 | 新测试:① 同一张盘 Shoola 与 Niryaana Shoola 的首运或年数不再全同;② 白羊上升样例起运在金牛 / 天蝎中较强者;③ 奇偶方向各一例;④ 动 / 固 / 双年数各一例;⑤ 在上游测试样例上(样例直接给定二宫 / 八宫力量)与上游 `pvr_niryaana_shoola_v1` 的主运序列一致(把期望值抄过来作 golden,写明出处);⑥ 两位公开名人各一条:记录我们选出的起运星座与上游 `cba6a031` 的差异,写进 PROGRESS,不作为失败 |
| T2 | full-reading 的 `niryaana_shoola_dasha` 键改用 T1 结果(替换 #81 的占位) | 上游合盘校验器读这一行仍为 `not_applicable`(合盘不用);个人报告节表显示新表 |
| T3 | 记录:BUG_HISTORY(新编号,关联 BUG-1283,写明「推翻其所选 profile」)、CHANGELOG(报告里 Niryaana Shoola 大运表变了)、PROGRESS | 文档齐 |
## 开工前置命令
```bash
git fetch origin --prune
git worktree add -b codex/niryaana-shoola-pvr-20261018 .worktrees/niryaana-shoola-pvr-20261018 origin/staging
cd .worktrees/niryaana-shoola-pvr-20261018
grep -n "goel_nirayana_shoola_v1\|_nine_year" scripts/rashi_dashas_native.py
grep -n "BUG-1[0-9][0-9][0-9]" docs/BUG_HISTORY.md | tail -3
```
验证:全量 `pytest tests -n 4` 与开工基线逐条比对;`run_quality_gate.py --profile quick`;屏蔽 PyJHora 后跑 `tests/test_rashi_dashas_native.py`。
## BUG 编号
开工时最大号 + 1(10-18 staging 最大为 BUG-1391,#80 在用 1380 段)。
@@ -4,7 +4,7 @@
- `origin/staging` 以开工时最新为准(`git fetch origin --prune`)。
- 分支:`codex/synastry-full-report-20261018`;工作树:`.worktrees/synastry-full-report-20261018`。
- **前置**:#78(档案上限 20 人)、#79(引擎、证据包接口、存储;第六版起不做页面)、#81(full-reading 补齐 5 个模块)三单都合入后开工。三单彼此不冲突,可并行。
- **前置**:#78(档案上限 20 人)、#79(引擎、证据包接口、存储;第六版起不做页面)、#81(full-reading 补齐 5 个模块)三单都合入后开工。上游参照以 `7a773d13` 为准(见实证 8b)。三单彼此不冲突,可并行。
- **上线**:#79 与本单要一起提升到 `main`(#79 单独上线时旧入口还在、新入口不存在)。本单从 #79 合入后的 `origin/staging` 开分支,复用 #79 的引擎、证据包接口、选人页和 `synastry_reports` 新列。
- **正文写作首要准则**:`docs/research/synastry_relationship_language_guide_2026_10_18.md`(产品负责人 10-18 提供,「从星曜语言到关系语言」)。凡与上游写作规范冲突处,以它为准。
- 原型(v13,关系画像 + 折叠的分析逻辑 + 时间线 + 技术依据摘要):https://claude.ai/artifact/Tq6vhXFMnGitDZhJdamaHe 。原型里的人名和正文全是虚构示例。
@@ -22,7 +22,7 @@
| 6 | 阅读页轮询 `/api/reports/:id`,有等待面板与阶段文案;语言切换写 `?lang=en` | `components/personal-report/personal-report-page.tsx`、`personal-report-progress.ts`、`report-language.ts` |
| 7 | 上游证据包在我们引擎上能直接跑:Claude 10-18 用两位公开名人(布拉德·皮特、安吉丽娜·朱莉,出生资料来自公开星盘库)在 `origin/staging` 上跑 `full-reading` 两次再跑上游 `build_full_synastry_dossier.py`,`delivery_readiness = ready_for_narrative`,九节写作素材齐全。整个证据包约 56 万字符;九节素材 3k~96k 字符不等 | 本地实测,未入库 |
| 8 | 与上游引擎相比我们缺 5 个模块(上游同一对人能算出):Lagna Vargottamamsa、Navamsha Dasha、Niryaana Shoola Dasha、Punarphoo、西占对照层。结果是「次要大运互证」一项 `blocked`,其余照常 | 由前置单 `TASK-full-reading-module-parity-20261018.md` 补齐 |
| 8b | 上游自己的缺陷:合盘校验器要的分盘键是 `varga_full.D4_Chaturthamsa`,两边引擎实际输出的是 `D4_Turyamsa`,所以「共同生活的收益与压力」那条双人任务在上游和我们这边都永远 `blocked`。已请占星师修;修好前该节照常写,只是少 D4 这层 | 上游 `validate_full_synastry_packets.py` 的 `PAIR_TASKS` |
| 8b | 上游 10-18 `7a773d13` 已修:D4 键名统一为 `D4_Turyamsa`,「共同生活」那条双人比对不再恒为 blocked;另加了单语言交付、把 Niryaana Shoola 从恋爱婚姻应期里排除。#79 搬的是 `bde5d615` 版本,本单开工第一步把 Skill 同步到 `7a773d13`(改动的 6 个文件与测试副本在 `docs/research/upstream_full_synastry_7a773d13/`;`marriage_match.py` 两版相同) | 上游 main `7a773d13` |
| 9 | 上游五个测试在我们这边 46 / 49 通过。3 条失败:① `scripts/aspects.py` 比上游旧(上游 09-24 改为按方向角和速度判相位,我们没同步);② 一条读上游桌面端 `jyotish-app/main.js`,我们没有该文件;③ 一条调上游 API 主文件里的 `_compute_marriage_match`,我们按薄注册另写 | 同上 |
## 决策记录(产品负责人 2026-10-18)
@@ -119,7 +119,7 @@
| T6 | **英文版按需生成**:`POST /api/reports/:id/english` 建任务(不扣点);同一报告只会有一个英文任务;失败可重试 | 路由与 worker 测试;英文提示里不出现中文稿内容(断言提示文本不含中文九节正文) |
| T7 | **阅读页**:合盘报告标题「A × B · 婚恋合盘」;九节正文(第一到七节结论后一个默认收起的「分析逻辑」;时间线每段同样带一个)→「对哪一段想再问清楚?」入口(E5d,对话侧未合入时隐藏)→「可以一起聊聊」四问 →「技术依据摘要」(默认折叠,E5b 六块,程序生成)→ 脚注(「合盘描述的是互动的质地,走不走、怎么走,仍然是你们两个人的选择」)。切英文时若还没生成,原地显示等待面板(一次揭幕),生成好整份替换 | 渲染测试用名人 golden;结论段落与「所以」行过一遍质检第 2、3 条禁表为空;分析逻辑与技术摘要默认收起;技术摘要六块齐全,第 ⑥ 块状态与证据包 `execution_receipt` 逐行一致;手机 390 px 无横向滚动;`next build` 后 `/` 仍 Static,首屏 gzip ±2% |
| T8 | **婚恋合盘页(从 #79 移来)与入口**:侧栏「婚恋合盘」;选人页(两个自定义下拉,底行「+ 新增星盘」复用星盘档案新建表;居中按钮,按钮下点数说明读 `synastry.full` 定价);「开始合盘」创建报告并跳到阅读页;「之前的合盘」列出合盘报告;档案页「和我合盘」改跳本页并预选两人;报告中心列表里合盘报告的卡片写「A × B 婚恋合盘」;删人确认句加「K 份合盘」 | 渲染与路由测试;20 人时下拉可滚动 |
| T9 | **上游测试搬入**:五个测试文件 + #79 已有的 `test_marriage_match_product.py`;`aspects.py` 同步上游 `bde5d615` 版本(方向角与速度判相位),跑全量 pytest 与基线逐条比对(相位值变了,报告类测试可能受影响,逐条写原因);读 `jyotish-app/main.js` 那条与调 `_compute_marriage_match` 那条改写为我们的等价断言(三栏说明) | 上游 49 条全过或有三栏说明;全量 pytest 无新增失败 |
| T9 | **上游同步与测试搬入**:先把 `skills/jyotish-full-synastry-reading/` 同步到 `7a773d13`(见实证 8b),`diff` 为空;搬入上游合盘测试(以 `7a773d13` 副本为准)+ #79 已有的 `test_marriage_match_product.py`;`aspects.py` 同步上游版本(方向角与速度判相位),跑全量 pytest 与基线逐条比对(相位值变了,报告类测试可能受影响,逐条写原因);依赖上游独有文件的测试改写为我们的等价断言(三栏说明) | 上游合盘测试全过或有三栏说明;全量 pytest 无新增失败;Niryaana Shoola 一行在名人证据包里为 `not_applicable` |
| T10 | **记录**:BUG-1373(#79 登记为 investigating,本单修复后改 resolved);DESIGN.md(选人页、阅读页、英文按需生成的等待态)、VOICE.md(引用准则文件;报告禁表、标题格式)、CHANGELOG、PROGRESS(与上游的差异:质检包装层、两条改写的测试)、`docs/testing/synastry-full-report-20261018.md` 真人清单(生成一份、再次打开不扣点、切英文、删人连带删除、失败退点、手机宽) | 文档齐 |
## 让步顺序