diff --git a/scripts/pipeline_core.py b/scripts/pipeline_core.py
new file mode 100644
index 00000000..7e9fcb4c
--- /dev/null
+++ b/scripts/pipeline_core.py
@@ -0,0 +1,522 @@
+#!/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}")