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