TheTree_mock/docs/plans/original-completion-2026-10-02/validate_facility_value_contracts.py

#!/usr/bin/env python3
"""Independent static contract consistency checks, not Runtime acceptance."""
import copy
import hashlib
import json
import re
from collections import Counter
from pathlib import Path

OUT = Path(__file__).resolve().parent
ROOT = OUT.parents[2]
TARGET = OUT / "facility_value_contracts.json"
EXPECTED_NAMES = ["一般客室", "上級客室", "VIPスイート", "ドミトリー", "大衆食堂", "レストラン", "酒場", "屋台", "大浴場", "カジノ", "闘技場", "礼拝堂", "風俗個室", "高級ソープ", "SM拷問室", "闇市場", "厨房", "倉庫", "待機室", "従業員寮", "警備員詰所", "清掃員室", "医務室", "覗き部屋", "地下牢", "マナ炉", "蔓の昇降機", "業務用EV", "ワープ", "魔法砲台"]
EXPECTED_SIZES = [[2,1], [4,1], [6,2], [4,1], [4,2], [6,2], [4,2], [2,1], [6,2], [6,3], [8,4], [4,3], [2,1], [4,2], [4,2], [4,2], [4,2], [4,2], [4,2], [2,1], [2,2], [2,1], [4,2], [2,1], [4,2], [6,3], [2,2], [3,2], [2,2], None]
EXPECTED_CUSTOMER = {"chapel": 8, "cleaning_room": 0}
MANDATORY_PARAMETER_KEYS = {"construction_cost", "build_minutes", "maintenance_gold", "maintenance_mana", "base_price", "footprint", "layer", "zone", "required_roles", "primary_skill", "staff_min", "service_types", "default_service_type", "service_profiles", "base_service_minutes", "customer_capacity", "capacity", "queue_capacity", "staff_max", "service_attribute", "max_level", "upgrade_costs", "unlock_rank", "research_requirements", "policy_schema", "faith_per_service"}


def validate(doc, check_files=True):
    errors = []
    def require(test, label):
        if not test:
            errors.append(label)
    rows = doc.get("original_facilities", [])
    require(len(rows) == 30, "original30count")
    require([r.get("name") for r in rows] == EXPECTED_NAMES, "original30identity_order")
    require(len({r.get("id") for r in rows}) == len(rows), "unique_contract_ids")
    require(sum(r.get("definition_exists", False) for r in rows) == 17, "defined17")
    require(sum(r.get("normal_registered_static", False) for r in rows) == 16, "normal16")
    require([r.get("definition_id") for r in rows if r.get("registration_state") == "definition_only"] == ["black_market"], "black_market_definition_only")
    require(len(doc.get("missing_definition_blueprints", [])) == 13, "blueprints13")
    require(len(doc.get("supplemental_current_definitions", [])) == 20, "other20definitions")
    customer_states = Counter()
    for i, r in enumerate(rows):
        parms = r.get("parameters", {})
        require(set(parms) == MANDATORY_PARAMETER_KEYS, f"{r.get('id')}:parameter_keys")
        footprint = parms.get("footprint", {}).get("original")
        require((footprint or {}).get("value") == EXPECTED_SIZES[i], f"{r.get('id')}:source_size")
        require(set(r.get("major_items", {})) == {f"F{i:02d}" for i in range(1, 11)}, f"{r.get('id')}:major10")
        require([lv.get("level") for lv in r.get("level_contracts", [])] == [1,2,3], f"{r.get('id')}:level123")
        for lv in r.get("level_contracts", []):
            for k in ["authoritative_level_specific_price", "authoritative_level_specific_customer_capacity", "authoritative_level_specific_staff_min", "authoritative_level_specific_staff_max", "unlock_condition", "policy_unlock_assignment"]:
                require(lv.get(k) is None, f"{r.get('id')}:no_invented_level_value:{k}")
            require(lv.get("base_approval_has_explicit_level_scope") is False, f"{r.get('id')}:no_invented_approval_scope")
        cp = parms.get("customer_capacity", {})
        customer_states[cp.get("value_status")] += 1
        fid = r.get("definition_id")
        approval = cp.get("individual_or_formula_approval")
        if fid in EXPECTED_CUSTOMER:
            require((approval or {}).get("value") == EXPECTED_CUSTOMER[fid], f"{fid}:approved_customer")
        else:
            require(approval is None, f"{fid}:no_customer_approval_inferred")
        require(r.get("generic_capacity_contract", {}).get("is_customer_capacity") is False, f"{fid}:capacity_distinct")
        require(r.get("runtime_acceptance_state") == "unverified_no_engine_run_in_this_audit", f"{fid}:no_runtime_pass")
        match_inputs = r.get("need_match_inputs", {})
        require(match_inputs.get("deficit_threshold", {}).get("value") is None, f"{fid}:threshold_unset")
        require(match_inputs.get("surcharge_amount_or_rate", {}).get("value") is None, f"{fid}:surcharge_unset")
        if check_files and fid:
            path = ROOT / r["definition_path"]
            raw_fields = {m[1]: m[2] for line in path.read_text().splitlines() if (m := re.match(r"(\w+) = (.+)$", line))}
            for k, parm in parms.items():
                observed = parm.get("current", {})
                if observed.get("origin") == "explicit_tres":
                    require(raw_fields.get(k) == observed.get("raw"), f"{fid}:{k}:raw_current_match")
    require(customer_states == Counter({"approved_initial_trial_value":2, "current_value_without_individual_approval":2, "explicitly_unset_required_contract":13, "not_set_no_definition":13}), "customer_state_counts")
    for bp in doc.get("missing_definition_blueprints", []):
        require(bp.get("production_id") is None and bp.get("register_and_allow_build_now") is False, "blueprint_not_playable")
    common = doc.get("common_contracts", {})
    require(common.get("operation_state", {}).get("authoritative_states") == ["INACTIVE", "OPERATING", "SUSPENDED"], "three_operation_states")
    for item in [common.get("operation_state", {}).get("resumption_conditions", {}), common.get("related_facility", {}).get("multi_relation_AND_OR", {})]:
        require(item.get("value") is None and item.get("decision_id") == "D03" and item.get("is_mandatory_missing") is False, "D03_not_mandatory_gap")
    require(common.get("need_match_surcharge", {}).get("multiple_attributes_and_duplicate_surcharge", {}).get("is_mandatory_missing") is False, "optional_multiple_attributes_not_gap")
    require(common.get("need_match_surcharge", {}).get("guest_need_effects_are_attribute_assignment") is False, "effect_not_attribute")
    require(common.get("need_match_surcharge", {}).get("guest_need_0_100_is_deficit_threshold") is False, "range_not_threshold")
    summary = doc.get("summary", {})
    require(summary.get("is_final_all_facility_catalog") is False and summary.get("approved_runtime_acceptances_counted") == 0, "no_final_acceptance_claim")
    require(summary.get("optional_unset_D03_counted_as_mandatory_missing") == 0, "no_optional_gap_count")
    require(doc.get("acceptance", {}).get("completed") is False and doc.get("acceptance", {}).get("runtime_runs_in_this_audit") == 0, "audit_not_complete_runtime")
    require(set(doc.get("approvals", {})) == {f"A{i:02d}" for i in range(1,6)}, "A01_A05_preserved")
    pins = doc.get("source_pins", {})
    if check_files:
        for path, sha in pins.items():
            require(hashlib.sha256((ROOT / path).read_bytes()).hexdigest() == sha, f"source_sha:{path}")
    def walk(value):
        if isinstance(value, dict):
            if all(k in value for k in ["path", "lines", "sha256", "authority"]):
                require(pins.get(value["path"]) == value["sha256"], f"anchor_pin:{value['path']}")
                if check_files:
                    count = len((ROOT / value["path"]).read_text().splitlines())
                    require(bool(value["lines"]) and all(1 <= n <= count for n in value["lines"]), f"anchor_lines:{value['path']}")
            for v in value.values():
                walk(v)
        elif isinstance(value, list):
            for v in value:
                walk(v)
    walk(doc)
    return errors


