#!/usr/bin/env python3 """ pipeline_core.py — Module-swappable six-stage search engine. Each stage is a Filter with a standard interface: input: List[Config] → output: List[Config] Stages can be swapped without rewriting the pipeline. All arithmetic is integer-based (Q16_16 raw where ratios needed). No floats. No native_decide. Pure Python stdlib. Usage: from pipeline_core import Pipeline, BraidStorm, TreeBraid, AngrySphinx, MultisurfacePacker, COUCHFilter, SidonFilter pipe = Pipeline([BraidStorm(k=8), TreeBraid(), AngrySphinx(budget=128), MultisurfacePacker(max_surfaces=64), COUCHFilter(threshold=49152), SidonFilter()]) result = pipe.run(labels=[1,2,4,8,16,32,64,128], S=128, moduli=[7,3,5,11,13,17,19,23,29]) To add a custom filter: class MyFilter(Filter): def apply(self, configs, ctx): # filter logic here return [c for c in configs if ...] @property def name(self): return "MyFilter" """ import sys, math, json, time, hashlib, random from abc import ABC, abstractmethod from dataclasses import dataclass, field from pathlib import Path from itertools import product from collections import Counter from typing import Any REPO_ROOT = Path(__file__).resolve().parent.parent ARTIFACTS_DIR = REPO_ROOT / ".openresearch" / "artifacts" ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True) # Q16_16 constants (no floats) Q16_ONE = 65536 Q16_THRESHOLD_COUCH = 49152 # 0.75 × 65536 Q16_SUBLEQ_SELFLOOP = 53908 # 0.823 × 65536 Q16_AVX_SELFLOOP = 57942 # 0.885 × 65536 Q16_RING_SELFLOOP = 0 # 0.0 # ── Config: the unit that flows through the pipeline ────────────────── @dataclass class Config: """A single configuration flowing through the pipeline.""" chiral: tuple # binary tuple (0=over, 1=under) per crossing labels: tuple # Sidon label set (integers) S: int # reflection point moduli: tuple # CRT moduli (L0, L1, ..., Lk) cost: int = 0 # compute cost (AngrySphinx) self_loop: int = 0 # contention proxy (COUCH, Q16_16 raw) sidon_score: int = 0 # Sidon score (Q16_16 raw: 65536 = perfect) collisions: int = 0 # collision count metadata: dict = field(default_factory=dict) # stage-specific data # ── Pipeline Context: shared state ───────────────────────────────────── @dataclass class PipelineContext: """Shared context across all stages.""" crossing_pairs: tuple = () # which strands cross: [(i,j), ...] groups: tuple = () # TreeBraid factorization seed: int = 0 stage_timings: dict = field(default_factory=dict) # ── Filter: the standard interface ──────────────────────────────────── class Filter(ABC): """Abstract base: every pipeline stage implements this.""" @abstractmethod def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: """Filter input configs → output configs.""" ... @property @abstractmethod def name(self) -> str: """Stage name for reporting.""" ... def run_stage(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: """Apply with timing.""" t0 = time.time() result = self.apply(configs, ctx) elapsed = time.time() - t0 ctx.stage_timings[self.name] = { "input": len(configs), "output": len(result), "time_s": round(elapsed, 4), } return result # ── Stage 1: BraidStorm — Generate ──────────────────────────────────── class BraidStorm(Filter): """Generates all 2^k chiral configurations.""" def __init__(self, k: int = 8): self.k = k @property def name(self) -> str: return f"BraidStorm(k={self.k})" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: if configs: # Use first config as template template = configs[0] else: return [] all_chiral = list(product([0, 1], repeat=self.k)) return [ Config( chiral=c, labels=template.labels, S=template.S, moduli=template.moduli, ) for c in all_chiral ] # ── Stage 2: TreeBraid — Factorize ──────────────────────────────────── class TreeBraid(Filter): """Factorizes crossing space via braid relations. σ_i σ_j = σ_j σ_i when |i-j| >= 2 (independent). Marks configs with their factorization group. Does NOT filter — just annotates. Actual reduction happens in subsequent stages that can use the group structure. """ @property def name(self) -> str: return "TreeBraid" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: if not configs: return [] k = len(configs[0].chiral) pairs = ctx.crossing_pairs if not pairs: pairs = tuple((i, i+1) for i in range(k)) ctx.crossing_pairs = pairs groups = self._factorize(k, pairs) ctx.groups = groups # Annotate each config with its group signature for c in configs: # Group signature: which groups have at least one under-crossing sig = tuple( any(c.chiral[idx] for idx in group) for group in groups ) c.metadata["group_sig"] = sig c.metadata["groups"] = groups return configs # no filtering, just annotation def _factorize(self, k: int, pairs: tuple) -> tuple: groups = [] remaining = list(range(k)) while remaining: group = [remaining[0]] for idx in remaining[1:]: si, sj = pairs[idx] independent = True for gidx in group: gi, gj = pairs[gidx] if (abs(si - gi) < 2 or abs(si - gj) < 2 or abs(sj - gi) < 2 or abs(sj - gj) < 2): independent = False break if independent: group.append(idx) for g in group: remaining.remove(g) groups.append(tuple(group)) return tuple(groups) # ── Stage 3: AngrySphinx — Resource Budget ──────────────────────────── class AngrySphinx(Filter): """Filters by compute budget. Cost = 2^(under-crossings).""" def __init__(self, budget: int = 128): self.budget = budget @property def name(self) -> str: return f"AngrySphinx(budget={self.budget})" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: result = [] for c in configs: under_count = sum(c.chiral) cost = 1 << under_count # 2^under_count — integer, no floats c.cost = cost if cost <= self.budget: result.append(c) return result # ── Stage 4: MultisurfacePacker — Spatial Fit ───────────────────────── class MultisurfacePacker(Filter): """Packs configs into available surfaces. Greedy by cost.""" def __init__(self, max_surfaces: int = 64): self.max_surfaces = max_surfaces @property def name(self) -> str: return f"MultisurfacePacker(max={self.max_surfaces})" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: if len(configs) <= self.max_surfaces: return configs # Sort by cost (cheapest first = most efficient packing) sorted_configs = sorted(configs, key=lambda c: c.cost) return sorted_configs[:self.max_surfaces] # ── Stage 5: COUCH — Geometric Stability ────────────────────────────── class COUCHFilter(Filter): """COUCH gate: contention below threshold. Self-loop proxy: under-crossing count → contention level. 0 under = ring dispatch (self_loop=0, always passes) k/2 under = SUBLEQ (self_loop=53908) all under = AVX-512 (self_loop=57942) """ def __init__(self, threshold: int = Q16_THRESHOLD_COUCH): self.threshold = threshold @property def name(self) -> str: return f"COUCH(threshold={self.threshold})" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: result = [] for c in configs: under_count = sum(c.chiral) k = len(c.chiral) # Self-loop: linear interpolation between ring (0) and AVX (57942) # All integer arithmetic: self_loop = 57942 * under_count // k c.self_loop = (Q16_AVX_SELFLOOP * under_count) // max(k, 1) if c.self_loop < self.threshold: result.append(c) return result # ── Stage 6: Sidon Filter — Algebraic Uniqueness ────────────────────── class SidonFilter(Filter): """Checks Sidon property via CRT reconstruction. NOTE: CRT sum-based Sidon check is chiral-invariant (proven — the negation x→-x is a ring automorphism). To actually discriminate chiral configs, swap this for DualQuaternionSidonFilter. This filter is kept as the default because it's the proven baseline. """ @property def name(self) -> str: return "SidonFilter" def apply(self, configs: list[Config], ctx: PipelineContext) -> list[Config]: result = [] for c in configs: embedded = self._embed_chiral(c) collisions = self._sidon_check(embedded, c.moduli) c.collisions = collisions total_pairs = len(c.labels) * (len(c.labels) + 1) // 2 # Sidon score: Q16_16 raw (65536 = perfect, 0 = all collide) c.sidon_score = Q16_ONE - (Q16_ONE * collisions) // max(total_pairs, 1) if collisions == 0: result.append(c) return result def _embed_chiral(self, c: Config) -> list[list[int]]: embedded = [] for a in c.labels: row = [a % c.moduli[0]] for j in range(1, len(c.moduli)): if c.chiral[j-1] == 0: row.append((c.S - a) % c.moduli[j]) else: row.append((a - c.S) % c.moduli[j]) embedded.append(row) return embedded def _sidon_check(self, embedded: list[list[int]], moduli: tuple) -> int: M = 1 for m in moduli: M *= m n = len(embedded) vals = [self._crt_reconstruct(row, moduli) for row in embedded] sums = [] for i in range(n): for j in range(i, n): sums.append((vals[i] + vals[j]) % M) counts = Counter(sums) return sum(c_count - 1 for c_count in counts.values()) def _crt_reconstruct(self, residues: list[int], moduli: tuple) -> int: M = 1 for m in moduli: M *= m x = 0 for r, m in zip(residues, moduli): Mi = M // m inv = self._modinv(Mi % m, m) if inv is None: return 0 x = (x + r * Mi * inv) % M return x def _egcd(self, a: int, b: int) -> tuple: if b == 0: return a, 1, 0 g, x, y = self._egcd(b, a % b) return g, y, x - (a // b) * y def _modinv(self, a: int, m: int) -> int | None: g, x, _ = self._egcd(a % m, m) return x % m if g == 1 else None # ── Swappable: Dual Quaternion Sidon Filter ─────────────────────────── class DualQuaternionSidonFilter(SidonFilter): """Sidon filter using dual quaternion products instead of CRT sums. Unlike CRT sums (which are chiral-invariant), dual quaternion products involve quaternion multiplication, which is NOT negation-invariant. This filter CAN discriminate chiral configs. Dual quaternion: q = r + ε·t where r=rotation, t=translation. For CRT: r = a mod L0 (identity/poloidal), t = (S-a) mod L1 (reflection/toroidal) Chiral flip: t → -t (negation of reflection component) Product: q_i ⊛ q_j = r_i·r_j + ε·(r_i·t_j + t_i·r_j) The product's translation part changes under chiral flip because it involves CROSS terms (r_i·t_j), not just sums. """ @property def name(self) -> str: return "DualQuaternionSidonFilter" def _embed_chiral(self, c: Config) -> list[list[int]]: """Embed as [r, t] pairs (dual quaternion components). r = a mod L0 (rotation/poloidal) t = (S - a) mod L1 or (a - S) mod L1 (translation/toroidal, chiral) """ embedded = [] for a in c.labels: r = a % c.moduli[0] if len(c.moduli) > 1: if c.chiral[0] == 0: t = (c.S - a) % c.moduli[1] else: t = (a - c.S) % c.moduli[1] else: t = 0 embedded.append([r, t]) return embedded def _sidon_check(self, embedded: list[list[int]], moduli: tuple) -> int: """Check Sidon on dual quaternion PRODUCTS (not sums). Product: q_i ⊛ q_j = r_i·r_j + ε·(r_i·t_j + t_i·r_j) We check if all products are distinct. """ n = len(embedded) L0 = moduli[0] L1 = moduli[1] if len(moduli) > 1 else 1 products = [] for i in range(n): for j in range(i, n): ri, ti = embedded[i] rj, tj = embedded[j] # Product: r_i*r_j (rotation part) + r_i*t_j + t_i*r_j (translation part) # Encode as a pair — two products are equal iff both parts match rot_part = (ri * rj) % L0 trans_part = (ri * tj + ti * rj) % L1 products.append((rot_part, trans_part)) counts = Counter(products) return sum(c_count - 1 for c_count in counts.values()) # ── Pipeline: chains filters together ────────────────────────────────── class Pipeline: """Chains filters into a pipeline.""" def __init__(self, filters: list[Filter]): self.filters = filters def run(self, labels: list[int], S: int, moduli: list[int], crossing_pairs: tuple = ()) -> dict: t0 = time.time() ctx = PipelineContext( crossing_pairs=crossing_pairs, seed=hash((tuple(labels), S, tuple(moduli))) % (2**31), ) # Seed config configs = [Config( chiral=tuple(0 for _ in range(len(moduli) - 1)), labels=tuple(labels), S=S, moduli=tuple(moduli), )] # Run each stage for f in self.filters: configs = f.run_stage(configs, ctx) elapsed = time.time() - t0 result = { "experiment": "pipeline_core", "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), "stages": [f.name for f in self.filters], "labels": list(labels), "S": S, "moduli": list(moduli), "stage_timings": ctx.stage_timings, "total_input": 1, "total_output": len(configs), "reduction": "N/A", "elapsed_s": round(elapsed, 4), "final_configs": [ { "chiral": list(c.chiral), "cost": c.cost, "self_loop": c.self_loop, "sidon_score": c.sidon_score, "collisions": c.collisions, } for c in configs ], } result["reduction"] = f"{result['total_input']} → {result['total_output']}" content = json.dumps(result, indent=2, sort_keys=True, default=str) result["sha256"] = hashlib.sha256(content.encode()).hexdigest() # Print summary print(f"\n{'='*60}") print(f" PIPELINE: {' → '.join(f.name for f in self.filters)}") print(f"{'='*60}") for s, t in ctx.stage_timings.items(): print(f" {s:30s} {t['input']:6d} → {t['output']:6d} ({t['time_s']:.4f}s)") print(f"{'='*60}") print(f" Total: {result['total_input']} → {result['total_output']} ({elapsed:.2f}s)") print(f"{'='*60}") return result # ── Main: default pipeline ───────────────────────────────────────────── if __name__ == "__main__": import argparse parser = argparse.ArgumentParser(description="Module-swappable pipeline") parser.add_argument("--strands", type=int, default=8) parser.add_argument("--budget", type=int, default=128) parser.add_argument("--surfaces", type=int, default=64) parser.add_argument("--filter", choices=["crt", "dq"], default="crt", help="Sidon filter: crt (sum-based) or dq (dual quaternion)") parser.add_argument("--output", default="pipeline_result.json") args = parser.parse_args() labels = [1, 2, 4, 8, 16, 32, 64, 128] S = 128 moduli = [7, 3, 5, 11, 13, 17, 19, 23, 29] # Select Sidon filter sidon_filter = SidonFilter() if args.filter == "crt" else DualQuaternionSidonFilter() # Build swappable pipeline pipe = Pipeline([ BraidStorm(k=args.strands), TreeBraid(), AngrySphinx(budget=args.budget), MultisurfacePacker(max_surfaces=args.surfaces), COUCHFilter(), sidon_filter, ]) result = pipe.run( labels=labels, S=S, moduli=moduli, crossing_pairs=tuple((i, i+1) for i in range(args.strands)), ) out_path = ARTIFACTS_DIR / args.output out_path.write_text(json.dumps(result, indent=2, default=str)) print(f"\nResults → {out_path}")