From 83d45ae1f04ba635f06eb416c4928eeb47bace65 Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Thu, 9 Jul 2026 10:39:51 +0800 Subject: [PATCH] feat: score active rectification answers --- docs/research/pre_work_error_ledger.md | 1 + scripts/active_rectification_questions.py | 50 +++++++++++++++++++- tests/test_active_rectification_questions.py | 23 ++++++++- 3 files changed, 72 insertions(+), 2 deletions(-) diff --git a/docs/research/pre_work_error_ledger.md b/docs/research/pre_work_error_ledger.md index e7e5d79c..9640ca07 100644 --- a/docs/research/pre_work_error_ledger.md +++ b/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 diff --git a/scripts/active_rectification_questions.py b/scripts/active_rectification_questions.py index 4c7bdc46..c1813e41 100644 --- a/scripts/active_rectification_questions.py +++ b/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 diff --git a/tests/test_active_rectification_questions.py b/tests/test_active_rectification_questions.py index 36527e09..aa11c3dc 100644 --- a/tests/test_active_rectification_questions.py +++ b/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"]