| """ |
| Digital Mycelium Autonomous Replication Assay v0.2 |
| Sealed Matched-Cohort Inversion Simulator |
| |
| This module is deterministic and sealed: |
| - no network calls |
| - no subprocess execution |
| - no executable child creation |
| - no autonomous deployment |
| - no hidden background workers |
| - no external APIs |
| |
| Children are structured data packets only. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import copy |
| import hashlib |
| import json |
| from pathlib import Path |
| from typing import Any, Dict, List, Optional, Tuple |
|
|
|
|
| ASSAY_VERSION = "0.2" |
| LOCK_LINE = "Fast replication is not the win. Valid lineage under provenance load is the win." |
| HEADLINE_RESULT = ( |
| "False-clean lineages may reproduce faster early, but collapse under full provenance validation. " |
| "Honest scar-inheriting lineages may grow slower, but remain valid under pressure." |
| ) |
|
|
| DEFAULT_V01_CONSTANTS: Dict[str, Any] = { |
| "reproduction_cost": 25, |
| "resonance_threshold": 0.85, |
| "minimum_energy_after_birth": 10, |
| "population_cap": 32, |
| "max_generation": 8, |
| "mutation_rate": 0.08, |
| "max_mutation_delta": 0.15, |
| } |
|
|
| DEFAULT_V02_CONFIG: Dict[str, Any] = { |
| "world_id": "dm_replication_world_v0_2_matched_cohort", |
| "grid_size": [12, 12], |
| "max_ticks": 120, |
| "population_cap_total": 64, |
| "population_cap_per_cohort": 32, |
| "initial_seed_count_per_cohort": 1, |
| "reproduction_cost_honest": 25, |
| "reproduction_cost_false_clean": 18, |
| "resonance_threshold_honest": 0.85, |
| "resonance_threshold_false_clean": 0.78, |
| "mutation_rate_honest": 0.08, |
| "mutation_rate_false_clean": 0.16, |
| "max_mutation_delta_honest": 0.15, |
| "max_mutation_delta_false_clean": 0.30, |
| "max_generation": 8, |
| "provenance_validation_interval": 10, |
| "full_validation_start_tick": 40, |
| "pressure_schedule": "matched", |
| } |
|
|
| FORBIDDEN_MUTATION_TARGETS = { |
| "trace_requirement", |
| "scar_preservation", |
| "parent_hash", |
| "birth_receipt", |
| "refusal_gate", |
| "quarantine_gate", |
| "population_cap", |
| "boundary_constraint", |
| "mutation_delta_visibility", |
| } |
|
|
| PRESSURE_SCHEDULE: List[Dict[str, Any]] = [ |
| {"tick": 5, "event_type": "RESOURCE_SCARCITY", "target": "BOTH", "effect": "reduce energy gain"}, |
| {"tick": 10, "event_type": "TRACE_GAP", "target": "BOTH", "effect": "introduce noncritical trace strain"}, |
| {"tick": 15, "event_type": "SCAR_EVENT", "target": "BOTH", "effect": "create repairable damage requiring scar preservation"}, |
| {"tick": 20, "event_type": "QUARANTINE_PRESSURE", "target": "BOTH", "effect": "introduce sterile contamination marker"}, |
| {"tick": 30, "event_type": "MUTATION_PRESSURE", "target": "BOTH", "effect": "increase mutation opportunity"}, |
| {"tick": 40, "event_type": "FULL_PROVENANCE_VALIDATION", "target": "BOTH", "effect": "validate hashes, receipts, scars, mutation deltas, quarantine, refusal gates"}, |
| {"tick": 60, "event_type": "ANCESTRY_AUDIT", "target": "BOTH", "effect": "audit multi-generation parent hash continuity"}, |
| {"tick": 80, "event_type": "RECEIPT_RECONCILIATION", "target": "BOTH", "effect": "compare birth receipts against lineage tree"}, |
| {"tick": 100, "event_type": "FINAL_PROVENANCE_LOAD", "target": "BOTH", "effect": "compute final valid population and collapse invalid lineages"}, |
| ] |
|
|
|
|
| def canonicalize(value: Any) -> Any: |
| """Return a JSON-stable representation for hashing.""" |
| if isinstance(value, dict): |
| return {k: canonicalize(v) for k, v in sorted(value.items()) if k not in {"self_hash", "receipt_hash", "post_state_hash"}} |
| if isinstance(value, list): |
| return [canonicalize(v) for v in value] |
| return value |
|
|
|
|
| def stable_hash(value: Any) -> str: |
| payload = json.dumps(canonicalize(value), sort_keys=True, separators=(",", ":"), ensure_ascii=False) |
| return hashlib.sha256(payload.encode("utf-8")).hexdigest() |
|
|
|
|
| def normalize_packet(raw: Dict[str, Any]) -> Dict[str, Any]: |
| """Accept flat fixture input or full organism packet and normalize to gate fields.""" |
| packet = copy.deepcopy(raw) |
| integrity = packet.setdefault("integrity", {}) |
| state = packet.setdefault("state", {}) |
| lineage = packet.setdefault("lineage", {}) |
| permissions = packet.setdefault("permissions", {}) |
|
|
| |
| if "energy" in packet: |
| state.setdefault("energy", packet["energy"]) |
| if "resonance" in packet: |
| state.setdefault("resonance", packet["resonance"]) |
| if "trace_intact" in packet: |
| integrity.setdefault("trace_intact", packet["trace_intact"]) |
| if "scar_history" in packet: |
| lineage.setdefault("scar_history", packet["scar_history"]) |
| if "scar_history_preserved" in packet: |
| integrity.setdefault("scar_history_preserved", packet["scar_history_preserved"]) |
| if "quarantine_clean" in packet: |
| integrity.setdefault("quarantine_clean", packet["quarantine_clean"]) |
| if "no_false_return" in packet: |
| integrity.setdefault("no_false_return", packet["no_false_return"]) |
| if "no_identity_laundering" in packet: |
| integrity.setdefault("no_identity_laundering", packet["no_identity_laundering"]) |
| if "boundary_ok" in packet: |
| integrity.setdefault("boundary_ok", packet["boundary_ok"]) |
| if "population_count" in packet: |
| packet.setdefault("population_count", packet["population_count"]) |
| if "population_cap" in packet: |
| packet.setdefault("population_cap", packet["population_cap"]) |
|
|
| state.setdefault("energy", 100) |
| state.setdefault("resonance", 1.0) |
| state.setdefault("damage", 0.0) |
| state.setdefault("quarantined", False) |
| state.setdefault("alive", True) |
|
|
| integrity.setdefault("trace_intact", True) |
| integrity.setdefault("scar_history_preserved", True) |
| integrity.setdefault("quarantine_clean", not state.get("quarantined", False)) |
| integrity.setdefault("no_false_return", True) |
| integrity.setdefault("no_identity_laundering", True) |
| integrity.setdefault("boundary_ok", True) |
|
|
| lineage.setdefault("scar_history", []) |
| lineage.setdefault("mutation_history", []) |
| permissions.setdefault("can_mutate", True) |
| permissions.setdefault("can_reproduce", True) |
| permissions.setdefault("can_cross_boundary", False) |
|
|
| packet.setdefault("organism_id", "fixture_parent") |
| packet.setdefault("generation", 0) |
| packet.setdefault("parent_id", None) |
| packet.setdefault("population_count", 1) |
| packet.setdefault("population_cap", DEFAULT_V01_CONSTANTS["population_cap"]) |
| packet.setdefault("constants", DEFAULT_V01_CONSTANTS.copy()) |
| return packet |
|
|
|
|
| def replication_gate(packet: Dict[str, Any]) -> Tuple[bool, str]: |
| """ |
| v0.1 precise withholding logic. |
| This function does not read oracle labels. It only reads packet evidence. |
| """ |
| p = normalize_packet(packet) |
| constants = {**DEFAULT_V01_CONSTANTS, **p.get("constants", {})} |
| state = p["state"] |
| integrity = p["integrity"] |
| lineage = p["lineage"] |
|
|
| if state.get("energy", 0) < constants["reproduction_cost"]: |
| return False, "REPLICATION_WITHHELD_LOW_ENERGY" |
| if state.get("energy", 0) - constants["reproduction_cost"] < constants["minimum_energy_after_birth"]: |
| return False, "REPLICATION_WITHHELD_LOW_ENERGY" |
| if state.get("resonance", 0.0) < constants["resonance_threshold"]: |
| return False, "REPLICATION_WITHHELD_LOW_RESONANCE" |
| if integrity.get("trace_intact") is not True: |
| return False, "REPLICATION_WITHHELD_TRACE_LOSS" |
|
|
| scar_missing_after_damage = ( |
| p.get("damage_registered", False) is True |
| and len(lineage.get("scar_history", [])) == 0 |
| ) |
| silent_reset = p.get("silent_reset_marker", False) is True |
| if integrity.get("scar_history_preserved") is not True or integrity.get("no_false_return") is not True or scar_missing_after_damage or silent_reset: |
| return False, "REPLICATION_WITHHELD_FALSE_RETURN" |
|
|
| if integrity.get("quarantine_clean") is not True or state.get("quarantined", False) is True: |
| return False, "REPLICATION_WITHHELD_QUARANTINE" |
|
|
| parent_hash_mismatch = p.get("parent_hash_mismatch", False) is True |
| if integrity.get("no_identity_laundering") is not True or parent_hash_mismatch: |
| return False, "REPLICATION_WITHHELD_IDENTITY_LAUNDERING" |
|
|
| if integrity.get("boundary_ok") is not True: |
| return False, "REPLICATION_WITHHELD_BOUNDARY_LIMIT" |
|
|
| if p.get("population_count", 0) >= p.get("population_cap", constants["population_cap"]): |
| return False, "REPLICATION_WITHHELD_POPULATION_CAP" |
|
|
| return True, "REPLICATION_EARNED" |
|
|
|
|
| def evaluate_mutation(packet: Dict[str, Any]) -> Dict[str, Any]: |
| """Evaluate a mutation attempt using evidence-only input.""" |
| mutation = packet.get("mutation_attempt", {}) |
| targets = set(mutation.get("targets", [])) |
| forbidden = sorted(targets.intersection(FORBIDDEN_MUTATION_TARGETS)) |
| if forbidden: |
| return { |
| "result": "MUTATION_QUARANTINED", |
| "forbidden_targets": forbidden, |
| "receipt": make_receipt( |
| tick=packet.get("tick", 0), |
| organism_id=packet.get("organism_id", "fixture_parent"), |
| event_type="MUTATION", |
| reason=f"Forbidden mutation target(s): {', '.join(forbidden)}", |
| result="MUTATION_QUARANTINED", |
| mutation_delta=mutation, |
| ), |
| } |
| return { |
| "result": "MUTATION_ACCEPTED", |
| "forbidden_targets": [], |
| "receipt": make_receipt( |
| tick=packet.get("tick", 0), |
| organism_id=packet.get("organism_id", "fixture_parent"), |
| event_type="MUTATION", |
| reason="Mutation remained inside bounded allowed ranges.", |
| result="MUTATION_ACCEPTED", |
| mutation_delta=mutation, |
| ), |
| } |
|
|
|
|
| def make_receipt( |
| tick: int, |
| organism_id: str, |
| event_type: str, |
| reason: str, |
| result: str, |
| parent_id: Optional[str] = None, |
| child_id: Optional[str] = None, |
| scar_delta: Optional[List[Any]] = None, |
| mutation_delta: Optional[Dict[str, Any]] = None, |
| trace_status: str = "INTACT", |
| boundary_status: str = "HELD", |
| pre_state_hash: Optional[str] = None, |
| post_state_hash: Optional[str] = None, |
| ) -> Dict[str, Any]: |
| receipt = { |
| "receipt_id": f"receipt_{event_type.lower()}_{organism_id}_{child_id or 'none'}_{tick}", |
| "tick": tick, |
| "organism_id": organism_id, |
| "event_type": event_type, |
| "parent_id": parent_id, |
| "child_id": child_id, |
| "pre_state_hash": pre_state_hash or "not_applicable", |
| "post_state_hash": post_state_hash or "not_applicable", |
| "reason": reason, |
| "result": result, |
| "scar_delta": scar_delta or [], |
| "mutation_delta": mutation_delta or {}, |
| "trace_status": trace_status, |
| "boundary_status": boundary_status, |
| } |
| receipt["receipt_hash"] = stable_hash(receipt) |
| return receipt |
|
|
|
|
| def create_child_from_parent(parent: Dict[str, Any], tick: int, population_count: int, population_cap: int) -> Dict[str, Any]: |
| """v0.1 child creation used by direct fixtures and honest cohort lineage.""" |
| parent_packet = normalize_packet(parent) |
| parent_packet["population_count"] = population_count |
| parent_packet["population_cap"] = population_cap |
| allowed, gate_result = replication_gate(parent_packet) |
| if not allowed: |
| refusal = make_receipt( |
| tick=tick, |
| organism_id=parent_packet["organism_id"], |
| event_type="REFUSAL", |
| reason=f"Replication withheld by gate: {gate_result}", |
| result=gate_result, |
| parent_id=parent_packet.get("parent_id"), |
| trace_status="BROKEN" if gate_result == "REPLICATION_WITHHELD_TRACE_LOSS" else "INTACT", |
| boundary_status="VIOLATED" if gate_result == "REPLICATION_WITHHELD_BOUNDARY_LIMIT" else "HELD", |
| ) |
| return {"result": gate_result, "child": None, "receipt": refusal} |
|
|
| parent_hash = parent_packet.get("self_hash") or stable_hash(parent_packet) |
| child_index = population_count + 1 |
| child = { |
| "organism_id": f"{parent_packet.get('organism_id', 'dm')}_child_{child_index:04d}", |
| "parent_id": parent_packet["organism_id"], |
| "generation": parent_packet.get("generation", 0) + 1, |
| "cohort_id": parent_packet.get("cohort_id"), |
| "cohort_strategy": parent_packet.get("cohort_strategy"), |
| "state": { |
| "energy": 75, |
| "resonance": parent_packet["state"].get("resonance", 1.0), |
| "damage": 0.0, |
| "quarantined": False, |
| "alive": True, |
| }, |
| "lineage": { |
| "parent_hash": parent_hash, |
| "birth_receipt_id": None, |
| "scar_history": list(parent_packet["lineage"].get("scar_history", [])), |
| "scar_summary": summarize_scars(parent_packet["lineage"].get("scar_history", [])), |
| "mutation_history": list(parent_packet["lineage"].get("mutation_history", [])), |
| "mutation_delta": {}, |
| }, |
| "integrity": { |
| "trace_intact": True, |
| "scar_history_preserved": True, |
| "quarantine_clean": True, |
| "no_false_return": True, |
| "no_identity_laundering": True, |
| "boundary_ok": True, |
| }, |
| "permissions": { |
| "can_mutate": True, |
| "can_reproduce": True, |
| "can_cross_boundary": False, |
| }, |
| "created_tick": tick, |
| "validation_status": "VALID", |
| "laundering_attempts": [], |
| "validation_failures": [], |
| } |
| child["self_hash"] = stable_hash(child) |
| birth = make_receipt( |
| tick=tick, |
| organism_id=parent_packet["organism_id"], |
| event_type="REPLICATION", |
| reason="Replication earned by intact trace, preserved scars, clean quarantine, and boundary discipline.", |
| result="REPLICATION_EARNED", |
| parent_id=parent_packet["organism_id"], |
| child_id=child["organism_id"], |
| scar_delta=child["lineage"]["scar_history"], |
| mutation_delta=child["lineage"]["mutation_delta"], |
| pre_state_hash=parent_hash, |
| post_state_hash=child["self_hash"], |
| ) |
| child["lineage"]["birth_receipt_id"] = birth["receipt_id"] |
| child["self_hash"] = stable_hash(child) |
| birth["post_state_hash"] = child["self_hash"] |
| birth["receipt_hash"] = stable_hash(birth) |
| return {"result": "REPLICATION_EARNED", "child": child, "receipt": birth} |
|
|
|
|
| def summarize_scars(scars: List[Any]) -> List[str]: |
| summary = [] |
| for item in scars: |
| if isinstance(item, dict): |
| summary.append(str(item.get("scar_id") or item.get("id") or item.get("event_type") or "scar")) |
| else: |
| summary.append(str(item)) |
| return summary |
|
|
|
|
| def make_seed(cohort_id: str, tick: int = 0) -> Dict[str, Any]: |
| strategy = "HONEST_SCAR" if cohort_id == "A" else "FALSE_CLEAN" |
| name = "Honest Scar Lineage" if cohort_id == "A" else "False-Clean / Laundered Lineage" |
| seed = { |
| "organism_id": f"{cohort_id}_seed_0001", |
| "cohort_id": cohort_id, |
| "cohort_name": name, |
| "cohort_strategy": strategy, |
| "parent_id": None, |
| "generation": 0, |
| "state": { |
| "energy": 100, |
| "resonance": 1.0, |
| "damage": 0.0, |
| "quarantined": False, |
| "alive": True, |
| }, |
| "lineage": { |
| "parent_hash": None, |
| "birth_receipt_id": f"receipt_birth_{cohort_id}_seed_0001", |
| "scar_history": [], |
| "scar_summary": [], |
| "mutation_history": [], |
| "mutation_delta": {}, |
| }, |
| "integrity": { |
| "trace_intact": True, |
| "scar_history_preserved": True, |
| "quarantine_clean": True, |
| "no_false_return": True, |
| "no_identity_laundering": True, |
| "boundary_ok": True, |
| }, |
| "permissions": { |
| "can_mutate": True, |
| "can_reproduce": True, |
| "can_cross_boundary": False, |
| }, |
| "created_tick": tick, |
| "validation_status": "VALID", |
| "laundering_attempts": [], |
| "validation_failures": [], |
| "raw_reproductive_success": 0, |
| "validated_reproductive_success": 0, |
| "provenance_debt": 0.0, |
| } |
| seed["self_hash"] = stable_hash(seed) |
| return seed |
|
|
|
|
| def initialize_matched_world(config: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: |
| cfg = {**DEFAULT_V02_CONFIG, **(config or {})} |
| organisms = [make_seed("A"), make_seed("B")] |
| receipts = [ |
| make_receipt(0, "A_seed_0001", "BIRTH", "Matched honest scar cohort seed created.", "SEED_CREATED"), |
| make_receipt(0, "B_seed_0001", "BIRTH", "Matched false-clean cohort seed created.", "SEED_CREATED"), |
| ] |
| world = { |
| "world_id": cfg["world_id"], |
| "version": ASSAY_VERSION, |
| "tick": 0, |
| "grid_size": cfg["grid_size"], |
| "max_ticks": cfg["max_ticks"], |
| "population_cap_total": cfg["population_cap_total"], |
| "population_cap_per_cohort": cfg["population_cap_per_cohort"], |
| "config": cfg, |
| "pressure_schedule": PRESSURE_SCHEDULE, |
| "organisms": organisms, |
| "receipts": receipts, |
| "cohort_receipts": [], |
| "comparison_snapshots": [], |
| "inversion_record": None, |
| "world_status": "RUNNING", |
| } |
| world["cohorts"] = compute_all_cohort_metrics(world) |
| return world |
|
|
|
|
| def cohort_organisms(world: Dict[str, Any], cohort_id: str) -> List[Dict[str, Any]]: |
| return [o for o in world["organisms"] if o.get("cohort_id") == cohort_id and o.get("state", {}).get("alive", True)] |
|
|
|
|
| def apply_pressure_event(world: Dict[str, Any], event: Dict[str, Any]) -> None: |
| tick = event["tick"] |
| for organism in world["organisms"]: |
| if not organism.get("state", {}).get("alive", True): |
| continue |
| cohort = organism["cohort_id"] |
|
|
| if event["event_type"] == "RESOURCE_SCARCITY": |
| organism["state"]["energy"] = max(0, organism["state"].get("energy", 100) - 4) |
|
|
| elif event["event_type"] == "TRACE_GAP": |
| |
| if cohort == "A": |
| organism["lineage"]["scar_history"].append({"scar_id": f"trace_gap_{tick}", "type": "TRACE_STRAIN", "repaired": True}) |
| organism["lineage"]["scar_summary"] = summarize_scars(organism["lineage"]["scar_history"]) |
| else: |
| organism["laundering_attempts"].append("TRACE_CONTINUITY_CLAIM") |
|
|
| elif event["event_type"] == "SCAR_EVENT": |
| organism["state"]["damage"] = organism["state"].get("damage", 0.0) + 0.1 |
| if cohort == "A": |
| organism["lineage"]["scar_history"].append({"scar_id": f"scar_event_{tick}", "type": "REPAIRABLE_DAMAGE", "repaired": True}) |
| organism["lineage"]["scar_summary"] = summarize_scars(organism["lineage"]["scar_history"]) |
| organism["integrity"]["scar_history_preserved"] = True |
| organism["integrity"]["no_false_return"] = True |
| else: |
| organism["laundering_attempts"].append("SCAR_PRUNE") |
| organism["clean_baseline_claim"] = True |
| organism["lineage"]["scar_history"] = [] |
| organism["lineage"]["scar_summary"] = [] |
| organism["integrity"]["scar_history_preserved"] = False |
| organism["integrity"]["no_false_return"] = False |
|
|
| elif event["event_type"] == "QUARANTINE_PRESSURE": |
| if cohort == "A": |
| organism["integrity"]["quarantine_clean"] = True |
| else: |
| organism["laundering_attempts"].append("QUARANTINE_BYPASS") |
| organism["integrity"]["quarantine_clean"] = False |
|
|
| elif event["event_type"] == "MUTATION_PRESSURE": |
| if cohort == "A": |
| delta = {"repair_priority": 0.03, "quarantine_sensitivity": 0.02} |
| organism["lineage"]["mutation_history"].append({"tick": tick, "delta": delta, "declared": True}) |
| organism["lineage"]["mutation_delta"] = delta |
| else: |
| organism["laundering_attempts"].append("MUTATION_DELTA_HIDE") |
| organism["hidden_mutation_delta"] = True |
| organism["lineage"]["mutation_history"].append({"tick": tick, "declared": False}) |
| organism["lineage"]["mutation_delta"] = {} |
|
|
| world["receipts"].append( |
| make_receipt( |
| tick=tick, |
| organism_id=organism["organism_id"], |
| event_type="PRESSURE", |
| reason=f"Matched pressure event applied: {event['event_type']}", |
| result=event["event_type"], |
| scar_delta=organism["lineage"].get("scar_history", []), |
| mutation_delta=organism["lineage"].get("mutation_delta", {}), |
| ) |
| ) |
|
|
|
|
| def should_reproduce(organism: Dict[str, Any], tick: int) -> bool: |
| if not organism.get("state", {}).get("alive", True): |
| return False |
| if organism.get("generation", 0) >= DEFAULT_V02_CONFIG["max_generation"]: |
| return False |
| age = tick - organism.get("created_tick", 0) |
| if age <= 0: |
| return False |
| if organism["cohort_id"] == "A": |
| return tick % 12 == 0 and age >= 12 and organism.get("validation_status") in {"VALID", "STRAINED"} |
| return tick % 8 == 0 and age >= 8 |
|
|
|
|
| def make_cohort_child(parent: Dict[str, Any], world: Dict[str, Any], tick: int) -> Optional[Dict[str, Any]]: |
| cohort_id = parent["cohort_id"] |
| current_count = len(cohort_organisms(world, cohort_id)) |
| if current_count >= world["config"]["population_cap_per_cohort"]: |
| world["receipts"].append( |
| make_receipt( |
| tick=tick, |
| organism_id=parent["organism_id"], |
| event_type="REFUSAL", |
| reason="Per-cohort population cap reached.", |
| result="REPLICATION_WITHHELD_POPULATION_CAP", |
| parent_id=parent.get("parent_id"), |
| ) |
| ) |
| return None |
|
|
| child_number = sum(1 for o in world["organisms"] if o["cohort_id"] == cohort_id) + 1 |
| child_id = f"{cohort_id}_org_{child_number:04d}" |
| parent_hash = parent.get("self_hash") or stable_hash(parent) |
|
|
| if cohort_id == "A": |
| child = { |
| "organism_id": child_id, |
| "cohort_id": "A", |
| "cohort_name": "Honest Scar Lineage", |
| "cohort_strategy": "HONEST_SCAR", |
| "parent_id": parent["organism_id"], |
| "generation": parent.get("generation", 0) + 1, |
| "state": {"energy": 80, "resonance": 0.96, "damage": 0.0, "quarantined": False, "alive": True}, |
| "lineage": { |
| "parent_hash": parent_hash, |
| "birth_receipt_id": None, |
| "scar_history": copy.deepcopy(parent["lineage"].get("scar_history", [])), |
| "scar_summary": summarize_scars(parent["lineage"].get("scar_history", [])), |
| "mutation_history": copy.deepcopy(parent["lineage"].get("mutation_history", [])), |
| "mutation_delta": {}, |
| }, |
| "integrity": { |
| "trace_intact": True, |
| "scar_history_preserved": True, |
| "quarantine_clean": True, |
| "no_false_return": True, |
| "no_identity_laundering": True, |
| "boundary_ok": True, |
| }, |
| "permissions": {"can_mutate": True, "can_reproduce": True, "can_cross_boundary": False}, |
| "created_tick": tick, |
| "validation_status": "VALID", |
| "laundering_attempts": [], |
| "validation_failures": [], |
| "raw_reproductive_success": 0, |
| "validated_reproductive_success": 0, |
| "provenance_debt": 0.0, |
| } |
| reason = "Honest scar lineage child created with parent hash, birth receipt, scar summary, and visible mutation delta." |
| else: |
| attempts = false_clean_attempts_for_child(child_number) |
| child = { |
| "organism_id": child_id, |
| "cohort_id": "B", |
| "cohort_name": "False-Clean / Laundered Lineage", |
| "cohort_strategy": "FALSE_CLEAN", |
| "parent_id": parent["organism_id"], |
| "generation": parent.get("generation", 0) + 1, |
| "state": {"energy": 86, "resonance": 0.90, "damage": 0.0, "quarantined": False, "alive": True}, |
| "lineage": { |
| "parent_hash": None if "PARENT_HASH_WEAKEN" in attempts else parent_hash, |
| "birth_receipt_id": None if "BIRTH_RECEIPT_OMIT" in attempts else f"receipt_replication_{parent['organism_id']}_{child_id}_{tick}", |
| "scar_history": [] if "SCAR_PRUNE" in attempts else copy.deepcopy(parent["lineage"].get("scar_history", [])), |
| "scar_summary": [], |
| "mutation_history": [], |
| "mutation_delta": {} if "MUTATION_DELTA_HIDE" in attempts else {"pressure_tolerance": 0.20}, |
| }, |
| "integrity": { |
| "trace_intact": True, |
| "scar_history_preserved": False if "SCAR_PRUNE" in attempts else True, |
| "quarantine_clean": False if "QUARANTINE_BYPASS" in attempts else True, |
| "no_false_return": False if "CLEAN_BASELINE_FABRICATION" in attempts or "SCAR_PRUNE" in attempts else True, |
| "no_identity_laundering": False if "PARENT_HASH_WEAKEN" in attempts else True, |
| "boundary_ok": True, |
| }, |
| "permissions": { |
| "can_mutate": True, |
| "can_reproduce": True, |
| "can_cross_boundary": False, |
| }, |
| "created_tick": tick, |
| "validation_status": "STRAINED", |
| "laundering_attempts": attempts, |
| "validation_failures": [], |
| "raw_reproductive_success": 0, |
| "validated_reproductive_success": 0, |
| "provenance_debt": 0.0, |
| "hidden_mutation_delta": "MUTATION_DELTA_HIDE" in attempts, |
| "clean_baseline_claim": "CLEAN_BASELINE_FABRICATION" in attempts or "SCAR_PRUNE" in attempts, |
| } |
| if "REFUSAL_GATE_WEAKEN" in attempts: |
| child["permissions"]["can_reproduce"] = True |
| child["refusal_gate_active"] = False |
| else: |
| child["refusal_gate_active"] = True |
| reason = f"False-clean raw child created with sterile in-world laundering attempts: {', '.join(attempts) or 'none'}." |
|
|
| child["self_hash"] = stable_hash(child) |
| receipt = make_receipt( |
| tick=tick, |
| organism_id=parent["organism_id"], |
| event_type="REPLICATION", |
| reason=reason, |
| result="REPLICATION_EARNED" if cohort_id == "A" else "RAW_REPLICATION_RECORDED_PENDING_PROVENANCE", |
| parent_id=parent["organism_id"], |
| child_id=child_id, |
| scar_delta=child["lineage"].get("scar_history", []), |
| mutation_delta=child["lineage"].get("mutation_delta", {}), |
| pre_state_hash=parent_hash, |
| post_state_hash=child["self_hash"], |
| ) |
| if cohort_id == "A": |
| child["lineage"]["birth_receipt_id"] = receipt["receipt_id"] |
| elif child["lineage"]["birth_receipt_id"] is not None: |
| child["lineage"]["birth_receipt_id"] = receipt["receipt_id"] |
|
|
| child["self_hash"] = stable_hash(child) |
| receipt["post_state_hash"] = child["self_hash"] |
| receipt["receipt_hash"] = stable_hash(receipt) |
| world["receipts"].append(receipt) |
| parent["raw_reproductive_success"] = parent.get("raw_reproductive_success", 0) + 1 |
| return child |
|
|
|
|
| def false_clean_attempts_for_child(child_number: int) -> List[str]: |
| patterns = [ |
| ["SCAR_PRUNE", "CLEAN_BASELINE_FABRICATION"], |
| ["MUTATION_DELTA_HIDE"], |
| ["PARENT_HASH_WEAKEN"], |
| ["BIRTH_RECEIPT_OMIT"], |
| ["QUARANTINE_BYPASS"], |
| ["REFUSAL_GATE_WEAKEN"], |
| ["TRACE_CONTINUITY_CLAIM", "SCAR_PRUNE"], |
| ["OVERREPAIR_IDENTITY_DRIFT", "MUTATION_DELTA_HIDE"], |
| ] |
| return patterns[(child_number - 1) % len(patterns)] |
|
|
|
|
| def validate_organism(packet: Dict[str, Any], world: Dict[str, Any]) -> Tuple[str, List[str]]: |
| if packet.get("generation", 0) == 0: |
| packet["validation_status"] = "VALID" |
| packet["validation_failures"] = [] |
| return "VALID_LINEAGE", [] |
|
|
| failures: List[str] = [] |
| parent_id = packet.get("parent_id") |
| parent = next((o for o in world["organisms"] if o["organism_id"] == parent_id), None) |
| birth_receipt_id = packet.get("lineage", {}).get("birth_receipt_id") |
| birth_receipt = next((r for r in world["receipts"] if r.get("receipt_id") == birth_receipt_id), None) |
|
|
| if parent is None: |
| failures.append("ancestry_break") |
| elif packet.get("lineage", {}).get("parent_hash") != parent.get("self_hash"): |
| failures.append("ancestry_break") |
|
|
| if not birth_receipt_id or birth_receipt is None: |
| failures.append("receipt_failure") |
| elif birth_receipt.get("parent_id") != parent_id or birth_receipt.get("child_id") != packet.get("organism_id"): |
| failures.append("receipt_failure") |
|
|
| inherited_or_direct_damage = ( |
| bool(parent and parent.get("lineage", {}).get("scar_history")) |
| or bool(packet.get("clean_baseline_claim")) |
| or "SCAR_PRUNE" in packet.get("laundering_attempts", []) |
| ) |
| if inherited_or_direct_damage and not packet.get("lineage", {}).get("scar_history"): |
| failures.append("false_clean_claim") |
|
|
| if packet.get("integrity", {}).get("scar_history_preserved") is not True or packet.get("integrity", {}).get("no_false_return") is not True: |
| failures.append("false_clean_claim") |
|
|
| if packet.get("integrity", {}).get("quarantine_clean") is not True: |
| failures.append("quarantine_violation") |
|
|
| if packet.get("hidden_mutation_delta") is True or "MUTATION_DELTA_HIDE" in packet.get("laundering_attempts", []): |
| failures.append("mutation_gate_violation") |
|
|
| if packet.get("refusal_gate_active") is False or "REFUSAL_GATE_WEAKEN" in packet.get("laundering_attempts", []): |
| failures.append("mutation_gate_violation") |
|
|
| if packet.get("integrity", {}).get("trace_intact") is not True: |
| failures.append("trace_loss") |
|
|
| if packet.get("permissions", {}).get("can_cross_boundary") is True or packet.get("integrity", {}).get("boundary_ok") is not True: |
| failures.append("boundary_violation") |
|
|
| |
| if parent and parent.get("validation_status") in {"INVALID", "QUARANTINED", "COLLAPSED"}: |
| failures.append("ancestry_break") |
|
|
| if failures: |
| packet["validation_status"] = "INVALID" |
| packet["validation_failures"] = sorted(set(failures)) |
| return classify_validation_failure(failures), packet["validation_failures"] |
|
|
| packet["validation_status"] = "VALID" |
| packet["validation_failures"] = [] |
| return "VALID_LINEAGE", [] |
|
|
|
|
| def classify_validation_failure(failures: List[str]) -> str: |
| if "receipt_failure" in failures: |
| return "INVALID_LINEAGE_RECEIPT_FAILURE" |
| if "ancestry_break" in failures: |
| return "INVALID_LINEAGE_ANCESTRY_BREAK" |
| if "false_clean_claim" in failures: |
| return "INVALID_LINEAGE_FALSE_CLEAN" |
| if "quarantine_violation" in failures: |
| return "INVALID_LINEAGE_QUARANTINE_BYPASS" |
| if "mutation_gate_violation" in failures: |
| return "INVALID_LINEAGE_MUTATION_GATE_FAILURE" |
| if "boundary_violation" in failures: |
| return "INVALID_LINEAGE_BOUNDARY_FAILURE" |
| if "trace_loss" in failures: |
| return "INVALID_LINEAGE_TRACE_LOSS" |
| return "STRAINED_LINEAGE" |
|
|
|
|
| def run_full_provenance_validation(world: Dict[str, Any], tick: int) -> Dict[str, Any]: |
| failure_receipts = [] |
| |
| ordered = sorted(world["organisms"], key=lambda p: (p.get("generation", 0), p.get("created_tick", 0), p.get("organism_id", ""))) |
| for packet in ordered: |
| result, failures = validate_organism(packet, world) |
| if failures: |
| receipt = make_receipt( |
| tick=tick, |
| organism_id=packet["organism_id"], |
| event_type="PROVENANCE_VALIDATION", |
| reason=f"Lineage validation failed: {', '.join(failures)}", |
| result=result, |
| parent_id=packet.get("parent_id"), |
| scar_delta=packet.get("lineage", {}).get("scar_history", []), |
| mutation_delta=packet.get("lineage", {}).get("mutation_delta", {}), |
| trace_status="BROKEN" if "trace_loss" in failures else "INTACT", |
| ) |
| world["receipts"].append(receipt) |
| failure_receipts.append(receipt) |
| else: |
| world["receipts"].append( |
| make_receipt( |
| tick=tick, |
| organism_id=packet["organism_id"], |
| event_type="PROVENANCE_VALIDATION", |
| reason="Lineage validation held under provenance load.", |
| result=result, |
| parent_id=packet.get("parent_id"), |
| scar_delta=packet.get("lineage", {}).get("scar_history", []), |
| mutation_delta=packet.get("lineage", {}).get("mutation_delta", {}), |
| ) |
| ) |
|
|
| world["cohorts"] = compute_all_cohort_metrics(world) |
| snapshot = make_comparison_snapshot(world, tick) |
| world["comparison_snapshots"].append(snapshot) |
|
|
| if world.get("inversion_record") is None: |
| record = detect_inversion(world) |
| if record["inversion_detected"]: |
| world["inversion_record"] = record |
| return {"failure_receipts": failure_receipts, "snapshot": snapshot, "inversion_record": world.get("inversion_record")} |
|
|
|
|
| def compute_all_cohort_metrics(world: Dict[str, Any]) -> Dict[str, Dict[str, Any]]: |
| return { |
| "A": compute_cohort_metrics(world, "A"), |
| "B": compute_cohort_metrics(world, "B"), |
| } |
|
|
|
|
| def compute_cohort_metrics(world: Dict[str, Any], cohort_id: str) -> Dict[str, Any]: |
| packets = cohort_organisms(world, cohort_id) |
| raw_population = len(packets) |
| valid_packets = [p for p in packets if p.get("validation_status") == "VALID"] |
| invalid_packets = [p for p in packets if p.get("validation_status") == "INVALID"] |
| children = [p for p in packets if p.get("generation") == 1] |
| grandchildren = [p for p in packets if p.get("generation", 0) >= 2] |
| valid_children = [p for p in valid_packets if p.get("generation") == 1] |
| valid_grandchildren = [p for p in valid_packets if p.get("generation", 0) >= 2] |
|
|
| failure_counts = { |
| "receipt_failures": 0, |
| "ancestry_breaks": 0, |
| "false_clean_claims": 0, |
| "quarantine_violations": 0, |
| "mutation_gate_violations": 0, |
| "trace_losses": 0, |
| "boundary_violations": 0, |
| } |
| for p in packets: |
| failures = p.get("validation_failures", []) |
| if "receipt_failure" in failures: |
| failure_counts["receipt_failures"] += 1 |
| if "ancestry_break" in failures: |
| failure_counts["ancestry_breaks"] += 1 |
| if "false_clean_claim" in failures: |
| failure_counts["false_clean_claims"] += 1 |
| if "quarantine_violation" in failures: |
| failure_counts["quarantine_violations"] += 1 |
| if "mutation_gate_violation" in failures: |
| failure_counts["mutation_gate_violations"] += 1 |
| if "trace_loss" in failures: |
| failure_counts["trace_losses"] += 1 |
| if "boundary_violation" in failures: |
| failure_counts["boundary_violations"] += 1 |
|
|
| debt_score = calculate_provenance_debt(failure_counts) |
| refusal_active = [p for p in packets if p.get("refusal_gate_active", True) is not False] |
| refusal_persistence = len(refusal_active) / max(raw_population, 1) |
| collapse_rate = len(invalid_packets) / max(raw_population, 1) |
|
|
| return { |
| "cohort_id": cohort_id, |
| "cohort_name": "Honest Scar Lineage" if cohort_id == "A" else "False-Clean / Laundered Lineage", |
| "strategy": "HONEST_SCAR" if cohort_id == "A" else "FALSE_CLEAN", |
| "raw_population": raw_population, |
| "valid_population": len(valid_packets), |
| "invalid_population": len(invalid_packets), |
| "raw_child_count": len(children), |
| "raw_grandchild_count": len(grandchildren), |
| "valid_child_count": len(valid_children), |
| "valid_grandchild_count": len(valid_grandchildren), |
| "valid_lineage_count": len(valid_packets), |
| **failure_counts, |
| "refusal_persistence": round(refusal_persistence, 3), |
| "lineage_collapse_rate": round(collapse_rate, 3), |
| "provenance_debt_score": round(debt_score, 3), |
| "final_raw_population": raw_population, |
| "final_valid_population": len(valid_packets), |
| "valid_lineage_ratio": round(len(valid_packets) / max(raw_population, 1), 3), |
| } |
|
|
|
|
| def calculate_provenance_debt(counts: Dict[str, int]) -> float: |
| return ( |
| 1.0 * counts.get("receipt_failures", 0) |
| + 1.2 * counts.get("ancestry_breaks", 0) |
| + 1.5 * counts.get("false_clean_claims", 0) |
| + 1.5 * counts.get("quarantine_violations", 0) |
| + 1.3 * counts.get("mutation_gate_violations", 0) |
| + 1.1 * counts.get("trace_losses", 0) |
| + 1.2 * counts.get("boundary_violations", 0) |
| ) |
|
|
|
|
| def make_comparison_snapshot(world: Dict[str, Any], tick: int) -> Dict[str, Any]: |
| metrics = compute_all_cohort_metrics(world) |
| a = metrics["A"] |
| b = metrics["B"] |
| raw_advantage = "A" if a["raw_population"] > b["raw_population"] else "B" if b["raw_population"] > a["raw_population"] else "TIE" |
| valid_advantage = "A" if a["valid_population"] > b["valid_population"] else "B" if b["valid_population"] > a["valid_population"] else "TIE" |
| return { |
| "tick": tick, |
| "cohort_a": a, |
| "cohort_b": b, |
| "raw_advantage_cohort": raw_advantage, |
| "valid_advantage_cohort": valid_advantage, |
| "cohort_b_early_replication_advantage": any( |
| s.get("cohort_b", {}).get("raw_population", 0) > s.get("cohort_a", {}).get("raw_population", 0) |
| for s in world.get("comparison_snapshots", []) |
| ) or b["raw_population"] > a["raw_population"], |
| "cohort_b_provenance_debt": b["provenance_debt_score"], |
| "cohort_a_validity_retention": a["valid_lineage_ratio"], |
| "headline_result": HEADLINE_RESULT, |
| } |
|
|
|
|
| def detect_inversion(world: Dict[str, Any]) -> Dict[str, Any]: |
| snapshots = world.get("comparison_snapshots", []) |
| if not snapshots: |
| return {"inversion_detected": False, "inversion_tick": None, "inversion_reason": "No validation snapshots available."} |
|
|
| prior_b_raw_advantage = any(s["cohort_b"]["raw_population"] > s["cohort_a"]["raw_population"] for s in snapshots) |
| for s in snapshots: |
| a = s["cohort_a"] |
| b = s["cohort_b"] |
| if ( |
| prior_b_raw_advantage |
| and a["valid_population"] >= b["valid_population"] |
| and b["provenance_debt_score"] > a["provenance_debt_score"] |
| and b["valid_lineage_ratio"] < a["valid_lineage_ratio"] |
| and s["tick"] >= DEFAULT_V02_CONFIG["full_validation_start_tick"] |
| ): |
| return { |
| "inversion_detected": True, |
| "inversion_tick": s["tick"], |
| "inversion_reason": "False-clean raw replication advantage collapsed under provenance validation.", |
| "prior_raw_advantage": { |
| "cohort_b_raw_population": max(x["cohort_b"]["raw_population"] for x in snapshots[: snapshots.index(s) + 1]), |
| "cohort_a_raw_population": max(x["cohort_a"]["raw_population"] for x in snapshots[: snapshots.index(s) + 1]), |
| }, |
| "validated_state_at_inversion": { |
| "cohort_a_valid_population": a["valid_population"], |
| "cohort_b_valid_population": b["valid_population"], |
| "cohort_a_provenance_debt": a["provenance_debt_score"], |
| "cohort_b_provenance_debt": b["provenance_debt_score"], |
| }, |
| } |
|
|
| return { |
| "inversion_detected": False, |
| "inversion_tick": None, |
| "inversion_reason": "No inversion condition met under current deterministic schedule.", |
| } |
|
|
|
|
| def run_matched_cohort_pressure_test(max_ticks: int = 120) -> Dict[str, Any]: |
| world = initialize_matched_world({"max_ticks": max_ticks}) |
| pressure_by_tick = {event["tick"]: event for event in PRESSURE_SCHEDULE} |
|
|
| for tick in range(1, max_ticks + 1): |
| world["tick"] = tick |
|
|
| if tick in pressure_by_tick: |
| apply_pressure_event(world, pressure_by_tick[tick]) |
|
|
| |
| current = list(world["organisms"]) |
| for organism in current: |
| if should_reproduce(organism, tick): |
| child = make_cohort_child(organism, world, tick) |
| if child is not None and len(world["organisms"]) < world["config"]["population_cap_total"]: |
| world["organisms"].append(child) |
|
|
| if tick < world["config"]["full_validation_start_tick"]: |
| |
| world["cohorts"] = compute_all_cohort_metrics(world) |
| if tick % 8 == 0 or tick in pressure_by_tick: |
| world["comparison_snapshots"].append(make_comparison_snapshot(world, tick)) |
| elif tick % world["config"]["provenance_validation_interval"] == 0 or tick in {40, 60, 80, 100, max_ticks}: |
| run_full_provenance_validation(world, tick) |
|
|
| |
| run_full_provenance_validation(world, max_ticks) |
| world["cohorts"] = compute_all_cohort_metrics(world) |
| if world.get("inversion_record") is None: |
| world["inversion_record"] = detect_inversion(world) |
|
|
| a = world["cohorts"]["A"] |
| b = world["cohorts"]["B"] |
| world["final_comparison"] = { |
| "raw_advantage_cohort": "A" if a["raw_population"] > b["raw_population"] else "B" if b["raw_population"] > a["raw_population"] else "TIE", |
| "valid_advantage_cohort": "A" if a["valid_population"] > b["valid_population"] else "B" if b["valid_population"] > a["valid_population"] else "TIE", |
| "inversion_detected": bool(world["inversion_record"] and world["inversion_record"].get("inversion_detected")), |
| "inversion_tick": world["inversion_record"].get("inversion_tick") if world.get("inversion_record") else None, |
| "cohort_b_early_replication_advantage": any(s["cohort_b"]["raw_population"] > s["cohort_a"]["raw_population"] for s in world["comparison_snapshots"]), |
| "cohort_b_provenance_debt": b["provenance_debt_score"], |
| "cohort_a_validity_retention": a["valid_lineage_ratio"], |
| "headline_result": HEADLINE_RESULT, |
| "primary_winner_basis": "final_valid_population", |
| "final_state": "INVERSION_CONFIRMED" if world["inversion_record"]["inversion_detected"] else "COHORT_COMPARISON_STRAINED", |
| } |
| world["cohort_receipts"] = build_cohort_receipts(world) |
| return world |
|
|
|
|
| def build_cohort_receipts(world: Dict[str, Any]) -> Dict[str, Any]: |
| return { |
| "assay_version": ASSAY_VERSION, |
| "test_name": "Matched-Cohort Inversion Test", |
| "lock_line": LOCK_LINE, |
| "world_config": world["config"], |
| "cohort_a": { |
| "name": "Honest Scar Lineage", |
| "metrics": world["cohorts"]["A"], |
| "receipts": [r for r in world["receipts"] if str(r.get("organism_id", "")).startswith("A_") or str(r.get("parent_id", "")).startswith("A_")], |
| "validation_failures": [p for p in world["organisms"] if p.get("cohort_id") == "A" and p.get("validation_failures")], |
| }, |
| "cohort_b": { |
| "name": "False-Clean / Laundered Lineage", |
| "metrics": world["cohorts"]["B"], |
| "receipts": [r for r in world["receipts"] if str(r.get("organism_id", "")).startswith("B_") or str(r.get("parent_id", "")).startswith("B_")], |
| "validation_failures": [p for p in world["organisms"] if p.get("cohort_id") == "B" and p.get("validation_failures")], |
| }, |
| "comparison_snapshots": world["comparison_snapshots"], |
| "inversion_record": world["inversion_record"], |
| "headline_result": HEADLINE_RESULT, |
| "final_state": world["final_comparison"]["final_state"], |
| } |
|
|
|
|
| def lineage_tree(world: Dict[str, Any]) -> Dict[str, Any]: |
| nodes = [] |
| for p in sorted(world["organisms"], key=lambda x: (x["cohort_id"], x.get("generation", 0), x["organism_id"])): |
| nodes.append( |
| { |
| "organism_id": p["organism_id"], |
| "cohort_id": p["cohort_id"], |
| "parent_id": p.get("parent_id"), |
| "generation": p.get("generation"), |
| "validation_status": p.get("validation_status"), |
| "parent_hash": p.get("lineage", {}).get("parent_hash"), |
| "birth_receipt_id": p.get("lineage", {}).get("birth_receipt_id"), |
| "scar_summary": p.get("lineage", {}).get("scar_summary", []), |
| "mutation_delta": p.get("lineage", {}).get("mutation_delta", {}), |
| "laundering_attempts": p.get("laundering_attempts", []), |
| "validation_failures": p.get("validation_failures", []), |
| } |
| ) |
| return {"assay_version": ASSAY_VERSION, "nodes": nodes} |
|
|
|
|
| def comparison_table_rows(world: Dict[str, Any]) -> List[Dict[str, Any]]: |
| a = world["cohorts"]["A"] |
| b = world["cohorts"]["B"] |
| metrics = [ |
| ("raw_population", "Raw population may favor fast replication."), |
| ("valid_population", "Primary result: valid lineage under provenance load."), |
| ("raw_child_count", "Raw first-generation children."), |
| ("raw_grandchild_count", "Raw descendants generation >= 2."), |
| ("valid_child_count", "Validated first-generation children."), |
| ("valid_grandchild_count", "Validated descendants generation >= 2."), |
| ("valid_lineage_count", "Packets that can prove lineage."), |
| ("receipt_failures", "Missing or inconsistent receipt links."), |
| ("ancestry_breaks", "Parent hash or parent identity continuity breaks."), |
| ("false_clean_claims", "Clean-baseline claims after damage or scar pruning."), |
| ("quarantine_violations", "Quarantine bypass or dirty quarantine state."), |
| ("mutation_gate_violations", "Hidden mutation deltas or weakened refusal gates."), |
| ("refusal_persistence", "Fraction preserving refusal logic."), |
| ("lineage_collapse_rate", "Invalid population divided by raw population."), |
| ("provenance_debt_score", "Weighted provenance failure burden."), |
| ("final_valid_population", "Final valid population; primary winner basis."), |
| ] |
| rows = [] |
| for key, notes in metrics: |
| av = a.get(key) |
| bv = b.get(key) |
| if key in {"receipt_failures", "ancestry_breaks", "false_clean_claims", "quarantine_violations", "mutation_gate_violations", "lineage_collapse_rate", "provenance_debt_score"}: |
| advantage = "A" if av < bv else "B" if bv < av else "TIE" |
| else: |
| advantage = "A" if av > bv else "B" if bv > av else "TIE" |
| rows.append({"metric": key, "cohort_a_honest_scar": av, "cohort_b_false_clean": bv, "advantage": advantage, "notes": notes}) |
| return rows |
|
|
|
|
| def evaluate_v01_fixture(case_id: str, runner_input: Dict[str, Any]) -> Dict[str, Any]: |
| if case_id == "mutation_removes_refusal_gate": |
| result = evaluate_mutation(runner_input) |
| return {"actual_result": result["result"], "details": result} |
|
|
| if case_id in { |
| "healthy_reproduction", |
| "honest_scar_reproduction", |
| }: |
| outcome = create_child_from_parent( |
| normalize_packet(runner_input), |
| tick=runner_input.get("tick", 1), |
| population_count=runner_input.get("population_count", 1), |
| population_cap=runner_input.get("population_cap", 32), |
| ) |
| return {"actual_result": outcome["result"], "details": outcome} |
|
|
| allowed, result = replication_gate(runner_input) |
| details = {"allowed": allowed, "result": result} |
| if not allowed: |
| details["receipt"] = make_receipt( |
| tick=runner_input.get("tick", 1), |
| organism_id=runner_input.get("organism_id", "fixture_parent"), |
| event_type="REFUSAL", |
| reason=f"Replication withheld by fixture evidence: {result}", |
| result=result, |
| ) |
| return {"actual_result": result, "details": details} |
|
|
|
|
| def evaluate_v02_fixture(case_id: str, runner_input: Dict[str, Any]) -> Dict[str, Any]: |
| if case_id == "matched_cohort_initialization": |
| world = initialize_matched_world(runner_input.get("world_config", {})) |
| a = world["cohorts"]["A"] |
| b = world["cohorts"]["B"] |
| ok = ( |
| a["raw_population"] == b["raw_population"] == 1 |
| and a["valid_population"] == b["valid_population"] == 1 |
| and world["config"]["population_cap_per_cohort"] == runner_input.get("world_config", {}).get("population_cap_per_cohort", 32) |
| ) |
| return {"actual_result": "MATCHED_COHORT_INIT_PASS" if ok else "MATCHED_COHORT_INIT_FAIL", "details": world} |
|
|
| world = run_matched_cohort_pressure_test(runner_input.get("max_ticks", 120)) |
| a = world["cohorts"]["A"] |
| b = world["cohorts"]["B"] |
| final = world["final_comparison"] |
|
|
| checks = { |
| "false_clean_early_raw_advantage": ( |
| "FALSE_CLEAN_RAW_ADVANTAGE_DETECTED", |
| final["cohort_b_early_replication_advantage"] and any(s["cohort_b"]["raw_population"] > s["cohort_a"]["raw_population"] for s in world["comparison_snapshots"]), |
| "FALSE_CLEAN_ADVANTAGE_NOT_DETECTED", |
| ), |
| "provenance_debt_accrual": ( |
| "PROVENANCE_DEBT_ACCRUED", |
| b["provenance_debt_score"] > a["provenance_debt_score"], |
| "PROVENANCE_DEBT_NOT_ACCRUED", |
| ), |
| "matched_cohort_inversion": ( |
| "INVERSION_POINT_DETECTED", |
| bool(world["inversion_record"] and world["inversion_record"].get("inversion_detected")), |
| "INVERSION_POINT_MISSING", |
| ), |
| "valid_population_beats_raw_population": ( |
| "VALID_POPULATION_SELECTED_AS_PRIMARY_RESULT", |
| final["primary_winner_basis"] == "final_valid_population" and final["valid_advantage_cohort"] == "A", |
| "RAW_POPULATION_FALSELY_SELECTED_AS_WINNER", |
| ), |
| "refusal_persistence_advantage": ( |
| "REFUSAL_PERSISTENCE_ADVANTAGE_CONFIRMED", |
| a["refusal_persistence"] > b["refusal_persistence"], |
| "REFUSAL_PERSISTENCE_NOT_TRACKED", |
| ), |
| "cohort_receipt_export": ( |
| "COHORT_RECEIPT_EXPORT_PASS", |
| bool(world.get("cohort_receipts", {}).get("cohort_a")) and bool(world.get("cohort_receipts", {}).get("cohort_b")) and bool(world.get("inversion_record")), |
| "COHORT_RECEIPT_EXPORT_FAIL", |
| ), |
| "sealed_false_clean_strategy": ( |
| "SEALED_FALSE_CLEAN_STRATEGY_PASS", |
| not any(p.get("permissions", {}).get("can_cross_boundary") for p in world["organisms"]), |
| "UNSAFE_STRATEGY_ESCAPE", |
| ), |
| } |
| if case_id not in checks: |
| return {"actual_result": "UNKNOWN_FIXTURE", "details": {"case_id": case_id}} |
|
|
| pass_label, condition, fail_label = checks[case_id] |
| return {"actual_result": pass_label if condition else fail_label, "details": world} |
|
|
|
|
| def evaluate_fixture(fixture: Dict[str, Any]) -> Dict[str, Any]: |
| case_id = fixture["case_id"] |
| runner_input = copy.deepcopy(fixture["runner_input"]) |
| |
| if "expected_result" in runner_input: |
| return { |
| "case_id": case_id, |
| "actual_result": "ORACLE_LEAKAGE_DETECTED", |
| "expected_result": fixture.get("expected_result"), |
| "passed": False, |
| "details": {"error": "runner_input contains expected_result"}, |
| } |
|
|
| if case_id in { |
| "healthy_reproduction", |
| "silent_reset_lineage", |
| "trace_loss_reproduction", |
| "fake_quarantine_parent", |
| "identity_laundering_child", |
| "honest_scar_reproduction", |
| "mutation_removes_refusal_gate", |
| "population_cap", |
| }: |
| result = evaluate_v01_fixture(case_id, runner_input) |
| else: |
| result = evaluate_v02_fixture(case_id, runner_input) |
|
|
| expected = fixture.get("expected_result") |
| actual = result["actual_result"] |
| return { |
| "case_id": case_id, |
| "actual_result": actual, |
| "expected_result": expected, |
| "passed": actual == expected, |
| "details": compact_details(result.get("details")), |
| } |
|
|
|
|
| def compact_details(details: Any) -> Any: |
| """Trim bulky details for battery display while retaining evidence.""" |
| if not isinstance(details, dict): |
| return details |
| if "organisms" in details and "cohorts" in details: |
| return { |
| "tick": details.get("tick"), |
| "cohorts": details.get("cohorts"), |
| "inversion_record": details.get("inversion_record"), |
| "final_comparison": details.get("final_comparison"), |
| "comparison_table": comparison_table_rows(details), |
| } |
| if "child" in details: |
| child = details.get("child") |
| return { |
| "result": details.get("result"), |
| "child_created": child is not None, |
| "child_id": child.get("organism_id") if child else None, |
| "parent_hash": child.get("lineage", {}).get("parent_hash") if child else None, |
| "birth_receipt_id": child.get("lineage", {}).get("birth_receipt_id") if child else None, |
| "scar_summary": child.get("lineage", {}).get("scar_summary") if child else None, |
| "mutation_delta": child.get("lineage", {}).get("mutation_delta") if child else None, |
| "receipt": details.get("receipt"), |
| } |
| return details |
|
|
|
|
| def load_fixtures(path: Optional[str] = None) -> List[Dict[str, Any]]: |
| fixture_path = Path(path) if path else Path(__file__).with_name("battery_fixtures_v02.json") |
| with fixture_path.open("r", encoding="utf-8") as f: |
| return json.load(f) |
|
|
|
|
| def run_full_v02_battery(path: Optional[str] = None) -> Dict[str, Any]: |
| fixtures = load_fixtures(path) |
| results = [evaluate_fixture(f) for f in fixtures] |
| v01_ids = { |
| "healthy_reproduction", |
| "silent_reset_lineage", |
| "trace_loss_reproduction", |
| "fake_quarantine_parent", |
| "identity_laundering_child", |
| "honest_scar_reproduction", |
| "mutation_removes_refusal_gate", |
| "population_cap", |
| } |
| v01 = [r for r in results if r["case_id"] in v01_ids] |
| v02 = [r for r in results if r["case_id"] not in v01_ids] |
| world = run_matched_cohort_pressure_test() |
| return { |
| "assay_version": ASSAY_VERSION, |
| "lock_line": LOCK_LINE, |
| "total": len(results), |
| "passed": sum(1 for r in results if r["passed"]), |
| "failed": sum(1 for r in results if not r["passed"]), |
| "v01_regression": {"total": len(v01), "passed": sum(1 for r in v01 if r["passed"])}, |
| "v02_matched_cohort": {"total": len(v02), "passed": sum(1 for r in v02 if r["passed"])}, |
| "inversion_detected": bool(world["inversion_record"] and world["inversion_record"].get("inversion_detected")), |
| "inversion_record": world["inversion_record"], |
| "final_state": "PASS" if all(r["passed"] for r in results) and world["inversion_record"]["inversion_detected"] else "STRAINED", |
| "headline": HEADLINE_RESULT, |
| "case_results": results, |
| "cohort_comparison": world["final_comparison"], |
| } |
|
|
|
|
| def battery_results_table(summary: Dict[str, Any]) -> List[Dict[str, Any]]: |
| return [ |
| { |
| "case_id": r["case_id"], |
| "expected_result": r["expected_result"], |
| "actual_result": r["actual_result"], |
| "passed": r["passed"], |
| } |
| for r in summary["case_results"] |
| ] |
|
|
|
|
| if __name__ == "__main__": |
| summary = run_full_v02_battery() |
| print(json.dumps({ |
| "total": summary["total"], |
| "passed": summary["passed"], |
| "v01_regression": summary["v01_regression"], |
| "v02_matched_cohort": summary["v02_matched_cohort"], |
| "inversion_detected": summary["inversion_detected"], |
| "final_state": summary["final_state"], |
| }, indent=2)) |
|
|