doc = json.loads(TARGET.read_text())
errors = validate(doc)
mutations = {
    "drop_original": lambda d: d["original_facilities"].pop(),
    "duplicate_id": lambda d: d["original_facilities"][1].update(id=d["original_facilities"][0]["id"]),
    "invent_cannon_size": lambda d: d["original_facilities"][-1]["parameters"]["footprint"].update(original={"value":[2,2]}),
    "infer_approval_from_capacity": lambda d: d["original_facilities"][0]["parameters"]["customer_capacity"].update(individual_or_formula_approval={"value":2}),
    "invent_lv2_value": lambda d: d["original_facilities"][0]["level_contracts"][1].update(authoritative_level_specific_price=150),
    "D03_mandatory_missing": lambda d: d["common_contracts"]["operation_state"]["resumption_conditions"].update(is_mandatory_missing=True),
    "invent_threshold": lambda d: d["original_facilities"][0]["need_match_inputs"]["deficit_threshold"].update(value=75),
    "runtime_pass": lambda d: d["original_facilities"][0].update(runtime_acceptance_state="PASS"),
    "final_catalog": lambda d: d["summary"].update(is_final_all_facility_catalog=True),
    "playable_undefined": lambda d: d["missing_definition_blueprints"][0].update(register_and_allow_build_now=True),
}
mutant_results = []
for name, mutate in mutations.items():
    changed = copy.deepcopy(doc)
    mutate(changed)
    mutant_results.append({"name":name, "rejected":bool(validate(changed, check_files=False))})
receipt = {"status": "PASS" if not errors and all(m["rejected"] for m in mutant_results) else "FAIL",
    "scope": "static audit consistency only; not game/UI/save acceptance",
    "audit_sha256": hashlib.sha256(TARGET.read_bytes()).hexdigest(),
    "source_pins_checked": len(doc["source_pins"]), "original_facilities_checked":30,
    "parameter_records_checked":780, "level_records_checked":90,
    "errors":errors, "negative_mutants":mutant_results, "runtime_runs":0}
(OUT / "facility_value_contracts_validation.json").write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + "\n")
print(json.dumps(receipt, ensure_ascii=False, indent=2))
raise SystemExit(0 if receipt["status"] == "PASS" else 1)