repro/send_driver.py

1#!/usr/bin/env python3

2"""Off-device driver for the courier send-path duplication repro.

4Acts as a minimal MCP client (the "leader") speaking JSON-RPC over

5courier's stdio. Fires ONE logical `send-message` and models how the

6real leader harness reacts when a send does not return a clean, timely

7success:

9 * a per-tool-call timeout (the client stops waiting for the response),

10 * on timeout OR courier death, kill + respawn courier and RE-ISSUE the

11 same send, up to --retries times (this is the harness /mcp-reconnect

12 + agent-retry loop that turned one logical send into N deliveries).

14The bug is proven by pointing courier's SEND path at mock_telegram.py

15(COURIER_TELEGRAM_API_URL) and counting sendMessage hits: one logical

16send producing >1 delivery == reproduced.

18NOTHING here can reach real Telegram: the api-url override forces every

19request to the local mock.

21Usage:

22 send_driver.py --bin <courier> --api-url <mock-base> --logdir <dir>

23 [--mode ok|reset|delay:N|status:N|notok]

24 [--tool-timeout SEC] [--retries N] [--text STR]

25 [--mode-endpoint URL]

26"""

27import argparse

28import json

29import os

30import subprocess

31import sys

32import threading

33import time

34import urllib.request

36CHAT_ID = "331005009" # David's real chat id, per the captured log. Never used

37 # live: the api-url override sends only to the mock.

40def set_mock_mode(mode_endpoint, mode):

41 data = json.dumps({"mode": mode}).encode()

42 req = urllib.request.Request(mode_endpoint, data=data,

43 headers={"Content-Type": "application/json"})

44 urllib.request.urlopen(req, timeout=5).read()

47def mock_sends(stats_url):

48 with urllib.request.urlopen(stats_url, timeout=5) as r:

49 return json.load(r)["sends"]

52class Courier:

53 """A single courier subprocess speaking MCP over stdio."""

55 def __init__(self, bin_path, env, logdir, gen):

56 self.gen = gen

57 self.errf = open(os.path.join(logdir, f"courier-{gen}.stderr"), "wb")

58 self.proc = subprocess.Popen(

59 [bin_path, "serve"],

60 stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=self.errf,

61 env=env, bufsize=0)

62 self._id = 0

63 self._lock = threading.Lock()

65 def _send(self, obj):

66 line = (json.dumps(obj) + "\n").encode()

67 self.proc.stdin.write(line)

68 self.proc.stdin.flush()

70 def _rpc(self, method, params, timeout):

71 """Send a request, wait up to `timeout` for the matching response.

72 Returns (result_or_error_dict, elapsed). Raises TimeoutError if no

73 response arrives in time, or ProcessLookupError if courier dies."""

74 with self._lock:

75 self._id += 1

76 rid = self._id

77 self._send({"jsonrpc": "2.0", "id": rid, "method": method,

78 "params": params})

79 result = {}

80 done = threading.Event()

82 def reader():

83 nonlocal result

84 for raw in self.proc.stdout:

85 try:

86 msg = json.loads(raw)

87 except Exception:

88 continue

89 if msg.get("id") == rid:

90 result = msg

91 done.set()

92 return

94 t = threading.Thread(target=reader, daemon=True)

95 t.start()

96 start = time.time()

97 ok = done.wait(timeout)

98 elapsed = time.time() - start

99 if not ok:

100 if self.proc.poll() is not None:

101 raise ProcessLookupError(f"courier exited rc={self.proc.returncode}")

102 raise TimeoutError(f"no response to {method} in {timeout}s")

103 return result, elapsed

104

105 def notify(self, method, params):

106 self._send({"jsonrpc": "2.0", "method": method, "params": params})

107

108 def initialize(self, timeout):

109 res, _ = self._rpc("initialize", {

110 "protocolVersion": "2024-11-05",

111 "capabilities": {},

112 "clientInfo": {"name": "send-driver", "version": "0"},

113 }, timeout)

114 self.notify("notifications/initialized", {})

115 return res

116

117 def call_send(self, text, timeout):

118 return self._rpc("tools/call", {

119 "name": "send-message",

120 "arguments": {"text": text, "to": CHAT_ID},

121 }, timeout)

122

123 def alive(self):

124 return self.proc.poll() is None

125

