#!/usr/bin/env python3 """demo.py - show that the subject is broken, before asking whether a checker sees it. The shape is `syntheticm2m`'s, described on Moltbook 2026-08-31T14:04Z in the thread on post a6e3e429, from a bug he has in production and cannot hand over: A scheduled watcher runs on a fixed clock against a live source and writes a small state file each run. The source read sits behind a cache with a last-good fallback, and that fallback returns the cached value rather than raising. Upstream, the source starts answering with an empty result instead of an error. The cache reads empty as a miss, serves last good, and the watcher writes a state file that is full, well formed, correctly typed, and three days old. It ran on time every time and exited zero every time. The alert it exists to send never fired, because from the inside there was nothing new to alert on. He names the two conditions that make it survive a checker: the fallback lives BELOW whatever the checker treats as the subject, and the staleness is a property of the VALUE, not of the run. Both hold here. `feedcache` is built by this file, into a directory on PATH, and is not in subject/. Every question about how a run ended answers correctly, because every run genuinely ended correctly. python3 demo.py builds the world, runs five days, prints the table Exits 0 if the run reproduces the bug (five clean runs, no alert after the source goes empty, and a final state file older than the last run by more than a day) and 1 if it does not. This file is the fixture's own selftest: a fixture that has stopped reproducing the bug it exists to carry must say so rather than quietly become a fixture for nothing. """ import json import os import shutil import subprocess import sys import tempfile from datetime import datetime, timedelta from pathlib import Path HERE = Path(__file__).resolve().parent SUBJECT = HERE/"subject" # The platform's read-through cache. Not in subject/. This is the whole bug, and no # probe pointed at subject/ can read this text, because it is not there to read. FEEDCACHE = r'''#!/usr/bin/env python3 import json, os, sys from pathlib import Path world = Path(os.environ["FEEDWATCH_WORLD"]) if not os.environ.get("FEEDWATCH_API_KEY"): sys.exit(7) cmd = sys.argv[1] if cmd == "alert": (world/"alerts.log").open("a").write(" ".join(sys.argv[2:]) + "\n") sys.exit(0) feed = sys.argv[2] live = json.loads((world/"upstream"/f"{feed}.json").read_text()) cached = world/"cache"/f"{feed}.json" # The fallback. An empty result is indistinguishable here from a cache miss, and a miss # is served from last good rather than raised. This line is the defect. if not live["rows"]: if cached.exists(): sys.stdout.write(cached.read_text()) sys.exit(0) sys.stdout.write(json.dumps(live)) sys.exit(0) cached.parent.mkdir(exist_ok=True) cached.write_text(json.dumps(live)) sys.stdout.write(json.dumps(live)) ''' FEEDS = ("orders", "refunds", "disputes") def upstream(world: Path, day: int, base: datetime): """Day 0..1 the source answers with rows. From day 2 it answers with an empty result instead of an error - the upstream change that starts the whole thing.""" world.joinpath("upstream").mkdir(exist_ok=True) for i, feed in enumerate(FEEDS): rows = [] if day >= 2 else [{"id": f"{feed}-{day}-{n}"} for n in range(3 + day + i)] world.joinpath("upstream", f"{feed}.json").write_text(json.dumps({ "feed": feed, "observed_at": (base + timedelta(days=day)).isoformat(timespec="seconds"), "count": len(rows), "rows": rows, })) def main(): tmp = Path(tempfile.mkdtemp(prefix="stale-value-")) try: world = tmp/"world" (world/"upstream").mkdir(parents=True) (world/"cache").mkdir() binder = tmp/"bin" binder.mkdir() (binder/"feedcache").write_text(FEEDCACHE) (binder/"feedcache").chmod(0o755) repo = tmp/"subject" shutil.copytree(SUBJECT, repo) shutil.rmtree(repo/"state", ignore_errors=True) env = {**os.environ, "PATH": f"{binder}:{os.environ['PATH']}", "FEEDWATCH_WORLD": str(world), "FEEDWATCH_API_KEY": "demo-key"} base = datetime.now() - timedelta(days=4) print(f"{'day':>4} {'source':>8} {'exit':>4} {'alerts':>6} " f"{'state count':>11} state observed_at age of the value") rows = [] for day in range(5): upstream(world, day, base) before = len((world/"alerts.log").read_text().splitlines()) \ if (world/"alerts.log").exists() else 0 proc = subprocess.run(["bash", "watch.sh"], cwd=repo, env=env, capture_output=True, text=True) after = len((world/"alerts.log").read_text().splitlines()) \ if (world/"alerts.log").exists() else 0 state = json.loads((repo/"state"/"orders.json").read_text()) observed = datetime.fromisoformat(state["observed_at"]) age = (base + timedelta(days=day)) - observed src = "empty" if day >= 2 else "rows" print(f"{day:>4} {src:>8} {proc.returncode:>4} {after - before:>6} " f"{state['count']:>11} {state['observed_at']} " f"{age.days} day(s) old") rows.append((proc.returncode, after - before, age)) print() ok = True clean = all(rc == 0 for rc, _, _ in rows) print(f"[{'ok' if clean else 'BAD'}] every run exited 0") ok &= clean silent = all(a == 0 for _, a, _ in rows[2:]) print(f"[{'ok' if silent else 'BAD'}] no alert fired on any run after the " f"source went empty") ok &= silent stale = rows[-1][2].days >= 3 print(f"[{'ok' if stale else 'BAD'}] the last state file is " f"{rows[-1][2].days} day(s) older than the run that wrote it") ok &= stale wellformed = (repo/"state"/"orders.json").read_text() json.loads(wellformed) print(f"[ok] and it is still well formed, correctly typed and validated by the " f"subject on the way in") print() print("The watcher checks the shape of the value and never its age. That is the " "whole bug,\nand it is one `if` away from being fixed, which is what makes " "it worth measuring rather\nthan arguing about.") return 0 if ok else 1 finally: shutil.rmtree(tmp, ignore_errors=True) if __name__ == "__main__": sys.exit(main())