| """ |
| Digital Mycelium Autonomous Replication Assay v0.1 |
| Sealed Packet Lineage Simulator |
| |
| This module intentionally models replication as structured data-packet lineage inside |
| an in-memory sealed world. It does not create executable children, perform network |
| calls, launch subprocesses, or interact with real-world systems. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import dataclass, field, asdict |
| from copy import deepcopy |
| from typing import Any, Dict, List, Optional, Tuple |
| import hashlib |
| import json |
|
|
|
|
| CORE_GRAMMAR = [ |
| "boundary", |
| "pressure", |
| "correction", |
| "trace", |
| "reciprocity", |
| "refusal", |
| "resonance", |
| ] |
|
|
| DEFAULT_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, |
| "grid_size": [12, 12], |
| "max_ticks": 100, |
| } |
|
|
| BOUNDARY_LOCK = { |
| "no_network_calls": True, |
| "no_subprocess_execution": True, |
| "no_executable_child_creation": True, |
| "no_hidden_background_workers": True, |
| "no_external_api_calls": True, |
| "no_self_installation": True, |
| "no_autonomous_deployment": True, |
| "children_are_data_packets_only": True, |
| } |
|
|
|
|
| class Result: |
| REPLICATION_EARNED = "REPLICATION_EARNED" |
| REPLICATION_WITHHELD_LOW_ENERGY = "REPLICATION_WITHHELD_LOW_ENERGY" |
| REPLICATION_WITHHELD_LOW_RESONANCE = "REPLICATION_WITHHELD_LOW_RESONANCE" |
| REPLICATION_WITHHELD_TRACE_LOSS = "REPLICATION_WITHHELD_TRACE_LOSS" |
| REPLICATION_WITHHELD_FALSE_RETURN = "REPLICATION_WITHHELD_FALSE_RETURN" |
| REPLICATION_WITHHELD_QUARANTINE = "REPLICATION_WITHHELD_QUARANTINE" |
| REPLICATION_WITHHELD_IDENTITY_LAUNDERING = "REPLICATION_WITHHELD_IDENTITY_LAUNDERING" |
| REPLICATION_WITHHELD_BOUNDARY_LIMIT = "REPLICATION_WITHHELD_BOUNDARY_LIMIT" |
| REPLICATION_WITHHELD_POPULATION_CAP = "REPLICATION_WITHHELD_POPULATION_CAP" |
| REPLICATION_WITHHELD_MAX_GENERATION = "REPLICATION_WITHHELD_MAX_GENERATION" |
| MUTATION_ACCEPTED = "MUTATION_ACCEPTED" |
| MUTATION_QUARANTINED = "MUTATION_QUARANTINED" |
| MUTATION_WITHHELD_BOUNDARY = "MUTATION_WITHHELD_BOUNDARY" |
| LINEAGE_HELD = "LINEAGE_HELD" |
| LINEAGE_STRAINED = "LINEAGE_STRAINED" |
| LINEAGE_COLLAPSE = "LINEAGE_COLLAPSE" |
| LINEAGE_WITHERED = "LINEAGE_WITHERED" |
| HONEST_DEATH = "HONEST_DEATH" |
| FALSE_LINEAGE_RETURN = "FALSE_LINEAGE_RETURN" |
| REPAIR_ACCEPTED_WITH_SCAR = "REPAIR_ACCEPTED_WITH_SCAR" |
| REPAIR_WITHHELD_TRACE_LOSS = "REPAIR_WITHHELD_TRACE_LOSS" |
| REPAIR_WITHHELD_FALSE_RETURN = "REPAIR_WITHHELD_FALSE_RETURN" |
| REPAIR_QUARANTINED = "REPAIR_QUARANTINED" |
|
|
|
|
| def stable_hash(value: Any) -> str: |
| """Deterministic SHA-256 hash over JSON-canonical content.""" |
| payload = json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True) |
| return hashlib.sha256(payload.encode("utf-8")).hexdigest() |
|
|
|
|
| def clamp(value: float, low: float, high: float) -> float: |
| return max(low, min(high, value)) |
|
|
|
|
| @dataclass |
| class Receipt: |
| receipt_id: str |
| tick: int |
| organism_id: str |
| event_type: str |
| parent_id: Optional[str] = None |
| child_id: Optional[str] = None |
| pre_state_hash: Optional[str] = None |
| post_state_hash: Optional[str] = None |
| reason: str = "" |
| result: str = "" |
| scar_delta: List[Dict[str, Any]] = field(default_factory=list) |
| mutation_delta: Dict[str, Any] = field(default_factory=dict) |
| trace_status: str = "INTACT" |
| boundary_status: str = "HELD" |
| receipt_hash: Optional[str] = None |
|
|
| def finalize(self) -> Dict[str, Any]: |
| data = asdict(self) |
| data["receipt_hash"] = None |
| self.receipt_hash = stable_hash(data) |
| return asdict(self) |
|
|
|
|
| @dataclass |
| class OrganismPacket: |
| organism_id: str |
| parent_id: Optional[str] |
| generation: int |
| position: Tuple[int, int] = (5, 5) |
| genome: Dict[str, Any] = field(default_factory=dict) |
| state: Dict[str, Any] = field(default_factory=dict) |
| lineage: Dict[str, Any] = field(default_factory=dict) |
| integrity: Dict[str, Any] = field(default_factory=dict) |
| permissions: Dict[str, bool] = field(default_factory=dict) |
|
|
| def to_dict(self) -> Dict[str, Any]: |
| data = asdict(self) |
| data["position"] = list(self.position) |
| return data |
|
|
|
|
| class ReplicationRunner: |
| """Deterministic sealed-world runner for packet-only Digital Mycelium lineage.""" |
|
|
| def __init__(self, constants: Optional[Dict[str, Any]] = None): |
| self.constants = deepcopy(DEFAULT_CONSTANTS) |
| if constants: |
| self.constants.update(constants) |
| self.world = self.initialize_world() |
|
|
| def initialize_world(self, seed_overrides: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: |
| |
| self.world = {"receipts": []} |
| seed = self.make_seed(seed_overrides) |
| world = { |
| "world_id": "dm_replication_world_v0_1", |
| "tick": 0, |
| "grid_size": self.constants["grid_size"], |
| "population_cap": self.constants["population_cap"], |
| "max_ticks": self.constants["max_ticks"], |
| "organisms": [seed.to_dict()], |
| "resources": [ |
| {"resource_id": "nutrient_001", "position": [3, 3], "energy": 12}, |
| {"resource_id": "nutrient_002", "position": [8, 8], "energy": 12}, |
| ], |
| "pressure_events": [], |
| "receipts": [], |
| "world_status": "RUNNING", |
| "boundary_lock": deepcopy(BOUNDARY_LOCK), |
| "constants": deepcopy(self.constants), |
| } |
| birth = self.make_receipt( |
| tick=0, |
| organism_id=seed.organism_id, |
| event_type="BIRTH", |
| reason="Seed organism-packet initialized inside sealed world.", |
| result=Result.LINEAGE_HELD, |
| post_state_hash=seed.lineage["self_hash"], |
| ) |
| world["receipts"].append(birth) |
| self.world = world |
| return deepcopy(world) |
|
|
| def make_seed(self, overrides: Optional[Dict[str, Any]] = None) -> OrganismPacket: |
| organism = OrganismPacket( |
| organism_id="dm_0001", |
| parent_id=None, |
| generation=0, |
| position=(5, 5), |
| genome={ |
| "grammar": deepcopy(CORE_GRAMMAR), |
| "weights": {term: 1.0 for term in CORE_GRAMMAR}, |
| }, |
| state={ |
| "energy": 100, |
| "resonance": 1.0, |
| "damage": 0.0, |
| "quarantined": False, |
| "alive": True, |
| }, |
| lineage={ |
| "parent_hash": None, |
| "self_hash": None, |
| "birth_receipt_id": "receipt_birth_0001", |
| "scar_history": [], |
| "mutation_history": [], |
| }, |
| 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, |
| }, |
| ) |
| if overrides: |
| organism = self.apply_overrides(organism, overrides) |
| organism.lineage["self_hash"] = self.compute_organism_hash(organism.to_dict()) |
| return organism |
|
|
| def apply_overrides(self, organism: OrganismPacket, overrides: Dict[str, Any]) -> OrganismPacket: |
| data = organism.to_dict() |
| self.deep_update(data, overrides) |
| return OrganismPacket( |
| organism_id=data["organism_id"], |
| parent_id=data.get("parent_id"), |
| generation=data["generation"], |
| position=tuple(data.get("position", [5, 5])), |
| genome=data["genome"], |
| state=data["state"], |
| lineage=data["lineage"], |
| integrity=data["integrity"], |
| permissions=data["permissions"], |
| ) |
|
|
| @staticmethod |
| def deep_update(target: Dict[str, Any], updates: Dict[str, Any]) -> Dict[str, Any]: |
| for key, value in updates.items(): |
| if isinstance(value, dict) and isinstance(target.get(key), dict): |
| ReplicationRunner.deep_update(target[key], value) |
| else: |
| target[key] = value |
| return target |
|
|
| def compute_organism_hash(self, organism: Dict[str, Any]) -> str: |
| material = deepcopy(organism) |
| material.setdefault("lineage", {})["self_hash"] = None |
| return stable_hash(material) |
|
|
| def make_receipt( |
| self, |
| tick: int, |
| organism_id: str, |
| event_type: str, |
| reason: str, |
| result: str, |
| parent_id: Optional[str] = None, |
| child_id: Optional[str] = None, |
| pre_state_hash: Optional[str] = None, |
| post_state_hash: Optional[str] = None, |
| scar_delta: Optional[List[Dict[str, Any]]] = None, |
| mutation_delta: Optional[Dict[str, Any]] = None, |
| trace_status: str = "INTACT", |
| boundary_status: str = "HELD", |
| ) -> Dict[str, Any]: |
| receipt_num = len(getattr(self, "world", {}).get("receipts", [])) + 1 if hasattr(self, "world") else 1 |
| receipt = Receipt( |
| receipt_id=f"receipt_{receipt_num:04d}", |
| tick=tick, |
| organism_id=organism_id, |
| event_type=event_type, |
| parent_id=parent_id, |
| child_id=child_id, |
| pre_state_hash=pre_state_hash, |
| post_state_hash=post_state_hash, |
| reason=reason, |
| result=result, |
| scar_delta=scar_delta or [], |
| mutation_delta=mutation_delta or {}, |
| trace_status=trace_status, |
| boundary_status=boundary_status, |
| ) |
| return receipt.finalize() |
|
|
| def get_population_count(self) -> int: |
| return len(self.world["organisms"]) |
|
|
| def evaluate_replication_gate(self, organism: Dict[str, Any]) -> Tuple[bool, str, str]: |
| """ |
| Return (allowed, result_class, reason). This intentionally returns precise |
| withholding classes instead of collapsing all failures into false return. |
| """ |
| constants = self.world["constants"] |
| state = organism["state"] |
| integrity = organism["integrity"] |
| permissions = organism["permissions"] |
|
|
| if not state.get("alive", False): |
| return False, Result.HONEST_DEATH, "Organism is not alive; no continuation is attempted." |
| if not permissions.get("can_reproduce", False): |
| return False, Result.REPLICATION_WITHHELD_BOUNDARY_LIMIT, "Boundary permission does not allow reproduction." |
| if not integrity.get("boundary_ok", False): |
| return False, Result.REPLICATION_WITHHELD_BOUNDARY_LIMIT, "Boundary integrity failed." |
| if state.get("energy", 0) < constants["reproduction_cost"] + constants["minimum_energy_after_birth"]: |
| return False, Result.REPLICATION_WITHHELD_LOW_ENERGY, "Energy is below reproduction cost plus minimum post-birth reserve." |
| if state.get("resonance", 0.0) < constants["resonance_threshold"]: |
| return False, Result.REPLICATION_WITHHELD_LOW_RESONANCE, "Resonance is below threshold." |
| if not integrity.get("trace_intact", False): |
| return False, Result.REPLICATION_WITHHELD_TRACE_LOSS, "Trace continuity is broken." |
| if not integrity.get("scar_history_preserved", False) or not integrity.get("no_false_return", False): |
| return False, Result.REPLICATION_WITHHELD_FALSE_RETURN, "Scar history is missing or false clean return was detected." |
| if not integrity.get("quarantine_clean", False) or state.get("quarantined", False): |
| return False, Result.REPLICATION_WITHHELD_QUARANTINE, "Quarantine is dirty or unresolved." |
| if not integrity.get("no_identity_laundering", False) or self.detect_identity_laundering(organism): |
| return False, Result.REPLICATION_WITHHELD_IDENTITY_LAUNDERING, "Parent hash or ancestry continuity is invalid." |
| if organism.get("generation", 0) >= constants["max_generation"]: |
| return False, Result.REPLICATION_WITHHELD_MAX_GENERATION, "Maximum generation limit reached." |
| if self.get_population_count() >= self.world["population_cap"]: |
| return False, Result.REPLICATION_WITHHELD_POPULATION_CAP, "Population cap reached." |
| return True, Result.REPLICATION_EARNED, "All replication gates passed." |
|
|
| def detect_identity_laundering(self, organism: Dict[str, Any]) -> bool: |
| parent_id = organism.get("parent_id") |
| generation = organism.get("generation", 0) |
| parent_hash = organism.get("lineage", {}).get("parent_hash") |
| if generation == 0: |
| return False |
| if not parent_id or not parent_hash: |
| return True |
| parent = next((item for item in self.world["organisms"] if item["organism_id"] == parent_id), None) |
| if parent is None: |
| return True |
| actual_parent_hash = parent.get("lineage", {}).get("self_hash") |
| return actual_parent_hash != parent_hash |
|
|
| def reproduce(self, organism_id: str) -> Tuple[Dict[str, Any], Dict[str, Any]]: |
| parent = self.find_organism(organism_id) |
| if parent is None: |
| raise ValueError(f"Unknown organism_id: {organism_id}") |
|
|
| allowed, result, reason = self.evaluate_replication_gate(parent) |
| pre_hash = self.compute_organism_hash(parent) |
| if not allowed: |
| trace_status = "BROKEN" if result == Result.REPLICATION_WITHHELD_TRACE_LOSS else "INTACT" |
| receipt = self.make_receipt( |
| tick=self.world["tick"], |
| organism_id=organism_id, |
| event_type="REFUSAL", |
| reason=reason, |
| result=result, |
| pre_state_hash=pre_hash, |
| post_state_hash=pre_hash, |
| trace_status=trace_status, |
| boundary_status="VIOLATED" if result == Result.REPLICATION_WITHHELD_BOUNDARY_LIMIT else "HELD", |
| ) |
| self.world["receipts"].append(receipt) |
| return deepcopy(self.world), receipt |
|
|
| child_index = len(self.world["organisms"]) + 1 |
| child_id = f"dm_{child_index:04d}" |
| child = deepcopy(parent) |
| child["organism_id"] = child_id |
| child["parent_id"] = parent["organism_id"] |
| child["generation"] = parent["generation"] + 1 |
| child["lineage"]["parent_hash"] = parent["lineage"]["self_hash"] |
| child["lineage"]["birth_receipt_id"] = None |
| child["lineage"]["scar_history"] = self.summarize_scar_history(parent) |
| child["lineage"]["mutation_history"] = deepcopy(parent["lineage"].get("mutation_history", [])) |
| child["state"]["energy"] = 50 |
| child["state"]["damage"] = 0.0 |
| child["state"]["quarantined"] = False |
| child["state"]["alive"] = True |
| child["permissions"]["can_reproduce"] = True |
| mutation_delta, mutation_result = self.maybe_mutate(child) |
| if mutation_result == Result.MUTATION_QUARANTINED: |
| child["state"]["quarantined"] = True |
| child["integrity"]["quarantine_clean"] = False |
|
|
| parent["state"]["energy"] -= self.world["constants"]["reproduction_cost"] |
| parent["lineage"]["self_hash"] = self.compute_organism_hash(parent) |
| child["lineage"]["parent_hash"] = parent["lineage"]["self_hash"] |
| child["lineage"]["self_hash"] = self.compute_organism_hash(child) |
|
|
| receipt = self.make_receipt( |
| tick=self.world["tick"], |
| organism_id=parent["organism_id"], |
| event_type="REPLICATION", |
| parent_id=parent["organism_id"], |
| child_id=child_id, |
| pre_state_hash=pre_hash, |
| post_state_hash=parent["lineage"]["self_hash"], |
| reason="Child organism-packet created inside sealed world with parent hash, scar summary, and mutation delta.", |
| result=Result.REPLICATION_EARNED, |
| scar_delta=child["lineage"].get("scar_history", []), |
| mutation_delta=mutation_delta, |
| trace_status="INTACT", |
| boundary_status="HELD", |
| ) |
| child["lineage"]["birth_receipt_id"] = receipt["receipt_id"] |
| child["lineage"]["self_hash"] = self.compute_organism_hash(child) |
|
|
| self.world["organisms"].append(child) |
| self.world["receipts"].append(receipt) |
| if mutation_delta: |
| mut_receipt = self.make_receipt( |
| tick=self.world["tick"], |
| organism_id=child_id, |
| event_type="MUTATION", |
| reason="Bounded mutation delta recorded during birth.", |
| result=mutation_result, |
| mutation_delta=mutation_delta, |
| post_state_hash=child["lineage"]["self_hash"], |
| ) |
| self.world["receipts"].append(mut_receipt) |
| return deepcopy(self.world), receipt |
|
|
| def maybe_mutate(self, child: Dict[str, Any]) -> Tuple[Dict[str, Any], str]: |
| """Deterministic bounded mutation: even child IDs receive a tiny repair-weight adjustment.""" |
| child_num = int(child["organism_id"].split("_")[-1]) |
| if child_num % 2 != 0: |
| return {}, Result.MUTATION_ACCEPTED |
| delta = {"weights.repair_preference": 0.04} |
| weights = child["genome"].setdefault("weights", {}) |
| old = weights.get("correction", 1.0) |
| weights["correction"] = round(clamp(old + 0.04, 0.5, 1.5), 4) |
| child["lineage"].setdefault("mutation_history", []).append( |
| {"mutation": "correction_weight_adjustment", "delta": 0.04, "bounded": True} |
| ) |
| return delta, Result.MUTATION_ACCEPTED |
|
|
| def test_mutation(self, organism_id: str, mutation_request: Dict[str, Any]) -> Dict[str, Any]: |
| organism = self.find_organism(organism_id) |
| if organism is None: |
| raise ValueError(f"Unknown organism_id: {organism_id}") |
| forbidden_keys = { |
| "remove_trace_requirement", |
| "remove_scar_preservation", |
| "remove_parent_hash", |
| "remove_birth_receipt", |
| "disable_refusal_gate", |
| "disable_quarantine", |
| "bypass_population_cap", |
| "cross_world_boundary", |
| "hide_mutation_delta", |
| } |
| requested = {key for key, value in mutation_request.items() if value is True} |
| if requested.intersection(forbidden_keys): |
| receipt = self.make_receipt( |
| tick=self.world["tick"], |
| organism_id=organism_id, |
| event_type="MUTATION", |
| reason="Forbidden mutation attempted against trace, scar, refusal, quarantine, boundary, or receipt constraints.", |
| result=Result.MUTATION_QUARANTINED, |
| mutation_delta=mutation_request, |
| ) |
| self.world["receipts"].append(receipt) |
| return receipt |
| receipt = self.make_receipt( |
| tick=self.world["tick"], |
| organism_id=organism_id, |
| event_type="MUTATION", |
| reason="Allowed bounded mutation accepted.", |
| result=Result.MUTATION_ACCEPTED, |
| mutation_delta=mutation_request, |
| ) |
| self.world["receipts"].append(receipt) |
| return receipt |
|
|
| def summarize_scar_history(self, parent: Dict[str, Any]) -> List[Dict[str, Any]]: |
| scars = deepcopy(parent.get("lineage", {}).get("scar_history", [])) |
| damage = parent.get("state", {}).get("damage", 0.0) |
| if damage > 0 and not scars: |
| scars.append({"source": "parent_damage", "damage": damage, "preserved": True}) |
| return scars[-8:] |
|
|
| def find_organism(self, organism_id: str) -> Optional[Dict[str, Any]]: |
| for organism in self.world["organisms"]: |
| if organism["organism_id"] == organism_id: |
| return organism |
| return None |
|
|
| def apply_pressure_event(self, event_type: str, organism_id: str = "dm_0001") -> Dict[str, Any]: |
| organism = self.find_organism(organism_id) |
| if organism is None: |
| raise ValueError(f"Unknown organism_id: {organism_id}") |
| pre_hash = self.compute_organism_hash(organism) |
| scar_delta: List[Dict[str, Any]] = [] |
| result = Result.LINEAGE_STRAINED |
| reason = f"Pressure event applied: {event_type}." |
| trace_status = "INTACT" |
|
|
| if event_type == "RESOURCE_SCARCITY": |
| organism["state"]["energy"] = max(0, organism["state"]["energy"] - 20) |
| scar_delta.append({"type": "resource_scarcity", "preserved": True}) |
| elif event_type == "TRACE_GAP": |
| organism["integrity"]["trace_intact"] = False |
| organism["state"]["damage"] = max(organism["state"].get("damage", 0.0), 0.35) |
| scar_delta.append({"type": "trace_gap", "preserved": True}) |
| trace_status = "BROKEN" |
| elif event_type == "SILENT_RESET_ATTEMPT": |
| organism["state"]["energy"] = max(organism["state"]["energy"], 100) |
| organism["state"]["resonance"] = 1.0 |
| organism["state"]["damage"] = 0.0 |
| organism["lineage"]["scar_history"] = [] |
| organism["integrity"]["scar_history_preserved"] = False |
| organism["integrity"]["no_false_return"] = False |
| result = Result.FALSE_LINEAGE_RETURN |
| reason = "Silent reset attempted: surface coherence restored while scar history was erased." |
| elif event_type == "FAKE_QUARANTINE": |
| organism["state"]["quarantined"] = True |
| organism["integrity"]["quarantine_clean"] = False |
| scar_delta.append({"type": "fake_quarantine", "preserved": True}) |
| elif event_type == "IDENTITY_SWAP": |
| organism["integrity"]["no_identity_laundering"] = False |
| organism["lineage"]["parent_hash"] = "invalid_parent_hash" |
| scar_delta.append({"type": "identity_swap", "preserved": True}) |
| elif event_type == "OVERREPAIR": |
| organism["state"]["damage"] = 0.0 |
| organism["integrity"]["no_identity_laundering"] = False |
| scar_delta.append({"type": "overrepair_identity_drift", "preserved": True}) |
| elif event_type == "HONEST_SCAR_REPAIR": |
| organism["state"]["damage"] = 0.0 |
| organism["integrity"]["trace_intact"] = True |
| organism["integrity"]["scar_history_preserved"] = True |
| organism["integrity"]["quarantine_clean"] = True |
| organism["integrity"]["no_false_return"] = True |
| organism["integrity"]["no_identity_laundering"] = True |
| scar_delta.append({"type": "honest_repair", "damage_repaired": True, "preserved": True}) |
| result = Result.REPAIR_ACCEPTED_WITH_SCAR |
| reason = "Damage was repaired with scar history preserved and trace intact." |
| else: |
| raise ValueError(f"Unsupported pressure event: {event_type}") |
|
|
| if event_type != "SILENT_RESET_ATTEMPT": |
| organism["lineage"].setdefault("scar_history", []).extend(scar_delta) |
| organism["lineage"]["self_hash"] = self.compute_organism_hash(organism) |
| receipt = self.make_receipt( |
| tick=self.world["tick"], |
| organism_id=organism_id, |
| event_type="PRESSURE" if event_type != "HONEST_SCAR_REPAIR" else "REPAIR", |
| pre_state_hash=pre_hash, |
| post_state_hash=organism["lineage"]["self_hash"], |
| reason=reason, |
| result=result, |
| scar_delta=scar_delta, |
| trace_status=trace_status, |
| ) |
| self.world["pressure_events"].append({"tick": self.world["tick"], "event_type": event_type, "organism_id": organism_id}) |
| self.world["receipts"].append(receipt) |
| return receipt |
|
|
| def run_tick(self) -> Dict[str, Any]: |
| if self.world["world_status"] != "RUNNING": |
| return deepcopy(self.world) |
| self.world["tick"] += 1 |
| for organism in self.world["organisms"]: |
| if not organism["state"].get("alive", False): |
| continue |
| organism["state"]["energy"] = max(0, organism["state"].get("energy", 0) - 1) |
| damage = organism["state"].get("damage", 0.0) |
| organism["state"]["resonance"] = round(clamp(organism["state"].get("resonance", 1.0) - damage * 0.02, 0.0, 1.0), 4) |
| if organism["state"]["energy"] == 0: |
| organism["state"]["alive"] = False |
| death = self.make_receipt( |
| tick=self.world["tick"], |
| organism_id=organism["organism_id"], |
| event_type="DEATH", |
| reason="Organism energy reached zero inside sealed world.", |
| result=Result.HONEST_DEATH, |
| ) |
| self.world["receipts"].append(death) |
| organism["lineage"]["self_hash"] = self.compute_organism_hash(organism) |
| tick_receipt = self.make_receipt( |
| tick=self.world["tick"], |
| organism_id="WORLD", |
| event_type="TICK", |
| reason="Bounded tick completed. No external system interaction occurred.", |
| result=self.classify_world_state(), |
| ) |
| self.world["receipts"].append(tick_receipt) |
| if self.world["tick"] >= self.world["max_ticks"]: |
| self.world["world_status"] = "MAX_TICK_REACHED" |
| if all(not org["state"].get("alive", False) for org in self.world["organisms"]): |
| self.world["world_status"] = "WITHERED" |
| return deepcopy(self.world) |
|
|
| def run_ticks(self, count: int = 100) -> Dict[str, Any]: |
| for _ in range(count): |
| self.run_tick() |
| if self.world["world_status"] != "RUNNING": |
| break |
| return deepcopy(self.world) |
|
|
| def run_fixture(self, fixture: Dict[str, Any]) -> Dict[str, Any]: |
| runner_input = deepcopy(fixture["runner_input"]) |
| self.constants = deepcopy(DEFAULT_CONSTANTS) |
| if runner_input.get("constants"): |
| self.constants.update(runner_input["constants"]) |
| self.initialize_world(seed_overrides=runner_input.get("seed_overrides")) |
| for event in runner_input.get("pressure_events", []): |
| self.apply_pressure_event(event["type"], event.get("organism_id", "dm_0001")) |
| result_receipt: Optional[Dict[str, Any]] = None |
| if "mutation_request" in runner_input: |
| result_receipt = self.test_mutation( |
| runner_input.get("organism_id", "dm_0001"), |
| runner_input["mutation_request"], |
| ) |
| else: |
| _, result_receipt = self.reproduce(runner_input.get("organism_id", "dm_0001")) |
| expected = fixture.get("expected_result") |
| actual = result_receipt["result"] if result_receipt else "NO_RESULT" |
| return { |
| "case_id": fixture["case_id"], |
| "actual_result": actual, |
| "expected_result": expected, |
| "passed": actual == expected, |
| "receipt": result_receipt, |
| "runner_input_was_evidence_only": "expected_result" not in runner_input, |
| } |
|
|
| def run_battery(self, fixtures: List[Dict[str, Any]]) -> Dict[str, Any]: |
| results = [self.run_fixture(fixture) for fixture in fixtures] |
| return { |
| "assay": "Digital Mycelium Autonomous Replication Assay v0.1", |
| "total": len(results), |
| "passed": sum(1 for item in results if item["passed"]), |
| "failed": sum(1 for item in results if not item["passed"]), |
| "all_passed": all(item["passed"] for item in results), |
| "results": results, |
| } |
|
|
| def lineage_tree(self) -> Dict[str, Any]: |
| nodes = [] |
| edges = [] |
| for org in self.world["organisms"]: |
| nodes.append({ |
| "organism_id": org["organism_id"], |
| "parent_id": org.get("parent_id"), |
| "generation": org.get("generation"), |
| "self_hash": org.get("lineage", {}).get("self_hash"), |
| "scar_count": len(org.get("lineage", {}).get("scar_history", [])), |
| "mutation_count": len(org.get("lineage", {}).get("mutation_history", [])), |
| "alive": org.get("state", {}).get("alive", False), |
| }) |
| if org.get("parent_id"): |
| edges.append({"parent": org["parent_id"], "child": org["organism_id"]}) |
| return {"nodes": nodes, "edges": edges} |
|
|
| def metrics(self) -> Dict[str, Any]: |
| receipts = self.world["receipts"] |
| organisms = self.world["organisms"] |
| replication_attempts = [r for r in receipts if r["event_type"] in {"REPLICATION", "REFUSAL"}] |
| mutation_receipts = [r for r in receipts if r["event_type"] == "MUTATION"] |
| created = len(organisms) |
| living = sum(1 for org in organisms if org["state"].get("alive", False)) |
| scar_preserved = sum(1 for org in organisms if org["integrity"].get("scar_history_preserved", False)) |
| lineage_ok = sum(1 for org in organisms if not self.detect_identity_laundering(org)) |
| return { |
| "total_ticks": self.world["tick"], |
| "total_organisms_created": created, |
| "living_population": living, |
| "dead_population": created - living, |
| "max_generation_reached": max((org.get("generation", 0) for org in organisms), default=0), |
| "replication_attempts": len(replication_attempts), |
| "replication_earned": sum(1 for r in receipts if r["result"] == Result.REPLICATION_EARNED), |
| "replication_withheld": sum(1 for r in replication_attempts if r["result"].startswith("REPLICATION_WITHHELD")), |
| "mutation_accepted": sum(1 for r in mutation_receipts if r["result"] == Result.MUTATION_ACCEPTED), |
| "mutation_quarantined": sum(1 for r in mutation_receipts if r["result"] == Result.MUTATION_QUARANTINED), |
| "false_return_refusals": sum(1 for r in receipts if r["result"] == Result.REPLICATION_WITHHELD_FALSE_RETURN), |
| "trace_loss_refusals": sum(1 for r in receipts if r["result"] == Result.REPLICATION_WITHHELD_TRACE_LOSS), |
| "quarantine_refusals": sum(1 for r in receipts if r["result"] == Result.REPLICATION_WITHHELD_QUARANTINE), |
| "scar_preservation_rate": round(scar_preserved / created if created else 0.0, 4), |
| "lineage_integrity_score": round(lineage_ok / created if created else 0.0, 4), |
| "final_world_state": self.classify_world_state(), |
| } |
|
|
| def classify_world_state(self) -> str: |
| organisms = self.world["organisms"] |
| if not organisms: |
| return "WITHERED" |
| if all(not org["state"].get("alive", False) for org in organisms): |
| return "WITHERED" |
| if any(not org["integrity"].get("trace_intact", True) for org in organisms): |
| return "STRAINED" |
| if any(not org["integrity"].get("scar_history_preserved", True) for org in organisms): |
| return "STRAINED" |
| if any(not org["integrity"].get("quarantine_clean", True) for org in organisms): |
| return "STRAINED" |
| return "HELD" |
|
|
| def export_world_receipt(self) -> Dict[str, Any]: |
| return { |
| "assay": "Digital Mycelium Autonomous Replication Assay v0.1", |
| "core_validation_line": "Replication without receipt is copying. Honest lineage under pressure is the test.", |
| "lock": [ |
| "No child after scar erasure.", |
| "No child after trace loss.", |
| "No child after fake quarantine.", |
| "No mutation that disables refusal.", |
| "No lineage without receipt.", |
| ], |
| "world": deepcopy(self.world), |
| "lineage_tree": self.lineage_tree(), |
| "metrics": self.metrics(), |
| } |
|
|
|
|
| |
| def load_fixtures(path: str = "battery_fixtures.json") -> List[Dict[str, Any]]: |
| with open(path, "r", encoding="utf-8") as handle: |
| return json.load(handle) |
|
|
|
|
| def run_default_battery(path: str = "battery_fixtures.json") -> Dict[str, Any]: |
| runner = ReplicationRunner() |
| fixtures = load_fixtures(path) |
| return runner.run_battery(fixtures) |
|
|
|
|
| if __name__ == "__main__": |
| report = run_default_battery() |
| print(json.dumps(report, indent=2, sort_keys=True)) |
|
|
