140's 10, RFC 6298's retransmission timer, written in aup with no new opcode (aup's examples/rto.aup, the timer a deadline in the state seen at each periodic tick): 20 scenarios (10 seeds, ticks every 10 and 100 ms; a closed loop: first sendings lost one in five, the receiver acknowledging cumulatively) and their 241,612 events right against rto.py, the model of the same rules (no known implementation takes given events), on the three paths, the paths the same; module_specs in make check: one scenario at both periods (tests/specs/rto, 2,297 events); bars: one workaround (the periodic tick: 2,795.8 ticks a timeout at 10 ms, 281.6 at 100, seen 4.27 and 49.23 ms late on average; ticks 93% and 57% of the counted work), 56 lines against Python's 30, three frictions (an input's period not readable, no abs, send a keyword); the one-shot timer's shape as a decision point, one spec needing it: noted, not made; the suite table's row 10

3 files changed, 186 insertions, 0 deletions

diff --git a/research/experiments/044-specs/2026-10-19-rto.jsonl b/research/experiments/044-specs/2026-10-19-rto.jsonl
new file mode 100644
index 0000000..71d73d7
--- /dev/null
+++ b/research/experiments/044-specs/2026-10-19-rto.jsonl
@@ -0,0 +1 @@
+{"scenarios": 20, "events": 241612, "timeouts": 156, "right": {"batches": 241612, "rounds": 241612, "all": 241612}, "paths_same": true, "costs": {"10": {"ticks": 218070, "timeouts": 78, "ticks_a_timeout": 2795.8, "late_mean_ms": 4.27, "late_most_ms": 9.875}, "100": {"ticks": 21968, "timeouts": 78, "ticks_a_timeout": 281.6, "late_mean_ms": 49.23, "late_most_ms": 99.0}}, "aup_lines": 56, "aup_brace_lines": 11, "python_lines": 30}
diff --git a/research/experiments/044-specs/rto.py b/research/experiments/044-specs/rto.py
new file mode 100644
index 0000000..e4a4a57
--- /dev/null
+++ b/research/experiments/044-specs/rto.py
@@ -0,0 +1,40 @@
+# RFC 6298's retransmission timer in plain Python: the reference 140 counts aup's lines against. A sender's segments
+# are sent and acknowledged (cumulatively) at given times; the round-trip time of each acknowledged segment that was
+# never sent again (Karn) updates SRTT, RTTVAR and RTO (2.2-2.5: 1 s at least, 60 s at most), and the timer runs as
+# 5.1-5.6 say: started when one is sent and none runs, stopped when all are acknowledged, restarted when new data is,
+# and on expiry the earliest is sent again and RTO doubles. The timer is seen at ticks of period P, as an app sees time.
+LOWEST, HIGHEST = 1000, 60000  # milliseconds; the clock's granularity G is the tick's period
+
+
+class Sender:
+    def __init__(self, period):
+        self.period, self.srtt, self.rttvar, self.rto, self.measured = period, 0.0, 0.0, 1000.0, False
+        self.deadline, self.outstanding, self.retransmits = -1.0, [], []  # outstanding: [seq, sent, again]
+
+    def send(self, now, seq):
+        self.outstanding.append([seq, now, False])
+        if self.deadline < 0:
+            self.deadline = now + self.rto  # 5.1
+
+    def ack(self, now, seq):
+        acked = [s for s in self.outstanding if s[0] <= seq]
+        if not acked:
+            return
+        self.outstanding = [s for s in self.outstanding if s[0] > seq]
+        last = acked[-1]
+        if not last[2]:  # Karn: no sample from a segment sent again
+            r = now - last[1]
+            if not self.measured:
+                self.srtt, self.rttvar, self.measured = r, r / 2, True  # 2.2
+            else:
+                self.rttvar = 0.75 * self.rttvar + 0.25 * abs(self.srtt - r)  # 2.3
+                self.srtt = 0.875 * self.srtt + 0.125 * r
+            self.rto = min(max(self.srtt + max(self.period, 4 * self.rttvar), LOWEST), HIGHEST)  # 2.4, 2.5
+        self.deadline = now + self.rto if self.outstanding else -1.0  # 5.3, 5.2
+
+    def tick(self, now):
+        if self.deadline >= 0 and now >= self.deadline:  # 5.4-5.6
+            self.outstanding[0][2] = True
+            self.retransmits.append((self.outstanding[0][0], now))
+            self.rto = min(self.rto * 2, HIGHEST)
+            self.deadline = now + self.rto
diff --git a/research/experiments/044-specs/rto_check.py b/research/experiments/044-specs/rto_check.py
new file mode 100644
index 0000000..be41ffb
--- /dev/null
+++ b/research/experiments/044-specs/rto_check.py
@@ -0,0 +1,145 @@
+#!/usr/bin/env python3
+"""Research 140: RFC 6298's retransmission timer written in aup (aup's examples/rto.aup), with periodic ticks.
+
+Each scenario (a seed and a tick period P) is made closed-loop by rto.py's model and a receiver: segments sent at random
+gaps, each arriving half a random round trip later unless its first sending is lost (one in five), the receiver
+acknowledging cumulatively all it holds without a gap, the acknowledgement arriving after the other half; a segment the
+timer sends again arrives. The events (sends, acknowledgements, ticks every P ms) are then given to the app
+through pure-run (time MS before each), on Pure's three paths: after every event its SRTT, RTTVAR, RTO, deadline, the
+segments outstanding and the retransmissions must be the model's, the paths the same. No known implementation is
+compared: the oracle is the model of the same rules. What the ticks cost is measured: ticks a timeout (a one-shot timer
+would be one event), and how late a timeout is seen (below P). Then counts lines of rto.aup and of rto.py's model. One
+JSON line."""
+import json, os, random, struct, subprocess, sys, tempfile
+
+here = os.path.dirname(os.path.abspath(__file__))
+pure = os.path.normpath(os.path.join(here, '../../..'))
+aup = os.environ.get('AUP', os.path.join(pure, '..', 'aup'))
+sys.path.insert(0, here)
+import rto  # noqa: E402
+
+START = 1_000_000  # the first event's time (ms since 1970, as now is)
+
+
+def lines(path, stop=None):
+    code = braces = 0
+    for line in open(path):
+        if stop and line.startswith(stop): break
+        text = line.strip()
+        if not text or text.startswith('#'): continue
+        code += 1
+        braces += text == '}'
+    return code, braces
+
+
+def scenario(seed, period, segments=40):
+    """The events the app is given (time, kind, seq), the model's state after each, and the timeouts' lateness. The
+    network: a sending arrives half a random round trip later unless it is lost (a first sending, one in five), and the
+    receiver acknowledges at once, cumulatively, all it holds without a gap; the acknowledgement arrives after the other
+    half."""
+    chooser = random.Random(seed)
+    sender = rto.Sender(period)
+    pending, t = [], START + period  # (time, order, kind, seq): arrivals, then acknowledgements, sends, then ticks
+    for seq in range(1, segments + 1):
+        t += chooser.choice([0, 0, 5, 40, 120, 300, 900])
+        pending.append((t, 2, 'transmit', seq))
+    held, through = set(), 0  # the receiver's segments, and the last one it holds every one up to
+
+    def sent(time, seq, first):
+        if not first or chooser.random() > 0.2:
+            half = chooser.randrange(20, 350)
+            pending.append((time + half, 0, 'arrive', seq * 1000 + half))
+
+    events, states, late = [], [], []
+    tick = START
+    while pending or sender.outstanding:
+        pending.sort()
+        if pending and pending[0][0] < tick + period:  # the host ticks every P, the timer running or not
+            time, _, kind, seq = pending.pop(0)
+        else:
+            time, kind, seq = tick + period, 'tick', 0
+            tick = time
+        if kind == 'arrive':  # the receiver's: not the app's
+            seq, half = divmod(seq, 1000)
+            held.add(seq)
+            while through + 1 in held:
+                through += 1
+            if through:
+                pending.append((time + half, 1, 'ack', through))
+            continue
+        if kind == 'transmit':
+            sender.send(time, seq)
+            sent(time, seq, True)
+        elif kind == 'ack':
+            sender.ack(time, seq)
+        else:
+            before, deadline = len(sender.retransmits), sender.deadline
+            sender.tick(time)
+            if len(sender.retransmits) > before:
+                late.append(time - deadline)
+                sent(time, sender.retransmits[-1][0], False)
+        events.append((time, kind, seq))
+        states.append((sender.srtt, sender.rttvar, sender.rto, sender.deadline, len(sender.outstanding), len(sender.retransmits)))
+        if len(events) > 200000:
+            raise RuntimeError('a scenario without end')
+    return events, states, late
+
+
+def run(module, events, env, work):
+    commands = os.path.join(work, 'rto.commands')
+    with open(commands, 'w') as f:
+        for time, kind, seq in events:
+            f.write(f'time {time}\n')
+            f.write('tick tick\n' if kind == 'tick' else f"message {kind} {struct.pack('<I', seq).hex()}\n")
+    got = []
+    with open(commands) as stdin:
+        process = subprocess.Popen([os.path.join(pure, 'build', 'pure-run'), module], stdin=stdin, stdout=subprocess.PIPE,
+                                   text=True, env=dict(os.environ, **env))
+        for k, line in enumerate(process.stdout):
+            if k % 2 == 0: continue  # a time command's
+            state = json.loads(line)
+            s = state['states']
+            got.append((state['fault'], s['srtt'], s['rttvar'], s['rto'], s['deadline'], len(s['outstanding']),
+                        len(s['retransmits'])))
+        process.wait()
+    return got
+
+
+if __name__ == '__main__':
+    report = {'scenarios': 0, 'events': 0, 'timeouts': 0, 'right': {}, 'paths_same': True}
+    costs = {}
+    seeds = int(sys.argv[1]) if len(sys.argv) > 1 else 10
+    with tempfile.TemporaryDirectory() as work:
+        source = open(os.path.join(aup, 'examples', 'rto.aup')).read()
+        for period in (10, 100):
+            module = os.path.join(work, f'rto-{period}.pmod')  # its tick and its G the scenario's period
+            with open(os.path.join(work, f'rto-{period}.aup'), 'w') as f:
+                f.write(source.replace('every 10 ms', f'every {period} ms').replace('fn PERIOD = 10 ', f'fn PERIOD = {period} '))
+            subprocess.run([os.path.join(aup, 'build', 'aup'), 'build', os.path.join(work, f'rto-{period}.aup'), '-o', module], check=True)
+            ticks = timeouts = 0
+            lateness = []
+            for seed in range(seeds):
+                events, states, late = scenario(seed, period)
+                outputs = {}
+                for name, env in (('batches', {'PURE_JIT': '0'}), ('rounds', {'PURE_JIT': '0', 'PURE_BATCH': '0'}),
+                                  ('all', {'PURE_JIT': 'all'})):
+                    got = run(module, events, env, work)
+                    want = [('none',) + s for s in states]
+                    report['right'][name] = report['right'].get(name, 0) + sum(g == w for g, w in zip(got, want))
+                    outputs[name] = got
+                report['paths_same'] = report['paths_same'] and outputs['rounds'] == outputs['batches'] == outputs['all']
+                report['scenarios'] += 1
+                report['events'] += len(events)
+                ticks += sum(kind == 'tick' for _, kind, _ in events)
+                timeouts += len(late)
+                lateness += late
+            report['timeouts'] += timeouts
+            costs[period] = {'ticks': ticks, 'timeouts': timeouts, 'ticks_a_timeout': round(ticks / max(timeouts, 1), 1),
+                             'late_mean_ms': round(sum(lateness) / max(len(lateness), 1), 2), 'late_most_ms': max(lateness, default=0)}
+    report['costs'] = costs
+    aup_code, aup_braces = lines(os.path.join(aup, 'examples', 'rto.aup'))
+    python_code, _ = lines(os.path.join(here, 'rto.py'))
+    report.update({'aup_lines': aup_code, 'aup_brace_lines': aup_braces, 'python_lines': python_code})
+    ok = all(n == report['events'] for n in report['right'].values()) and report['paths_same']
+    print(json.dumps(report))
+    sys.exit(0 if ok else 1)