# -*- coding: utf-8 -*- """修：被顶替的会话抢走了本该给新会话的数据。 证据（Java 客户端打点）： [dbg] attach ok session=ad9283e7 <- 新手表接入 [dbg] send 122 bytes -> 200 <- 新手表发 auth [dbg] poll -> 200 bytes=53 session=ad9283e7 <- 新手表其实收到了 [dbg] poll -> 200 bytes=50 session=a7c11203 <- 但旧手表也收到了一份 根因： device.to_client 是**按设备共享**的一个缓冲区。 旧会话被顶掉后，它那个 40 秒的长轮询还挂在 device.event 上。 新数据一来，谁先醒谁拿走 —— 旧会话就把它截胡了。 修法： 1. Session 各自带一个 event 2. 顶替旧会话时，把它的 event set() 一下 —— 它的长轮询立刻醒来 3. client_take 每轮先看 session.replaced，是就抛 _Replaced 4. 接口层把 _Replaced 翻成 409 + replaced，客户端据此断开（不抢） 5. 数据只唤醒**当前**会话（device.active_session） """ from __future__ import annotations import sys from pathlib import Path RELAY = Path(r"C:\Users\shu\Desktop\deepseekagent\LanRemote\relay\relay.py") # ---------------- 1. Device 记住当前会话对象 ---------------- OLD_SLOTS = ''' "event", "bytes_in", "bytes_out", "current_session")''' NEW_SLOTS = ''' "event", "bytes_in", "bytes_out", "current_session", "active_session")''' OLD_INIT = ''' self.current_session = ""''' NEW_INIT = ''' self.current_session = "" # 当前挂在这台设备上的 Session 对象。数据来了只唤醒它， # 不要让已经被顶掉的旧会话把数据截胡。 self.active_session = None''' # ---------------- 2. push_to_client 只唤醒当前会话 ---------------- OLD_PUSH = ''' def push_to_client(self, data: bytes) -> None: if not data: return self.bytes_out += len(data) self._append(self.to_client, data) self.event.set()''' NEW_PUSH = ''' def push_to_client(self, data: bytes) -> None: if not data: return self.bytes_out += len(data) self._append(self.to_client, data) self.event.set() # 同时叫醒当前那个会话的长轮询（旧会话已经被替换，不该再拿数据） session = self.active_session if session is not None: session.event.set()''' # ---------------- 3. Session 带自己的 event + replaced ---------------- OLD_SESSION = ''' __slots__ = ("token", "device_id", "code", "name", "ip", "created", "last_seen") def __init__(self, device_id: str, code: str, name: str, ip: str) -> None: self.token = secrets.token_hex(24) self.device_id = device_id self.code = code self.name = name self.ip = ip self.created = time.time() self.last_seen = time.time()''' NEW_SESSION = ''' __slots__ = ("token", "device_id", "code", "name", "ip", "created", "last_seen", "event", "replaced") def __init__(self, device_id: str, code: str, name: str, ip: str) -> None: self.token = secrets.token_hex(24) self.device_id = device_id self.code = code self.name = name self.ip = ip self.created = time.time() self.last_seen = time.time() # 自己的唤醒事件。每个会话独立等待，避免被顶掉的旧会话截胡数据。 self.event = threading.Event() self.replaced = False''' # ---------------- 4. attach：踢的时候唤醒旧会话 ---------------- OLD_KICK = ''' if stale.device_id == device_id: self.sessions.pop(stale_token, None) self.retired[stale_token] = "replaced" log(f"踢掉旧会话 {stale_token[:8]}…（同一设备只允许一个）")''' NEW_KICK = ''' if stale.device_id == device_id: self.sessions.pop(stale_token, None) self.retired[stale_token] = "replaced" stale.replaced = True # 立刻叫醒它挂着的长轮询 —— 否则它会一直等下去， # 等新数据来时把本该给新手表的数据抢走。 stale.event.set() log(f"踢掉旧会话 {stale_token[:8]}…（同一设备只允许一个）")''' OLD_BIND = ''' device.current_session = session.token # 换会话了：把上一条留在队列里的半截数据清掉， # 不然新手表会收到上一个手表没取走的残留字节。 device.to_client.clear()''' NEW_BIND = ''' device.current_session = session.token device.active_session = session # 换会话了：把上一条留在队列里的半截数据清掉， # 不然新手表会收到上一个手表没取走的残留字节。 device.to_client.clear()''' # ---------------- 5. client_take：独立事件 + 让位 ---------------- OLD_TAKE = ''' def client_take(self, session: Session, wait: float) -> bytes: """手表取走「被控端发来的」数据。""" with self.lock: device = self.devices.get(session.device_id) if device is None: return b"" return self._take(device, device.to_client, wait, "client")''' NEW_TAKE = ''' def client_take(self, session: Session, wait: float) -> bytes: """手表取走「被控端发来的」数据。 注意这里**不能**复用 _take：那个是按设备的事件等的， 会把被顶掉的旧会话和新会话混在一起等，导致数据被旧会话截胡。 每个会话等自己的 event，并且每轮先看自己是不是已经被顶替了。 """ deadline = time.time() + wait while True: if session.replaced: raise SessionReplaced() with self.lock: device = self.devices.get(session.device_id) if device is None: return b"" if device.to_client: data = bytes(device.to_client) device.to_client.clear() device.event.clear() session.event.clear() return data remaining = deadline - time.time() if remaining <= 0: return b"" session.event.wait(min(remaining, 0.5)) session.event.clear()''' # ---------------- 6. _Replaced 异常 ---------------- OLD_EXC_ANCHOR = '''class Relay:''' NEW_EXC = '''class SessionReplaced(Exception): """这个会话已经被新的手表顶替了 —— 它必须让位，不能继续取数据。""" class Relay:''' # ---------------- 7. 接口层翻成 409 ---------------- OLD_POLL = ''' data = RELAY.client_take(session, POLL_SECONDS) self._send(200, data, "application/octet-stream") return''' NEW_POLL = ''' try: data = RELAY.client_take(session, POLL_SECONDS) except SessionReplaced: self._json(409, {"ok": False, "replaced": True, "error": "已被另一台手表接管"}) return self._send(200, data, "application/octet-stream") return''' # ---------------- 8. push_to_agent 也唤醒对应会话（agent 侧不用改） ---------------- def main() -> int: text = RELAY.read_text(encoding="utf-8") ok = True for old, new, tag in ((OLD_SLOTS, NEW_SLOTS, "Device 加 active_session"), (OLD_INIT, NEW_INIT, "初始化 active_session"), (OLD_PUSH, NEW_PUSH, "push 只唤醒当前会话"), (OLD_SESSION, NEW_SESSION, "Session 带 event/replaced"), (OLD_KICK, NEW_KICK, "踢的时候唤醒旧会话"), (OLD_BIND, NEW_BIND, "绑定 active_session"), (OLD_TAKE, NEW_TAKE, "client_take 独立等待 + 让位"), (OLD_EXC_ANCHOR, NEW_EXC, "SessionReplaced 异常"), (OLD_POLL, NEW_POLL, "poll 把让位翻成 409")): if old not in text: print(f" [失败] {tag} —— 锚点没找到") ok = False continue text = text.replace(old, new, 1) print(f" {tag}") RELAY.write_text(text, encoding="utf-8") import ast try: ast.parse(text) print(" 语法 OK") except SyntaxError as exc: print(f" 语法错误: {exc}") ok = False return 0 if ok else 1 if __name__ == "__main__": sys.exit(main())