查看收件箱中有多少消息(不取出)
(self, agent_name: str)
| 244 | return [AgentMessage.from_dict(m) for m in raw_messages] |
| 245 | |
| 246 | def peek(self, agent_name: str) -> int: |
| 247 | """查看收件箱中有多少消息(不取出)""" |
| 248 | inbox_path = self._inbox_path(agent_name) |
| 249 | if not os.path.exists(inbox_path): |
| 250 | return 0 |
| 251 | with open(inbox_path, "r") as f: |
| 252 | try: |
| 253 | return len(json.load(f)) |
| 254 | except json.JSONDecodeError: |
| 255 | return 0 |
| 256 | |
| 257 | |
| 258 | # ============================================================ |
nothing calls this directly
no test coverage detected