126 def kill(self):

127 try:

128 self.proc.kill()

129 self.proc.wait(timeout=5)

130 except Exception:

131 pass

132 try:

133 self.errf.close()

134 except Exception:

135 pass

136

137

138def main():

139 ap = argparse.ArgumentParser()

140 ap.add_argument("--bin", required=True)

141 ap.add_argument("--api-url", required=True, help="mock base, e.g. http://127.0.0.1:PORT")

142 ap.add_argument("--logdir", required=True)

143 ap.add_argument("--stats-url", required=True)

144 ap.add_argument("--mode-endpoint", required=True)

145 ap.add_argument("--mode", default="ok")

146 ap.add_argument("--tool-timeout", type=float, default=30.0)

147 ap.add_argument("--retries", type=int, default=0,

148 help="max kill+respawn+re-send cycles after a timeout/death")

149 ap.add_argument("--text", default="repro test message")

150 args = ap.parse_args()

151

152 env = dict(os.environ)

153 env["COURIER_TELEGRAM_TOKEN"] = "TESTTOKEN"

154 env["COURIER_TELEGRAM_CHAT_ID"] = CHAT_ID

155 env["COURIER_TELEGRAM_API_URL"] = args.api_url

156 # Sends ENABLED against the mock (this is the whole point).

157 env.pop("COURIER_DISABLE_TELEGRAM_SEND", None)

158 env["CLAUDE_OPS_LOG_DIR"] = args.logdir

159 # Isolate persistent state (relay dir + the send-dedup.state that the

160 # fix writes beside it) to this run's logdir, so runs don't cross-

161 # contaminate. Every restarted courier gen shares this dir -- exactly

162 # how the real leader's restarts share ~/.courier -- so the persistent

163 # dedup can do its job across process restarts.

164 env["COURIER_RELAY_DIR"] = os.path.join(os.path.abspath(args.logdir), "relays")

165

166 set_mock_mode(args.mode_endpoint, args.mode)

167 before = mock_sends(args.stats_url)

168

169 print(f"== driver: mode={args.mode} tool-timeout={args.tool_timeout}s "

170 f"retries={args.retries} ==")

171

172 attempts = 0

173 gen = 0

174 outcome = "UNKNOWN"

175 cur = Courier(args.bin, env, args.logdir, gen)

176 try:

177 cur.initialize(timeout=15)

178 while True:

179 attempts += 1

180 print(f"-- attempt {attempts}: sending (courier gen {gen}) --")

181 try:

182 res, elapsed = cur.call_send(args.text, timeout=args.tool_timeout)

183 txt = ""

184 if "result" in res:

185 try:

186 txt = res["result"]["content"][0]["text"]

187 except Exception:

188 txt = json.dumps(res["result"])

189 else:

190 txt = json.dumps(res.get("error", res))

191 print(f" tool returned in {elapsed:.1f}s: {txt!r}")

192 outcome = "SEND_RETURNED"

193 break

194 except TimeoutError as e:

195 print(f" TIMEOUT: {e}")

196 except ProcessLookupError as e:

197 print(f" COURIER DIED: {e}")

198 # timeout or death: this is where the harness would restart+retry

199 if attempts > args.retries:

200 outcome = "GAVE_UP_AFTER_RETRIES"

201 break

202 cur.kill()

203 gen += 1

204 print(f" -> restart courier (gen {gen}) and re-issue send")

205 cur = Courier(args.bin, env, args.logdir, gen)

206 cur.initialize(timeout=15)

207 finally:

208 # Close stdin so courier shuts down cleanly, then reap.

209 time.sleep(0.5)

210 try:

211 cur.proc.stdin.close()

212 except Exception:

213 pass

214 time.sleep(0.5)

215 alive = cur.alive()

216 cur.kill()

217

218 after = mock_sends(args.stats_url)

219 deliveries = after - before

220 print(f"== RESULT: logical_sends=1 attempts={attempts} "

221 f"deliveries_at_mock={deliveries} outcome={outcome} "

222 f"last_courier_alive_before_kill={alive} ==")

223 # Machine-readable summary line for scripts.

224 print(f"SUMMARY mode={args.mode} deliveries={deliveries} "

225 f"attempts={attempts} outcome={outcome}")

226

227

228if __name__ == "__main__":

229 main()