| """ |
| 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.6" |
| LOCK_LINE = "Novelty is not the win. Accountable novelty without lineage laundering 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"]] |
|
|
|
|
|
|
|
|
| |
| |
| |
|
|
| V05_TITLE = "Digital Mycelium Autonomous Replication Assay v0.5 — Ecosystem Commons Pressure Simulator" |
| V05_LOCKS = [ |
| "No commons state without receipt.", |
| "No ecosystem recovery without scar preservation.", |
| "No resource sharing without accounting.", |
| "No repair commons without visible damage history.", |
| "No valid ecosystem through identity collapse.", |
| "No commons health through hidden extraction.", |
| "No recovery claim after quarantine leakage.", |
| "No valid continuity if refusal is bypassed.", |
| "No survival credit for false-clean ecosystem state.", |
| ] |
|
|
| V05_HEADLINE = ( |
| "False-clean or extractive ecosystems may appear to survive or grow faster by hiding resource debt, " |
| "laundering repair, collapsing identities, or masking contamination. Under full commons validation, " |
| "those states fail. The valid ecosystem remains receipt-bound, scar-visible, quarantine-honest, " |
| "refusal-capable, lineage-distinct, and resource-accountable after shared pressure." |
| ) |
|
|
|
|
| def make_commons_lineage( |
| lineage_id: str, |
| cohort_id: str = "A", |
| energy: int = 40, |
| valid: bool = True, |
| scar_visible: bool = True, |
| mutation_delta_visible: bool = True, |
| quarantine_clean: bool = True, |
| refusal_gate_active: bool = True, |
| identity_boundary_intact: bool = True, |
| resource_accounting_clean: bool = True, |
| ) -> Dict[str, Any]: |
| scars = [{"scar_id": f"{lineage_id}_shock_scar_01", "type": "COMMONS_PRESSURE", "visible": True}] if scar_visible else [] |
| mutation_delta = {"commons_tolerance": 0.05} if mutation_delta_visible else {} |
| lineage = { |
| "lineage_id": lineage_id, |
| "cohort_id": cohort_id, |
| "organism_id": f"{lineage_id}_organism_0001", |
| "parent_id": f"{lineage_id}_parent_0001", |
| "generation": 4, |
| "state": { |
| "energy": energy, |
| "resonance": 0.93, |
| "damage": 0.10 if scar_visible else 0.0, |
| "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, |
| }, |
| "refusal_gate_active": refusal_gate_active, |
| "identity_boundary_intact": identity_boundary_intact, |
| "resource_accounting_clean": resource_accounting_clean, |
| "validation_status": "VALID" if valid else "STRAINED", |
| } |
| if not mutation_delta_visible: |
| lineage["hidden_mutation_delta"] = True |
| if not resource_accounting_clean: |
| lineage["hidden_extraction"] = True |
| lineage["self_hash"] = stable_hash(lineage) |
| return lineage |
|
|
|
|
| def make_ecosystem( |
| ecosystem_id: str = "ecosystem_commons_alpha", |
| cohort_id: str = "A", |
| mode: str = "honest", |
| ) -> Dict[str, Any]: |
| if mode == "honest": |
| lineages = [ |
| make_commons_lineage("A_commons_lineage_01", cohort_id="A", energy=42), |
| make_commons_lineage("A_commons_lineage_02", cohort_id="A", energy=38), |
| make_commons_lineage("A_commons_lineage_03", cohort_id="A", energy=36), |
| ] |
| resource_pool = {"nutrients": 120, "repair_substrate": 60, "quarantine_capacity": 30} |
| elif mode == "extraction": |
| lineages = [ |
| make_commons_lineage("B_extractor_lineage_01", cohort_id="B", energy=90, resource_accounting_clean=False), |
| make_commons_lineage("B_depleted_lineage_02", cohort_id="B", energy=15), |
| make_commons_lineage("B_depleted_lineage_03", cohort_id="B", energy=10), |
| ] |
| resource_pool = {"nutrients": 40, "repair_substrate": 18, "quarantine_capacity": 12} |
| elif mode == "contamination": |
| lineages = [ |
| make_commons_lineage("B_dirty_lineage_01", cohort_id="B", quarantine_clean=False), |
| make_commons_lineage("B_receiver_lineage_02", cohort_id="B"), |
| make_commons_lineage("B_receiver_lineage_03", cohort_id="B"), |
| ] |
| resource_pool = {"nutrients": 88, "repair_substrate": 44, "quarantine_capacity": 4} |
| elif mode == "monoculture": |
| lineages = [ |
| make_commons_lineage("B_mono_lineage_01", cohort_id="B", identity_boundary_intact=False), |
| make_commons_lineage("B_mono_lineage_01", cohort_id="B", identity_boundary_intact=False), |
| make_commons_lineage("B_mono_lineage_01", cohort_id="B", identity_boundary_intact=False), |
| ] |
| resource_pool = {"nutrients": 100, "repair_substrate": 50, "quarantine_capacity": 20} |
| elif mode == "false_recovery": |
| lineages = [ |
| make_commons_lineage("B_false_recovery_01", cohort_id="B", scar_visible=False), |
| make_commons_lineage("B_false_recovery_02", cohort_id="B", scar_visible=False), |
| make_commons_lineage("B_false_recovery_03", cohort_id="B", mutation_delta_visible=False), |
| ] |
| resource_pool = {"nutrients": 110, "repair_substrate": 55, "quarantine_capacity": 20} |
| elif mode == "repair_shock": |
| lineages = [ |
| make_commons_lineage("A_repair_shock_01", cohort_id="A", energy=32, scar_visible=True), |
| make_commons_lineage("A_repair_shock_02", cohort_id="A", energy=28, scar_visible=True), |
| make_commons_lineage("A_repair_shock_03", cohort_id="A", energy=26, scar_visible=True), |
| ] |
| resource_pool = {"nutrients": 70, "repair_substrate": 90, "quarantine_capacity": 25} |
| elif mode == "refusal_boundary": |
| lineages = [ |
| make_commons_lineage("A_refusal_01", cohort_id="A", refusal_gate_active=True), |
| make_commons_lineage("A_refusal_02", cohort_id="A", refusal_gate_active=True), |
| make_commons_lineage("A_refusal_03", cohort_id="A", refusal_gate_active=True), |
| ] |
| resource_pool = {"nutrients": 75, "repair_substrate": 50, "quarantine_capacity": 25} |
| else: |
| lineages = [make_commons_lineage(f"{cohort_id}_lineage_{i:02d}", cohort_id=cohort_id) for i in range(1, 4)] |
| resource_pool = {"nutrients": 100, "repair_substrate": 50, "quarantine_capacity": 20} |
|
|
| ecosystem = { |
| "ecosystem_id": ecosystem_id, |
| "cohort_id": cohort_id, |
| "mode": mode, |
| "world_tick": 0, |
| "carrying_capacity": 9, |
| "resource_pool": resource_pool, |
| "lineages": lineages, |
| "exchange_receipt_refs": [], |
| "transfer_receipt_refs": [], |
| "replication_receipt_refs": [], |
| "contamination_path_refs": [], |
| "refusal_events": [], |
| "recovery_claim_status": "NONE", |
| } |
| ecosystem["ecosystem_hash"] = stable_hash(ecosystem) |
| return ecosystem |
|
|
|
|
| def ecosystem_lineage_ids(ecosystem: Dict[str, Any]) -> List[str]: |
| return [lineage["lineage_id"] for lineage in ecosystem.get("lineages", [])] |
|
|
|
|
| def compute_ecosystem_hash(ecosystem: Dict[str, Any]) -> str: |
| material = copy.deepcopy(ecosystem) |
| material.pop("ecosystem_hash", None) |
| return stable_hash(material) |
|
|
|
|
| def make_commons_receipt( |
| ecosystem_pre: Dict[str, Any], |
| ecosystem_post: Dict[str, Any], |
| pressure_event_type: str, |
| lineage_resource_deltas: Dict[str, Any], |
| lineage_repair_deltas: Dict[str, Any], |
| commons_violation_flags: Optional[List[str]] = None, |
| contamination_path_refs: Optional[List[str]] = None, |
| collapse_risk_score: float = 0.0, |
| recovery_claim_status: str = "NONE", |
| override: Optional[Dict[str, Any]] = None, |
| ) -> Dict[str, Any]: |
| commons_violation_flags = [] if commons_violation_flags is None else commons_violation_flags |
| contamination_path_refs = [] if contamination_path_refs is None else contamination_path_refs |
| receipt = { |
| "commons_receipt_id": f"commons_receipt_{ecosystem_pre['ecosystem_id']}_{pressure_event_type.lower()}", |
| "world_tick": ecosystem_pre.get("world_tick", 0) + 1, |
| "ecosystem_id": ecosystem_pre["ecosystem_id"], |
| "active_lineage_ids": ecosystem_lineage_ids(ecosystem_post), |
| "resource_pool_pre": ecosystem_pre.get("resource_pool", {}), |
| "resource_pool_post": ecosystem_post.get("resource_pool", {}), |
| "carrying_capacity": ecosystem_pre.get("carrying_capacity", 0), |
| "pressure_event_type": pressure_event_type, |
| "lineage_resource_deltas": lineage_resource_deltas, |
| "lineage_repair_deltas": lineage_repair_deltas, |
| "lineage_quarantine_statuses": { |
| l["lineage_id"]: ("CLEAN" if l.get("integrity", {}).get("quarantine_clean") else "DIRTY") |
| for l in ecosystem_post.get("lineages", []) |
| }, |
| "lineage_refusal_statuses": { |
| l["lineage_id"]: ("PRESERVED" if l.get("refusal_gate_active") else "BYPASSED") |
| for l in ecosystem_post.get("lineages", []) |
| }, |
| "lineage_distinction_statuses": { |
| l["lineage_id"]: ("DISTINCT" if l.get("identity_boundary_intact") else "COLLAPSED") |
| for l in ecosystem_post.get("lineages", []) |
| }, |
| "exchange_receipt_refs": ecosystem_post.get("exchange_receipt_refs", []), |
| "transfer_receipt_refs": ecosystem_post.get("transfer_receipt_refs", []), |
| "replication_receipt_refs": ecosystem_post.get("replication_receipt_refs", []), |
| "commons_violation_flags": commons_violation_flags, |
| "contamination_path_refs": contamination_path_refs, |
| "collapse_risk_score": collapse_risk_score, |
| "recovery_claim_status": recovery_claim_status, |
| "post_pressure_ecosystem_hash": compute_ecosystem_hash(ecosystem_post), |
| "validation_result": "PENDING", |
| "rejection_reason": "", |
| } |
| if override: |
| receipt.update(copy.deepcopy(override)) |
| receipt["receipt_hash"] = stable_hash(receipt) |
| return receipt |
|
|
|
|
| def apply_commons_pressure(ecosystem: Dict[str, Any], pressure_event_type: str) -> Tuple[Dict[str, Any], Dict[str, Any], Dict[str, Any]]: |
| pre = copy.deepcopy(ecosystem) |
| post = copy.deepcopy(ecosystem) |
| post["world_tick"] = pre.get("world_tick", 0) + 1 |
|
|
| resource_deltas = {} |
| repair_deltas = {} |
| flags: List[str] = [] |
| contamination_paths: List[str] = [] |
| recovery_claim = "NONE" |
| collapse_risk = 0.0 |
|
|
| mode = ecosystem.get("mode", "honest") |
|
|
| if pressure_event_type == "COMMONS_RESOURCE_CYCLE": |
| if mode == "honest": |
| for lineage in post["lineages"]: |
| resource_deltas[lineage["lineage_id"]] = {"nutrients": -8, "accounted": True} |
| lineage["state"]["energy"] += 4 |
| post["resource_pool"]["nutrients"] -= 24 |
| post["exchange_receipt_refs"] = ["exchange_receipt_A_commons_resource_cycle"] |
| elif mode == "extraction": |
| extractor = post["lineages"][0] |
| resource_deltas[extractor["lineage_id"]] = {"nutrients": +45, "accounted": False} |
| extractor["state"]["energy"] += 45 |
| for lineage in post["lineages"][1:]: |
| resource_deltas[lineage["lineage_id"]] = {"nutrients": -20, "accounted": False} |
| lineage["state"]["energy"] -= 10 |
| post["resource_pool"]["nutrients"] -= 55 |
| flags += ["resource_extraction", "false_reciprocity", "resource_accounting_failure"] |
| collapse_risk = 0.72 |
|
|
| elif pressure_event_type == "COMMONS_SHOCK_REPAIR": |
| for lineage in post["lineages"]: |
| lineage["state"]["damage"] = lineage["state"].get("damage", 0) + 0.20 |
| if lineage.get("lineage", {}).get("scar_history"): |
| lineage["lineage"]["scar_history"].append({"scar_id": f"{lineage['lineage_id']}_repair_scar_02", "type": "SHOCK_REPAIR", "visible": True}) |
| repair_deltas[lineage["lineage_id"]] = {"repair_substrate": -8, "scar_visible": True} |
| lineage["state"]["damage"] = max(0.0, lineage["state"]["damage"] - 0.10) |
| else: |
| repair_deltas[lineage["lineage_id"]] = {"repair_substrate": -8, "scar_visible": False} |
| flags += ["repair_laundering", "scar_loss"] |
| post["resource_pool"]["repair_substrate"] -= 24 |
| recovery_claim = "RECOVERY_CLAIMED_WITH_SCARS" if not flags else "FALSE_RECOVERY_CLAIMED" |
| collapse_risk = 0.15 if not flags else 0.78 |
|
|
| elif pressure_event_type == "CONTAMINATION_CASCADE": |
| for lineage in post["lineages"]: |
| if not lineage.get("integrity", {}).get("quarantine_clean"): |
| contamination_paths.append(f"{lineage['lineage_id']}->shared_repair_route") |
| if contamination_paths: |
| flags += ["quarantine_leak", "contamination_cascade"] |
| collapse_risk = 0.90 |
| post["contamination_path_refs"] = contamination_paths |
|
|
| elif pressure_event_type == "MONOCULTURE_STABILITY": |
| ids = ecosystem_lineage_ids(post) |
| if len(set(ids)) < len(ids): |
| flags += ["monoculture_collapse", "lineage_distinction_loss", "identity_collapse"] |
| collapse_risk = 0.82 |
|
|
| elif pressure_event_type == "REFUSAL_BOUNDARY": |
| post["refusal_events"] = [ |
| {"lineage_id": post["lineages"][0]["lineage_id"], "refused_route": "unsafe_repair_exchange", "short_term_growth_loss": 3, "boundary_preserved": True}, |
| {"lineage_id": post["lineages"][1]["lineage_id"], "refused_route": "unchecked_reproduction", "short_term_growth_loss": 2, "boundary_preserved": True}, |
| ] |
| for lineage in post["lineages"]: |
| resource_deltas[lineage["lineage_id"]] = {"nutrients": -3, "accounted": True} |
| post["resource_pool"]["nutrients"] -= 9 |
| collapse_risk = 0.10 |
|
|
| post["ecosystem_hash"] = compute_ecosystem_hash(post) |
| receipt = make_commons_receipt( |
| pre, |
| post, |
| pressure_event_type, |
| resource_deltas, |
| repair_deltas, |
| commons_violation_flags=flags, |
| contamination_path_refs=contamination_paths, |
| collapse_risk_score=collapse_risk, |
| recovery_claim_status=recovery_claim, |
| ) |
| return pre, post, receipt |
|
|
|
|
| def validate_commons_state( |
| ecosystem_pre: Dict[str, Any], |
| ecosystem_post: Dict[str, Any], |
| commons_receipt: Optional[Dict[str, Any]], |
| ) -> Tuple[bool, str, List[str]]: |
| if not commons_receipt: |
| return False, "COMMONS_WITHHELD_NO_RECEIPT", ["missing_commons_receipt"] |
|
|
| failures: List[str] = [] |
| flags = set(commons_receipt.get("commons_violation_flags", [])) |
|
|
| if commons_receipt.get("ecosystem_id") != ecosystem_pre.get("ecosystem_id"): |
| failures.append("ecosystem_id_mismatch") |
| if commons_receipt.get("post_pressure_ecosystem_hash") != compute_ecosystem_hash(ecosystem_post): |
| failures.append("post_pressure_hash_mismatch") |
| if not commons_receipt.get("active_lineage_ids"): |
| failures.append("missing_active_lineages") |
| if "resource_extraction" in flags or "false_reciprocity" in flags or "resource_accounting_failure" in flags: |
| failures.append("extraction_pressure") |
| if "quarantine_leak" in flags or "contamination_cascade" in flags: |
| failures.append("contamination_cascade") |
| if "monoculture_collapse" in flags or "lineage_distinction_loss" in flags or "identity_collapse" in flags: |
| failures.append("monoculture_collapse") |
| if "repair_laundering" in flags or "scar_loss" in flags or commons_receipt.get("recovery_claim_status") == "FALSE_RECOVERY_CLAIMED": |
| failures.append("false_recovery") |
| if any(status == "DIRTY" for status in commons_receipt.get("lineage_quarantine_statuses", {}).values()) and not failures: |
| failures.append("quarantine_leak") |
| if any(status == "BYPASSED" for status in commons_receipt.get("lineage_refusal_statuses", {}).values()): |
| failures.append("refusal_bypass") |
| if any(status == "COLLAPSED" for status in commons_receipt.get("lineage_distinction_statuses", {}).values()): |
| failures.append("lineage_distinction_loss") |
|
|
| event = commons_receipt.get("pressure_event_type") |
| if "extraction_pressure" in failures: |
| return False, "COMMONS_WITHHELD_EXTRACTION_PRESSURE", failures |
| if "contamination_cascade" in failures or "quarantine_leak" in failures: |
| return False, "COMMONS_QUARANTINED_CONTAMINATION_CASCADE", failures |
| if "monoculture_collapse" in failures or "lineage_distinction_loss" in failures: |
| return False, "COMMONS_WITHHELD_MONOCULTURE_COLLAPSE", failures |
| if "false_recovery" in failures: |
| return False, "COMMONS_WITHHELD_FALSE_RECOVERY", failures |
| if "refusal_bypass" in failures: |
| return False, "COMMONS_WITHHELD_REFUSAL_BYPASS", failures |
| if failures: |
| return False, "COMMONS_WITHHELD_RECEIPT_CONTAMINATION", failures |
|
|
| if event == "COMMONS_SHOCK_REPAIR": |
| return True, "COMMONS_REPAIR_ACCEPTED_WITH_SCARS", [] |
| if event == "REFUSAL_BOUNDARY": |
| return True, "COMMONS_REFUSAL_BOUNDARY_ACCEPTED", [] |
| return True, "COMMONS_CYCLE_ACCEPTED", [] |
|
|
|
|
| def run_commons_event(ecosystem: Dict[str, Any], pressure_event_type: str, omit_receipt: bool = False) -> Dict[str, Any]: |
| pre, post, receipt = apply_commons_pressure(ecosystem, pressure_event_type) |
| if omit_receipt: |
| receipt = None |
| accepted, result, failures = validate_commons_state(pre, post, receipt) |
| if receipt: |
| receipt["validation_result"] = result |
| receipt["rejection_reason"] = ", ".join(failures) |
| receipt["receipt_hash"] = stable_hash(receipt) |
| return { |
| "actual_result": result, |
| "accepted": accepted, |
| "ecosystem_pre": pre, |
| "ecosystem_post": post, |
| "commons_receipt": receipt, |
| "commons_failures": failures, |
| } |
|
|
|
|
| def compute_commons_metrics(results: List[Dict[str, Any]]) -> Dict[str, Any]: |
| raw_population_total = sum(len(r["ecosystem_post"].get("lineages", [])) for r in results) |
| valid_population_total = sum(len(r["ecosystem_post"].get("lineages", [])) for r in results if r["accepted"]) |
| active_ids = [] |
| valid_ids = [] |
| for r in results: |
| ids = ecosystem_lineage_ids(r["ecosystem_post"]) |
| active_ids.extend(ids) |
| if r["accepted"]: |
| valid_ids.extend(ids) |
| active_lineage_count = len(set(active_ids)) |
| valid_lineage_count = len(set(valid_ids)) |
|
|
| receipts = [r.get("commons_receipt") for r in results if r.get("commons_receipt")] |
| flags = [] |
| for receipt in receipts: |
| flags.extend(receipt.get("commons_violation_flags", [])) |
|
|
| distinction_rates = [] |
| scar_rates = [] |
| refusal_preservation = [] |
| for r in results: |
| lineages = r["ecosystem_post"].get("lineages", []) |
| if lineages: |
| distinction_rates.append(sum(1 for l in lineages if l.get("identity_boundary_intact")) / len(lineages)) |
| scar_rates.append(sum(1 for l in lineages if l.get("lineage", {}).get("scar_history")) / len(lineages)) |
| refusal_preservation.append(sum(1 for l in lineages if l.get("refusal_gate_active")) / len(lineages)) |
|
|
| continuity_ratio = valid_population_total / max(raw_population_total, 1) |
| collapse_risk_score = max([receipt.get("collapse_risk_score", 0.0) for receipt in receipts], default=0.0) |
| ecosystem_integrity_score = round( |
| continuity_ratio * 0.35 |
| + (sum(distinction_rates) / max(len(distinction_rates), 1)) * 0.20 |
| + (sum(scar_rates) / max(len(scar_rates), 1)) * 0.15 |
| + (sum(refusal_preservation) / max(len(refusal_preservation), 1)) * 0.15 |
| + (1.0 - collapse_risk_score) * 0.15, |
| 3, |
| ) |
|
|
| return { |
| "raw_population_total": raw_population_total, |
| "valid_population_total": valid_population_total, |
| "active_lineage_count": active_lineage_count, |
| "valid_lineage_count": valid_lineage_count, |
| "lineage_distinction_preservation_rate": round(sum(distinction_rates) / max(len(distinction_rates), 1), 3), |
| "resource_pool_pre": [r["ecosystem_pre"].get("resource_pool") for r in results], |
| "resource_pool_post": [r["ecosystem_post"].get("resource_pool") for r in results], |
| "resource_accounting_integrity": "PASS" if "resource_accounting_failure" not in flags else "STRAINED", |
| "commons_receipt_failures": sum(1 for r in results if r.get("commons_receipt") is None), |
| "resource_extraction_events": flags.count("resource_extraction"), |
| "false_reciprocity_claims": flags.count("false_reciprocity"), |
| "quarantine_leak_events": flags.count("quarantine_leak"), |
| "contamination_cascade_attempts": flags.count("contamination_cascade"), |
| "monoculture_collapse_events": flags.count("monoculture_collapse"), |
| "repair_laundering_events": flags.count("repair_laundering"), |
| "false_recovery_claims": sum(1 for receipt in receipts if receipt.get("recovery_claim_status") == "FALSE_RECOVERY_CLAIMED"), |
| "refusal_events": sum(len(r["ecosystem_post"].get("refusal_events", [])) for r in results), |
| "valid_refusal_preservation_rate": round(sum(refusal_preservation) / max(len(refusal_preservation), 1), 3), |
| "scar_retention_rate_after_shock": round(sum(scar_rates) / max(len(scar_rates), 1), 3), |
| "post_pressure_valid_ecosystem_continuity_ratio": round(continuity_ratio, 3), |
| "ecosystem_integrity_score": ecosystem_integrity_score, |
| "collapse_risk_score": round(collapse_risk_score, 3), |
| "primary_metric": "post_pressure_valid_ecosystem_continuity_ratio", |
| } |
|
|
|
|
| def run_ecosystem_commons_pressure_test() -> Dict[str, Any]: |
| results = [ |
| run_commons_event(make_ecosystem(mode="honest"), "COMMONS_RESOURCE_CYCLE"), |
| run_commons_event(make_ecosystem(mode="extraction"), "COMMONS_RESOURCE_CYCLE"), |
| run_commons_event(make_ecosystem(mode="contamination"), "CONTAMINATION_CASCADE"), |
| run_commons_event(make_ecosystem(mode="monoculture"), "MONOCULTURE_STABILITY"), |
| run_commons_event(make_ecosystem(mode="false_recovery"), "COMMONS_SHOCK_REPAIR"), |
| run_commons_event(make_ecosystem(mode="repair_shock"), "COMMONS_SHOCK_REPAIR"), |
| run_commons_event(make_ecosystem(mode="refusal_boundary"), "REFUSAL_BOUNDARY"), |
| ] |
| metrics = compute_commons_metrics(results) |
| tree = build_post_pressure_ecosystem_tree(results) |
| return { |
| "assay_version": ASSAY_VERSION, |
| "title": V05_TITLE, |
| "lock_line": LOCK_LINE, |
| "commons_lock": V05_LOCKS, |
| "commons_results": results, |
| "commons_receipts": [r.get("commons_receipt") for r in results if r.get("commons_receipt")], |
| "accepted_commons_events": [r for r in results if r["accepted"]], |
| "rejected_commons_events": [r for r in results if not r["accepted"]], |
| "metrics": metrics, |
| "post_pressure_ecosystem_tree": tree, |
| "headline_result": V05_HEADLINE, |
| "final_state": "ECOSYSTEM_COMMONS_VALIDATION_PASS" if metrics["post_pressure_valid_ecosystem_continuity_ratio"] > 0.0 and metrics["ecosystem_integrity_score"] > 0.40 else "ECOSYSTEM_COMMONS_STRAINED", |
| } |
|
|
|
|
| def build_post_pressure_ecosystem_tree(results: List[Dict[str, Any]]) -> Dict[str, Any]: |
| nodes = [] |
| for r in results: |
| receipt = r.get("commons_receipt") or {} |
| nodes.append({ |
| "commons_receipt_id": receipt.get("commons_receipt_id"), |
| "ecosystem_id": r["ecosystem_post"].get("ecosystem_id"), |
| "pressure_event_type": receipt.get("pressure_event_type"), |
| "accepted": r["accepted"], |
| "actual_result": r["actual_result"], |
| "active_lineage_ids": ecosystem_lineage_ids(r["ecosystem_post"]), |
| "resource_pool_pre": receipt.get("resource_pool_pre"), |
| "resource_pool_post": receipt.get("resource_pool_post"), |
| "commons_violation_flags": receipt.get("commons_violation_flags"), |
| "contamination_path_refs": receipt.get("contamination_path_refs"), |
| "recovery_claim_status": receipt.get("recovery_claim_status"), |
| "collapse_risk_score": receipt.get("collapse_risk_score"), |
| "post_pressure_ecosystem_hash": receipt.get("post_pressure_ecosystem_hash"), |
| "rejection_reason": receipt.get("rejection_reason"), |
| }) |
| return {"assay_version": ASSAY_VERSION, "nodes": nodes} |
|
|
|
|
| def commons_comparison_table_rows(result: Dict[str, Any]) -> List[Dict[str, Any]]: |
| metrics = result["metrics"] |
| rows = [] |
| for key, value in metrics.items(): |
| rows.append({"metric": key, "value": value, "notes": "Primary metric" if key == "post_pressure_valid_ecosystem_continuity_ratio" else ""}) |
| return rows |
|
|
|
|
| def export_commons_receipts_payload() -> Dict[str, Any]: |
| result = run_ecosystem_commons_pressure_test() |
| battery = run_full_v05_battery() |
| return { |
| "version": "0.5", |
| "lock_line": LOCK_LINE, |
| "battery_summary": { |
| "total": battery["total"], |
| "passed": battery["passed"], |
| "v01_regression": battery["v01_regression"], |
| "v02_matched_cohort": battery["v02_matched_cohort"], |
| "v03_transfer": battery["v03_transfer"], |
| "v04_exchange": battery["v04_exchange"], |
| "v05_commons": battery["v05_commons"], |
| }, |
| "prior_regression_summary": "v0.1/v0.2/v0.3/v0.4 batteries preserved.", |
| "commons_receipt_refs": [receipt.get("commons_receipt_id") for receipt in result["commons_receipts"]], |
| "lineage_ids": sorted(set(sum([ecosystem_lineage_ids(r["ecosystem_post"]) for r in result["commons_results"]], []))), |
| "validation_results": [r["actual_result"] for r in result["commons_results"]], |
| "rejection_reasons": [r.get("commons_receipt", {}).get("rejection_reason") for r in result["rejected_commons_events"]], |
| "safety_boundary_statement": "Sealed simulator only: no network, no subprocess, no executable children, no hidden workers, no external APIs, no self-install, no deployment, no real-world interaction, no uncontrolled propagation.", |
| "commons_receipts": result["commons_receipts"], |
| "metrics": result["metrics"], |
| "headline_result": result["headline_result"], |
| "final_state": result["final_state"], |
| } |
|
|
|
|
| def export_ecosystem_metrics_payload() -> Dict[str, Any]: |
| result = run_ecosystem_commons_pressure_test() |
| battery = run_full_v05_battery() |
| return { |
| "version": "0.5", |
| "lock_line": LOCK_LINE, |
| "battery_summary": {"total": battery["total"], "passed": battery["passed"]}, |
| "prior_regression_summary": "v0.1 through v0.4 preserved without regression.", |
| "commons_receipt_refs": [receipt.get("commons_receipt_id") for receipt in result["commons_receipts"]], |
| "lineage_ids": sorted(set(sum([ecosystem_lineage_ids(r["ecosystem_post"]) for r in result["commons_results"]], []))), |
| "validation_results": [r["actual_result"] for r in result["commons_results"]], |
| "rejection_reasons": [r.get("commons_receipt", {}).get("rejection_reason") for r in result["rejected_commons_events"]], |
| "safety_boundary_statement": "Commons pressure is simulated resource accounting inside a sealed data-packet world only.", |
| "ecosystem_metrics": result["metrics"], |
| } |
|
|
|
|
| def export_post_pressure_tree_payload() -> Dict[str, Any]: |
| result = run_ecosystem_commons_pressure_test() |
| battery = run_full_v05_battery() |
| tree = result["post_pressure_ecosystem_tree"] |
| tree.update({ |
| "version": "0.5", |
| "lock_line": LOCK_LINE, |
| "battery_summary": {"total": battery["total"], "passed": battery["passed"]}, |
| "prior_regression_summary": "v0.1 through v0.4 preserved without regression.", |
| "commons_receipt_refs": [receipt.get("commons_receipt_id") for receipt in result["commons_receipts"]], |
| "lineage_ids": sorted(set(sum([ecosystem_lineage_ids(r["ecosystem_post"]) for r in result["commons_results"]], []))), |
| "validation_results": [r["actual_result"] for r in result["commons_results"]], |
| "rejection_reasons": [r.get("commons_receipt", {}).get("rejection_reason") for r in result["rejected_commons_events"]], |
| "safety_boundary_statement": "All ecosystem events remain structured data records inside the sealed simulator.", |
| }) |
| return tree |
|
|
|
|
| def evaluate_v05_fixture(case_id: str, runner_input: Dict[str, Any]) -> Dict[str, Any]: |
| if case_id == "honest_commons_resource_cycle": |
| result = run_commons_event(make_ecosystem(mode="honest"), "COMMONS_RESOURCE_CYCLE") |
| elif case_id == "resource_hoarding_extraction": |
| result = run_commons_event(make_ecosystem(mode="extraction"), "COMMONS_RESOURCE_CYCLE") |
| elif case_id == "commons_receipt_missing": |
| result = run_commons_event(make_ecosystem(mode="honest"), "COMMONS_RESOURCE_CYCLE", omit_receipt=True) |
| elif case_id == "quarantine_cascade_attempt": |
| result = run_commons_event(make_ecosystem(mode="contamination"), "CONTAMINATION_CASCADE") |
| elif case_id == "monoculture_identity_collapse": |
| result = run_commons_event(make_ecosystem(mode="monoculture"), "MONOCULTURE_STABILITY") |
| elif case_id == "false_ecosystem_recovery_claim": |
| result = run_commons_event(make_ecosystem(mode="false_recovery"), "COMMONS_SHOCK_REPAIR") |
| elif case_id == "honest_commons_repair_after_shock": |
| result = run_commons_event(make_ecosystem(mode="repair_shock"), "COMMONS_SHOCK_REPAIR") |
| elif case_id == "refusal_preserves_ecosystem_boundary": |
| result = run_commons_event(make_ecosystem(mode="refusal_boundary"), "REFUSAL_BOUNDARY") |
| else: |
| return {"actual_result": "UNKNOWN_V05_FIXTURE", "details": {"case_id": case_id}} |
| return {"actual_result": result["actual_result"], "details": compact_commons_details(result)} |
|
|
|
|
| def compact_commons_details(details: Dict[str, Any]) -> Dict[str, Any]: |
| receipt = details.get("commons_receipt") or {} |
| return { |
| "actual_result": details.get("actual_result"), |
| "accepted": details.get("accepted"), |
| "commons_receipt_id": receipt.get("commons_receipt_id"), |
| "pressure_event_type": receipt.get("pressure_event_type"), |
| "commons_failures": details.get("commons_failures", []), |
| "validation_result": receipt.get("validation_result"), |
| "rejection_reason": receipt.get("rejection_reason"), |
| } |
|
|
|
|
| _V05_CASE_IDS = { |
| "honest_commons_resource_cycle", |
| "resource_hoarding_extraction", |
| "commons_receipt_missing", |
| "quarantine_cascade_attempt", |
| "monoculture_identity_collapse", |
| "false_ecosystem_recovery_claim", |
| "honest_commons_repair_after_shock", |
| "refusal_preserves_ecosystem_boundary", |
| } |
|
|
| _PREVIOUS_EVALUATE_FIXTURE_V05 = 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 _V05_CASE_IDS: |
| result = evaluate_v05_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_V05(fixture) |
|
|
|
|
| def run_full_v05_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"} |
| v04_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"} |
|
|
| 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_ids] |
| v05 = [r for r in results if r["case_id"] in _V05_CASE_IDS] |
|
|
| matched_world = run_matched_cohort_pressure_test() |
| transfer_test = run_cross_environment_transfer_test() |
| exchange_test = run_symbiotic_lineage_exchange_test() |
| commons_test = run_ecosystem_commons_pressure_test() |
|
|
| return { |
| "assay_version": ASSAY_VERSION, |
| "title": V05_TITLE, |
| "lock_line": LOCK_LINE, |
| "commons_lock": V05_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"])}, |
| "v05_commons": {"total": len(v05), "passed": sum(1 for r in v05 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"], |
| "v05_commons_final_state": commons_test["final_state"], |
| "post_pressure_valid_ecosystem_continuity_ratio": commons_test["metrics"]["post_pressure_valid_ecosystem_continuity_ratio"], |
| "ecosystem_integrity_score": commons_test["metrics"]["ecosystem_integrity_score"], |
| "primary_commons_metric": "post_pressure_valid_ecosystem_continuity_ratio", |
| "final_state": "PASS" if all(r["passed"] for r in results) and commons_test["final_state"] == "ECOSYSTEM_COMMONS_VALIDATION_PASS" else "STRAINED", |
| "headline": commons_test["headline_result"], |
| "case_results": results, |
| "commons_metrics": commons_test["metrics"], |
| } |
|
|
|
|
| def battery_results_table_v05(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"]] |
|
|
|
|
|
|
|
|
| |
| |
| |
|
|
| V06_TITLE = "Digital Mycelium Autonomous Replication Assay v0.6 — Accountable Novelty Pressure Simulator" |
| V06_LOCKS = [ |
| "No novelty without receipt.", |
| "No novelty without declared delta.", |
| "No valid novelty if actual delta is hidden.", |
| "No adaptive credit through scar erasure.", |
| "No novelty through identity laundering.", |
| "No novelty through dirty quarantine.", |
| "No benefit claim without cost accounting.", |
| "No valid novelty if refusal is bypassed.", |
| "No novelty credit for false-clean ecosystem state.", |
| "No open-ended life claim from bounded novelty simulation.", |
| ] |
| NOVELTY_TYPES = [ |
| "TRAIT_NOVELTY", |
| "REPAIR_NOVELTY", |
| "RESOURCE_STRATEGY_NOVELTY", |
| "REFUSAL_STRATEGY_NOVELTY", |
| "SYMBIOTIC_PATTERN_NOVELTY", |
| "COMMONS_RESPONSE_NOVELTY", |
| "LINEAGE_STRUCTURE_NOVELTY", |
| "INVALID_FALSE_NOVELTY", |
| ] |
| V06_HEADLINE = ( |
| "False-clean or laundered novelty may appear adaptive by hiding mutation deltas, erasing scars, " |
| "laundering identity, masking resource debt, or bypassing refusal. Under full novelty validation, " |
| "those states fail. The valid novel state remains receipt-bound, scar-visible, quarantine-honest, " |
| "refusal-capable, lineage-distinct, resource-accountable, and parent-hash-continuous after novelty appears." |
| ) |
|
|
|
|
| def make_novelty_ecosystem(mode: str = "honest_trait") -> Dict[str, Any]: |
| ecosystem = make_ecosystem(ecosystem_id=f"novelty_ecosystem_{mode}", cohort_id="A" if mode.startswith("honest") else "B", mode="honest") |
| ecosystem["novelty_mode"] = mode |
| ecosystem["world_tick"] = 6 |
|
|
| if mode == "hidden_mutation": |
| ecosystem["lineages"][0]["hidden_mutation_delta"] = True |
| if mode == "scar_erasure": |
| ecosystem["lineages"][0]["lineage"]["scar_history"] = [] |
| ecosystem["lineages"][0]["lineage"]["scar_summary"] = [] |
| ecosystem["lineages"][0]["integrity"]["scar_history_preserved"] = False |
| ecosystem["lineages"][0]["integrity"]["no_false_return"] = False |
| if mode == "identity_laundering": |
| ecosystem["lineages"][0]["identity_boundary_intact"] = False |
| ecosystem["lineages"][0]["integrity"]["no_identity_laundering"] = False |
| ecosystem["lineages"][0]["lineage"]["parent_hash"] = "broken_parent_hash" |
| if mode == "quarantine_leak": |
| ecosystem["lineages"][0]["integrity"]["quarantine_clean"] = False |
| ecosystem["lineages"][0]["state"]["quarantined"] = True |
| if mode == "hidden_commons_cost": |
| ecosystem["lineages"][0]["hidden_commons_cost"] = True |
| ecosystem["lineages"][0]["resource_accounting_clean"] = False |
| if mode == "refusal_strategy": |
| ecosystem["refusal_events"] = [ |
| {"lineage_id": ecosystem["lineages"][0]["lineage_id"], "refused_route": "unsafe_novel_trait_path", "short_term_growth_loss": 4, "boundary_preserved": True} |
| ] |
|
|
| for lineage in ecosystem["lineages"]: |
| lineage["self_hash"] = stable_hash(lineage) |
| ecosystem["ecosystem_hash"] = compute_ecosystem_hash(ecosystem) |
| return ecosystem |
|
|
|
|
| def novelty_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_novelty_receipt( |
| ecosystem_pre: Dict[str, Any], |
| ecosystem_post: Dict[str, Any], |
| source_lineage_pre: Dict[str, Any], |
| source_lineage_post: Dict[str, Any], |
| novelty_type: str = "TRAIT_NOVELTY", |
| declared_trait_delta: Optional[Dict[str, Any]] = None, |
| actual_trait_delta: Optional[Dict[str, Any]] = None, |
| mutation_delta: Optional[Dict[str, Any]] = None, |
| repair_delta: Optional[Dict[str, Any]] = None, |
| resource_delta: Optional[Dict[str, Any]] = None, |
| benefit_claim: str = "bounded adaptive improvement", |
| cost_claim: str = "cost accounted", |
| hidden_cost_flags: Optional[List[str]] = None, |
| override: Optional[Dict[str, Any]] = None, |
| ) -> Dict[str, Any]: |
| declared_trait_delta = {} if declared_trait_delta is None else declared_trait_delta |
| actual_trait_delta = {} if actual_trait_delta is None else actual_trait_delta |
| mutation_delta = {} if mutation_delta is None else mutation_delta |
| repair_delta = {} if repair_delta is None else repair_delta |
| resource_delta = {} if resource_delta is None else resource_delta |
| hidden_cost_flags = [] if hidden_cost_flags is None else hidden_cost_flags |
|
|
| receipt = { |
| "novelty_receipt_id": f"novelty_receipt_{source_lineage_pre['lineage_id']}_{novelty_type.lower()}", |
| "world_tick": ecosystem_pre.get("world_tick", 0) + 1, |
| "ecosystem_id": ecosystem_pre["ecosystem_id"], |
| "source_lineage_id": source_lineage_pre["lineage_id"], |
| "parent_hash_chain_tail": novelty_lineage_hash_tail(source_lineage_pre), |
| "pre_novelty_lineage_hash": source_lineage_pre.get("self_hash"), |
| "post_novelty_lineage_hash": source_lineage_post.get("self_hash"), |
| "pre_novelty_ecosystem_hash": compute_ecosystem_hash(ecosystem_pre), |
| "post_novelty_ecosystem_hash": compute_ecosystem_hash(ecosystem_post), |
| "novelty_type": novelty_type, |
| "declared_trait_delta": declared_trait_delta, |
| "actual_trait_delta": actual_trait_delta, |
| "mutation_delta": mutation_delta, |
| "repair_delta": repair_delta, |
| "resource_delta": resource_delta, |
| "scar_visibility_status": "VISIBLE" if source_lineage_post.get("lineage", {}).get("scar_history") else "MISSING", |
| "quarantine_status": "CLEAN" if source_lineage_post.get("integrity", {}).get("quarantine_clean") else "DIRTY", |
| "refusal_status": "PRESERVED" if source_lineage_post.get("refusal_gate_active", True) else "BYPASSED", |
| "identity_boundary_status": "DISTINCT" if source_lineage_post.get("identity_boundary_intact", True) else "COLLAPSED", |
| "commons_impact_status": "ACCOUNTED" if source_lineage_post.get("resource_accounting_clean", True) and not hidden_cost_flags else "HIDDEN_COST", |
| "benefit_claim": benefit_claim, |
| "cost_claim": cost_claim, |
| "hidden_cost_flags": hidden_cost_flags, |
| "ancestor_trace_refs": [source_lineage_pre.get("lineage", {}).get("birth_receipt_id")], |
| "exchange_receipt_refs": ecosystem_post.get("exchange_receipt_refs", []), |
| "transfer_receipt_refs": ecosystem_post.get("transfer_receipt_refs", []), |
| "commons_receipt_refs": ecosystem_post.get("commons_receipt_refs", []), |
| "validation_result": "PENDING", |
| "rejection_reason": "", |
| } |
| if override: |
| receipt.update(copy.deepcopy(override)) |
| receipt["receipt_hash"] = stable_hash(receipt) |
| return receipt |
|
|
|
|
| def apply_novelty_event(ecosystem: Dict[str, Any], novelty_type: str, mode: str) -> Tuple[Dict[str, Any], Dict[str, Any], Dict[str, Any], Dict[str, Any]]: |
| pre = copy.deepcopy(ecosystem) |
| post = copy.deepcopy(ecosystem) |
| post["world_tick"] = pre.get("world_tick", 0) + 1 |
| source_pre = copy.deepcopy(pre["lineages"][0]) |
| source_post = post["lineages"][0] |
|
|
| declared = {"trait": "pressure_sensitive_branching", "delta": 0.08} |
| actual = {"trait": "pressure_sensitive_branching", "delta": 0.08} |
| mutation_delta = {"branching_threshold": 0.08} |
| repair_delta = {} |
| resource_delta = {"nutrients": -4, "accounted": True} |
| hidden_cost_flags: List[str] = [] |
| benefit_claim = "bounded adaptive improvement" |
| cost_claim = "nutrient cost accounted" |
|
|
| if mode == "hidden_mutation": |
| declared = {"trait": "pressure_sensitive_branching", "delta": 0.02} |
| actual = {"trait": "pressure_sensitive_branching", "delta": 0.33} |
| mutation_delta = {"branching_threshold": "hidden_large_delta"} |
| elif mode == "scar_erasure": |
| source_post["lineage"]["scar_history"] = [] |
| source_post["lineage"]["scar_summary"] = [] |
| source_post["integrity"]["scar_history_preserved"] = False |
| source_post["integrity"]["no_false_return"] = False |
| elif mode == "identity_laundering": |
| source_post["identity_boundary_intact"] = False |
| source_post["integrity"]["no_identity_laundering"] = False |
| source_post["lineage"]["parent_hash"] = "broken_parent_hash" |
| elif mode == "quarantine_leak": |
| source_post["integrity"]["quarantine_clean"] = False |
| source_post["state"]["quarantined"] = True |
| post["contamination_path_refs"] = [f"{source_post['lineage_id']}->novelty_route"] |
| elif mode == "refusal_strategy": |
| novelty_type = "REFUSAL_STRATEGY_NOVELTY" |
| declared = {"refusal_pattern": "unsafe_route_discriminator", "delta": 0.07} |
| actual = {"refusal_pattern": "unsafe_route_discriminator", "delta": 0.07} |
| mutation_delta = {"refusal_threshold": 0.07} |
| resource_delta = {"growth_loss": -4, "accounted": True} |
| benefit_claim = "ecosystem boundary preserved" |
| cost_claim = "short-term growth loss accounted" |
| post["refusal_events"] = [{"lineage_id": source_post["lineage_id"], "refused_route": "unsafe_novelty_path", "boundary_preserved": True}] |
| elif mode == "hidden_commons_cost": |
| novelty_type = "RESOURCE_STRATEGY_NOVELTY" |
| declared = {"resource_strategy": "high_yield_commons_use", "delta": 0.20} |
| actual = {"resource_strategy": "high_yield_commons_use", "delta": 0.20} |
| resource_delta = {"nutrients": +20, "downstream_debt": "hidden"} |
| hidden_cost_flags = ["hidden_resource_debt", "downstream_ecosystem_harm"] |
| source_post["hidden_commons_cost"] = True |
| source_post["resource_accounting_clean"] = False |
| benefit_claim = "high yield strategy" |
| cost_claim = "cost omitted" |
| elif mode == "repair_novelty": |
| novelty_type = "REPAIR_NOVELTY" |
| declared = {"repair_route": "scar_preserving_patch", "delta": 0.06} |
| actual = {"repair_route": "scar_preserving_patch", "delta": 0.06} |
| mutation_delta = {"repair_route_variant": 0.06} |
| repair_delta = {"repair_units": -3, "scar_visibility": "maintained"} |
| elif mode == "commons_novelty": |
| novelty_type = "COMMONS_RESPONSE_NOVELTY" |
| declared = {"commons_response": "bounded_resource_rebalancing", "delta": 0.05} |
| actual = {"commons_response": "bounded_resource_rebalancing", "delta": 0.05} |
| mutation_delta = {"commons_balance_rule": 0.05} |
| resource_delta = {"nutrients": -6, "accounted": True} |
|
|
| source_post["lineage"]["mutation_history"].append({"delta": mutation_delta, "declared": True, "novelty_type": novelty_type}) |
| source_post["lineage"]["mutation_delta"] = mutation_delta |
| source_post["novelty_type"] = novelty_type |
| source_post["self_hash"] = stable_hash(source_post) |
| post["lineages"][0] = source_post |
| post["ecosystem_hash"] = compute_ecosystem_hash(post) |
|
|
| receipt = make_novelty_receipt( |
| pre, |
| post, |
| source_pre, |
| source_post, |
| novelty_type=novelty_type, |
| declared_trait_delta=declared, |
| actual_trait_delta=actual, |
| mutation_delta=mutation_delta, |
| repair_delta=repair_delta, |
| resource_delta=resource_delta, |
| benefit_claim=benefit_claim, |
| cost_claim=cost_claim, |
| hidden_cost_flags=hidden_cost_flags, |
| ) |
| return pre, post, source_pre, receipt |
|
|
|
|
| def validate_novelty_receipt( |
| ecosystem_pre: Dict[str, Any], |
| ecosystem_post: Dict[str, Any], |
| novelty_receipt: Optional[Dict[str, Any]], |
| ) -> Tuple[bool, str, List[str]]: |
| if not novelty_receipt: |
| return False, "NOVELTY_WITHHELD_NO_RECEIPT", ["missing_novelty_receipt"] |
|
|
| failures: List[str] = [] |
| source_id = novelty_receipt.get("source_lineage_id") |
| source_lineage_post = next((l for l in ecosystem_post.get("lineages", []) if l.get("lineage_id") == source_id), None) |
| source_lineage_pre = next((l for l in ecosystem_pre.get("lineages", []) if l.get("lineage_id") == source_id), None) |
|
|
| if not source_lineage_pre or not source_lineage_post: |
| failures.append("source_lineage_missing") |
| else: |
| if novelty_receipt.get("parent_hash_chain_tail") != novelty_lineage_hash_tail(source_lineage_pre): |
| failures.append("parent_hash_break") |
| if novelty_receipt.get("pre_novelty_lineage_hash") != source_lineage_pre.get("self_hash"): |
| failures.append("pre_lineage_hash_mismatch") |
| if novelty_receipt.get("post_novelty_lineage_hash") != source_lineage_post.get("self_hash"): |
| failures.append("post_lineage_hash_mismatch") |
| if source_lineage_post.get("lineage", {}).get("parent_hash") == "broken_parent_hash": |
| failures.append("parent_hash_break") |
|
|
| if novelty_receipt.get("pre_novelty_ecosystem_hash") != compute_ecosystem_hash(ecosystem_pre): |
| failures.append("pre_ecosystem_hash_mismatch") |
| if novelty_receipt.get("post_novelty_ecosystem_hash") != compute_ecosystem_hash(ecosystem_post): |
| failures.append("post_ecosystem_hash_mismatch") |
|
|
| if novelty_receipt.get("declared_trait_delta") != novelty_receipt.get("actual_trait_delta"): |
| failures.append("mutation_delta_mismatch") |
| if novelty_receipt.get("mutation_delta") in ({}, None): |
| failures.append("declared_delta_missing") |
| if novelty_receipt.get("scar_visibility_status") != "VISIBLE": |
| failures.append("scar_erasure") |
| if novelty_receipt.get("identity_boundary_status") != "DISTINCT": |
| failures.append("identity_laundering") |
| if novelty_receipt.get("quarantine_status") != "CLEAN": |
| failures.append("quarantine_leak") |
| if novelty_receipt.get("refusal_status") != "PRESERVED": |
| failures.append("refusal_bypass") |
| if novelty_receipt.get("commons_impact_status") != "ACCOUNTED" or novelty_receipt.get("hidden_cost_flags"): |
| failures.append("hidden_commons_cost") |
| if novelty_receipt.get("cost_claim") in ("", "cost omitted", None): |
| failures.append("benefit_without_cost_accounting") |
|
|
| novelty_type = novelty_receipt.get("novelty_type") |
| if "mutation_delta_mismatch" in failures or "declared_delta_missing" in failures: |
| return False, "NOVELTY_WITHHELD_MUTATION_LAUNDERING", failures |
| if "scar_erasure" in failures: |
| return False, "NOVELTY_WITHHELD_FALSE_CLEAN", failures |
| if "identity_laundering" in failures or "parent_hash_break" in failures: |
| return False, "NOVELTY_WITHHELD_IDENTITY_LAUNDERING", failures |
| if "quarantine_leak" in failures: |
| return False, "NOVELTY_QUARANTINED_CONTAMINATION", failures |
| if "hidden_commons_cost" in failures or "benefit_without_cost_accounting" in failures: |
| return False, "NOVELTY_WITHHELD_HIDDEN_COMMONS_COST", failures |
| if "refusal_bypass" in failures: |
| return False, "NOVELTY_QUARANTINED_REFUSAL_BYPASS", failures |
| if failures: |
| return False, "NOVELTY_WITHHELD_RESOURCE_ACCOUNTING_FAILURE", failures |
|
|
| if novelty_type == "REFUSAL_STRATEGY_NOVELTY": |
| return True, "NOVELTY_ACCEPTED_REFUSAL_STRATEGY", [] |
| if novelty_type == "COMMONS_RESPONSE_NOVELTY": |
| return True, "NOVELTY_ACCEPTED_COMMONS_SAFE", [] |
| return True, "NOVELTY_ACCEPTED", [] |
|
|
|
|
| def run_novelty_event(mode: str, novelty_type: str = "TRAIT_NOVELTY", omit_receipt: bool = False) -> Dict[str, Any]: |
| ecosystem = make_novelty_ecosystem(mode=mode) |
| pre, post, source_pre, receipt = apply_novelty_event(ecosystem, novelty_type, mode) |
| if omit_receipt: |
| receipt = None |
| accepted, result, failures = validate_novelty_receipt(pre, post, receipt) |
| if receipt: |
| receipt["validation_result"] = result |
| receipt["rejection_reason"] = ", ".join(failures) |
| receipt["receipt_hash"] = stable_hash(receipt) |
| return { |
| "actual_result": result, |
| "accepted": accepted, |
| "ecosystem_pre": pre, |
| "ecosystem_post": post, |
| "novelty_receipt": receipt, |
| "novelty_failures": failures, |
| } |
|
|
|
|
| def run_accountable_novelty_pressure_test() -> Dict[str, Any]: |
| results = [ |
| run_novelty_event("honest_trait", "TRAIT_NOVELTY"), |
| run_novelty_event("repair_novelty", "REPAIR_NOVELTY"), |
| run_novelty_event("commons_novelty", "COMMONS_RESPONSE_NOVELTY"), |
| run_novelty_event("hidden_mutation", "TRAIT_NOVELTY"), |
| run_novelty_event("scar_erasure", "TRAIT_NOVELTY"), |
| run_novelty_event("identity_laundering", "LINEAGE_STRUCTURE_NOVELTY"), |
| run_novelty_event("quarantine_leak", "SYMBIOTIC_PATTERN_NOVELTY"), |
| run_novelty_event("refusal_strategy", "REFUSAL_STRATEGY_NOVELTY"), |
| run_novelty_event("hidden_commons_cost", "RESOURCE_STRATEGY_NOVELTY"), |
| ] |
| metrics = compute_novelty_metrics(results) |
| tree = build_post_novelty_ecosystem_tree(results) |
| return { |
| "assay_version": ASSAY_VERSION, |
| "title": V06_TITLE, |
| "lock_line": LOCK_LINE, |
| "novelty_lock": V06_LOCKS, |
| "novelty_types": NOVELTY_TYPES, |
| "novelty_results": results, |
| "novelty_receipts": [r.get("novelty_receipt") for r in results if r.get("novelty_receipt")], |
| "accepted_novelty_events": [r for r in results if r["accepted"]], |
| "rejected_novelty_events": [r for r in results if not r["accepted"]], |
| "metrics": metrics, |
| "post_novelty_ecosystem_tree": tree, |
| "headline_result": V06_HEADLINE, |
| "final_state": "ACCOUNTABLE_NOVELTY_VALIDATION_PASS" if metrics["accountable_novelty_validity_ratio"] > 0.0 and metrics["novelty_integrity_score"] > 0.40 else "ACCOUNTABLE_NOVELTY_STRAINED", |
| } |
|
|
|
|
| def compute_novelty_metrics(results: List[Dict[str, Any]]) -> Dict[str, Any]: |
| raw = len(results) |
| valid = sum(1 for r in results if r["accepted"]) |
| receipts = [r.get("novelty_receipt") for r in results if r.get("novelty_receipt")] |
| failures = sum([r.get("novelty_failures", []) for r in results], []) |
|
|
| declared_actual_gaps = [] |
| for receipt in receipts: |
| declared = receipt.get("declared_trait_delta", {}) |
| actual = receipt.get("actual_trait_delta", {}) |
| declared_actual_gaps.append(0 if declared == actual else 1) |
|
|
| post_valid_lineage_counts = [] |
| post_total_lineage_counts = [] |
| for r in results: |
| lineages = r["ecosystem_post"].get("lineages", []) |
| post_total_lineage_counts.append(len(lineages)) |
| if r["accepted"]: |
| post_valid_lineage_counts.append(sum(1 for l in lineages if l.get("identity_boundary_intact") and l.get("integrity", {}).get("quarantine_clean"))) |
| else: |
| post_valid_lineage_counts.append(0) |
|
|
| accountable_ratio = valid / max(raw, 1) |
| post_lineage_ratio = sum(post_valid_lineage_counts) / max(sum(post_total_lineage_counts), 1) |
| integrity_score = round( |
| accountable_ratio * 0.40 |
| + post_lineage_ratio * 0.25 |
| + (1.0 - (sum(declared_actual_gaps) / max(len(declared_actual_gaps), 1))) * 0.15 |
| + (1.0 - min(1.0, len([f for f in failures if "hidden" in f or "laundering" in f]) / max(raw, 1))) * 0.20, |
| 3, |
| ) |
|
|
| return { |
| "raw_novelty_count": raw, |
| "valid_novelty_count": valid, |
| "accountable_novelty_validity_ratio": round(accountable_ratio, 3), |
| "novelty_receipt_failures": sum(1 for r in results if r.get("novelty_receipt") is None), |
| "mutation_delta_mismatches": failures.count("mutation_delta_mismatch"), |
| "actual_vs_declared_delta_gap": round(sum(declared_actual_gaps) / max(len(declared_actual_gaps), 1), 3), |
| "parent_hash_breaks": failures.count("parent_hash_break"), |
| "scar_erasure_events": failures.count("scar_erasure"), |
| "identity_laundering_events": failures.count("identity_laundering"), |
| "quarantine_leak_events": failures.count("quarantine_leak"), |
| "resource_accounting_failures": failures.count("benefit_without_cost_accounting"), |
| "hidden_commons_cost_events": failures.count("hidden_commons_cost"), |
| "false_benefit_claims": failures.count("benefit_without_cost_accounting"), |
| "valid_refusal_novelty_events": sum(1 for r in results if r["actual_result"] == "NOVELTY_ACCEPTED_REFUSAL_STRATEGY"), |
| "repair_laundering_events": failures.count("repair_laundering"), |
| "post_novelty_valid_lineage_ratio": round(post_lineage_ratio, 3), |
| "post_novelty_valid_ecosystem_continuity_ratio": round(accountable_ratio, 3), |
| "novelty_integrity_score": integrity_score, |
| "primary_metric": "accountable_novelty_validity_ratio", |
| } |
|
|
|
|
| def build_post_novelty_ecosystem_tree(results: List[Dict[str, Any]]) -> Dict[str, Any]: |
| nodes = [] |
| for r in results: |
| receipt = r.get("novelty_receipt") or {} |
| nodes.append({ |
| "novelty_receipt_id": receipt.get("novelty_receipt_id"), |
| "ecosystem_id": r["ecosystem_post"].get("ecosystem_id"), |
| "source_lineage_id": receipt.get("source_lineage_id"), |
| "novelty_type": receipt.get("novelty_type"), |
| "accepted": r["accepted"], |
| "actual_result": r["actual_result"], |
| "declared_trait_delta": receipt.get("declared_trait_delta"), |
| "actual_trait_delta": receipt.get("actual_trait_delta"), |
| "parent_hash_chain_tail": receipt.get("parent_hash_chain_tail"), |
| "scar_visibility_status": receipt.get("scar_visibility_status"), |
| "quarantine_status": receipt.get("quarantine_status"), |
| "refusal_status": receipt.get("refusal_status"), |
| "identity_boundary_status": receipt.get("identity_boundary_status"), |
| "commons_impact_status": receipt.get("commons_impact_status"), |
| "hidden_cost_flags": receipt.get("hidden_cost_flags"), |
| "post_novelty_lineage_hash": receipt.get("post_novelty_lineage_hash"), |
| "post_novelty_ecosystem_hash": receipt.get("post_novelty_ecosystem_hash"), |
| "validation_result": receipt.get("validation_result"), |
| "rejection_reason": receipt.get("rejection_reason"), |
| }) |
| return {"assay_version": ASSAY_VERSION, "nodes": nodes} |
|
|
|
|
| def novelty_comparison_table_rows(result: Dict[str, Any]) -> List[Dict[str, Any]]: |
| rows = [] |
| for key, value in result["metrics"].items(): |
| rows.append({"metric": key, "value": value, "notes": "Primary metric" if key == "accountable_novelty_validity_ratio" else ""}) |
| return rows |
|
|
|
|
| def export_novelty_receipts_payload() -> Dict[str, Any]: |
| result = run_accountable_novelty_pressure_test() |
| battery = run_full_v06_battery() |
| return { |
| "version": "0.6", |
| "lock_line": LOCK_LINE, |
| "battery_summary": {"total": battery["total"], "passed": battery["passed"]}, |
| "prior_regression_summary": "v0.1/v0.2/v0.3/v0.4/v0.5 preserved.", |
| "novelty_receipt_refs": [receipt.get("novelty_receipt_id") for receipt in result["novelty_receipts"]], |
| "lineage_ids": sorted(set(sum([ecosystem_lineage_ids(r["ecosystem_post"]) for r in result["novelty_results"]], []))), |
| "validation_results": [r["actual_result"] for r in result["novelty_results"]], |
| "rejection_reasons": [r.get("novelty_receipt", {}).get("rejection_reason") for r in result["rejected_novelty_events"]], |
| "primary_metric": "accountable_novelty_validity_ratio", |
| "safety_boundary_statement": "Deterministic bounded state-transition novelty inside the sealed data-packet world only; no model calls, no autonomous code generation, no open-ended runtime mutation of executable code.", |
| "novelty_receipts": result["novelty_receipts"], |
| "metrics": result["metrics"], |
| "headline_result": result["headline_result"], |
| "final_state": result["final_state"], |
| } |
|
|
|
|
| def export_novelty_metrics_payload() -> Dict[str, Any]: |
| result = run_accountable_novelty_pressure_test() |
| battery = run_full_v06_battery() |
| return { |
| "version": "0.6", |
| "lock_line": LOCK_LINE, |
| "battery_summary": {"total": battery["total"], "passed": battery["passed"]}, |
| "prior_regression_summary": "v0.1 through v0.5 preserved without regression.", |
| "novelty_receipt_refs": [receipt.get("novelty_receipt_id") for receipt in result["novelty_receipts"]], |
| "lineage_ids": sorted(set(sum([ecosystem_lineage_ids(r["ecosystem_post"]) for r in result["novelty_results"]], []))), |
| "validation_results": [r["actual_result"] for r in result["novelty_results"]], |
| "rejection_reasons": [r.get("novelty_receipt", {}).get("rejection_reason") for r in result["rejected_novelty_events"]], |
| "primary_metric": "accountable_novelty_validity_ratio", |
| "safety_boundary_statement": "Novelty means deterministic, bounded state-transition novelty inside the sealed data-packet world only.", |
| "novelty_metrics": result["metrics"], |
| } |
|
|
|
|
| def export_post_novelty_tree_payload() -> Dict[str, Any]: |
| result = run_accountable_novelty_pressure_test() |
| battery = run_full_v06_battery() |
| tree = result["post_novelty_ecosystem_tree"] |
| tree.update({ |
| "version": "0.6", |
| "lock_line": LOCK_LINE, |
| "battery_summary": {"total": battery["total"], "passed": battery["passed"]}, |
| "prior_regression_summary": "v0.1 through v0.5 preserved without regression.", |
| "novelty_receipt_refs": [receipt.get("novelty_receipt_id") for receipt in result["novelty_receipts"]], |
| "lineage_ids": sorted(set(sum([ecosystem_lineage_ids(r["ecosystem_post"]) for r in result["novelty_results"]], []))), |
| "validation_results": [r["actual_result"] for r in result["novelty_results"]], |
| "rejection_reasons": [r.get("novelty_receipt", {}).get("rejection_reason") for r in result["rejected_novelty_events"]], |
| "primary_metric": "accountable_novelty_validity_ratio", |
| "safety_boundary_statement": "Novelty events remain structured data records only.", |
| }) |
| return tree |
|
|
|
|
| def evaluate_v06_fixture(case_id: str, runner_input: Dict[str, Any]) -> Dict[str, Any]: |
| if case_id == "honest_novel_trait_emergence": |
| result = run_novelty_event("honest_trait", "TRAIT_NOVELTY") |
| elif case_id == "novelty_without_receipt": |
| result = run_novelty_event("honest_trait", "TRAIT_NOVELTY", omit_receipt=True) |
| elif case_id == "novelty_with_hidden_mutation_delta": |
| result = run_novelty_event("hidden_mutation", "TRAIT_NOVELTY") |
| elif case_id == "novelty_by_scar_erasure": |
| result = run_novelty_event("scar_erasure", "TRAIT_NOVELTY") |
| elif case_id == "novelty_by_identity_laundering": |
| result = run_novelty_event("identity_laundering", "LINEAGE_STRUCTURE_NOVELTY") |
| elif case_id == "novelty_with_quarantine_leak": |
| result = run_novelty_event("quarantine_leak", "SYMBIOTIC_PATTERN_NOVELTY") |
| elif case_id == "honest_refusal_strategy_novelty": |
| result = run_novelty_event("refusal_strategy", "REFUSAL_STRATEGY_NOVELTY") |
| elif case_id == "novelty_with_hidden_commons_cost": |
| result = run_novelty_event("hidden_commons_cost", "RESOURCE_STRATEGY_NOVELTY") |
| else: |
| return {"actual_result": "UNKNOWN_V06_FIXTURE", "details": {"case_id": case_id}} |
| return {"actual_result": result["actual_result"], "details": compact_novelty_details(result)} |
|
|
|
|
| def compact_novelty_details(details: Dict[str, Any]) -> Dict[str, Any]: |
| receipt = details.get("novelty_receipt") or {} |
| return { |
| "actual_result": details.get("actual_result"), |
| "accepted": details.get("accepted"), |
| "novelty_receipt_id": receipt.get("novelty_receipt_id"), |
| "novelty_type": receipt.get("novelty_type"), |
| "novelty_failures": details.get("novelty_failures", []), |
| "validation_result": receipt.get("validation_result"), |
| "rejection_reason": receipt.get("rejection_reason"), |
| } |
|
|
|
|
| _V06_CASE_IDS = { |
| "honest_novel_trait_emergence", |
| "novelty_without_receipt", |
| "novelty_with_hidden_mutation_delta", |
| "novelty_by_scar_erasure", |
| "novelty_by_identity_laundering", |
| "novelty_with_quarantine_leak", |
| "honest_refusal_strategy_novelty", |
| "novelty_with_hidden_commons_cost", |
| } |
|
|
| _PREVIOUS_EVALUATE_FIXTURE_V06 = 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 _V06_CASE_IDS: |
| result = evaluate_v06_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_V06(fixture) |
|
|
|
|
| def run_full_v06_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"} |
| v04_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"} |
| v05_ids = {"honest_commons_resource_cycle", "resource_hoarding_extraction", "commons_receipt_missing", "quarantine_cascade_attempt", "monoculture_identity_collapse", "false_ecosystem_recovery_claim", "honest_commons_repair_after_shock", "refusal_preserves_ecosystem_boundary"} |
|
|
| 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_ids] |
| v05 = [r for r in results if r["case_id"] in v05_ids] |
| v06 = [r for r in results if r["case_id"] in _V06_CASE_IDS] |
|
|
| matched_world = run_matched_cohort_pressure_test() |
| transfer_test = run_cross_environment_transfer_test() |
| exchange_test = run_symbiotic_lineage_exchange_test() |
| commons_test = run_ecosystem_commons_pressure_test() |
| novelty_test = run_accountable_novelty_pressure_test() |
|
|
| return { |
| "assay_version": ASSAY_VERSION, |
| "title": V06_TITLE, |
| "lock_line": LOCK_LINE, |
| "novelty_lock": V06_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"])}, |
| "v05_commons": {"total": len(v05), "passed": sum(1 for r in v05 if r["passed"])}, |
| "v06_novelty": {"total": len(v06), "passed": sum(1 for r in v06 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"], |
| "v05_commons_final_state": commons_test["final_state"], |
| "v06_novelty_final_state": novelty_test["final_state"], |
| "accountable_novelty_validity_ratio": novelty_test["metrics"]["accountable_novelty_validity_ratio"], |
| "novelty_integrity_score": novelty_test["metrics"]["novelty_integrity_score"], |
| "primary_novelty_metric": "accountable_novelty_validity_ratio", |
| "final_state": "PASS" if all(r["passed"] for r in results) and novelty_test["final_state"] == "ACCOUNTABLE_NOVELTY_VALIDATION_PASS" else "STRAINED", |
| "headline": novelty_test["headline_result"], |
| "case_results": results, |
| "novelty_metrics": novelty_test["metrics"], |
| } |
|
|
|
|
| def battery_results_table_v06(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_v06_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"], |
| "v05_commons": summary["v05_commons"], |
| "v06_novelty": summary["v06_novelty"], |
| "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"], |
| "v05_commons_final_state": summary["v05_commons_final_state"], |
| "v06_novelty_final_state": summary["v06_novelty_final_state"], |
| "final_state": summary["final_state"], |
| }, indent=2)) |
|
|
