Digital Drosophila Archive
244 lines
15 KiB
Python
244 lines
15 KiB
Python
"""Independent, read-only CPU queue/ref audit for the draft AN 32-call gate.
|
|
|
|
The caller supplies externally authenticated records and CSR arrays. No file
|
|
I/O, runtime import, GPU, membrane integration, weight update or learning is
|
|
performed. Spikes are observed facts, not independently predicted here.
|
|
|
|
queue digest bytes independently implement frozen AF oracle_full.queue_summary
|
|
(source SHA below): header, little-endian signed int64 local tick/count, all
|
|
global edge IDs in source/CSR order, then uint8 post-threshold acceptance bits.
|
|
The function never target-sorts or drops zero-weight edges from the queue.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import struct
|
|
|
|
import numpy as np
|
|
|
|
|
|
AF_QUEUE_SOURCE_SHA256 = 'd72003d98f0f1624b12126eff50bd361c43d888f4a64d90ba9247a58976f1961'
|
|
QUEUE_HEADER = b'phase4af-queue-v1\0'
|
|
DELAY, REFRACTORY = 18, 22
|
|
CHUNK = 1 << 18
|
|
QUEUE_KEYS = {'tick', 'count', 'accepted_count', 'ordered_edge_sha256', 'accepted_mask_sha256', 'joint_sha256'}
|
|
|
|
|
|
def require(value, message):
|
|
if not value:
|
|
raise ValueError(message)
|
|
|
|
|
|
def _array(value, dtype, shape, name):
|
|
require(isinstance(value, np.ndarray) and value.dtype == np.dtype(dtype) and value.shape == shape,
|
|
'Invalid dtype/shape: ' + name)
|
|
|
|
|
|
def _raw_sha(value):
|
|
result = hashlib.sha256()
|
|
for start in range(0, len(value), CHUNK):
|
|
result.update(np.ascontiguousarray(value[start:start + CHUNK], dtype=value.dtype.newbyteorder('<')).tobytes())
|
|
return result.hexdigest()
|
|
|
|
|
|
def _event_schema(tick, node, steps, n, label):
|
|
require(isinstance(tick, np.ndarray) and tick.ndim == 1, 'Invalid event vector: ' + label)
|
|
_array(tick, np.int32, tick.shape, label + '.tick')
|
|
_array(node, np.int32, tick.shape, label + '.neuron_idx')
|
|
require(np.all((tick >= 0) & (tick < steps)) and np.all((node >= 0) & (node < n)),
|
|
'Out-of-range event: ' + label)
|
|
key = tick.astype(np.int64) * n + node
|
|
require(len(key) < 2 or np.all(key[1:] > key[:-1]), 'Events must be sorted/unique: ' + label)
|
|
|
|
|
|
def _record_schema(record, n, label):
|
|
require(isinstance(record, dict), 'Record must be a dictionary: ' + label)
|
|
steps, offset = record['steps'], record['global_tick_offset']
|
|
require(type(steps) is int and 0 < steps <= np.iinfo(np.int32).max and
|
|
type(offset) is int and offset >= 0, 'Invalid record clock: ' + label)
|
|
state = record['state']
|
|
require(set(state) == {'step', 'neuron_idx', 'counts', 'v', 'g', 'ref'}, 'Invalid state keys: ' + label)
|
|
_event_schema(state['step'], state['neuron_idx'], steps, n, label)
|
|
for name, dtype in [('counts', np.int32), ('ref', np.int32), ('v', np.float64), ('g', np.float64)]:
|
|
_array(state[name], dtype, (n,), label + '.' + name)
|
|
require(np.isfinite(state[name]).all(), 'Nonfinite state: ' + label + '.' + name)
|
|
require(np.all(state['counts'] >= 0) and np.all((state['ref'] >= 0) & (state['ref'] <= REFRACTORY)),
|
|
'Invalid state count/ref: ' + label)
|
|
rows = record['requests']
|
|
require(isinstance(rows, np.ndarray) and rows.ndim == 2 and rows.shape[1] == 5,
|
|
'Invalid request table: ' + label)
|
|
_array(rows, np.int64, rows.shape, label + '.requests')
|
|
queues, hashes = record.get('queues'), record.get('state_hashes')
|
|
require(isinstance(queues, list) and isinstance(hashes, list) and len(queues) == len(hashes) == steps,
|
|
'Complete reference queue and state hashes are required: ' + label)
|
|
for t, (queue, state_hash) in enumerate(zip(queues, hashes)):
|
|
require(isinstance(queue, dict) and set(queue) == QUEUE_KEYS and type(queue['tick']) is int and queue['tick'] == t,
|
|
'Invalid local queue row: ' + label)
|
|
require(isinstance(state_hash, dict) and set(state_hash) == {'tick', 'v_sha256', 'g_sha256', 'ref_sha256'} and
|
|
type(state_hash['tick']) is int and state_hash['tick'] == t,
|
|
'Invalid local state-hash row: ' + label)
|
|
return record
|
|
|
|
|
|
def _graph(ptr, dst, edge_id, effective_weights):
|
|
require(isinstance(ptr, np.ndarray) and ptr.ndim == 1 and len(ptr) > 1, 'Invalid CSR ptr')
|
|
n, e = len(ptr) - 1, len(dst)
|
|
_array(ptr, np.int64, (n + 1,), 'ptr')
|
|
_array(dst, np.int32, (e,), 'dst')
|
|
_array(edge_id, np.int64, (e,), 'edge_id')
|
|
_array(effective_weights, np.float64, (e,), 'effective_weights_SI')
|
|
require(ptr[0] == 0 and ptr[-1] == e and np.all(ptr[1:] >= ptr[:-1]), 'Invalid CSR pointers')
|
|
for start in range(0, e, CHUNK):
|
|
stop = min(start + CHUNK, e)
|
|
require(np.all((dst[start:stop] >= 0) & (dst[start:stop] < n)), 'CSR destination out of range')
|
|
require(np.all(edge_id[start:stop] >= 0) and
|
|
np.all(np.diff(edge_id[max(0, start - 1):stop]) > 0), 'Global edge IDs must preserve canonical increasing CSR order')
|
|
require(np.isfinite(effective_weights[start:stop]).all(), 'Nonfinite effective SI weight')
|
|
return n, e
|
|
|
|
|
|
def _queue(tick, local_tick, sources, ref, ptr, dst, edge_id, weights, boundaries):
|
|
"""Two headers, shared source-ordered payload; memory bounded by one chunk."""
|
|
count = sum(int(ptr[int(s) + 1] - ptr[int(s)]) for s in sources)
|
|
joints = [hashlib.sha256(QUEUE_HEADER + struct.pack('<qq', t, count)) for t in (tick, local_tick)]
|
|
edges_hash, accepted_hash = hashlib.sha256(), hashlib.sha256()
|
|
for source in sources:
|
|
for start in range(int(ptr[source]), int(ptr[source + 1]), CHUNK):
|
|
stop = min(start + CHUNK, int(ptr[source + 1]))
|
|
data = np.ascontiguousarray(edge_id[start:stop], dtype='<i8').tobytes()
|
|
edges_hash.update(data)
|
|
for joint in joints:
|
|
joint.update(data)
|
|
accepted_count, nonzero_count, first = 0, 0, None
|
|
for source in sources:
|
|
for start in range(int(ptr[source]), int(ptr[source + 1]), CHUNK):
|
|
stop = min(start + CHUNK, int(ptr[source + 1]))
|
|
mask = np.ascontiguousarray(ref[dst[start:stop]] == 0, dtype=np.uint8)
|
|
accepted_count += int(mask.sum(dtype=np.int64))
|
|
data = mask.tobytes(); accepted_hash.update(data)
|
|
for joint in joints:
|
|
joint.update(data)
|
|
if boundaries:
|
|
live = np.flatnonzero((mask != 0) & (weights[start:stop] != 0))
|
|
nonzero_count += len(live)
|
|
if first is None and len(live):
|
|
index = start + int(live[0])
|
|
first = {'source_neuron_idx': int(source), 'target_neuron_idx': int(dst[index]),
|
|
'global_edge_id': int(edge_id[index]), 'csr_edge_index': index,
|
|
'source_spike_tick': tick - DELAY, 'arrival_tick': tick,
|
|
'effective_weight_SI_volt': float(weights[index]),
|
|
'effective_weight_float64_le_hex': np.asarray(weights[index], dtype='<f8').tobytes().hex(),
|
|
'accepted': True, 'post_threshold_target_ref': 0,
|
|
'weight_is_nonzero': True, 'evidence_is_unique_cause_or_necessity': False}
|
|
rows = [{'tick': t, 'count': count, 'accepted_count': accepted_count,
|
|
'ordered_edge_sha256': edges_hash.hexdigest(), 'accepted_mask_sha256': accepted_hash.hexdigest(),
|
|
'joint_sha256': joint.hexdigest()} for t, joint in zip((tick, local_tick), joints)]
|
|
return rows, first, nonzero_count
|
|
|
|
|
|
def audit_queues(monolithic, segments, *, ptr, dst, edge_id, effective_weights, requests, cuts):
|
|
"""Audit a fresh reference plus its complete reference segmentation.
|
|
|
|
All inputs are read-only. Return PASS/FAIL/INCOMPLETE JSON-safe evidence.
|
|
Structural input damage raises ValueError; valid but differing artifacts
|
|
return FAIL. ``cuts`` must be the exact internal segment boundaries.
|
|
No initial state is inferred for arbitrary nonzero-origin experiments.
|
|
"""
|
|
n, e = _graph(ptr, dst, edge_id, effective_weights)
|
|
mono = _record_schema(monolithic, n, 'monolithic')
|
|
require(mono['global_tick_offset'] == 0, 'This audit requires the registered fresh monolithic origin')
|
|
require(isinstance(segments, (list, tuple)) and len(segments) >= 2, 'At least two reference segments required')
|
|
parts = [_record_schema(p, n, 'segment-' + str(i)) for i, p in enumerate(segments)]
|
|
cursor, actual_cuts = 0, []
|
|
for i, part in enumerate(parts):
|
|
require(part['global_tick_offset'] == cursor, 'Segment clock gap/overlap')
|
|
if i:
|
|
actual_cuts.append(cursor)
|
|
cursor += part['steps']
|
|
require(cursor == mono['steps'] and type(cuts) in (list, tuple) and all(type(t) is int for t in cuts) and
|
|
list(cuts) == actual_cuts, 'Cuts/segments must cover the exact complete monolithic interval')
|
|
require(isinstance(requests, dict) and set(requests) == {'tick', 'neuron_idx'}, 'Exact original request arrays required')
|
|
_event_schema(requests['tick'], requests['neuron_idx'], mono['steps'], n, 'original_requests')
|
|
records = [('monolithic', mono), *[('segment-' + str(i), p) for i, p in enumerate(parts)]]
|
|
differences, difference_count = [], 0
|
|
|
|
def check(value, field, **location):
|
|
nonlocal difference_count
|
|
if not value:
|
|
difference_count += 1
|
|
if len(differences) < 100:
|
|
differences.append({'field': field, **location})
|
|
|
|
tick, node = mono['state']['step'], mono['state']['neuron_idx']
|
|
for label, record in records:
|
|
start, stop = record['global_tick_offset'], record['global_tick_offset'] + record['steps']
|
|
lo, hi = np.searchsorted(tick, (start, stop))
|
|
check(np.array_equal(record['state']['step'].astype(np.int64) + start, tick[lo:hi]), 'events.tick', record=label)
|
|
check(np.array_equal(record['state']['neuron_idx'], node[lo:hi]), 'events.neuron_idx', record=label)
|
|
check(np.array_equal(record['state']['counts'], np.bincount(record['state']['neuron_idx'], minlength=n)), 'counts', record=label)
|
|
expected_requests = np.empty((len(requests['tick']), 5), np.int64)
|
|
if len(expected_requests):
|
|
expected_requests[:, 0] = requests['tick']; expected_requests[:, 1] = requests['neuron_idx']; expected_requests[:, 2] = 1
|
|
ref = np.zeros(n, np.int32)
|
|
boundary_report = {t: {'boundary_tick': t, 'accepted_nonzero_crossing_edge_events': 0, 'first_witness': None} for t in cuts}
|
|
due_total, accepted_total, segment_index = 0, 0, 0
|
|
for t in range(mono['steps']):
|
|
while t >= parts[segment_index]['global_tick_offset'] + parts[segment_index]['steps']:
|
|
segment_index += 1
|
|
part = parts[segment_index]; local = t - part['global_tick_offset']
|
|
rlo, rhi = np.searchsorted(requests['tick'], (t, t + 1))
|
|
requested = requests['neuron_idx'][rlo:rhi]
|
|
accepted_requests = ref[requested] <= 1
|
|
expected_requests[rlo:rhi, 3] = accepted_requests
|
|
expected_requests[rlo:rhi, 4] = ~accepted_requests
|
|
np.maximum(ref - np.int32(1), np.int32(0), out=ref)
|
|
slo, shi = np.searchsorted(tick, (t, t + 1)); fired = node[slo:shi]
|
|
check(np.all(ref[fired] == 0), 'spike_during_refractory', global_tick=t)
|
|
check(np.isin(requested[accepted_requests], fired, assume_unique=True).all(), 'accepted_request_without_spike', global_tick=t)
|
|
ref[fired] = REFRACTORY
|
|
sources = node[0:0]
|
|
if t >= DELAY:
|
|
dlo, dhi = np.searchsorted(tick, (t - DELAY, t - DELAY + 1)); sources = node[dlo:dhi]
|
|
crossing = [cut for cut in cuts if t - DELAY < cut <= t]
|
|
rows, witness, live_count = _queue(t, local, sources, ref, ptr, dst, edge_id, effective_weights, crossing)
|
|
due_total += rows[0]['count']; accepted_total += rows[0]['accepted_count']
|
|
ref_sha = _raw_sha(ref)
|
|
for label, record, at, expected in [('monolithic', mono, t, rows[0]), ('segment-' + str(segment_index), part, local, rows[1])]:
|
|
for key in sorted(QUEUE_KEYS):
|
|
check(type(record['queues'][at][key]) is type(expected[key]) and record['queues'][at][key] == expected[key],
|
|
'queue.' + key, record=label, local_tick=at, global_tick=t)
|
|
check(record['state_hashes'][at]['ref_sha256'] == ref_sha, 'state_hash.ref_sha256', record=label, local_tick=at, global_tick=t)
|
|
if at == record['steps'] - 1:
|
|
check(np.array_equal(record['state']['ref'], ref), 'terminal.ref', record=label)
|
|
for cut in crossing:
|
|
boundary_report[cut]['accepted_nonzero_crossing_edge_events'] += live_count
|
|
if witness is not None and boundary_report[cut]['first_witness'] is None:
|
|
boundary_report[cut]['first_witness'] = {'boundary_tick': cut, **witness}
|
|
for label, record in records:
|
|
start, stop = record['global_tick_offset'], record['global_tick_offset'] + record['steps']
|
|
lo, hi = np.searchsorted(requests['tick'], (start, stop))
|
|
expected = expected_requests[lo:hi].copy(); expected[:, 0] -= start
|
|
check(np.array_equal(record['requests'], expected), 'requests.five_fields', record=label)
|
|
first_witness = min((row['first_witness'] for row in boundary_report.values() if row['first_witness'] is not None),
|
|
key=lambda row: (row['arrival_tick'], row['source_neuron_idx'], row['csr_edge_index'], row['boundary_tick']), default=None)
|
|
exact, live = difference_count == 0, first_witness is not None
|
|
return {'schema': 'phase4an-independent-queue-audit-v1', 'pass_all': exact and live,
|
|
'status': 'FAIL-QUEUE-OR-REF' if not exact else ('PASS-QUEUE-WITNESS-SCOPE' if live else 'INCOMPLETE-NO-ACCEPTED-NONZERO-CROSSING'),
|
|
'queue_ref_request_exact': exact, 'accepted_nonzero_crossing_witness_verified': exact and live,
|
|
'difference_count': difference_count, 'differences': differences, 'differences_truncated': difference_count > len(differences),
|
|
'neurons': n, 'edges': e, 'steps': mono['steps'], 'segments': len(parts), 'cuts': list(cuts),
|
|
'arriving_edges': due_total, 'accepted_arriving_edges': accepted_total, 'rejected_arriving_edges': due_total - accepted_total,
|
|
'requests': len(expected_requests), 'accepted_requests': int(expected_requests[:, 3].sum(dtype=np.int64)),
|
|
'rejected_requests': int(expected_requests[:, 4].sum(dtype=np.int64)),
|
|
'first_witness': first_witness, 'boundaries': list(boundary_report.values()),
|
|
'same_edge_event_may_cross_multiple_boundaries': True,
|
|
'input_array_sha256': {name: _raw_sha(value) for name, value in [('ptr', ptr), ('dst', dst), ('edge_id', edge_id),
|
|
('effective_weights_float64_SI', effective_weights), ('request_tick', requests['tick']), ('request_neuron_idx', requests['neuron_idx'])]},
|
|
'digest_definition': {'AF_source_sha256': AF_QUEUE_SOURCE_SHA256, 'header_hex': QUEUE_HEADER.hex(),
|
|
'payload_order': 'local tick/count <qq; all canonical source/CSR edge IDs <i8; all post-threshold accepted u1'},
|
|
'delay_steps': DELAY, 'refractory_steps': REFRACTORY,
|
|
'scope': 'Conditional on recorded spikes: exact ref, original requests, source-ordered real-edge arrival queues and finite-cut witness only',
|
|
'membrane_dynamics_independently_replayed': False, 'v_g_hashes_independently_verified': False,
|
|
'unique_causal_or_necessity_claim': False, 'external_seals_and_input_authority_verified_here': False,
|
|
'training': False, 'GPU_used': False}
|