从 JSONL 文件加载会话。 逐行解析,跳过解析失败的行(崩溃安全)。 对应 Reference: EP09 §3 parseJSONL → Map
(self, session_id: str)
| 336 | f.write(line) |
| 337 | |
| 338 | def load_session(self, session_id: str) -> Session: |
| 339 | """ |
| 340 | 从 JSONL 文件加载会话。 |
| 341 | |
| 342 | 逐行解析,跳过解析失败的行(崩溃安全)。 |
| 343 | |
| 344 | 对应 Reference: EP09 §3 |
| 345 | parseJSONL → Map<UUID, TranscriptMessage> |
| 346 | """ |
| 347 | path = self._session_path(session_id) |
| 348 | session = Session(session_id=session_id) |
| 349 | metadata = {} |
| 350 | |
| 351 | with open(path, "r") as f: |
| 352 | for line_no, line in enumerate(f, 1): |
| 353 | line = line.strip() |
| 354 | if not line: |
| 355 | continue |
| 356 | try: |
| 357 | entry = json.loads(line) |
| 358 | except json.JSONDecodeError: |
| 359 | # 崩溃安全:跳过写到一半的行 |
| 360 | print(f" [WARNING] 第 {line_no} 行解析失败,跳过 " |
| 361 | "(可能是崩溃时写到一半的数据)") |
| 362 | continue |
| 363 | |
| 364 | entry_type = entry.get("type", "") |
| 365 | if entry_type in ("user", "assistant", "tool"): |
| 366 | session.messages.append( |
| 367 | ConversationMessage.from_dict(entry) |
| 368 | ) |
| 369 | elif entry_type == "custom-title": |
| 370 | metadata["title"] = entry.get("title", "") |
| 371 | elif entry_type == "last-prompt": |
| 372 | metadata["last_prompt"] = entry.get("text", "") |
| 373 | |
| 374 | return session |
| 375 | |
| 376 | def list_sessions(self) -> list[dict]: |
| 377 | """ |