repro/mock_telegram.py

1#!/usr/bin/env python3

2"""Mock Telegram Bot API for OFF-DEVICE courier repro.

4Models the two semantics that matter for the getUpdates-backlog spam bug:

6 * getUpdates(offset=N) CONFIRMS (drops) every pending update with

7 update_id < N, then returns the remaining pending updates. A call

8 with no offset (or offset 0) confirms nothing and returns the whole

9 backlog -- exactly how a fresh poller (offset reset to 0) re-fetches

10 Telegram's ~24h backlog.

11 * sendMessage records the outbound and returns ok. Nothing here ever

12 reaches real Telegram; the point is to COUNT would-be sends.

14Every request is appended to <logdir>/requests.log so the driver can

15assert on re-delivery. Control endpoints (not part of the Telegram API)

16let the driver seed and inject updates:

18 POST /_seed body: {"updates":[{...},...]} replace the backlog

19 POST /_add body: {"text":"...","sender_id":"..","chat_id":".."} append one update

20 POST /_dwell body: {"seconds":N} long-poll dwell (see below)

21 GET /_stats {"delivered":[ids], "sends":N,

22 "conflicts":N, ...}

24The SINGLE-CONSUMER rule (409 Conflict) is modelled too, because the

25"exactly one poller per token" fix cannot be verified against a mock that

26is incapable of producing the error it is supposed to eliminate -- "zero

27409s" would then pass for free, on a mock, with the bug fully present.

29Telegram permits ONE getUpdates consumer per bot token: a getUpdates that

30arrives while another is already in flight is answered

31 409 {"ok":false,"error_code":409,

32 "description":"Conflict: terminated by other getUpdates request..."}

33and that is the error stream observed live on 2026-08-04 with three

34concurrent couriers.

36`dwell` is what makes this observable off-device. Real Telegram HOLDS an

37empty getUpdates open for the caller's `timeout` seconds, which is what

38makes two independent pollers overlap almost continuously. courier polls

39with timeout=0, so a mock that answers instantly leaves an in-flight

40window of microseconds and two pollers would collide only by luck. Setting

41a dwell (e.g. 2.5s, longer than courier's 2s poll interval) restores the

42overlap deterministically:

44 * TWO pollers -> the second one's call always lands inside the first

45 one's dwell -> conflicts climb.

46 * ONE poller -> its calls are strictly sequential (poll, wait, poll),

47 so it can never overlap ITSELF -> conflicts stay 0.

49Dwell defaults to 0 (env GETUPDATES_DWELL), so every pre-existing repro

50keeps its original timing and behaviour.

52The sentinel-zero (2026-07-02) storm case is exercised by

53`repro-sentinel-zero.sh`: `_seed []` gives an EMPTY backlog so a cold start

54drains count=0, then `_add` accumulates a backlog before a restart -- the

55scenario where persisting offset=0 (pre-fix) causes re-delivery.

57Usage: mock_telegram.py <port> <logdir>

58"""

59import json

60import sys

61import os

62import time

63import threading

64from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

66PORT = int(sys.argv[1])

67LOGDIR = sys.argv[2]

68os.makedirs(LOGDIR, exist_ok=True)

69REQ_LOG = os.path.join(LOGDIR, "requests.log")

70SEND_LOG = os.path.join(LOGDIR, "sends.log")

72# ------------------------------------------------------------------

73# sendMessage failure-injection (send-path duplication repro).

74#

75# The default "ok" mode returns a normal Telegram success. The other

76# modes model the transport conditions that make courier's send handler

77# NOT return a clean, timely success -- the situations under which the

78# leader's MCP client would re-issue the send and (pre-fix) re-deliver:

79#

80# ok normal 200 {ok:true} response

81# delay:<sec> sleep <sec> BEFORE responding (models a slow ack read;

82# with <sec> > courier's *send-request-timeout* the read

83# times out AFTER Telegram already recorded the send)

84# reset record the send, then drop the connection with NO HTTP

85# response (models a delivered-but-ack-lost read failure)

86# status:<code> record the send, respond with an HTTP error status

87# notok record the send, respond 200 {ok:false} (API-level error)

88#

89# Set the initial mode via env SEND_MODE; change at runtime via

90# POST /_mode {"mode":"..."}. Every mode still COUNTS the send, because

91# in all of them Telegram has actually received (and would deliver) the

92# message -- that is the whole point of counting deliveries at the mock.

