2x2 FRONT warp at 3000G/30 materials/360 min, 15G/day and 1 Mana upkeep. Every completed warp pair is linked by one path connector, so crossing is a single step; a 3-Mana fare per rider is charged once in the step's tick. Warps close while either end is unpowered, the reserve is empty or below one fare, and blocked arrivals offer no step. Saves validate the derived links. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
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("all7_confirmed_decisions", {d["id"] for d in decisions["confirmed"]} == {f"A{i:02d}" for i in range(1, 8)})
|
|
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()
|