#!/usr/bin/env python3 """Independent finite audit of suffix-paid, input-built namespace reindexing. Sets, dictionaries and integer counters are semantic/audit oracles, not native instructions. The worker is a unit-operation service model; native timing rests on the written FIFO/capsule/compiler contract and its uniform primitive costs. """ from collections import deque from itertools import permutations, product import json import random def words(n): return tuple(''.join(x) for x in product('01', repeat=n)) def encode(s): return ''.join('0' + x for x in s) + '11' def parse_target(q, m): c = '' pos = 0 while pos + 2 <= len(q): pair = q[pos:pos + 2] pos += 2 if pair == '11': break if pair not in ('00', '01'): return None c += pair[1] else: return None r = (m - 1) // 2 suffix = q[pos:] if len(c) != m or not c.startswith('0') or len(suffix) != 2 * r: return None return c, suffix[:r], suffix[r:] def prepare(records, m): """Actual incidence ingestion; no supplied sigma dictionary is consulted.""" if m < 3 or m % 2 == 0 or not records: raise ValueError('width/right block') r = (m - 1) // 2 mapping, masks, closed = {}, {}, set() preceding = None for b, q in records: if len(b) != m or any(x not in '01' for x in b): raise ValueError('store spelling') if b != preceding: if b in closed: raise ValueError('reopened group') if preceding is not None: closed.add(preceding) preceding = b parsed = parse_target(q, m) if parsed is None: raise ValueError('target spelling') c, a, z = parsed oldmask = masks.get(q, 0) if b[0] == '0': if b != c: raise ValueError('wrong primary owner') masks[q] = oldmask | 1 else: alpha, beta = b[1:1 + r], b[1 + r:] if beta != z: raise ValueError('copied suffix mismatch') key = c, a if key in mapping and mapping[key] != alpha: raise ValueError('inconsistent partial map') mapping[key] = alpha masks[q] = oldmask | 2 return mapping, masks def service(alpha, z, rate=4): """One fixed setup, alpha steps, switch, then arriving/replayed z steps. Leading foreign bit 1 has already been consumed. Terminal membership costs one additional operation. Queue appends are a separate fixed ingestion lane. Each counted step is a macro with fixed primitive cost in the proof. """ backlog = deque() head = 0 consumed = '' phase = 'setup' finished = False operations = 0 trace = [] for bit in z: backlog.append(bit) used = 0 for _ in range(rate): if phase == 'setup': phase = 'alpha' elif phase == 'alpha': if head < len(alpha): consumed += alpha[head] head += 1 else: phase = 'suffix' elif phase == 'suffix': if backlog: consumed += backlog.popleft() else: break used += 1 operations += 1 trace.append((head, len(backlog), used)) # The ordinary final delimiter pays the already-reached terminal observation. if phase == 'suffix' and not backlog: operations += 1 finished = True return finished, consumed, operations, trace def read(prepared, active, q, m): mapping, masks = prepared mask = masks.get(q, 0) parsed = parse_target(q, m) if parsed is None or not mask: return False c, a, z = parsed primary = bool(mask & 1) and c in active foreign = False if mask & 2: alpha = mapping[c, a] complete, spelling, _, _ = service(alpha, z) assert complete and spelling == alpha + z foreign = '1' + spelling in active return primary or foreign def horn_reaches(records, active, q): edges = [(('source', 's'), ('store', b)) for b in active] edges += [(('store', b), ('target', t)) for b, t in records] reached = {('source', 's')} while True: nxt = reached | {y for x, y in edges if x in reached} if nxt == reached: return ('target', q) in reached reached = nxt def make_records(r, maps, sparse=False): m = 2 * r + 1 namespace = ['0' + w for w in words(2 * r)[:len(maps)]] rows = {} expected = {} for c, sigma in zip(namespace, maps): for k, a in enumerate(words(r)): for j, z in enumerate(words(r)): q = encode(c) + a + z b = '1' + sigma[k] + z expected[q] = b if not sparse or (k + j) % 3: rows.setdefault(c, []).append(q) if not sparse or (k + 2 * j) % 3: rows.setdefault(b, []).append(q) records = [(b, q) for b in sorted(rows) for q in rows[b]] return records, expected def must_reject(records, m): try: prepare(records, m) except ValueError: return raise AssertionError('invalid input accepted') def audit(): rng = random.Random(229) counts = dict(instances=0, queries=0, horn_queries=0, historical_queries=0, rejected_inputs=0, schedule_cases=0, reversal_width_witnesses=0) families = [] for r in (1, 2): perms = tuple(permutations(words(r))) combinations = list(product(perms, repeat=2)) # All two-namespace permutations at r<=2, plus noninjective partial maps. for maps in combinations: families.append((r, maps)) families.append((r, (tuple(words(r)[0] for _ in words(r)),))) for r, maps in families: m = 2 * r + 1 for sparse in (False, True): records, expected = make_records(r, maps, sparse) # Reorder groups and incidences; duplicates remain inside each group. groups = {} for b, q in records: groups.setdefault(b, []).append(q) keys = list(groups) rng.shuffle(keys) records = [] for b in keys: row = groups[b][:] rng.shuffle(row) records.extend((b, q) for q in row + row[:1]) prepared = prepare(records, m) counts['instances'] += 1 allstores = list(groups) candidates = list(expected) + ['0', '11', encode('0' * m), encode('0' * m) + '0' * (2*r+1)] snapshots = [frozenset()] for turn in range(12): prior = snapshots[-1] b = rng.choice(allstores) alias = turn % 4 != 0 snapshots.append(prior | {b} if alias else prior) snapshots += [frozenset(allstores), frozenset(b for b in allstores if b[0]=='1')] for active in snapshots: direct = {q for b, q in records if b in active} for q in candidates: answer = read(prepared, active, q, m) assert answer == (q in direct) counts['queries'] += 1 counts['historical_queries'] += 1 for q in candidates[:3]: assert read(prepared, active, q, m) == horn_reaches(records, active, q) counts['horn_queries'] += 1 # Earlier immutable snapshots survive all later source selections. assert snapshots[0] == frozenset() # Precise validation failures, including consistency across distinct stores. m = 5 c = '00000' q = encode(c) + '0010' invalid = [ [('10010', q), ('11110', q)], [('10011', q)], [('00001', q)], [('10010', q), ('00000', q), ('10010', q)], [('10010', q + '0')], [('1001', q)], ] for bad in invalid: must_reject(bad, m) counts['rejected_inputs'] += 1 # Service enumeration includes tiny r and long alpha, with every phase switch. for r in range(1, 513): for alpha, z in [('0'*r, '1'*r), ('01'*(r//2)+'0'*(r%2), '10'*(r//2)+'1'*(r%2))]: complete, out, ops, trace = service(alpha, z) assert complete and out == alpha + z assert ops == 2*r + 3 assert all(used <= 4 for _, _, used in trace) assert trace[-1][1] == 0 counts['schedule_cases'] += 1 # The direct decoded one-tick ยง16 clock must have >=k+1 emissions by this # prefix to exhaust 2r+1 source bits. Reversal exposes 2^min(k,r-k) choices. for r in range(2, 13): k = r // 2 fixed_prefix = '0' * k emitted = {'1' + a[::-1][:k] for a in words(r) if a.startswith(fixed_prefix)} assert len(emitted) == 2 ** min(k, r-k) counts['reversal_width_witnesses'] += 1 return dict(status='pass', counts=counts, scope='Finite semantics, retained sets, validation counterexamples and unit service model; not a native simulator.', service='2r+2 queued/setup/switch macrosteps over r symbols at rate4; ready terminal observation on ordinary delimiter.') if __name__ == '__main__': print(json.dumps(audit(), indent=2, sort_keys=True))