93# ------------------------------------------------------------------

94SEND_MODE_DEFAULT = os.environ.get("SEND_MODE", "ok")

96LOCK = threading.Lock()

97STATE = {

98 "updates": [], # pending updates (list of dicts with update_id)

99 "next_id": 1, # next update_id to assign via _add

100 "getupdates": 0, # count of getUpdates calls

101 "sends": 0, # count of sendMessage calls

102 "delivered_ids": [], # update_ids the mock RETURNED to a poller (per call)

103 "send_mode": SEND_MODE_DEFAULT,

104 # --- single-consumer (409) modelling; see module docstring ---

105 "inflight": 0, # getUpdates calls currently being served

106 "conflicts": 0, # getUpdates calls rejected with 409

107 "max_inflight": 0, # high-water mark, i.e. observed concurrent pollers

108 "dwell": float(os.environ.get("GETUPDATES_DWELL", "0")),

109}

110

111

112def logline(path, msg):

113 with open(path, "a") as f:

114 f.write(msg + "\n")

115 f.flush()

116

117

118def make_update(update_id, text, sender_id="42", chat_id="1001"):

119 return {

120 "update_id": update_id,

121 "message": {

122 "message_id": update_id,

123 "date": 1000000 + update_id,

124 "text": text,

125 "from": {"id": int(sender_id), "is_bot": False,

126 "first_name": "David", "username": "daviwil"},

127 "chat": {"id": int(chat_id), "type": "private"},

128 },

129 }

130

131

132class Handler(BaseHTTPRequestHandler):

133 def log_message(self, *a):

134 pass # silence default stderr logging

135

136 def _body(self):

137 n = int(self.headers.get("Content-Length", 0))

138 raw = self.rfile.read(n) if n else b""

139 try:

140 return json.loads(raw) if raw else {}

141 except Exception:

142 return {}

143

144 def _reply(self, obj, code=200):

145 data = json.dumps(obj).encode()

146 try:

147 self.send_response(code)

148 self.send_header("Content-Type", "application/json")

149 self.send_header("Content-Length", str(len(data)))

150 self.end_headers()

151 self.wfile.write(data)

152 except (BrokenPipeError, ConnectionResetError):

153 # courier was SIGKILLed mid-send (the storm's restart step) --

154 # the delivery was still counted; the lost ack is expected.

155 pass

156

157 def do_GET(self):

158 if self.path == "/_stats":

159 with LOCK:

160 self._reply({"delivered": STATE["delivered_ids"],

161 "sends": STATE["sends"],

162 "getupdates": STATE["getupdates"],

163 "conflicts": STATE["conflicts"],

164 "max_inflight": STATE["max_inflight"],

165 "dwell": STATE["dwell"],

166 "pending": [u["update_id"] for u in STATE["updates"]]})

167 return

168 self._reply({"ok": False, "description": "not found"}, 404)

169

170 def do_POST(self):

171 body = self._body()

172 path = self.path

173

174 # ---- control endpoints ----

175 if path == "/_seed":

176 with LOCK:

177 STATE["updates"] = list(body.get("updates", []))

178 STATE["next_id"] = (max([u["update_id"] for u in STATE["updates"]],

179 default=0) + 1)

180 STATE["getupdates"] = 0

181 STATE["sends"] = 0

182 STATE["delivered_ids"] = []

183 STATE["conflicts"] = 0

184 STATE["max_inflight"] = 0

185 self._reply({"ok": True})

186 return

187 if path == "/_dwell":

188 with LOCK:

189 STATE["dwell"] = float(body.get("seconds", 0))

190 d = STATE["dwell"]

191 logline(REQ_LOG, f"--- getUpdates dwell set to {d}s ---")

192 self._reply({"ok": True, "dwell": d})

193 return

194 if path == "/_add":

195 with LOCK:

196 uid = STATE["next_id"]

197 STATE["next_id"] += 1

198 STATE["updates"].append(make_update(

199 uid, body.get("text", "msg"),

200 body.get("sender_id", "42"), body.get("chat_id", "1001")))

201 self._reply({"ok": True, "update_id": uid})

202 return

203 if path == "/_mode":

204 with LOCK:

205 STATE["send_mode"] = body.get("mode", "ok")

206 mode = STATE["send_mode"]

207 logline(REQ_LOG, f"--- send_mode set to {mode!r} ---")

208 self._reply({"ok": True, "mode": mode})

