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 GET /_stats {"delivered":[ids], "sends":N}

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

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

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

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

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

28"""

29import json

30import sys

31import os

32import time

33import threading

34from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

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

37LOGDIR = sys.argv[2]

38os.makedirs(LOGDIR, exist_ok=True)

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

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

42# ------------------------------------------------------------------

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

44#

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

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

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

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

49#

50# ok normal 200 {ok:true} response

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

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

53# times out AFTER Telegram already recorded the send)

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

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

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

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

58#

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

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

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

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

63# ------------------------------------------------------------------

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

66LOCK = threading.Lock()

67STATE = {

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

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

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

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

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

73 "send_mode": SEND_MODE_DEFAULT,

74}

77def logline(path, msg):

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

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

80 f.flush()

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

84 return {

85 "update_id": update_id,

86 "message": {

87 "message_id": update_id,

88 "date": 1000000 + update_id,

89 "text": text,

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

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

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

93 },

94 }

97class Handler(BaseHTTPRequestHandler):

98 def log_message(self, *a):

99 pass # silence default stderr logging

100

101 def _body(self):

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

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

104 try:

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

106 except Exception:

107 return {}

108

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

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

111 try:

112 self.send_response(code)

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

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

115 self.end_headers()

116 self.wfile.write(data)

117 except (BrokenPipeError, ConnectionResetError):

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

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

120 pass

121

122 def do_GET(self):

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

124 with LOCK:

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

126 "sends": STATE["sends"],

127 "getupdates": STATE["getupdates"],

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

129 return

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

131

132 def do_POST(self):

133 body = self._body()

134 path = self.path

135

136 # ---- control endpoints ----

137 if path == "/_seed":

138 with LOCK:

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

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

141 default=0) + 1)

142 STATE["getupdates"] = 0

143 STATE["sends"] = 0

144 STATE["delivered_ids"] = []

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

146 return

147 if path == "/_add":

148 with LOCK:

149 uid = STATE["next_id"]

150 STATE["next_id"] += 1

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

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

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

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

155 return

156 if path == "/_mode":

157 with LOCK:

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

159 mode = STATE["send_mode"]

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

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

162 return

163

164 # ---- Telegram Bot API ----

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

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

167

168 if method == "getUpdates":

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

170 with LOCK:

171 STATE["getupdates"] += 1

172 if offset > 0:

173 # Confirm: drop everything below the offset.

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

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

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

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

178 STATE["delivered_ids"].append({"offset": offset, "returned": ids})

179 logline(REQ_LOG,

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

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

182 return

183

184 if method == "sendMessage":

185 chat_id = body.get("chat_id")

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

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

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

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

190 with LOCK:

191 STATE["sends"] += 1

192 n = STATE["sends"]

193 mode = STATE["send_mode"]

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

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

196

197 # ---- failure injection ----

198 if mode == "reset":

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

200 try:

201 self.connection.close()

202 except Exception:

203 pass

204 return

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

206 try:

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

208 except ValueError:

209 secs = 30.0

210 time.sleep(secs)

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

212 # client/courier read deadline already fired)

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

214 try:

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

216 except ValueError:

217 code = 500

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

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

220 return

221 if mode == "notok":

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

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

224 return

225

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

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

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

229 return

230

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

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

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

234

235

236if __name__ == "__main__":

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

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

239 srv.serve_forever()