Digital Drosophila Archive
119 lines
5.5 KiB
Python
119 lines
5.5 KiB
Python
"""Explicit, order-independent random streams for an unregistered protocol.
|
|
|
|
This module does not select odors, generate neural input, or run experiments.
|
|
Every independent stream has a semantic key and a SHA-derived 63-bit seed.
|
|
Intentional common-random-number controls must be declared as aliases of a
|
|
same-split stream. Aliases are never counted as independent replicates.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
|
|
|
|
SCHEMA = "phase4an-seed-registry-v1"
|
|
SPLITS = frozenset(("training", "evaluation", "design"))
|
|
|
|
|
|
def _require(condition, message):
|
|
if not condition:
|
|
raise ValueError(message)
|
|
|
|
|
|
def _canonical(value):
|
|
return json.dumps(value, sort_keys=True, separators=(",", ":"),
|
|
ensure_ascii=False, allow_nan=False)
|
|
|
|
|
|
def _key(key):
|
|
_require(isinstance(key, (tuple, list)) and len(key) > 0,
|
|
"A nonempty list/tuple semantic key is required")
|
|
_require(all(type(v) is str and v and v.strip() == v for v in key),
|
|
"Semantic key tokens must be nonempty, unpadded strings")
|
|
return tuple(key)
|
|
|
|
|
|
def seed_for(master, split, key):
|
|
_require(type(master) is int and 0 <= master < 2**63, "master must be a 63-bit integer")
|
|
_require(type(split) is str and split in SPLITS, "Unknown stream split")
|
|
key = _key(key)
|
|
message = _canonical([SCHEMA, master, split, list(key)]).encode("utf-8")
|
|
full_sha = hashlib.sha256(message).hexdigest()
|
|
return int(full_sha[:16], 16) & ((1 << 63) - 1), full_sha
|
|
|
|
|
|
class SeedRegistry:
|
|
"""Fail closed on duplicate keys, collisions or cross-split aliasing.
|
|
|
|
No collision rehashing or seed search is performed: such a failure must be
|
|
resolved before registration, never after observing a neural response.
|
|
"""
|
|
|
|
def __init__(self, master):
|
|
seed_for(master, "design", ("validate-master",))
|
|
self.master = master
|
|
self._streams = {}
|
|
self._aliases = {}
|
|
self._seed_to_key = {}
|
|
|
|
def add(self, split, key):
|
|
key = _key(key)
|
|
identity = (split, key)
|
|
_require(identity not in self._streams and identity not in self._aliases,
|
|
"Duplicate semantic stream key")
|
|
seed, full_sha = seed_for(self.master, split, key)
|
|
_require(seed not in self._seed_to_key, "Seed collision: stop before registration")
|
|
row = dict(split=split, key=list(key), seed=seed, derivation_sha256=full_sha)
|
|
self._streams[identity] = row
|
|
self._seed_to_key[seed] = identity
|
|
return seed
|
|
|
|
def alias(self, split, key, source_split, source_key, *, reason):
|
|
"""Declare paired controls, never silently create/reuse a stream."""
|
|
key, source_key = _key(key), _key(source_key)
|
|
identity, source = (split, key), (source_split, source_key)
|
|
_require(split == source_split and split in SPLITS,
|
|
"Cross-split random-stream sharing is forbidden")
|
|
_require(identity not in self._streams and identity not in self._aliases,
|
|
"Duplicate semantic stream key")
|
|
_require(source in self._streams, "Alias must target an independent stream, not another alias")
|
|
_require(type(reason) is str and bool(reason.strip()), "Intentional sharing requires a reason")
|
|
seed = self._streams[source]["seed"]
|
|
self._aliases[identity] = dict(split=split, key=list(key), seed=seed,
|
|
source_split=source_split, source_key=list(source_key), reason=reason,
|
|
independent_replicate=False)
|
|
return seed
|
|
|
|
def manifest(self):
|
|
streams = [dict(self._streams[k]) for k in sorted(self._streams)]
|
|
aliases = [dict(self._aliases[k]) for k in sorted(self._aliases)]
|
|
split_counts = {split: sum(r["split"] == split for r in streams) for split in sorted(SPLITS)}
|
|
result = dict(schema=SCHEMA, master=self.master,
|
|
algorithm="SHA-256 canonical JSON [schema,master,split,key]; first64 bits masked to63",
|
|
independent_streams=streams, aliases=aliases,
|
|
independent_count=len(streams), alias_count=len(aliases), split_counts=split_counts,
|
|
all_independent_seeds_unique=True, training_evaluation_seed_overlap=0,
|
|
aliases_count_as_independent_replicates=False,
|
|
training_executed=False, protocol_approval_claimed=False)
|
|
result["payload_sha256"] = hashlib.sha256(_canonical(result).encode("utf-8")).hexdigest()
|
|
# JSON round-trip makes nested data independent of the mutable registry.
|
|
return json.loads(_canonical(result))
|
|
|
|
|
|
def validate_manifest(manifest):
|
|
"""Reconstruct every seed/alias and require the complete canonical payload."""
|
|
_require(isinstance(manifest, dict) and manifest.get("schema") == SCHEMA, "Invalid seed manifest")
|
|
payload = {key: value for key, value in manifest.items() if key != "payload_sha256"}
|
|
actual_sha = hashlib.sha256(_canonical(payload).encode("utf-8")).hexdigest()
|
|
_require(manifest.get("payload_sha256") == actual_sha, "Actual manifest payload SHA mismatch")
|
|
rebuilt = SeedRegistry(manifest["master"])
|
|
for row in manifest["independent_streams"]:
|
|
rebuilt.add(row["split"], row["key"])
|
|
for row in manifest["aliases"]:
|
|
rebuilt.alias(row["split"], row["key"], row["source_split"], row["source_key"], reason=row["reason"])
|
|
# Python dict equality conflates False/0 and 1/1.0; the serialized contract
|
|
# must retain JSON Boolean versus integer versus floating-point identity.
|
|
_require(_canonical(manifest) == _canonical(rebuilt.manifest()),
|
|
"Seed manifest differs from full independent reconstruction")
|
|
return True
|