277 lines
19 KiB
Python
277 lines
19 KiB
Python
"""Validate planning data only; this does not execute or accept the game."""
|
|
|
|
from pathlib import Path
|
|
import argparse
|
|
import collections
|
|
import hashlib
|
|
import json
|
|
|
|
|
|
PLAN = Path(__file__).resolve().parent
|
|
ROOT = PLAN.parents[2]
|
|
|
|
|
|
def read(name):
|
|
return json.loads((PLAN / name).read_text())
|
|
|
|
|
|
def digest(path):
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def limited_view_registry_errors(atomic, declaration):
|
|
"""Validate declared view payloads against an immutable prior register."""
|
|
errors = []
|
|
|
|
def require(ok, message):
|
|
if not ok:
|
|
raise ValueError(message)
|
|
|
|
try:
|
|
ref = declaration["manifest"]
|
|
path = (PLAN / ref["path"]).resolve()
|
|
require(digest(path) == ref["sha256"], "limited view manifest SHA")
|
|
patch = json.loads(path.read_text())
|
|
require(type(patch["schema_version"]) is int and patch["schema_version"] == 1
|
|
and type(patch["whole_atom_status_changes"]) is int
|
|
and patch["whole_atom_status_changes"] == 0, "limited view manifest scope")
|
|
baseline_ref = patch["baseline_registry"]
|
|
baseline_path = (PLAN / baseline_ref["path"]).resolve()
|
|
require(digest(baseline_path) == baseline_ref["sha256"], "limited view baseline SHA")
|
|
baseline = json.loads(baseline_path.read_text())
|
|
expected = json.loads(json.dumps(baseline))
|
|
by_id = {row["id"]: row for row in expected["requirements"]}
|
|
matrix = patch["view_results_patch"]
|
|
require(type(matrix) is dict and bool(matrix), "limited view nonempty ID matrix")
|
|
require(set(matrix) <= set(by_id), "limited view unknown requirement ID")
|
|
allowed_views = set(baseline["completion_views"])
|
|
allowed_states = set(read("progress_contract.json")["view_states"])
|
|
declared_counts = collections.Counter()
|
|
reference_pins = patch["reference_pins"]
|
|
for rid, views in matrix.items():
|
|
require(type(views) is dict and bool(views) and set(views) <= allowed_views,
|
|
"limited view empty/unknown view: " + rid)
|
|
for view, result in views.items():
|
|
require(type(result) is dict and result.get("state") in allowed_states
|
|
and result.get("state") != "unverified", "limited view state: " + rid)
|
|
require(type(result.get("evidence_refs")) is list
|
|
and bool(result["evidence_refs"]) and bool(result.get("reviewer")),
|
|
"limited view missing evidence/reviewer: " + rid)
|
|
if result["state"] == "not_applicable":
|
|
require(bool(result.get("not_applicable_reason"))
|
|
and bool(result.get("source_or_explicit_user_decision")),
|
|
"limited view unsupported not-applicable: " + rid)
|
|
declared_counts[result["state"]] += 1
|
|
for evidence in result["evidence_refs"] + [result["current_adoption_evidence"]]:
|
|
require(type(evidence) is str, "limited view reference type: " + rid)
|
|
filename, separator, fragment = evidence.partition("#")
|
|
require(separator == "#" and bool(fragment) and evidence in reference_pins,
|
|
"limited view resolved reference: " + rid)
|
|
evidence_path = (PLAN / filename).resolve()
|
|
require(digest(evidence_path) == reference_pins[evidence],
|
|
"limited view evidence SHA: " + rid)
|
|
selected = json.loads(evidence_path.read_text())
|
|
for key in fragment.split("/"):
|
|
selected = selected[key]
|
|
require(bool(selected), "limited view empty referenced fragment: " + rid)
|
|
by_id[rid]["view_results"][view] = result
|
|
for key in ["atom_count", "view_count"]:
|
|
require(type(declaration[key]) is int and type(patch[key]) is int,
|
|
"limited view count type: " + key)
|
|
require(len(matrix) == declaration["atom_count"] == patch["atom_count"],
|
|
"limited view atom count")
|
|
require(sum(declared_counts.values()) == declaration["view_count"] == patch["view_count"],
|
|
"limited view observation count")
|
|
require(type(declaration["state_counts"]) is dict and type(patch["state_counts"]) is dict
|
|
and all(type(value) is int for value in declaration["state_counts"].values())
|
|
and dict(declared_counts) == declaration["state_counts"] == patch["state_counts"],
|
|
"limited view state counts")
|
|
require(json.dumps(atomic, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False)
|
|
== json.dumps(expected, ensure_ascii=False, sort_keys=True, separators=(",", ":"), allow_nan=False),
|
|
"limited view type-exact payload / untouched registry mismatch")
|
|
for row in atomic["requirements"]:
|
|
require(row["implementation_state"] == "not_reaudited"
|
|
and row["acceptance_state"] == row["status"] == "unverified",
|
|
"limited evidence changed whole-atom status: " + row["id"])
|
|
except (OSError, ValueError, KeyError, TypeError, AttributeError) as error:
|
|
errors.append(type(error).__name__ + ": " + str(error))
|
|
return errors
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--initial-snapshot", action="store_true", help="Also verify the immutable initial planning state")
|
|
parser.add_argument("--verify-baseline", action="store_true", help="Also compare current source bytes to the historical baseline")
|
|
args = parser.parse_args()
|
|
packages = read("execution/work_packages.initial.json" if args.initial_snapshot else "work_packages.json")
|
|
coverage = read("requirement_coverage.json")
|
|
baseline = read("baseline.json")
|
|
decisions = read("decisions.json")
|
|
contract = read("execution/progress_contract.initial.json" if args.initial_snapshot else "progress_contract.json")
|
|
checks = []
|
|
|
|
def check(name, condition):
|
|
checks.append({"name": name, "passed": bool(condition)})
|
|
|
|
if args.initial_snapshot:
|
|
initial = read("execution/PLANNING_DATA_INITIAL_SNAPSHOT.json")
|
|
check("immutable_initial_planning_data_pins", all(
|
|
(PLAN / name).is_file() and digest(PLAN / name) == value
|
|
for name, value in initial["sha256"].items()
|
|
))
|
|
|
|
ids = [row["id"] for row in packages]
|
|
expected = {f"P{i:02d}" for i in range(30)}
|
|
check("all30_unique_package_ids", len(ids) == 30 and set(ids) == expected)
|
|
tasks = [task for row in packages for task in row["remaining_deliverables"]]
|
|
task_ids = {task["id"] for task in tasks}
|
|
check("unique_delivery_ids", len(tasks) == len(task_ids) and all(p["remaining_deliverables"] for p in packages))
|
|
if args.initial_snapshot:
|
|
check("initial88_delivery_ids", len(tasks) == 88)
|
|
check("delivery_scope_and_acceptance", all(t["work"] and t["acceptance"] for t in tasks))
|
|
states = {"planned", "in_progress", "blocked", "completed"}
|
|
check("valid_delivery_states", all(t["state"] in states for t in tasks))
|
|
check("completed_delivery_requires_evidence", all(t["state"] != "completed" or t["evidence"] for t in tasks))
|
|
check("linked_delivery_evidence_exists", all(
|
|
isinstance(ref, dict) and (ROOT / ref["path"]).is_file()
|
|
for task in tasks for ref in task["evidence"]
|
|
))
|
|
if args.initial_snapshot:
|
|
check("initial_remaining_delivery_not_claimed_done", all(t["state"] == "planned" and not t["evidence"] for t in tasks))
|
|
check("known_dependency_ids", all(set(row["depends_on"]) <= expected for row in packages))
|
|
by_id = {row["id"]: row for row in packages}
|
|
visited, active = set(), set()
|
|
|
|
def acyclic(node):
|
|
if node in active:
|
|
return False
|
|
if node in visited:
|
|
return True
|
|
active.add(node)
|
|
if not all(acyclic(dep) for dep in by_id[node]["depends_on"] if dep in by_id):
|
|
return False
|
|
active.remove(node)
|
|
visited.add(node)
|
|
return True
|
|
|
|
check("dependency_graph_acyclic", all(acyclic(node) for node in ids))
|
|
old_requirements = json.loads((ROOT / "docs/traceability/original/requirements.json").read_text())
|
|
original_ids = {r["id"] for r in old_requirements}
|
|
mapped_ids = [r["requirement_id"] for r in coverage]
|
|
check("all75_original_groups_once", len(mapped_ids) == 75 and set(mapped_ids) == original_ids)
|
|
check("feature71_schedule4", sum(r["scope"] == "feature_or_content" for r in coverage) == 71 and sum(r["scope"] == "schedule" for r in coverage) == 4)
|
|
check("coverage_known_packages", all(set(r["work_packages"]) <= expected for r in coverage))
|
|
check("coverage_known_delivery_ids", all(set(r["remaining_deliverable_ids"]) <= task_ids for r in coverage))
|
|
check("coverage_exact_package_deliveries", all(set(r["remaining_deliverable_ids"]) == {d["id"] for p in packages if p["id"] in r["work_packages"] for d in p["remaining_deliverables"]} for r in coverage))
|
|
check("package_and_requirement_mapping_agree", all({p["id"] for p in packages if r["requirement_id"] in p["requirement_ids"]} == set(r["work_packages"]) for r in coverage))
|
|
check("source_lines_preserved", all(r["source_lines"] == next(x["source_lines"] for x in old_requirements if x["id"] == r["requirement_id"]) for r in coverage))
|
|
check("old_status_not_promoted", all("current_status" not in r and r["current_atomic_acceptance"] == "not_reaudited" for r in coverage))
|
|
supplements = {s for p in packages for s in p["supplement_ids"]}
|
|
check("all10_facility_supplements", supplements == {f"F-{i:02d}" for i in range(1, 11)})
|
|
check("P09_owns10_supplements", set(by_id["P09"]["supplement_ids"]) == supplements)
|
|
pending = {d["id"] for d in decisions["pending"]}
|
|
check("all10_pending_decision_groups", pending == {f"D{i:02d}" for i in range(1, 11)})
|
|
check("known_decision_references", all(set(p["decisions"]) <= pending for p in packages))
|
|
check("all14_confirmed_decisions", {d["id"] for d in decisions["confirmed"]} == {f"A{i:02d}" for i in range(1, 15)})
|
|
check("confirmed_decision_references_exist", all((PLAN / d["reference"]).is_file() for d in decisions["confirmed"] if "reference" in d))
|
|
state_counts = dict(collections.Counter(p["state"] for p in packages))
|
|
check("valid_package_states", all(p["state"] in states for p in packages))
|
|
check("new_completed_packages_require_evidence", all(p["state"] != "completed" or p["legacy_state"] == "completed" or p.get("completion_evidence") for p in packages))
|
|
check("acceptance_conditions_not_claimed_results", all("cross_cutting_done" not in p and p["cross_cutting_acceptance_scope"] for p in packages))
|
|
check("initial_revision_history_preserved", dict(collections.Counter(p["initial_revision_state"] for p in packages)) == baseline["revised_package_state_counts"])
|
|
if args.initial_snapshot:
|
|
check("initial_revised_state_counts", state_counts == baseline["revised_package_state_counts"] == {"completed": 1, "in_progress": 25, "planned": 4})
|
|
check("initial_no_new_completed_packages", all(p["state"] != "completed" or p["legacy_state"] == "completed" for p in packages))
|
|
check("initial_eight_partial_state_promotions", {p["id"] for p in packages if p["state"] != p["legacy_state"]} == {"P09", "P14", "P15", "P16", "P20", "P23", "P25", "P28"})
|
|
check("initial_atomic_denominator_not_invented", baseline["atomic_denominator"] is None and baseline["overall_atomic_implementation_percent"] is None and baseline["overall_atomic_acceptance_percent"] is None)
|
|
check("progress_contract8_views", len(contract["completion_views"]) == 8)
|
|
check("progress_no_partial_credit", contract["partial_credit"] is False)
|
|
known_denominator = None
|
|
final_denominator = None
|
|
if contract["denominator_frozen"]:
|
|
check("frozen_denominator_has_register", bool(contract["atomic_register"]) and (PLAN / contract["atomic_register"]).is_file())
|
|
atomic = read(contract["atomic_register"])
|
|
freeze = read(contract["atomic_denominator_freeze"])
|
|
index = read(contract["current_evidence_index"])
|
|
known_denominator = len(atomic["denominator_ids"])
|
|
final_denominator = contract["final_atomic_denominator"]
|
|
check("known_denominator_matches_frozen_register", contract["known_condition_denominator"] == freeze["denominator"] == known_denominator and freeze["denominator_ids"] == atomic["denominator_ids"])
|
|
check("progress_register_pin_matches", contract["atomic_register_sha256_at_integration"] == digest(PLAN / contract["atomic_register"]))
|
|
check("evidence_index_register_pin_matches", index["atomic_register_sha256"] == digest(PLAN / contract["atomic_register"]) and index["known_condition_denominator"] == known_denominator)
|
|
check("progress_index_pin_matches", contract["current_evidence_index_sha256_at_integration"] == digest(PLAN / contract["current_evidence_index"]))
|
|
check("known_counts_match_freeze", contract["current_whole_atom_implementation_count"] == len(freeze["implemented_ids"]) and contract["current_whole_atom_acceptance_count"] == len(freeze["accepted_ids"]))
|
|
check("unresolved_final_denominator_not_reported_as_total", contract["final_denominator_established"] or (
|
|
final_denominator is None and all(contract[key] is None for key in [
|
|
"overall_atomic_implementation_percent", "overall_atomic_acceptance_percent",
|
|
"total_game_implementation_percent", "total_game_acceptance_percent"
|
|
]) and atomic["final_denominator_established"] is False and freeze["final_denominator_established"] is False
|
|
))
|
|
check("static_leads_not_whole_atom_acceptance", index["whole_atom_acceptance_claimed"] is False and index["complete_runtime_ui_save_review"] is False)
|
|
if contract["atomic_implementation_state"] == "not_reaudited":
|
|
if contract.get("limited_view_review_manifest"):
|
|
check("declared_limited_views_whole_atoms_unreviewed", not limited_view_registry_errors(
|
|
atomic, contract["limited_view_review_manifest"]
|
|
) and index.get("limited_view_review_manifest") == contract["limited_view_review_manifest"])
|
|
else:
|
|
check("unreviewed_atomic_views_remain_unverified", all(
|
|
view["state"] == "unverified" and view["implementation_state"] == "not_reaudited"
|
|
for row in atomic["requirements"] for view in row["view_results"].values()
|
|
))
|
|
snapshot = read(contract["progress_snapshot"])
|
|
regression = snapshot["latest_regression_at_evidence_time"]
|
|
check("regression_evidence_pin_matches", digest(ROOT / regression["path"]) == regression["sha256"] and digest(ROOT / regression["source_map_path"]) == regression["source_map_sha256"])
|
|
check("regression_source_time_scope_preserved", regression["future_source_changes_auto_accepted"] is False and regression["original_full_acceptance"] is False and regression["windows_acceptance"] is False)
|
|
check("P01_P03_not_whole_package_completed", by_id["P01"]["state"] != "completed" and by_id["P03"]["state"] != "completed")
|
|
else:
|
|
check("unfrozen_denominator_not_assigned", contract["atomic_register"] is None)
|
|
check("progress_schedule_excluded", contract["excluded_from_function_denominator"] == ["B-14.1", "B-14.2", "B-14.3", "B-14.4"])
|
|
check("baseline_materials_exist", all((ROOT / path).is_file() for path in baseline["source_pins"]))
|
|
pins = baseline["facility_audit_current_pins_verified"]
|
|
if args.verify_baseline:
|
|
check("plan_source_pins_unchanged", all((ROOT / path).is_file() and digest(ROOT / path) == value for path, value in baseline["source_pins"].items()))
|
|
check("current_facility40_pins_unchanged", len(pins) == 40 and all(digest(ROOT / path) == value for path, value in pins.items()))
|
|
accepted = json.loads((ROOT / "artifacts/facility-need-pricing-production-scale-adoption-2026-10-01-r01/after.json").read_text())
|
|
if args.verify_baseline:
|
|
check("accepted1396_files_unchanged", len(accepted) == 1396 and all((ROOT / path).is_file() and digest(ROOT / path) == value for path, value in accepted.items()))
|
|
receipt = json.loads((ROOT / "artifacts/facility-need-pricing-production-scale-adoption-2026-10-01-r01/receipt.json").read_text())
|
|
check("retained_regression155_12310", baseline["full_regression"]["suites"] == receipt["accepted_proof"]["actual_full_suites"] == 155 and baseline["full_regression"]["checks"] == receipt["accepted_proof"]["actual_full_checks"] == 12310)
|
|
check("new_game_tests_not_claimed", baseline["full_regression"]["new_engine_runs_in_plan_creation"] == 0)
|
|
text = (PLAN / ("execution/IMPLEMENTATION_PLAN.initial.md" if args.initial_snapshot else "IMPLEMENTATION_PLAN.md")).read_text()
|
|
check("all_package_headings", all(f"### {identifier} " in text for identifier in ids))
|
|
check("all_group_rows", all(f"| {identifier} |" in text for identifier in original_ids))
|
|
check("all_delivery_ids_in_document", all(identifier in text for identifier in task_ids))
|
|
check("encyclopedia_and_hero_rival_scope", "図鑑の完成" in text and "ライバルは別集合" in text)
|
|
check("current_native_and_longplay_not_claimed", baseline["current_head_windows_native_verified"] is False and baseline["long_play_acceptance_verified"] is False)
|
|
failures = [c["name"] for c in checks if not c["passed"]]
|
|
result = {
|
|
"plan_date_jst": baseline["plan_date_jst"],
|
|
"scope": "planning consistency and static baseline reconciliation; not game acceptance",
|
|
"passed": not failures,
|
|
"checks": checks,
|
|
"failed_checks": failures,
|
|
"packages": len(packages),
|
|
"remaining_delivery_units": len(tasks),
|
|
"current_package_state_counts": state_counts,
|
|
"initial_snapshot_checked": args.initial_snapshot,
|
|
"historical_source_bytes_compared": args.verify_baseline,
|
|
"covered_original_groups": len(mapped_ids),
|
|
"missing_original_group_ids": sorted(original_ids - set(mapped_ids)),
|
|
"facility_supplements": len(supplements),
|
|
"atomic_denominator": final_denominator,
|
|
"known_condition_denominator": known_denominator,
|
|
"final_denominator_established": contract.get("final_denominator_established", False),
|
|
"total_game_implementation_percent": None,
|
|
"total_game_acceptance_percent": None,
|
|
"new_game_execution_count": 0,
|
|
"document_sha256": {p.name: digest(p) for p in sorted(PLAN.iterdir()) if p.is_file() and p.name not in ["plan_validation.json", "plan_validation.initial.json"]},
|
|
}
|
|
output = "plan_validation.initial.json" if args.initial_snapshot else "plan_validation.json"
|
|
(PLAN / output).write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n")
|
|
print(json.dumps({k: v for k, v in result.items() if k not in ["checks", "document_sha256"]}, ensure_ascii=False))
|
|
if failures:
|
|
raise SystemExit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|