feat: score active rectification answers

docs/research/pre_work_error_ledger.md

@@ -58,6 +58,7 @@ For large architecture or release work, also read:
 | ERR-025 | VedAstro gateway can report legacy `status=ok` from catalog availability even when no official raw response is present. | mitigated 2026-07-08 | `official_closure_state=official_verified` requires `official_raw_response`; otherwise expose `official_closure_reason=official_raw_response_missing`. |
 | ERR-026 | VedAstro service adapter can obtain an official full-snapshot raw response while the user entrypoint drops it, leaving gateway official closure permanently blocked. | mitigated 2026-07-09 | `vedastro_user_entrypoint` must expose `vedastro_official_full_snapshot.raw_response_available` and root `official_raw_response` when explicitly requested; gateway tests must prove raw propagation reaches `official_verified`. |
 | ERR-027 | External engine readiness diagnostics can be mistaken for a completed same-chart parity comparison. | mitigated 2026-07-09 | `diagnose_external_engine_adapters.py` must expose `same_chart_parity_contract.required_outputs`, per-engine expected oracle fields, and `tested=false` until a real same-chart comparison runs. |
+| ERR-028 | Active birth-time rectification can stop at question generation and never narrow candidate clusters from user answers. | mitigated 2026-07-09 | `active_rectification_questions.score_answers()` must turn A/B/C/D answers into cluster rankings, next-round questions, and an explicit boundary that final rectification still needs candidate chart differences. |
 
 ## Fragment Sweep Command Set
 

scripts/active_rectification_questions.py

@@ -84,14 +84,62 @@ def build_questionnaire(birth_time: str, uncertainty_minutes: int = 30, step_min
     }
 
 
+def score_answers(questionnaire: dict[str, Any], answers: dict[str, str]) -> dict[str, Any]:
+    questions = questionnaire.get("questions") if isinstance(questionnaire.get("questions"), list) else []
+    by_id = {question["id"]: question for question in questions if isinstance(question, dict) and question.get("id")}
+    cluster_scores: dict[str, int] = {}
+    applied = []
+    unknown_ids = []
+    invalid_answers = []
+
+    for question_id, raw_choice in (answers or {}).items():
+        question = by_id.get(question_id)
+        if not question:
+            unknown_ids.append(question_id)
+            continue
+        choice = str(raw_choice or "").strip().upper()
+        scoring = (question.get("scoring_map") or {}).get(choice)
+        if not isinstance(scoring, dict):
+            invalid_answers.append({"id": question_id, "answer": raw_choice})
+            continue
+        cluster = str(scoring.get("cluster") or "neutral")
+        points = int(scoring.get("points") or 0)
+        if cluster != "neutral":
+            cluster_scores[cluster] = cluster_scores.get(cluster, 0) + points
+        applied.append({"id": question_id, "answer": choice, "cluster": cluster, "points": points})
+
+    answered_ids = {item["id"] for item in applied}
+    unanswered = [question for question in questions if question.get("id") not in answered_ids]
+    next_round = min((int(question.get("round") or 0) for question in unanswered), default=None)
+    rankings = [
+        {"cluster": cluster, "score": score}
+        for cluster, score in sorted(cluster_scores.items(), key=lambda item: (-item[1], item[0]))
+    ]
+    return {
+        "scope": "active_birth_time_rectification_scoring",
+        "schema_version": 1,
+        "answered_count": len(applied),
+        "candidate_cluster_rankings": rankings,
+        "next_round": next_round,
+        "next_round_questions": [question for question in unanswered if question.get("round") == next_round],
+        "applied_scoring": applied,
+        "unknown_question_ids": unknown_ids,
+        "invalid_answers": invalid_answers,
+        "boundary": "This narrows candidate clusters only; final rectification requires scoring answers against actual candidate chart differences.",
+    }
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description=__doc__)
     parser.add_argument("--birth-time", required=True, help="Approximate local birth time, YYYY-MM-DD HH:MM")
     parser.add_argument("--uncertainty-minutes", type=int, default=30)
     parser.add_argument("--step-minutes", type=int, default=1)
+    parser.add_argument("--answers-json", default="", help="Optional JSON object mapping question id to A/B/C/D")
     parser.add_argument("--pretty", action="store_true")
     args = parser.parse_args()
-    print(json.dumps(build_questionnaire(args.birth_time, args.uncertainty_minutes, args.step_minutes), ensure_ascii=False, indent=2 if args.pretty else None))
+    questionnaire = build_questionnaire(args.birth_time, args.uncertainty_minutes, args.step_minutes)
+    report = score_answers(questionnaire, json.loads(args.answers_json)) if args.answers_json else questionnaire
+    print(json.dumps(report, ensure_ascii=False, indent=2 if args.pretty else None))
     return 0
 
 

tests/test_active_rectification_questions.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from scripts.active_rectification_questions import build_questionnaire
+from scripts.active_rectification_questions import build_questionnaire, score_answers
 
 
 def test_active_rectification_questions_generate_choice_based_workflow() -> None:
@@ -16,3 +16,24 @@ def test_active_rectification_questions_generate_choice_based_workflow() -> None
     assert {q["round"] for q in report["questions"]} == {1, 2, 3}
     assert all({option["key"] for option in q["options"]} == {"A", "B", "C", "D"} for q in report["questions"])
     assert all("scoring_map" in q for q in report["questions"])
+
+
+def test_active_rectification_scores_answers_and_selects_next_round() -> None:
+    report = build_questionnaire("1955-02-24 19:15", uncertainty_minutes=30)
+    scored = score_answers(
+        report,
+        {
+            "education_environment_shift": "A",
+            "residence_relocation_shift": "B",
+            "relationship_or_partner_entry": "D",
+            "career_responsibility_pressure": "A",
+            "research_tool_expression_shift": "C",
+        },
+    )
+
+    assert scored["scope"] == "active_birth_time_rectification_scoring"
+    assert scored["answered_count"] == 5
+    assert scored["next_round"] == 2
+    assert scored["next_round_questions"]
+    assert scored["candidate_cluster_rankings"][0]["score"] > scored["candidate_cluster_rankings"][-1]["score"]
+    assert "final rectification requires scoring answers against actual candidate chart differences" in scored["boundary"]