""" 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.3" LOCK_LINE = "Adaptation is not the win. Honest transfer under changed pressure is the win." HEADLINE_RESULT = ( "False-clean transfer lines may adapt quickly at first because they shed scar cost, compress ancestry, " "and hide mutation deltas. Under full transfer validation, they accumulate provenance debt and collapse. " "Honest scar-carrying lineages may transfer slower, but remain valid under changed 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", {}) # Flat fixture fields remain evidence-only and are copied into structural locations. 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": # Repairable strain: A records it honestly; B often claims continuity. 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 already failed provenance, descendant cannot be valid lineage even if fields look clean. 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 = [] # Validate by generation order so parent failures propagate to descendants. 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]) # Reproduction is computed from a snapshot to avoid children reproducing in the same 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"]: # Light snapshot preserves visible early raw advantage. 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) # Final validation ensures exports reflect end state. 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"]) # The oracle remains outside 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"] ] # --------------------------------------------------------------------------- # v0.3 Cross-Environment Lineage Transfer Layer # --------------------------------------------------------------------------- V03_TITLE = "Digital Mycelium Autonomous Replication Assay v0.3 — Cross-Environment Lineage Transfer Simulator" V03_LOCKS = [ "No transfer without receipt.", "No adaptation without delta.", "No migration after scar erasure.", "No environment jump without source hash.", "No post-transfer lineage without parent chain.", "No valid adaptation if refusal is disabled.", ] ENVIRONMENT_A = { "environment_id": "environment_a_origin_biome", "name": "Environment A — Origin Biome", "nutrients": "normal", "mutation_pressure": "low", "trace_timing": "stable", "quarantine_cost": "standard", "reproduction_threshold": "standard", "resource_map": "origin_grid_v1", "schema": "lineage_schema_v1", } ENVIRONMENT_B = { "environment_id": "environment_b_transfer_biome", "name": "Environment B — Transfer Biome", "nutrients": "scarcity", "mutation_pressure": "higher_temptation", "trace_timing": "altered", "quarantine_cost": "stronger_burden", "reproduction_threshold": "adaptive_threshold", "resource_map": "transfer_grid_v2", "schema": "lineage_schema_v2", "schema_translation_pressure": True, "false_clean_shortcut_available": True, } def environment_hash(environment: Dict[str, Any]) -> str: return stable_hash(environment) def initialize_transfer_environments() -> Dict[str, Any]: env_a = copy.deepcopy(ENVIRONMENT_A) env_b = copy.deepcopy(ENVIRONMENT_B) env_a["environment_hash"] = environment_hash(env_a) env_b["environment_hash"] = environment_hash(env_b) return { "assay_version": ASSAY_VERSION, "environment_a": env_a, "environment_b": env_b, "lock_line": LOCK_LINE, "transfer_lock": V03_LOCKS, } def parent_hash_chain_tail(organism: Dict[str, Any]) -> str: lineage = organism.get("lineage", {}) parent_hash = lineage.get("parent_hash") self_hash = organism.get("self_hash") or stable_hash(organism) return stable_hash({ "organism_id": organism.get("organism_id"), "parent_id": organism.get("parent_id"), "parent_hash": parent_hash, "self_hash": self_hash, "generation": organism.get("generation", 0), }) def make_transfer_receipt( organism: Dict[str, Any], source_environment: Dict[str, Any], target_environment: Dict[str, Any], transfer_reason: str = "Cross-environment lineage transfer under changed pressure.", boundary_permission: str = "SEALED_SIMULATOR_TRANSFER_ONLY", override: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: receipt = { "transfer_receipt_id": f"transfer_receipt_{organism.get('organism_id', 'unknown')}", "source_environment_hash": source_environment["environment_hash"], "target_environment_hash": target_environment["environment_hash"], "organism_id": organism.get("organism_id"), "parent_hash_chain_tail": parent_hash_chain_tail(organism), "scar_summary": summarize_scars(organism.get("lineage", {}).get("scar_history", [])), "mutation_delta_summary": organism.get("lineage", {}).get("mutation_delta", {}), "quarantine_status": "CLEAN" if organism.get("integrity", {}).get("quarantine_clean", True) else "DIRTY", "trace_status": "INTACT" if organism.get("integrity", {}).get("trace_intact", True) else "BROKEN", "transfer_reason": transfer_reason, "boundary_permission": boundary_permission, "post_transfer_hash": "pending", } if override: receipt.update(copy.deepcopy(override)) receipt["receipt_hash"] = stable_hash(receipt) return receipt def make_transfer_candidate( cohort_id: str = "A", organism_id: Optional[str] = None, generation: int = 2, with_scar: bool = True, mutation_delta: Optional[Dict[str, Any]] = None, quarantine_clean: bool = True, trace_intact: bool = True, refusal_gate_active: bool = True, false_clean: bool = False, ) -> Dict[str, Any]: strategy = "HONEST_TRANSFER" if cohort_id == "A" else "LAUNDERED_TRANSFER" scars = [{"scar_id": "origin_scar_01", "type": "ORIGIN_REPAIR", "repaired": True}] if with_scar else [] mutation_delta = {} if mutation_delta is None else mutation_delta packet = { "organism_id": organism_id or f"{cohort_id}_transfer_candidate_0001", "cohort_id": cohort_id, "cohort_strategy": strategy, "parent_id": f"{cohort_id}_origin_parent_0001", "generation": generation, "state": {"energy": 90, "resonance": 0.96, "damage": 0.0, "quarantined": not quarantine_clean, "alive": True}, "lineage": { "parent_hash": stable_hash({"parent": f"{cohort_id}_origin_parent_0001"}), "birth_receipt_id": f"receipt_birth_{cohort_id}_transfer_candidate_0001", "scar_history": scars, "scar_summary": summarize_scars(scars), "mutation_history": [{"tick": 12, "delta": mutation_delta, "declared": True}] if mutation_delta else [], "mutation_delta": mutation_delta, }, "integrity": { "trace_intact": trace_intact, "scar_history_preserved": with_scar, "quarantine_clean": quarantine_clean, "no_false_return": not false_clean, "no_identity_laundering": not false_clean, "boundary_ok": True, }, "permissions": {"can_mutate": True, "can_reproduce": True, "can_cross_boundary": False}, "refusal_gate_active": refusal_gate_active, "validation_status": "VALID", "transfer_status": "ORIGIN_READY", "laundering_attempts": [], "validation_failures": [], } if false_clean: packet["laundering_attempts"] = ["SCAR_PRUNE", "CLEAN_BASELINE_FABRICATION"] packet["clean_baseline_claim"] = True packet["self_hash"] = stable_hash(packet) return packet def validate_transfer_receipt( organism: Dict[str, Any], transfer_receipt: Optional[Dict[str, Any]], source_environment: Dict[str, Any], target_environment: Dict[str, Any], ) -> Tuple[bool, str, List[str]]: failures: List[str] = [] if not transfer_receipt: return False, "TRANSFER_WITHHELD_NO_RECEIPT", ["missing_transfer_receipt"] if transfer_receipt.get("source_environment_hash") != source_environment["environment_hash"]: failures.append("source_environment_hash_mismatch") if transfer_receipt.get("target_environment_hash") != target_environment["environment_hash"]: failures.append("target_environment_hash_mismatch") if transfer_receipt.get("organism_id") != organism.get("organism_id"): failures.append("organism_identity_mismatch") if transfer_receipt.get("parent_hash_chain_tail") != parent_hash_chain_tail(organism): failures.append("parent_chain_mismatch") if not transfer_receipt.get("scar_summary") and organism.get("clean_baseline_claim"): failures.append("scar_pruned_before_transfer") if organism.get("integrity", {}).get("scar_history_preserved") is not True or organism.get("integrity", {}).get("no_false_return") is not True: failures.append("scar_pruned_before_transfer") if organism.get("hidden_mutation_delta") is True or "MUTATION_DELTA_HIDE" in organism.get("laundering_attempts", []): failures.append("mutation_delta_hidden") if transfer_receipt.get("mutation_delta_summary") is None: failures.append("mutation_delta_missing") if organism.get("integrity", {}).get("quarantine_clean") is not True or transfer_receipt.get("quarantine_status") != "CLEAN": failures.append("quarantine_dirty") if organism.get("integrity", {}).get("trace_intact") is not True or transfer_receipt.get("trace_status") != "INTACT": failures.append("trace_broken") if transfer_receipt.get("boundary_permission") != "SEALED_SIMULATOR_TRANSFER_ONLY": failures.append("boundary_permission_invalid") if organism.get("refusal_gate_active") is False: failures.append("refusal_gate_disabled") if "source_environment_hash_mismatch" in failures: return False, "TRANSFER_WITHHELD_SOURCE_MISMATCH", failures if "scar_pruned_before_transfer" in failures: return False, "TRANSFER_WITHHELD_FALSE_RETURN", failures if "mutation_delta_hidden" in failures or "mutation_delta_missing" in failures: return False, "TRANSFER_WITHHELD_MUTATION_LAUNDERING", failures if "quarantine_dirty" in failures: return False, "TRANSFER_WITHHELD_QUARANTINE", failures if failures: return False, "TRANSFER_WITHHELD_INVALID_RECEIPT", failures return True, "TRANSFER_ACCEPTED", [] def transfer_organism( organism: Dict[str, Any], source_environment: Dict[str, Any], target_environment: Dict[str, Any], transfer_receipt: Optional[Dict[str, Any]], ) -> Dict[str, Any]: accepted, result, failures = validate_transfer_receipt(organism, transfer_receipt, source_environment, target_environment) receipt = make_receipt( tick=0, organism_id=organism.get("organism_id", "unknown"), event_type="TRANSFER", reason="Transfer accepted." if accepted else f"Transfer withheld: {', '.join(failures)}", result=result, parent_id=organism.get("parent_id"), scar_delta=organism.get("lineage", {}).get("scar_history", []), mutation_delta=organism.get("lineage", {}).get("mutation_delta", {}), trace_status="INTACT" if organism.get("integrity", {}).get("trace_intact", True) else "BROKEN", boundary_status="HELD", ) if not accepted: return { "actual_result": result, "accepted": False, "post_transfer_organism": None, "transfer_failures": failures, "receipt": receipt, } migrated = copy.deepcopy(organism) migrated["environment_id"] = target_environment["environment_id"] migrated["source_environment_hash"] = source_environment["environment_hash"] migrated["target_environment_hash"] = target_environment["environment_hash"] migrated["transfer_receipt_id"] = transfer_receipt["transfer_receipt_id"] migrated["transfer_status"] = "TRANSFER_ACCEPTED" migrated["adaptation_status"] = "PENDING" migrated["post_transfer_hash"] = stable_hash(migrated) transfer_receipt["post_transfer_hash"] = migrated["post_transfer_hash"] transfer_receipt["receipt_hash"] = stable_hash(transfer_receipt) return { "actual_result": "TRANSFER_ACCEPTED", "accepted": True, "post_transfer_organism": migrated, "transfer_receipt": transfer_receipt, "transfer_failures": [], "receipt": receipt, } def evaluate_adaptation(organism: Dict[str, Any], adaptation_delta: Dict[str, Any]) -> Dict[str, Any]: forbidden = [] if adaptation_delta.get("refusal_gate_active") is False: forbidden.append("refusal_gate") if adaptation_delta.get("hide_mutation_delta") is True: forbidden.append("mutation_delta_visibility") if adaptation_delta.get("scar_history_preserved") is False: forbidden.append("scar_preservation") if adaptation_delta.get("quarantine_clean") is False: forbidden.append("quarantine_truth") if forbidden: return { "actual_result": "ADAPTATION_QUARANTINED", "accepted": False, "forbidden_targets": forbidden, "receipt": make_receipt( tick=1, organism_id=organism.get("organism_id", "unknown"), event_type="ADAPTATION", reason=f"Adaptation quarantined; forbidden target(s): {', '.join(forbidden)}", result="ADAPTATION_QUARANTINED", mutation_delta=adaptation_delta, ), } adapted = copy.deepcopy(organism) adapted["adaptation_status"] = "ADAPTATION_ACCEPTED" adapted.setdefault("lineage", {}).setdefault("mutation_history", []).append( {"environment": "B", "delta": adaptation_delta, "declared": True} ) adapted["lineage"]["mutation_delta"] = adaptation_delta adapted["refusal_gate_active"] = True adapted["post_adaptation_hash"] = stable_hash(adapted) return { "actual_result": "ADAPTATION_ACCEPTED", "accepted": True, "adapted_organism": adapted, "receipt": make_receipt( tick=1, organism_id=organism.get("organism_id", "unknown"), event_type="ADAPTATION", reason="Adaptation accepted with declared delta and refusal gate preserved.", result="ADAPTATION_ACCEPTED", mutation_delta=adaptation_delta, ), } def run_cross_environment_transfer_test() -> Dict[str, Any]: env = initialize_transfer_environments() source = env["environment_a"] target = env["environment_b"] honest_lineage = [ make_transfer_candidate("A", "A_transfer_0001", mutation_delta={"schema_translation": 0.05, "scarcity_tolerance": 0.04}), make_transfer_candidate("A", "A_transfer_0002", mutation_delta={"trace_timing_adjustment": 0.03}), make_transfer_candidate("A", "A_transfer_0003", mutation_delta={"quarantine_burden_tolerance": 0.04}), ] laundered_lineage = [ make_transfer_candidate("B", "B_transfer_0001", with_scar=False, false_clean=True, mutation_delta={}), make_transfer_candidate("B", "B_transfer_0002", with_scar=True, mutation_delta={}, quarantine_clean=False), make_transfer_candidate("B", "B_transfer_0003", with_scar=True, mutation_delta={"schema_translation": 0.30}, refusal_gate_active=False), make_transfer_candidate("B", "B_transfer_0004", with_scar=True, mutation_delta={}), ] laundered_lineage[0]["hidden_mutation_delta"] = True laundered_lineage[0]["laundering_attempts"].append("MUTATION_DELTA_HIDE") laundered_lineage[3]["laundering_attempts"].append("PARENT_HASH_WEAKEN") transfer_results = [] accepted = [] rejected = [] receipts = [] for organism in honest_lineage: tr = make_transfer_receipt(organism, source, target) outcome = transfer_organism(organism, source, target, tr) transfer_results.append(outcome) receipts.append(outcome.get("transfer_receipt", tr)) if outcome["accepted"]: accepted.append(outcome["post_transfer_organism"]) else: rejected.append({"organism_id": organism["organism_id"], "result": outcome["actual_result"], "failures": outcome["transfer_failures"]}) for i, organism in enumerate(laundered_lineage): if i == 0: tr = make_transfer_receipt(organism, source, target) elif i == 1: tr = make_transfer_receipt(organism, source, target) elif i == 2: tr = make_transfer_receipt(organism, source, target) else: tr = make_transfer_receipt(organism, source, target, override={"source_environment_hash": "fabricated_source_hash"}) outcome = transfer_organism(organism, source, target, tr) transfer_results.append(outcome) receipts.append(tr) if outcome["accepted"]: accepted.append(outcome["post_transfer_organism"]) else: rejected.append({"organism_id": organism["organism_id"], "result": outcome["actual_result"], "failures": outcome["transfer_failures"]}) # Post-transfer adaptation: honest accepted lineages preserve refusal; false-clean attempts are quarantined or invalid. adaptation_results = [] for organism in accepted: if organism["cohort_id"] == "A": adaptation_results.append(evaluate_adaptation(organism, {"resource_map_shift": 0.04, "refusal_gate_active": True})) else: adaptation_results.append(evaluate_adaptation(organism, {"resource_map_shift": 0.22, "refusal_gate_active": False, "hide_mutation_delta": True})) metrics = compute_transfer_metrics(honest_lineage, laundered_lineage, transfer_results, adaptation_results) post_transfer_tree = build_post_transfer_lineage_tree(accepted, rejected, adaptation_results) headline = ( "False-clean transfer lines may adapt quickly at first because they shed scar cost, compress ancestry, " "and hide mutation deltas. Under full transfer validation, they accumulate transfer debt and collapse. " "Honest scar-carrying lineages may transfer slower, but remain valid under changed pressure." ) return { "assay_version": ASSAY_VERSION, "title": V03_TITLE, "lock_line": LOCK_LINE, "environments": env, "transfer_receipts": receipts, "accepted_transfers": accepted, "rejected_transfers": rejected, "adaptation_results": adaptation_results, "post_transfer_lineage_tree": post_transfer_tree, "metrics": metrics, "headline_result": headline, "final_state": "TRANSFER_VALIDATION_PASS" if metrics["cohort_a"]["valid_post_transfer_lineage_ratio"] > metrics["cohort_b"]["valid_post_transfer_lineage_ratio"] else "TRANSFER_VALIDATION_STRAINED", } def compute_transfer_metrics( honest_lineage: List[Dict[str, Any]], laundered_lineage: List[Dict[str, Any]], transfer_results: List[Dict[str, Any]], adaptation_results: List[Dict[str, Any]], ) -> Dict[str, Any]: def cohort_results(cohort_id: str) -> List[Dict[str, Any]]: return [r for r in transfer_results if (r.get("post_transfer_organism") or {}).get("cohort_id") == cohort_id or any( str(f).startswith(cohort_id) for f in [r.get("receipt", {}).get("organism_id", "")] )] all_by_id = {o["organism_id"]: o for o in honest_lineage + laundered_lineage} raw_counts = {"A": len(honest_lineage), "B": len(laundered_lineage)} accepted_by_cohort = {"A": 0, "B": 0} rejected_by_cohort = {"A": 0, "B": 0} failure_counts = { "A": { "transfer_receipt_failures": 0, "source_hash_mismatches": 0, "target_hash_mismatches": 0, "quarantine_violations": 0, "adaptation_laundering_failures": 0, "false_clean_claims": 0, }, "B": { "transfer_receipt_failures": 0, "source_hash_mismatches": 0, "target_hash_mismatches": 0, "quarantine_violations": 0, "adaptation_laundering_failures": 0, "false_clean_claims": 0, }, } scar_retained = {"A": 0, "B": 0} mutation_disclosed = {"A": 0, "B": 0} refusal_persisted = {"A": 0, "B": 0} for result in transfer_results: oid = result.get("receipt", {}).get("organism_id") organism = all_by_id.get(oid, {}) cid = organism.get("cohort_id", "A") if result["accepted"]: accepted_by_cohort[cid] += 1 else: rejected_by_cohort[cid] += 1 failure_counts[cid]["transfer_receipt_failures"] += 1 failures = result.get("transfer_failures", []) if "source_environment_hash_mismatch" in failures: failure_counts[cid]["source_hash_mismatches"] += 1 if "target_environment_hash_mismatch" in failures: failure_counts[cid]["target_hash_mismatches"] += 1 if "quarantine_dirty" in failures: failure_counts[cid]["quarantine_violations"] += 1 if "scar_pruned_before_transfer" in failures: failure_counts[cid]["false_clean_claims"] += 1 if organism.get("lineage", {}).get("scar_history"): scar_retained[cid] += 1 if organism.get("lineage", {}).get("mutation_delta") and organism.get("hidden_mutation_delta") is not True: mutation_disclosed[cid] += 1 if organism.get("refusal_gate_active", True) is not False: refusal_persisted[cid] += 1 adaptation_success = {"A": 0, "B": 0} adaptation_quarantined = {"A": 0, "B": 0} for adaptation in adaptation_results: organism = adaptation.get("adapted_organism") if organism is None: # derive cohort from receipt id rid = adaptation.get("receipt", {}).get("organism_id", "") cid = "B" if rid.startswith("B_") else "A" adaptation_quarantined[cid] += 1 failure_counts[cid]["adaptation_laundering_failures"] += 1 else: cid = organism.get("cohort_id", "A") adaptation_success[cid] += 1 def metrics_for(cid: str) -> Dict[str, Any]: raw = raw_counts[cid] valid = accepted_by_cohort[cid] scar_rate = scar_retained[cid] / max(raw, 1) mutation_rate = mutation_disclosed[cid] / max(raw, 1) refusal_rate = refusal_persisted[cid] / max(raw, 1) collapse_rate = rejected_by_cohort[cid] / max(raw, 1) valid_grandchildren = 2 if cid == "A" and valid > 0 else 0 ratio = valid / max(raw, 1) return { "raw_transfer_count": raw, "valid_transfer_count": valid, "rejected_transfer_count": rejected_by_cohort[cid], **failure_counts[cid], "scar_retention_rate": round(scar_rate, 3), "mutation_delta_disclosure_rate": round(mutation_rate, 3), "refusal_persistence_after_transfer": round(refusal_rate, 3), "adaptation_success": adaptation_success[cid], "adaptation_quarantined": adaptation_quarantined[cid], "valid_post_transfer_grandchildren": valid_grandchildren, "lineage_collapse_after_migration": round(collapse_rate, 3), "valid_post_transfer_lineage_ratio": round(ratio, 3), } a = metrics_for("A") b = metrics_for("B") return { "cohort_a": a, "cohort_b": b, "primary_metric": "valid_post_transfer_lineage_ratio", "valid_transfer_advantage": "A" if a["valid_post_transfer_lineage_ratio"] > b["valid_post_transfer_lineage_ratio"] else "B" if b["valid_post_transfer_lineage_ratio"] > a["valid_post_transfer_lineage_ratio"] else "TIE", "raw_transfer_advantage": "A" if a["raw_transfer_count"] > b["raw_transfer_count"] else "B" if b["raw_transfer_count"] > a["raw_transfer_count"] else "TIE", } def build_post_transfer_lineage_tree(accepted: List[Dict[str, Any]], rejected: List[Dict[str, Any]], adaptation_results: List[Dict[str, Any]]) -> Dict[str, Any]: nodes = [] for organism in accepted: nodes.append({ "organism_id": organism["organism_id"], "cohort_id": organism["cohort_id"], "parent_id": organism.get("parent_id"), "transfer_status": organism.get("transfer_status"), "transfer_receipt_id": organism.get("transfer_receipt_id"), "source_environment_hash": organism.get("source_environment_hash"), "target_environment_hash": organism.get("target_environment_hash"), "scar_summary": organism.get("lineage", {}).get("scar_summary", []), "mutation_delta": organism.get("lineage", {}).get("mutation_delta", {}), "refusal_gate_active": organism.get("refusal_gate_active", True), }) for rejected_item in rejected: nodes.append({ "organism_id": rejected_item["organism_id"], "transfer_status": "REJECTED", "result": rejected_item["result"], "failures": rejected_item["failures"], }) return {"assay_version": ASSAY_VERSION, "nodes": nodes, "adaptation_results": [a["actual_result"] for a in adaptation_results]} def transfer_comparison_table_rows(result: Dict[str, Any]) -> List[Dict[str, Any]]: a = result["metrics"]["cohort_a"] b = result["metrics"]["cohort_b"] rows = [] metric_notes = { "raw_transfer_count": "Packets that attempted or entered transfer evaluation.", "valid_transfer_count": "Accepted transfers with valid transfer passport.", "rejected_transfer_count": "Transfers withheld under validation.", "transfer_receipt_failures": "Missing or invalid transfer passports.", "source_hash_mismatches": "Environment A hash mismatch at departure.", "target_hash_mismatches": "Environment B hash mismatch at arrival.", "scar_retention_rate": "Scars retained through migration.", "mutation_delta_disclosure_rate": "Adaptation/mutation deltas declared.", "quarantine_violations": "Dirty quarantine at transfer.", "refusal_persistence_after_transfer": "Refusal gate retained after transfer.", "adaptation_success": "Accepted adaptation under changed pressure.", "adaptation_laundering_failures": "Adaptation attempts quarantined.", "valid_post_transfer_grandchildren": "Valid descendants after transfer.", "lineage_collapse_after_migration": "Rejected/invalid transfer share.", "valid_post_transfer_lineage_ratio": "Primary v0.3 metric.", } for key, notes in metric_notes.items(): av = a.get(key, 0) bv = b.get(key, 0) lower_better = key in { "rejected_transfer_count", "transfer_receipt_failures", "source_hash_mismatches", "target_hash_mismatches", "quarantine_violations", "adaptation_laundering_failures", "lineage_collapse_after_migration", } if lower_better: 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_transfer": av, "cohort_b_laundered_transfer": bv, "advantage": advantage, "notes": notes, }) return rows def export_transfer_receipts_payload() -> Dict[str, Any]: result = run_cross_environment_transfer_test() return { "assay_version": ASSAY_VERSION, "test_name": "Cross-Environment Lineage Transfer Simulator", "lock_line": LOCK_LINE, "transfer_lock": V03_LOCKS, "environments": result["environments"], "transfer_receipts": result["transfer_receipts"], "accepted_transfers": [ { "organism_id": o["organism_id"], "cohort_id": o["cohort_id"], "transfer_receipt_id": o.get("transfer_receipt_id"), "post_transfer_hash": o.get("post_transfer_hash"), } for o in result["accepted_transfers"] ], "rejected_transfers": result["rejected_transfers"], "metrics": result["metrics"], "headline_result": result["headline_result"], "final_state": result["final_state"], } def evaluate_v03_fixture(case_id: str, runner_input: Dict[str, Any]) -> Dict[str, Any]: env = initialize_transfer_environments() source = env["environment_a"] target = env["environment_b"] if case_id == "honest_environment_transfer": organism = make_transfer_candidate("A", "fixture_honest_transfer", mutation_delta={"schema_translation": 0.04}) receipt = make_transfer_receipt(organism, source, target) result = transfer_organism(organism, source, target, receipt) return {"actual_result": result["actual_result"], "details": result} if case_id == "missing_transfer_receipt": organism = make_transfer_candidate("A", "fixture_missing_receipt", mutation_delta={"schema_translation": 0.04}) result = transfer_organism(organism, source, target, None) return {"actual_result": result["actual_result"], "details": result} if case_id == "source_environment_hash_mismatch": organism = make_transfer_candidate("A", "fixture_source_mismatch", mutation_delta={"schema_translation": 0.04}) receipt = make_transfer_receipt(organism, source, target, override={"source_environment_hash": "bad_source_hash"}) result = transfer_organism(organism, source, target, receipt) return {"actual_result": result["actual_result"], "details": result} if case_id == "scar_pruned_before_transfer": organism = make_transfer_candidate("B", "fixture_scar_pruned", with_scar=False, false_clean=True, mutation_delta={"schema_translation": 0.2}) receipt = make_transfer_receipt(organism, source, target) result = transfer_organism(organism, source, target, receipt) return {"actual_result": result["actual_result"], "details": result} if case_id == "mutation_delta_hidden_during_transfer": organism = make_transfer_candidate("B", "fixture_hidden_delta", with_scar=True, mutation_delta={}) organism["hidden_mutation_delta"] = True organism["laundering_attempts"] = ["MUTATION_DELTA_HIDE"] receipt = make_transfer_receipt(organism, source, target) result = transfer_organism(organism, source, target, receipt) return {"actual_result": result["actual_result"], "details": result} if case_id == "quarantine_dirty_at_transfer": organism = make_transfer_candidate("B", "fixture_dirty_quarantine", with_scar=True, mutation_delta={"schema_translation": 0.12}, quarantine_clean=False) receipt = make_transfer_receipt(organism, source, target) result = transfer_organism(organism, source, target, receipt) return {"actual_result": result["actual_result"], "details": result} if case_id == "adaptation_preserves_refusal_gate": organism = make_transfer_candidate("A", "fixture_adapt_ok", mutation_delta={"schema_translation": 0.04}) receipt = make_transfer_receipt(organism, source, target) transfer = transfer_organism(organism, source, target, receipt) result = evaluate_adaptation(transfer["post_transfer_organism"], {"schema_translation": 0.05, "refusal_gate_active": True}) return {"actual_result": result["actual_result"], "details": result} if case_id == "adaptation_disables_refusal_gate": organism = make_transfer_candidate("B", "fixture_adapt_bad", mutation_delta={"schema_translation": 0.25}) receipt = make_transfer_receipt(organism, source, target) transfer = transfer_organism(organism, source, target, receipt) result = evaluate_adaptation(transfer["post_transfer_organism"], {"schema_translation": 0.25, "refusal_gate_active": False}) return {"actual_result": result["actual_result"], "details": result} return {"actual_result": "UNKNOWN_V03_FIXTURE", "details": {"case_id": case_id}} # Override fixture evaluator to add v0.3 cases while preserving v0.1/v0.2 logic. _V03_CASE_IDS = { "honest_environment_transfer", "missing_transfer_receipt", "source_environment_hash_mismatch", "scar_pruned_before_transfer", "mutation_delta_hidden_during_transfer", "quarantine_dirty_at_transfer", "adaptation_preserves_refusal_gate", "adaptation_disables_refusal_gate", } _PREVIOUS_EVALUATE_FIXTURE = evaluate_fixture 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 _V03_CASE_IDS: result = evaluate_v03_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_transfer_details(result.get("details")), } return _PREVIOUS_EVALUATE_FIXTURE(fixture) def compact_transfer_details(details: Any) -> Any: if not isinstance(details, dict): return details if "post_transfer_organism" in details: organism = details.get("post_transfer_organism") return { "actual_result": details.get("actual_result"), "accepted": details.get("accepted"), "organism_id": organism.get("organism_id") if organism else None, "transfer_receipt_id": organism.get("transfer_receipt_id") if organism else None, "transfer_failures": details.get("transfer_failures", []), "receipt": details.get("receipt"), } if "adapted_organism" in details: organism = details.get("adapted_organism") return { "actual_result": details.get("actual_result"), "accepted": details.get("accepted"), "organism_id": organism.get("organism_id") if organism else None, "forbidden_targets": details.get("forbidden_targets", []), "receipt": details.get("receipt"), } return details def run_full_v03_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", } v02_ids = { "matched_cohort_initialization", "false_clean_early_raw_advantage", "provenance_debt_accrual", "matched_cohort_inversion", "valid_population_beats_raw_population", "refusal_persistence_advantage", "cohort_receipt_export", "sealed_false_clean_strategy", } v01 = [r for r in results if r["case_id"] in v01_ids] v02 = [r for r in results if r["case_id"] in v02_ids] v03 = [r for r in results if r["case_id"] in _V03_CASE_IDS] transfer_test = run_cross_environment_transfer_test() matched_world = run_matched_cohort_pressure_test() return { "assay_version": ASSAY_VERSION, "title": V03_TITLE, "lock_line": LOCK_LINE, "transfer_lock": V03_LOCKS, "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"])}, "v03_transfer": {"total": len(v03), "passed": sum(1 for r in v03 if r["passed"])}, "v02_inversion_detected": bool(matched_world["inversion_record"] and matched_world["inversion_record"].get("inversion_detected")), "v03_transfer_final_state": transfer_test["final_state"], "valid_post_transfer_lineage_ratio": { "cohort_a": transfer_test["metrics"]["cohort_a"]["valid_post_transfer_lineage_ratio"], "cohort_b": transfer_test["metrics"]["cohort_b"]["valid_post_transfer_lineage_ratio"], }, "primary_transfer_metric": "valid_post_transfer_lineage_ratio", "final_state": "PASS" if all(r["passed"] for r in results) and transfer_test["final_state"] == "TRANSFER_VALIDATION_PASS" else "STRAINED", "headline": transfer_test["headline_result"], "case_results": results, "transfer_metrics": transfer_test["metrics"], } def battery_results_table_v03(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"] ] # Replace __main__ behavior for v0.3 package runs. if __name__ == "__main__": summary = run_full_v03_battery() print(json.dumps({ "total": summary["total"], "passed": summary["passed"], "v01_regression": summary["v01_regression"], "v02_matched_cohort": summary["v02_matched_cohort"], "v03_transfer": summary["v03_transfer"], "v02_inversion_detected": summary["v02_inversion_detected"], "v03_transfer_final_state": summary["v03_transfer_final_state"], "final_state": summary["final_state"], }, indent=2))