209 return

210

211 # ---- Telegram Bot API ----

212 # path looks like /bot<token>/<method>

213 method = path.rsplit("/", 1)[-1]

214

215 if method == "getUpdates":

216 offset = body.get("offset", 0) or 0

217 with LOCK:

218 STATE["getupdates"] += 1

219 # Single-consumer rule: a getUpdates arriving while another

220 # is in flight is a second poller on this token. Real

221 # Telegram terminates one of them with 409; we reject the

222 # newcomer, which is equivalent for counting purposes.

223 if STATE["inflight"] > 0:

224 STATE["conflicts"] += 1

225 STATE["max_inflight"] = max(STATE["max_inflight"],

226 STATE["inflight"] + 1)

227 n = STATE["conflicts"]

228 conflict = True

229 else:

230 STATE["inflight"] += 1

231 STATE["max_inflight"] = max(STATE["max_inflight"],

232 STATE["inflight"])

233 conflict = False

234 if conflict:

235 logline(REQ_LOG, f"getUpdates offset={offset} -> 409 CONFLICT (#{n})")

236 self._reply({"ok": False, "error_code": 409,

237 "description": "Conflict: terminated by other "

238 "getUpdates request; make sure that "

239 "only one bot instance is running"},

240 code=409)

241 return

242 try:

243 with LOCK:

244 if offset > 0:

245 # Confirm: drop everything below the offset.

246 STATE["updates"] = [u for u in STATE["updates"]

247 if u["update_id"] >= offset]

248 result = list(STATE["updates"])

249 ids = [u["update_id"] for u in result]

250 STATE["delivered_ids"].append({"offset": offset,

251 "returned": ids})

252 dwell = STATE["dwell"]

253 # Long-poll emulation: hold an EMPTY response open, exactly

254 # as Telegram holds the caller's `timeout`. This is what makes

255 # two independent pollers overlap observably; see the module

256 # docstring. A non-empty result returns immediately, as

257 # Telegram does.

258 if dwell > 0 and not result:

259 time.sleep(dwell)

260 logline(REQ_LOG,

261 f"getUpdates offset={offset} -> returned={ids}")

262 self._reply({"ok": True, "result": result})

263 finally:

264 with LOCK:

265 STATE["inflight"] -= 1

266 return

267

268 if method == "sendMessage":

269 chat_id = body.get("chat_id")

270 text = body.get("text", "")

271 # Count the send FIRST: in every mode Telegram has received the

272 # message and would deliver it to David's phone. That is exactly

273 # the count that matters -- how many times the phone buzzes.

274 with LOCK:

275 STATE["sends"] += 1

276 n = STATE["sends"]

277 mode = STATE["send_mode"]

278 logline(SEND_LOG, f"#{n} sendMessage chat_id={chat_id} text={text!r} mode={mode}")

279 logline(REQ_LOG, f"sendMessage chat_id={chat_id} text={text!r} mode={mode}")

280

281 # ---- failure injection ----

282 if mode == "reset":

283 # Delivered, but ack read fails: drop the socket, no response.

284 try:

285 self.connection.close()

286 except Exception:

287 pass

288 return

289 if mode.startswith("delay:"):

290 try:

291 secs = float(mode.split(":", 1)[1])

292 except ValueError:

293 secs = 30.0

294 time.sleep(secs)

295 # fall through to a normal ok response (may arrive after the

296 # client/courier read deadline already fired)

297 if mode.startswith("status:"):

298 try:

299 code = int(mode.split(":", 1)[1])

300 except ValueError:

301 code = 500

302 self._reply({"ok": False, "error_code": code,

303 "description": "injected error"}, code=code)

304 return

305 if mode == "notok":

306 self._reply({"ok": False, "error_code": 400,

307 "description": "injected api error"})

308 return

309

310 self._reply({"ok": True, "result": {

311 "message_id": 9000 + n, "date": 1,

312 "chat": {"id": chat_id, "type": "private"}, "text": text}})

313 return

314

315 # Any other Telegram method (getMe, etc.) -> generic ok.

316 logline(REQ_LOG, f"{method} (generic-ok)")

317 self._reply({"ok": True, "result": {}})

318

319

320if __name__ == "__main__":

321 logline(REQ_LOG, f"--- mock start on :{PORT} ---")

322 srv = ThreadingHTTPServer(("127.0.0.1", PORT), Handler)

323 srv.serve_forever()