Spaces:
Paused
Paused
File size: 5,143 Bytes
3d46076 3327a12 3d46076 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """Research ledger V4 (mission §35): every major event receives a full
provenance record with cryptographic hashes and a chained event stream.
Deterministic: event ids derive from content + sequence, never wall-clock
(timestamps are operational metadata only).
"""
import hashlib
import json
import os
import subprocess
import time
from typing import Any, Dict, List, Optional
LEDGER_VERSION = "research_ledger_v1"
EVENT_TYPES = ("birth", "death", "mutation", "crossover", "learning", "growth",
"pruning", "synapse_creation", "synapse_deletion", "reproduction",
"speciation", "checkpoint", "experiment", "milestone")
def code_sha() -> str:
"""Current commit as code identity; 'uncommitted' when dirty/unavailable."""
try:
commit = subprocess.run(["git", "rev-parse", "HEAD"], capture_output=True,
text=True, timeout=5).stdout.strip()
dirty = subprocess.run(["git", "status", "--porcelain"], capture_output=True,
text=True, timeout=5).stdout.strip()
return commit if (commit and not dirty) else f"{commit or 'unknown'}+dirty"
except Exception:
return "unknown"
def dataset_sha(soma_path: str, connections_path: str) -> str:
from src.connectome.loader import _file_sha256
return hashlib.sha256(
_file_sha256(soma_path).encode()
+ _file_sha256(connections_path).encode()).hexdigest()
def shader_sha(shader_dir: str = "shaders") -> str:
h = hashlib.sha256()
for name in sorted(os.listdir(shader_dir)) if os.path.isdir(shader_dir) else []:
p = os.path.join(shader_dir, name)
if name.endswith(".spv") and os.path.isfile(p):
with open(p, "rb") as f:
h.update(name.encode() + f.read())
return h.hexdigest()
def _record_hash(rec: Dict[str, Any]) -> str:
# Identity covers content + sequence + chain only. Operational metadata
# (ts wall-clock, record_hash itself) is NEVER identity: two identical
# appends must hash identically or deterministic replay breaks.
payload = json.dumps({k: v for k, v in rec.items()
if k not in ("record_hash", "ts")},
sort_keys=True, default=str)
return hashlib.sha256(payload.encode()).hexdigest()
class ResearchLedger:
"""Append-only, hash-chained research event ledger."""
def __init__(self, experiment_id: str, seed: int,
path: Optional[str] = None):
self.experiment_id = experiment_id
self.seed = int(seed)
self.path = path
self.records: List[Dict[str, Any]] = []
self._prev = "0" * 64
self._static = {
"experiment_id": experiment_id,
"seed": self.seed,
"code_sha": code_sha(),
"dataset_sha": "unknown",
"shader_sha": shader_sha(),
"ledger_version": LEDGER_VERSION,
}
def set_dataset_sha(self, sha: str) -> None:
self._static["dataset_sha"] = sha
def append(self, event_type: str, tick: int, generation: int,
payload: Dict[str, Any], genome_sha: str = "", brain_sha: str = "",
world_sha: str = "", result_sha: str = "", parent_event: str = "") \
-> Dict[str, Any]:
if event_type not in EVENT_TYPES:
raise ValueError(f"unknown ledger event type {event_type!r}")
seq = len(self.records)
rec = {
**self._static,
"event_id": hashlib.sha256(
f"{self.experiment_id}|{event_type}|{seq}|{self._prev}"
.encode()).hexdigest()[:16],
"parent_event": parent_event or (self.records[-1]["event_id"]
if self.records else ""),
"event_type": event_type,
"seq": seq,
"tick": int(tick),
"generation": int(generation),
"genome_sha": genome_sha or "none",
"brain_sha": brain_sha or "none",
"world_sha": world_sha or "none",
"result_sha": result_sha or "none",
"payload": payload,
"prev_hash": self._prev,
"ts": time.time(), # operational metadata only, never identity
}
rec["record_hash"] = _record_hash(rec)
self.records.append(rec)
self._prev = rec["record_hash"]
if self.path:
os.makedirs(os.path.dirname(self.path) or ".", exist_ok=True)
with open(self.path, "a", encoding="utf-8") as f:
f.write(json.dumps(rec, sort_keys=True, default=str) + "\n")
return rec
def verify_chain(self) -> bool:
prev = "0" * 64
for rec in self.records:
if rec["prev_hash"] != prev:
return False
if rec["record_hash"] != _record_hash(rec):
return False
prev = rec["record_hash"]
return True
def event_stream_hash(self) -> str:
h = hashlib.sha256()
for rec in self.records:
h.update(rec["record_hash"].encode())
return h.hexdigest()
|