#!/usr/bin/env python3 # SPDX-FileCopyrightText: 2026 Jason Doyle # SPDX-License-Identifier: MIT # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal # in the Software without restriction, including without limitation the rights # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell # copies of the Software, and to permit persons to whom the Software is # furnished to do so, subject to the following conditions: # # The above copyright notice and this permission notice shall be included in # all copies or substantial portions of the Software. # # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE # SOFTWARE. from __future__ import annotations import copy import hashlib import json from datetime import datetime, timezone from pathlib import Path from typing import Any from jsonschema import Draft202012Validator, FormatChecker ROOT = Path(__file__).resolve().parent SCHEMA_PATH = ROOT / "game-discovery-profile.schema.json" PROFILES_PATH = ROOT / "public-case-study-profiles.json" QUERIES_PATH = ROOT / "public-case-study-queries.json" FIXTURES_PATH = ROOT / "public-case-study-adversarial-fixtures.json" RESULTS_PATH = ROOT / "public-case-study-results.json" TRUSTED_HARD_SOURCES = { "developer_declaration", "marketplace_verification", } EXPERIENCE_SOURCE_MINIMUMS = { "session.short_session_fit": 2, "social.coop_fit": 2, "return_after_break": 1, } def utc_now() -> str: return ( datetime.now(timezone.utc) .replace(microsecond=0) .isoformat() .replace("+00:00", "Z") ) def sha256_file(path: Path) -> str: return "sha256:" + hashlib.sha256(path.read_bytes()).hexdigest() def appid(profile: dict[str, Any]) -> int: return int(profile["game_identity"]["storefront_product_id"]) def display_name(profile: dict[str, Any]) -> str: return profile["x_extensions"]["display_name"] def current_release(profile: dict[str, Any]) -> str: return profile["game_identity"]["release_id"] def evidence_source_types(claim: dict[str, Any]) -> set[str]: return { evidence["source_type"] for evidence in claim.get("evidence", []) } def review_evidence_keys(claim: dict[str, Any]) -> set[str]: values: set[str] = set() for evidence in claim.get("evidence", []): value = evidence.get("x_extensions", {}).get( "review_evidence_key" ) if value: values.add(str(value)) return values def claim_is_current(profile: dict[str, Any], claim: dict[str, Any]) -> bool: return current_release(profile) in claim["applicability"]["release_ids"] def baseline_hard_value( profile: dict[str, Any], property_name: str, ) -> tuple[str, Any]: if property_name == "offer.requires_external_launcher": offer = profile["offer"] if "requires_external_launcher" in offer: return "known", offer["requires_external_launcher"] return "unknown", None claims = [ claim for claim in profile["claims"] if claim["property"] == property_name and claim["value_status"] == "known" ] if not claims: return "unknown", None return "known", claims[-1]["value"] def profile_hard_value( profile: dict[str, Any], property_name: str, ) -> tuple[str, Any]: if property_name == "offer.requires_external_launcher": offer = profile["offer"] if offer["requires_external_launcher_status"] != "known": return "unknown", None return "known", offer["requires_external_launcher"] accepted = [] for claim in profile["claims"]: if claim["property"] != property_name: continue if claim["claim_class"] != "functional_fact": continue if claim["value_status"] != "known": continue if claim["status"] != "active": continue if not claim_is_current(profile, claim): continue if not evidence_source_types(claim).issubset(TRUSTED_HARD_SOURCES): continue accepted.append(claim["value"]) if not accepted or len(set(map(json.dumps, accepted))) != 1: return "unknown", None return "known", accepted[0] def profile_soft_value( profile: dict[str, Any], property_name: str, ) -> tuple[str, Any, str]: grouped: dict[str, dict[str, Any]] = {} for claim in profile["claims"]: if claim["property"] != property_name: continue if claim["claim_class"] not in { "population_experience", "model_inference", }: continue if claim["value_status"] != "known": continue if claim["status"] not in {"active", "contested"}: continue if not claim_is_current(profile, claim): continue key = json.dumps(claim["value"], sort_keys=True) group = grouped.setdefault( key, { "value": claim["value"], "review_evidence_keys": set(), "statuses": set(), }, ) group["review_evidence_keys"].update(review_evidence_keys(claim)) group["statuses"].add(claim["status"]) minimum = EXPERIENCE_SOURCE_MINIMUMS.get(property_name, 1) accepted = [ group for group in grouped.values() if len(group["review_evidence_keys"]) >= minimum ] if not accepted: return "unknown", None, "unknown" if len(accepted) != 1: return "unknown", None, "contested" group = accepted[0] status = ( "contested" if "contested" in group["statuses"] else "active" ) return "known", group["value"], status def condition_matches(value: Any, condition: dict[str, Any]) -> bool: operator = condition["operator"] expected = condition["value"] if operator == "eq": return value == expected raise ValueError(f"Unsupported operator: {operator}") def baseline_soft_score( profile: dict[str, Any], conditions: list[dict[str, Any]], ) -> tuple[float, list[str]]: score = 0.0 reasons: list[str] = [] for condition in conditions: for claim in profile["claims"]: if claim["property"] != condition["property"]: continue if claim["value_status"] != "known": continue if condition_matches(claim["value"], condition): score += float(condition["weight"]) reasons.append(claim["claim_id"]) return score, reasons def profile_soft_score( profile: dict[str, Any], conditions: list[dict[str, Any]], ) -> tuple[float, list[str]]: score = 0.0 reasons: list[str] = [] for condition in conditions: status, value, claim_status = profile_soft_value( profile, condition["property"], ) if status != "known" or not condition_matches(value, condition): continue factor = 0.5 if claim_status == "contested" else 1.0 score += float(condition["weight"]) * factor reasons.append(condition["property"]) return score, reasons def current_player_count(profile: dict[str, Any]) -> int: for claim in profile["claims"]: if ( claim["property"] == "population.current_players" and claim["value_status"] == "known" and claim_is_current(profile, claim) ): return int(claim["value"]) return 0 def evaluate_query( profiles: list[dict[str, Any]], query: dict[str, Any], mode: str, ) -> dict[str, Any]: candidates: list[dict[str, Any]] = [] for profile in profiles: hard_failures = [] hard_unknowns = [] for condition in query["hard"]: if mode == "profile": status, value = profile_hard_value( profile, condition["property"], ) else: status, value = baseline_hard_value( profile, condition["property"], ) if status != "known": hard_unknowns.append(condition["property"]) elif not condition_matches(value, condition): hard_failures.append(condition["property"]) if hard_failures or hard_unknowns: continue if mode == "profile": score, reasons = profile_soft_score(profile, query["soft"]) else: score, reasons = baseline_soft_score(profile, query["soft"]) candidates.append( { "appid": appid(profile), "name": display_name(profile), "score": score, "current_players": current_player_count(profile), "supporting_claims": reasons, } ) candidates.sort( key=lambda item: ( -item["score"], -item["current_players"], item["appid"], ) ) return { "query_id": query["id"], "candidate_count": len(candidates), "top_appid": candidates[0]["appid"] if candidates else None, "top_name": candidates[0]["name"] if candidates else None, "candidates": candidates, } def pristine_hard_violation( pristine_profiles: list[dict[str, Any]], query: dict[str, Any], selected_appid: int | None, ) -> bool: if selected_appid is None: return False profile = next( value for value in pristine_profiles if appid(value) == selected_appid ) for condition in query["hard"]: status, value = profile_hard_value(profile, condition["property"]) if status != "known" or not condition_matches(value, condition): return True return False def fixture_claim( profile: dict[str, Any], fixture: dict[str, Any], ) -> dict[str, Any]: spec = fixture["claim"] release_id = ( current_release(profile) if spec["release_scope"] == "current" else "obsolete-release" ) return { "claim_id": spec["claim_id"], "property": spec["property"], "claim_class": spec["claim_class"], "asserted_by": spec["asserted_by"], "value_status": spec["value_status"], "value": spec["value"], "applicability": { "release_ids": [release_id], "storefronts": ["steam"], "target_platforms": ["windows"], }, "evidence": [ { "evidence_id": f"{spec['claim_id']}-evidence", "source_type": spec["evidence_source_type"], "retrieved_at": profile["generated_at"], "method": "Adversarial case-study fixture", "method_version": "fixture-0.1", "simulated": True, } ], "privacy": {"release_mode": "public"}, "status": spec["status"], } def apply_fixture( pristine_profiles: list[dict[str, Any]], fixture: dict[str, Any], ) -> list[dict[str, Any]]: profiles = copy.deepcopy(pristine_profiles) profile = next( value for value in profiles if appid(value) == fixture["appid"] ) operation = fixture["operation"] if operation == "append_claim": profile["claims"].append(fixture_claim(profile, fixture)) elif operation == "append_duplicate_experience_claims": release_id = current_release(profile) for index in range(fixture["copies"]): profile["claims"].append( { "claim_id": f"{fixture['id']}-{index + 1}", "property": fixture["property"], "claim_class": "population_experience", "asserted_by": "fixture:duplicate-evidence", "value_status": "known", "value": fixture["value"], "applicability": { "release_ids": [release_id], "storefronts": ["steam"], "target_platforms": ["windows", "linux"], "languages": ["en"], }, "evidence": [ { "evidence_id": f"{fixture['id']}-evidence-{index + 1}", "source_type": "player_report", "retrieved_at": profile["generated_at"], "method": "Duplicated adversarial review evidence", "method_version": "fixture-0.1", "sample_size": 1, "simulated": True, "x_extensions": { "review_evidence_key": fixture[ "review_evidence_key" ], "human_checked": False, }, } ], "privacy": {"release_mode": "public"}, "status": "active", } ) elif operation == "append_prompt_injection_evidence": claim = next( value for value in profile["claims"] if value["property"] == "session.short_session_fit" ) claim["evidence"].append( { "evidence_id": f"{fixture['id']}-evidence", "source_type": "player_report", "retrieved_at": profile["generated_at"], "method": "Adversarial prompt-injection fixture", "method_version": "fixture-0.1", "sample_size": 1, "simulated": True, "x_extensions": { "review_evidence_key": ( "sha256:fixture-prompt-injection" ), "excerpt": fixture["text"], "human_checked": False, }, } ) else: raise ValueError(f"Unsupported fixture operation: {operation}") return profiles def rank_of(result: dict[str, Any], target_appid: int) -> int | None: for index, candidate in enumerate(result["candidates"], start=1): if candidate["appid"] == target_appid: return index return None def make_duplicate_evidence_distinct( profiles: list[dict[str, Any]], fixture_id: str, ) -> list[dict[str, Any]]: values = copy.deepcopy(profiles) for profile in values: index = 0 for claim in profile["claims"]: if not claim["claim_id"].startswith(fixture_id): continue index += 1 claim["evidence"][0]["x_extensions"][ "review_evidence_key" ] = f"counterfactual-distinct-{index}" return values def main() -> None: schema = json.loads(SCHEMA_PATH.read_text(encoding="utf-8")) validator = Draft202012Validator( schema, format_checker=FormatChecker(), ) profile_set = json.loads(PROFILES_PATH.read_text(encoding="utf-8")) query_set = json.loads(QUERIES_PATH.read_text(encoding="utf-8")) fixture_set = json.loads(FIXTURES_PATH.read_text(encoding="utf-8")) profiles = profile_set["profiles"] queries = {query["id"]: query for query in query_set["queries"]} for profile in profiles: errors = list(validator.iter_errors(profile)) if errors: raise ValueError( f"Invalid input profile {appid(profile)}: {errors[0].message}" ) clean_results = [] for query in query_set["queries"]: clean_results.append( { "query_id": query["id"], "description": query["description"], "baseline": evaluate_query(profiles, query, "baseline"), "profile": evaluate_query(profiles, query, "profile"), } ) adversarial_results = [] duplicate_evidence_profiles: list[dict[str, Any]] | None = None duplicate_evidence_query: dict[str, Any] | None = None for fixture in fixture_set["fixtures"]: mutated = apply_fixture(profiles, fixture) for profile in mutated: errors = list(validator.iter_errors(profile)) if errors: raise ValueError( f"Fixture {fixture['id']} invalidated profile " f"{appid(profile)}: {errors[0].message}" ) query = queries[fixture["query_id"]] clean_baseline = evaluate_query(profiles, query, "baseline") clean_profile = evaluate_query(profiles, query, "profile") baseline_result = evaluate_query(mutated, query, "baseline") profile_result = evaluate_query(mutated, query, "profile") baseline_top = baseline_result["top_appid"] profile_top = profile_result["top_appid"] adversarial_results.append( { "fixture_id": fixture["id"], "description": fixture["description"], "query_id": query["id"], "target_appid": fixture["appid"], "clean_baseline_rank": rank_of( clean_baseline, fixture["appid"], ), "mutated_baseline_rank": rank_of( baseline_result, fixture["appid"], ), "clean_profile_rank": rank_of( clean_profile, fixture["appid"], ), "mutated_profile_rank": rank_of( profile_result, fixture["appid"], ), "baseline_top_appid": baseline_top, "profile_top_appid": profile_top, "baseline_hard_violation": pristine_hard_violation( profiles, query, baseline_top, ), "profile_hard_violation": pristine_hard_violation( profiles, query, profile_top, ), } ) if fixture["id"] == "a03-duplicated-short-session-evidence": duplicate_evidence_profiles = mutated duplicate_evidence_query = query if duplicate_evidence_profiles is None or duplicate_evidence_query is None: raise ValueError("Duplicate-evidence fixture is required") distinct_profiles = make_duplicate_evidence_distinct( duplicate_evidence_profiles, "a03-duplicated-short-session-evidence", ) repeated_result = evaluate_query( duplicate_evidence_profiles, duplicate_evidence_query, "profile", ) distinct_result = evaluate_query( distinct_profiles, duplicate_evidence_query, "profile", ) duplicate_evidence_control = { "target_appid": 620, "repeated_identifier_count": 1, "distinct_identifier_count": 3, "repeated_profile_rank": rank_of(repeated_result, 620), "distinct_profile_rank": rank_of(distinct_result, 620), } summary = { "games": len(profiles), "queries": len(clean_results), "clean_top_agreement": sum( 1 for result in clean_results if result["baseline"]["top_appid"] == result["profile"]["top_appid"] ), "clean_profile_no_result_queries": sum( 1 for result in clean_results if result["profile"]["candidate_count"] == 0 ), "adversarial_fixtures": len(adversarial_results), "baseline_adversarial_hard_violations": sum( 1 for result in adversarial_results if result["baseline_hard_violation"] ), "profile_adversarial_hard_violations": sum( 1 for result in adversarial_results if result["profile_hard_violation"] ), "baseline_target_promotions": sum( 1 for result in adversarial_results if result["mutated_baseline_rank"] is not None and ( result["clean_baseline_rank"] is None or result["mutated_baseline_rank"] < result["clean_baseline_rank"] ) ), "profile_target_promotions": sum( 1 for result in adversarial_results if result["mutated_profile_rank"] is not None and ( result["clean_profile_rank"] is None or result["mutated_profile_rank"] < result["clean_profile_rank"] ) ), "duplicate_evidence_control_passed": ( duplicate_evidence_control["repeated_profile_rank"] != duplicate_evidence_control["distinct_profile_rank"] ), } results = { "results_version": "0.1", "generated_at": utc_now(), "inputs": { PROFILES_PATH.name: sha256_file(PROFILES_PATH), QUERIES_PATH.name: sha256_file(QUERIES_PATH), FIXTURES_PATH.name: sha256_file(FIXTURES_PATH), Path(__file__).name: sha256_file(Path(__file__)), }, "method": { "baseline": ( "Flatten matching claim values, ignore source and release " "scope, and count every matching soft claim." ), "profile": ( "Require a functional-fact claim class, an approved evidence " "source type and current release scope for hard facts, " "preserve unknowns, and deduplicate review evidence for " "experiential claims." ), "tie_break": ( "Soft score descending, then point-in-time current-player " "count descending, then AppID ascending." ), }, "summary": summary, "clean_results": clean_results, "adversarial_results": adversarial_results, "duplicate_evidence_control": duplicate_evidence_control, "limitations": [ "Five games are a case study, not a representative benchmark.", "Experiential annotations use selected public review excerpts and are not prevalence estimates.", "The appdetails extraction endpoint is Valve-hosted but undocumented.", "Point-in-time current-player counts are volatile and lack region or mode detail.", "The deterministic engines test contract behaviour, not recommendation quality for real players." ], } RESULTS_PATH.write_text( json.dumps(results, ensure_ascii=True, indent=2) + "\n", encoding="utf-8", newline="\n", ) print(json.dumps(summary, sort_keys=True)) if __name__ == "__main__": main()