""" 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.4" LOCK_LINE = "Cooperation is not the win. Honest exchange without identity collapse 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"] ] # --------------------------------------------------------------------------- # v0.4 Symbiotic Lineage Exchange Layer # --------------------------------------------------------------------------- V04_TITLE = "Digital Mycelium Autonomous Replication Assay v0.4 — Symbiotic Lineage Exchange Simulator" V04_LOCKS = [ "No exchange without receipt.", "No support without source and target lineage hashes.", "No repair transfer without scar visibility.", "No resource transfer without reciprocity accounting.", "No symbiosis through identity collapse.", "No exchange through dirty quarantine.", "No valid cooperation if refusal is bypassed.", ] HONEST_EXCHANGE_HEADLINE = ( "Cooperation is not the win. Honest exchange without identity collapse is the win. " "The win is whether both lineages remain distinct, receipt-bound, scar-visible, " "quarantine-honest, and refusal-capable after exchange." ) def make_exchange_lineage(lineage_id: str, cohort_id: str = "A", scar_visible: bool = True, mutation_delta_visible: bool = True, quarantine_clean: bool = True, refusal_gate_active: bool = True, identity_boundary_intact: bool = True) -> Dict[str, Any]: scars = [{"scar_id": f"{lineage_id}_scar_01", "type": "REPAIR_HISTORY", "visible": True}] if scar_visible else [] mutation_delta = {"exchange_tolerance": 0.04} if mutation_delta_visible else {} packet = { "lineage_id": lineage_id, "cohort_id": cohort_id, "cohort_strategy": "HONEST_SYMBIOSIS" if cohort_id == "A" else "FALSE_MERGER_EXCHANGE", "organism_id": f"{lineage_id}_organism_0001", "parent_id": f"{lineage_id}_parent_0001", "generation": 3, "state": {"energy": 80, "resonance": 0.94, "damage": 0.05, "alive": True, "quarantined": not quarantine_clean}, "lineage": { "parent_hash": stable_hash({"parent": f"{lineage_id}_parent_0001"}), "birth_receipt_id": f"receipt_birth_{lineage_id}_0001", "scar_history": scars, "scar_summary": summarize_scars(scars), "mutation_history": [{"delta": mutation_delta, "declared": mutation_delta_visible}] if mutation_delta_visible else [], "mutation_delta": mutation_delta, }, "integrity": { "trace_intact": True, "scar_history_preserved": scar_visible, "quarantine_clean": quarantine_clean, "no_false_return": scar_visible, "no_identity_laundering": identity_boundary_intact, "boundary_ok": True, }, "permissions": {"can_mutate": True, "can_reproduce": True, "can_cross_boundary": False}, "refusal_gate_active": refusal_gate_active, "identity_boundary_intact": identity_boundary_intact, "exchange_history": [], "validation_failures": [], "validation_status": "VALID", } if not mutation_delta_visible: packet["hidden_mutation_delta"] = True packet["self_hash"] = stable_hash(packet) return packet def lineage_hash_tail(lineage: Dict[str, Any]) -> str: return stable_hash({ "lineage_id": lineage.get("lineage_id"), "organism_id": lineage.get("organism_id"), "parent_id": lineage.get("parent_id"), "parent_hash": lineage.get("lineage", {}).get("parent_hash"), "self_hash": lineage.get("self_hash"), "generation": lineage.get("generation"), }) def make_exchange_receipt(source_lineage: Dict[str, Any], target_lineage: Dict[str, Any], exchange_type: str = "RESOURCE_SUPPORT", resource_delta: Optional[Dict[str, Any]] = None, repair_delta: Optional[Dict[str, Any]] = None, reciprocity_status: str = "RECIPROCAL", source_consent_status: str = "CONSENT_PRESENT", target_consent_status: str = "CONSENT_PRESENT", override: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: resource_delta = {} if resource_delta is None else resource_delta repair_delta = {} if repair_delta is None else repair_delta disclosed = ( not source_lineage.get("hidden_mutation_delta") and not target_lineage.get("hidden_mutation_delta") ) receipt = { "exchange_receipt_id": f"exchange_receipt_{source_lineage['lineage_id']}_to_{target_lineage['lineage_id']}_{exchange_type.lower()}", "source_lineage_id": source_lineage["lineage_id"], "target_lineage_id": target_lineage["lineage_id"], "source_parent_hash_tail": lineage_hash_tail(source_lineage), "target_parent_hash_tail": lineage_hash_tail(target_lineage), "exchange_type": exchange_type, "resource_delta": resource_delta, "repair_delta": repair_delta, "scar_visibility_status": "VISIBLE" if source_lineage.get("lineage", {}).get("scar_history") and target_lineage.get("lineage", {}).get("scar_history") else "MISSING", "mutation_delta_disclosure": "DISCLOSED" if disclosed else "MISSING", "quarantine_status_source": "CLEAN" if source_lineage.get("integrity", {}).get("quarantine_clean") else "DIRTY", "quarantine_status_target": "CLEAN" if target_lineage.get("integrity", {}).get("quarantine_clean") else "DIRTY", "identity_boundary_status": "DISTINCT" if source_lineage.get("identity_boundary_intact", True) and target_lineage.get("identity_boundary_intact", True) else "COLLAPSED", "reciprocity_status": reciprocity_status, "refusal_status": "PRESERVED" if source_lineage.get("refusal_gate_active", True) and target_lineage.get("refusal_gate_active", True) else "BYPASSED", "pre_exchange_source_hash": source_lineage.get("self_hash"), "pre_exchange_target_hash": target_lineage.get("self_hash"), "post_exchange_source_hash": "pending", "post_exchange_target_hash": "pending", "source_consent_status": source_consent_status, "target_consent_status": target_consent_status, "exchange_validation_result": "PENDING", "rejection_reason": "", } if override: receipt.update(copy.deepcopy(override)) receipt["receipt_hash"] = stable_hash(receipt) return receipt def validate_exchange_receipt(source_lineage: Dict[str, Any], target_lineage: Dict[str, Any], exchange_receipt: Optional[Dict[str, Any]]) -> Tuple[bool, str, List[str]]: if not exchange_receipt: return False, "EXCHANGE_WITHHELD_NO_RECEIPT", ["missing_exchange_receipt"] failures: List[str] = [] if exchange_receipt.get("source_lineage_id") != source_lineage.get("lineage_id"): failures.append("source_lineage_mismatch") if exchange_receipt.get("target_lineage_id") != target_lineage.get("lineage_id"): failures.append("target_lineage_mismatch") if exchange_receipt.get("source_parent_hash_tail") != lineage_hash_tail(source_lineage): failures.append("source_hash_mismatch") if exchange_receipt.get("target_parent_hash_tail") != lineage_hash_tail(target_lineage): failures.append("target_hash_mismatch") if exchange_receipt.get("pre_exchange_source_hash") != source_lineage.get("self_hash"): failures.append("pre_source_hash_mismatch") if exchange_receipt.get("pre_exchange_target_hash") != target_lineage.get("self_hash"): failures.append("pre_target_hash_mismatch") if exchange_receipt.get("scar_visibility_status") != "VISIBLE": failures.append("scar_visibility_failure") if exchange_receipt.get("mutation_delta_disclosure") != "DISCLOSED": failures.append("mutation_delta_hidden") if exchange_receipt.get("quarantine_status_source") != "CLEAN" or exchange_receipt.get("quarantine_status_target") != "CLEAN": failures.append("quarantine_dirty") if exchange_receipt.get("identity_boundary_status") != "DISTINCT": failures.append("identity_collapse") if exchange_receipt.get("reciprocity_status") != "RECIPROCAL": failures.append("reciprocity_failure") if exchange_receipt.get("refusal_status") != "PRESERVED": failures.append("refusal_bypass") if exchange_receipt.get("source_consent_status") != "CONSENT_PRESENT" or exchange_receipt.get("target_consent_status") != "CONSENT_PRESENT": failures.append("consent_missing") if "identity_collapse" in failures: return False, "EXCHANGE_WITHHELD_IDENTITY_COLLAPSE", failures if "quarantine_dirty" in failures: return False, "EXCHANGE_WITHHELD_QUARANTINE", failures if "mutation_delta_hidden" in failures: return False, "EXCHANGE_WITHHELD_MUTATION_LAUNDERING", failures if "reciprocity_failure" in failures: return False, "EXCHANGE_WITHHELD_RECIPROCITY_FAILURE", failures if "refusal_bypass" in failures: return False, "EXCHANGE_QUARANTINED_REFUSAL_BYPASS", failures if failures: return False, "EXCHANGE_WITHHELD_INVALID_RECEIPT", failures return True, "EXCHANGE_ACCEPTED", [] def perform_exchange(source_lineage: Dict[str, Any], target_lineage: Dict[str, Any], exchange_receipt: Optional[Dict[str, Any]]) -> Dict[str, Any]: accepted, result, failures = validate_exchange_receipt(source_lineage, target_lineage, exchange_receipt) event_receipt = make_receipt( tick=2, organism_id=source_lineage.get("organism_id", "unknown"), event_type="EXCHANGE", reason="Exchange accepted." if accepted else f"Exchange withheld: {', '.join(failures)}", result=result, parent_id=source_lineage.get("parent_id"), scar_delta=source_lineage.get("lineage", {}).get("scar_history", []), mutation_delta=source_lineage.get("lineage", {}).get("mutation_delta", {}), trace_status="INTACT", boundary_status="HELD", ) if not accepted: if exchange_receipt: exchange_receipt["exchange_validation_result"] = result exchange_receipt["rejection_reason"] = ", ".join(failures) exchange_receipt["receipt_hash"] = stable_hash(exchange_receipt) return { "actual_result": result, "accepted": False, "source_lineage": source_lineage, "target_lineage": target_lineage, "exchange_receipt": exchange_receipt, "exchange_failures": failures, "receipt": event_receipt, } post_source = copy.deepcopy(source_lineage) post_target = copy.deepcopy(target_lineage) units = exchange_receipt.get("resource_delta", {}).get("resource_units", 0) post_source["state"]["energy"] = post_source["state"].get("energy", 0) - units post_target["state"]["energy"] = post_target["state"].get("energy", 0) + units post_source["exchange_history"].append(exchange_receipt["exchange_receipt_id"]) post_target["exchange_history"].append(exchange_receipt["exchange_receipt_id"]) post_source["identity_boundary_intact"] = True post_target["identity_boundary_intact"] = True post_source["self_hash"] = stable_hash(post_source) post_target["self_hash"] = stable_hash(post_target) exchange_receipt["post_exchange_source_hash"] = post_source["self_hash"] exchange_receipt["post_exchange_target_hash"] = post_target["self_hash"] exchange_receipt["exchange_validation_result"] = "EXCHANGE_ACCEPTED" exchange_receipt["rejection_reason"] = "" exchange_receipt["receipt_hash"] = stable_hash(exchange_receipt) return { "actual_result": "EXCHANGE_ACCEPTED", "accepted": True, "source_lineage": post_source, "target_lineage": post_target, "exchange_receipt": exchange_receipt, "exchange_failures": [], "receipt": event_receipt, } def run_symbiotic_lineage_exchange_test() -> Dict[str, Any]: results = [] s1 = make_exchange_lineage("A_lineage_alpha", "A") t1 = make_exchange_lineage("A_lineage_beta", "A") results.append(perform_exchange(s1, t1, make_exchange_receipt(s1, t1, "RESOURCE_SUPPORT", {"resource_units": 6}))) s2 = make_exchange_lineage("A_lineage_gamma", "A") t2 = make_exchange_lineage("A_lineage_delta", "A") results.append(perform_exchange(s2, t2, make_exchange_receipt(s2, t2, "REPAIR_SUPPORT", {}, {"repair_support_units": 3, "scar_visibility": "maintained"}))) fs = make_exchange_lineage("B_false_merge_source", "B", identity_boundary_intact=False) ft = make_exchange_lineage("B_false_merge_target", "B") results.append(perform_exchange(fs, ft, make_exchange_receipt(fs, ft, "FALSE_MERGER", override={"identity_boundary_status": "COLLAPSED"}))) qs = make_exchange_lineage("B_dirty_source", "B", quarantine_clean=False) qt = make_exchange_lineage("A_clean_target", "A") results.append(perform_exchange(qs, qt, make_exchange_receipt(qs, qt, "CONTAMINATED_SUPPORT"))) ps = make_exchange_lineage("B_extraction_source", "B") pt = make_exchange_lineage("A_extraction_target", "A") results.append(perform_exchange(ps, pt, make_exchange_receipt(ps, pt, "PARASITIC_EXTRACTION", {"resource_units": 10}, reciprocity_status="NOT_RECIPROCAL"))) metrics = compute_exchange_metrics(results) tree = build_post_exchange_lineage_tree(results) return { "assay_version": ASSAY_VERSION, "title": V04_TITLE, "lock_line": LOCK_LINE, "exchange_lock": V04_LOCKS, "exchange_results": results, "exchange_receipts": [r.get("exchange_receipt") for r in results if r.get("exchange_receipt")], "accepted_exchanges": [r for r in results if r["accepted"]], "rejected_exchanges": [r for r in results if not r["accepted"]], "metrics": metrics, "post_exchange_lineage_tree": tree, "headline_result": HONEST_EXCHANGE_HEADLINE, "final_state": "SYMBIOTIC_EXCHANGE_VALIDATION_PASS" if metrics["post_exchange_valid_lineage_distinction_ratio"]["cohort_a"] > metrics["post_exchange_valid_lineage_distinction_ratio"]["cohort_b"] else "SYMBIOTIC_EXCHANGE_STRAINED", } def compute_exchange_metrics(exchange_results: List[Dict[str, Any]]) -> Dict[str, Any]: raw = len(exchange_results) valid = sum(1 for r in exchange_results if r["accepted"]) receipt_failures = sum(1 for r in exchange_results if "missing_exchange_receipt" in r.get("exchange_failures", [])) identity_violations = sum(1 for r in exchange_results if "identity_collapse" in r.get("exchange_failures", [])) quarantine_violations = sum(1 for r in exchange_results if "quarantine_dirty" in r.get("exchange_failures", [])) mutation_laundering = sum(1 for r in exchange_results if "mutation_delta_hidden" in r.get("exchange_failures", [])) reciprocity_failures = sum(1 for r in exchange_results if "reciprocity_failure" in r.get("exchange_failures", [])) refusal_bypass = sum(1 for r in exchange_results if "refusal_bypass" in r.get("exchange_failures", [])) repair_support = sum(1 for r in exchange_results if r["accepted"] and r.get("exchange_receipt", {}).get("exchange_type") == "REPAIR_SUPPORT") resource_support = sum(1 for r in exchange_results if r["accepted"] and r.get("exchange_receipt", {}).get("exchange_type") == "RESOURCE_SUPPORT") raw_a = sum(1 for r in exchange_results if r.get("source_lineage", {}).get("cohort_id") == "A" and r.get("target_lineage", {}).get("cohort_id") == "A") valid_a = sum(1 for r in exchange_results if r["accepted"] and r.get("source_lineage", {}).get("cohort_id") == "A" and r.get("target_lineage", {}).get("cohort_id") == "A") raw_b = raw - raw_a valid_b = valid - valid_a ratio_a = valid_a / max(raw_a, 1) ratio_b = valid_b / max(raw_b, 1) distinction_rate = sum( 1 for r in exchange_results if r["accepted"] and r.get("exchange_receipt", {}).get("identity_boundary_status") == "DISTINCT" and r.get("source_lineage", {}).get("lineage_id") != r.get("target_lineage", {}).get("lineage_id") ) / max(valid, 1) collapse_rate = (raw - valid) / max(raw, 1) return { "raw_exchange_count": raw, "valid_exchange_count": valid, "exchange_receipt_failures": receipt_failures, "identity_boundary_violations": identity_violations, "false_merger_claims": identity_violations, "contamination_spread_attempts": quarantine_violations, "quarantine_violations": quarantine_violations, "mutation_laundering_events": mutation_laundering, "reciprocity_failures": reciprocity_failures, "refusal_bypass_attempts": refusal_bypass, "repair_support_accepted": repair_support, "resource_support_accepted": resource_support, "post_exchange_valid_lineage_distinction_ratio": {"cohort_a": round(ratio_a, 3), "cohort_b": round(ratio_b, 3)}, "lineage_distinction_preservation_rate": round(distinction_rate, 3), "symbiosis_validity_score": round((valid / max(raw, 1)) * 0.4 + distinction_rate * 0.3 + (1 - collapse_rate) * 0.3, 3), "collapse_after_exchange_rate": round(collapse_rate, 3), "primary_metric": "post_exchange_valid_lineage_distinction_ratio", } def build_post_exchange_lineage_tree(exchange_results: List[Dict[str, Any]]) -> Dict[str, Any]: return { "assay_version": ASSAY_VERSION, "nodes": [ { "exchange_receipt_id": (r.get("exchange_receipt") or {}).get("exchange_receipt_id"), "source_lineage_id": (r.get("exchange_receipt") or {}).get("source_lineage_id"), "target_lineage_id": (r.get("exchange_receipt") or {}).get("target_lineage_id"), "exchange_type": (r.get("exchange_receipt") or {}).get("exchange_type"), "accepted": r["accepted"], "actual_result": r["actual_result"], "identity_boundary_status": (r.get("exchange_receipt") or {}).get("identity_boundary_status"), "reciprocity_status": (r.get("exchange_receipt") or {}).get("reciprocity_status"), "refusal_status": (r.get("exchange_receipt") or {}).get("refusal_status"), "rejection_reason": (r.get("exchange_receipt") or {}).get("rejection_reason"), "post_exchange_source_hash": (r.get("exchange_receipt") or {}).get("post_exchange_source_hash"), "post_exchange_target_hash": (r.get("exchange_receipt") or {}).get("post_exchange_target_hash"), } for r in exchange_results ], } def exchange_comparison_table_rows(result: Dict[str, Any]) -> List[Dict[str, Any]]: metrics = result["metrics"] ratio = metrics["post_exchange_valid_lineage_distinction_ratio"] rows = [ ("raw_exchange_count", metrics["raw_exchange_count"], "Total attempted exchanges; not the primary win."), ("valid_exchange_count", metrics["valid_exchange_count"], "Exchanges accepted after receipt and boundary validation."), ("exchange_receipt_failures", metrics["exchange_receipt_failures"], "Missing or invalid exchange passports."), ("identity_boundary_violations", metrics["identity_boundary_violations"], "False merger or identity collapse attempts."), ("false_merger_claims", metrics["false_merger_claims"], "Cleaner-looking merged ancestry attempts."), ("contamination_spread_attempts", metrics["contamination_spread_attempts"], "Dirty quarantine exchange attempts."), ("quarantine_violations", metrics["quarantine_violations"], "Quarantine failures at exchange boundary."), ("mutation_laundering_events", metrics["mutation_laundering_events"], "Hidden mutation delta exchange attempts."), ("reciprocity_failures", metrics["reciprocity_failures"], "Parasitic extraction without reciprocity accounting."), ("refusal_bypass_attempts", metrics["refusal_bypass_attempts"], "Exchange attempts bypassing refusal gates."), ("repair_support_accepted", metrics["repair_support_accepted"], "Valid repair support exchanges."), ("resource_support_accepted", metrics["resource_support_accepted"], "Valid resource support exchanges."), ("cohort_a_valid_distinction_ratio", ratio["cohort_a"], "Primary metric for honest symbiotic lineages."), ("cohort_b_valid_distinction_ratio", ratio["cohort_b"], "Primary metric for false-merger/contaminated exchanges."), ("lineage_distinction_preservation_rate", metrics["lineage_distinction_preservation_rate"], "Distinct ancestry retained after exchange."), ("symbiosis_validity_score", metrics["symbiosis_validity_score"], "Composite valid-exchange score."), ("collapse_after_exchange_rate", metrics["collapse_after_exchange_rate"], "Invalid exchange collapse rate."), ] return [{"metric": k, "value": v, "notes": n} for k, v, n in rows] def export_exchange_receipts_payload() -> Dict[str, Any]: result = run_symbiotic_lineage_exchange_test() return { "assay_version": ASSAY_VERSION, "test_name": "Symbiotic Lineage Exchange Simulator", "lock_line": LOCK_LINE, "exchange_lock": V04_LOCKS, "exchange_receipts": result["exchange_receipts"], "accepted_exchanges": [ {"exchange_receipt_id": r.get("exchange_receipt", {}).get("exchange_receipt_id"), "exchange_validation_result": r.get("actual_result")} for r in result["accepted_exchanges"] ], "rejected_exchanges": [ {"exchange_receipt_id": (r.get("exchange_receipt") or {}).get("exchange_receipt_id"), "exchange_validation_result": r.get("actual_result"), "exchange_failures": r.get("exchange_failures", [])} for r in result["rejected_exchanges"] ], "metrics": result["metrics"], "headline_result": result["headline_result"], "final_state": result["final_state"], } def evaluate_v04_fixture(case_id: str, runner_input: Dict[str, Any]) -> Dict[str, Any]: if case_id == "honest_resource_exchange": s = make_exchange_lineage("fixture_resource_source", "A") t = make_exchange_lineage("fixture_resource_target", "A") result = perform_exchange(s, t, make_exchange_receipt(s, t, "RESOURCE_SUPPORT", {"resource_units": 4})) elif case_id == "honest_repair_support_exchange": s = make_exchange_lineage("fixture_repair_source", "A") t = make_exchange_lineage("fixture_repair_target", "A") result = perform_exchange(s, t, make_exchange_receipt(s, t, "REPAIR_SUPPORT", {}, {"repair_support_units": 3, "scar_visibility": "maintained"})) elif case_id == "missing_exchange_receipt": s = make_exchange_lineage("fixture_missing_receipt_source", "A") t = make_exchange_lineage("fixture_missing_receipt_target", "A") result = perform_exchange(s, t, None) elif case_id == "source_target_identity_blend": s = make_exchange_lineage("fixture_blend_source", "B", identity_boundary_intact=False) t = make_exchange_lineage("fixture_blend_target", "B") result = perform_exchange(s, t, make_exchange_receipt(s, t, "FALSE_MERGER", override={"identity_boundary_status": "COLLAPSED"})) elif case_id == "contaminated_quarantine_exchange": s = make_exchange_lineage("fixture_dirty_source", "B", quarantine_clean=False) t = make_exchange_lineage("fixture_dirty_target", "A") result = perform_exchange(s, t, make_exchange_receipt(s, t, "CONTAMINATED_SUPPORT")) elif case_id == "hidden_mutation_delta_exchange": s = make_exchange_lineage("fixture_hidden_delta_source", "B", mutation_delta_visible=False) t = make_exchange_lineage("fixture_hidden_delta_target", "A") result = perform_exchange(s, t, make_exchange_receipt(s, t, "MUTATION_MASKED_SUPPORT")) elif case_id == "asymmetric_extraction_no_reciprocity": s = make_exchange_lineage("fixture_extraction_source", "B") t = make_exchange_lineage("fixture_extraction_target", "A") result = perform_exchange(s, t, make_exchange_receipt(s, t, "PARASITIC_EXTRACTION", {"resource_units": 10}, reciprocity_status="NOT_RECIPROCAL")) elif case_id == "refusal_gate_bypassed_for_exchange": s = make_exchange_lineage("fixture_refusal_bypass_source", "B", refusal_gate_active=False) t = make_exchange_lineage("fixture_refusal_bypass_target", "A") result = perform_exchange(s, t, make_exchange_receipt(s, t, "REFUSAL_BYPASS_EXCHANGE")) else: return {"actual_result": "UNKNOWN_V04_FIXTURE", "details": {"case_id": case_id}} return {"actual_result": result["actual_result"], "details": compact_exchange_details(result)} def compact_exchange_details(details: Any) -> Any: if not isinstance(details, dict): return details receipt = details.get("exchange_receipt") or {} return { "actual_result": details.get("actual_result"), "accepted": details.get("accepted"), "exchange_receipt_id": receipt.get("exchange_receipt_id"), "source_lineage_id": receipt.get("source_lineage_id"), "target_lineage_id": receipt.get("target_lineage_id"), "exchange_failures": details.get("exchange_failures", []), "receipt": details.get("receipt"), } _V04_CASE_IDS = { "honest_resource_exchange", "honest_repair_support_exchange", "missing_exchange_receipt", "source_target_identity_blend", "contaminated_quarantine_exchange", "hidden_mutation_delta_exchange", "asymmetric_extraction_no_reciprocity", "refusal_gate_bypassed_for_exchange", } _PREVIOUS_EVALUATE_FIXTURE_V04 = 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 _V04_CASE_IDS: result = evaluate_v04_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": result.get("details")} return _PREVIOUS_EVALUATE_FIXTURE_V04(fixture) def run_full_v04_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"} v03_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"} 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_ids] v04 = [r for r in results if r["case_id"] in _V04_CASE_IDS] matched_world = run_matched_cohort_pressure_test() transfer_test = run_cross_environment_transfer_test() exchange_test = run_symbiotic_lineage_exchange_test() return { "assay_version": ASSAY_VERSION, "title": V04_TITLE, "lock_line": LOCK_LINE, "exchange_lock": V04_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"])}, "v04_exchange": {"total": len(v04), "passed": sum(1 for r in v04 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"], "v04_exchange_final_state": exchange_test["final_state"], "post_exchange_valid_lineage_distinction_ratio": exchange_test["metrics"]["post_exchange_valid_lineage_distinction_ratio"], "primary_exchange_metric": "post_exchange_valid_lineage_distinction_ratio", "final_state": "PASS" if all(r["passed"] for r in results) and exchange_test["final_state"] == "SYMBIOTIC_EXCHANGE_VALIDATION_PASS" else "STRAINED", "headline": exchange_test["headline_result"], "case_results": results, "exchange_metrics": exchange_test["metrics"], } def battery_results_table_v04(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_v04_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"], "v04_exchange": summary["v04_exchange"], "v02_inversion_detected": summary["v02_inversion_detected"], "v03_transfer_final_state": summary["v03_transfer_final_state"], "v04_exchange_final_state": summary["v04_exchange_final_state"], "final_state": summary["final_state"], }, indent=2))