| 1 | """Session management for conversation history.""" |
| 2 | |
| 3 | import json |
| 4 | import os |
| 5 | import shutil |
| 6 | from dataclasses import dataclass, field |
| 7 | from datetime import datetime |
| 8 | from pathlib import Path |
| 9 | from typing import Any |
| 10 | |
| 11 | from loguru import logger |
| 12 | |
| 13 | from nanobot.config.paths import get_legacy_sessions_dir |
| 14 | from nanobot.session.auto_generate import get_auto_generate |
| 15 | from nanobot.session.source import get_source |
| 16 | from nanobot.utils.helpers import ( |
| 17 | ensure_dir, |
| 18 | find_legal_message_start, |
| 19 | image_placeholder_text, |
| 20 | safe_filename, |
| 21 | ) |
| 22 | |
| 23 | _PREVIEW_METADATA_KEY = "preview" |
| 24 | _PREVIEW_SOURCE_METADATA_KEY = "preview_source" |
| 25 | _INTERNAL_USER_PREFIXES = ( |
| 26 | "Internal workplace", |
| 27 | "Director remote result:", |
| 28 | "REFERENCE_IMAGE_GATE ", |
| 29 | "STORY_DIRECTION_CONFIRM ", |
| 30 | ) |
| 31 | |
| 32 | |
| 33 | @dataclass |
| 34 | class Session: |
| 35 | """A conversation session.""" |
| 36 | |
| 37 | key: str # channel:chat_id |
| 38 | messages: list[dict[str, Any]] = field(default_factory=list) |
| 39 | created_at: datetime = field(default_factory=datetime.now) |
| 40 | updated_at: datetime = field(default_factory=datetime.now) |
| 41 | metadata: dict[str, Any] = field(default_factory=dict) |
| 42 | last_consolidated: int = 0 # Number of messages already consolidated to files |
| 43 | |
| 44 | def add_message(self, role: str, content: str, **kwargs: Any) -> None: |
| 45 | """Add a message to the session.""" |
| 46 | if role == "user" and isinstance(content, str): |
| 47 | from nanobot.session.agent_inject import visible_user_content |
| 48 | |
| 49 | content = visible_user_content(content) |
| 50 | msg = { |
| 51 | "role": role, |
| 52 | "content": content, |
| 53 | "timestamp": datetime.now().isoformat(), |
| 54 | **kwargs |
| 55 | } |
| 56 | self.messages.append(msg) |
| 57 | self.updated_at = datetime.now() |
| 58 | |
| 59 | def get_history(self, max_messages: int = 500) -> list[dict[str, Any]]: |
| 60 | """Return unconsolidated messages for LLM input, aligned to a legal tool-call boundary.""" |
| 61 | unconsolidated = self.messages[self.last_consolidated:] |
| 62 | sliced = unconsolidated[-max_messages:] |
| 63 | |
| 64 | # Avoid starting mid-turn when possible. |
| 65 | for i, message in enumerate(sliced): |
| 66 | if message.get("role") == "user": |
| 67 | sliced = sliced[i:] |
| 68 | break |
| 69 | |
| 70 | # Drop orphan tool results at the front. |
| 71 | start = find_legal_message_start(sliced) |
| 72 | if start: |
| 73 | sliced = sliced[start:] |
| 74 | |
| 75 | out: list[dict[str, Any]] = [] |
| 76 | for message in sliced: |
| 77 | content = message.get("content", "") |
| 78 | if message.get("role") == "user" and isinstance(content, str): |
| 79 | from nanobot.session.agent_inject import visible_user_content |
| 80 | |
| 81 | content = visible_user_content(content) |
| 82 | # Synthesize an ``[image: path]`` breadcrumb from the persisted |
| 83 | # ``media`` kwarg so LLM replay still sees *something* where the |
| 84 | # image used to be. Without this, an image-only user turn |
| 85 | # replays as an empty user message — the assistant's reply then |
| 86 | # looks like it's responding to nothing. |
| 87 | media = message.get("media") |
| 88 | if isinstance(media, list) and media and isinstance(content, str): |
| 89 | breadcrumbs = "\n".join( |
| 90 | image_placeholder_text(p) for p in media if isinstance(p, str) and p |
| 91 | ) |
| 92 | content = f"{content}\n{breadcrumbs}" if content else breadcrumbs |
| 93 | entry: dict[str, Any] = {"role": message["role"], "content": content} |
| 94 | for key in ("tool_calls", "tool_call_id", "name", "reasoning_content"): |
| 95 | if key in message: |
| 96 | entry[key] = message[key] |
| 97 | out.append(entry) |
| 98 | return out |
| 99 | |
| 100 | def clear(self) -> None: |
| 101 | """Clear all messages and reset session to initial state.""" |
| 102 | self.messages = [] |
| 103 | self.last_consolidated = 0 |
| 104 | self.updated_at = datetime.now() |
| 105 | |
| 106 | def retain_recent_legal_suffix(self, max_messages: int) -> None: |
| 107 | """Keep a legal recent suffix, mirroring get_history boundary rules.""" |
| 108 | if max_messages <= 0: |
| 109 | self.clear() |
| 110 | return |
| 111 | if len(self.messages) <= max_messages: |
| 112 | return |
| 113 | |
| 114 | start_idx = max(0, len(self.messages) - max_messages) |
| 115 | |
| 116 | # If the cutoff lands mid-turn, extend backward to the nearest user turn. |
| 117 | while start_idx > 0 and self.messages[start_idx].get("role") != "user": |
| 118 | start_idx -= 1 |
| 119 | |
| 120 | retained = self.messages[start_idx:] |
| 121 | |
| 122 | # Mirror get_history(): avoid persisting orphan tool results at the front. |
| 123 | start = find_legal_message_start(retained) |
| 124 | if start: |
| 125 | retained = retained[start:] |
| 126 | |
| 127 | dropped = len(self.messages) - len(retained) |
| 128 | self.messages = retained |
| 129 | self.last_consolidated = max(0, self.last_consolidated - dropped) |
| 130 | self.updated_at = datetime.now() |
| 131 | |
| 132 | |
| 133 | class SessionManager: |
| 134 | """ |
| 135 | Manages conversation sessions. |
| 136 | |
| 137 | Sessions are stored as JSONL files in the sessions directory. |
| 138 | """ |
| 139 | |
| 140 | @staticmethod |
| 141 | def _preview_from_message(message: dict[str, Any]) -> str: |
| 142 | """Return a sidebar-friendly preview from one persisted message row.""" |
| 143 | if message.get("role") != "user": |
| 144 | return "" |
| 145 | content = message.get("content") |
| 146 | text = content.strip() if isinstance(content, str) else "" |
| 147 | if not text: |
| 148 | media = message.get("media") |
| 149 | if isinstance(media, list): |
| 150 | text = "\n".join( |
| 151 | image_placeholder_text(path) |
| 152 | for path in media |
| 153 | if isinstance(path, str) and path |
| 154 | ).strip() |
| 155 | if not text: |
| 156 | return "" |
| 157 | if any(text.startswith(prefix) for prefix in _INTERNAL_USER_PREFIXES): |
| 158 | return "" |
| 159 | return " ".join(text.split()) |
| 160 | |
| 161 | @classmethod |
| 162 | def _preview_from_session_path(cls, path: Path) -> str: |
| 163 | """Scan a JSONL session file for the first user-visible preview line.""" |
| 164 | try: |
| 165 | with open(path, encoding="utf-8") as f: |
| 166 | for line in f: |
| 167 | line = line.strip() |
| 168 | if not line: |
| 169 | continue |
| 170 | data = json.loads(line) |
| 171 | if data.get("_type") == "metadata": |
| 172 | continue |
| 173 | preview = cls._preview_from_message(data) |
| 174 | if preview: |
| 175 | return preview |
| 176 | except (OSError, json.JSONDecodeError): |
| 177 | return "" |
| 178 | return "" |
| 179 | |
| 180 | def __init__( |
| 181 | self, |
| 182 | workspace: Path, |
| 183 | *, |
| 184 | preview_generator: Any | None = None, |
| 185 | ): |
| 186 | self.workspace = workspace |
| 187 | self.sessions_dir = ensure_dir(self.workspace / "sessions") |
| 188 | self.legacy_sessions_dir = get_legacy_sessions_dir() |
| 189 | self._cache: dict[str, Session] = {} |
| 190 | self.preview_generator = preview_generator |
| 191 | |
| 192 | @classmethod |
| 193 | def first_user_source_text(cls, session: Session) -> str: |
| 194 | for message in session.messages: |
| 195 | text = cls._preview_from_message(message) |
| 196 | if text: |
| 197 | return text |
| 198 | return "" |
| 199 | |
| 200 | async def ensure_preview(self, session: Session) -> str: |
| 201 | """Persist a sidebar preview for *session* when the first user turn changes.""" |
| 202 | source = self.first_user_source_text(session) |
| 203 | if not source: |
| 204 | return "" |
| 205 | metadata = session.metadata if isinstance(session.metadata, dict) else {} |
| 206 | session.metadata = metadata |
| 207 | cached_source = metadata.get(_PREVIEW_SOURCE_METADATA_KEY) |
| 208 | cached_preview = metadata.get(_PREVIEW_METADATA_KEY) |
| 209 | if ( |
| 210 | isinstance(cached_source, str) |
| 211 | and cached_source == source |
| 212 | and isinstance(cached_preview, str) |
| 213 | and cached_preview.strip() |
| 214 | ): |
| 215 | return cached_preview |
| 216 | |
| 217 | preview = "" |
| 218 | generator = self.preview_generator |
| 219 | if generator is not None: |
| 220 | from nanobot.session.preview import SessionPreviewGenerator |
| 221 | |
| 222 | if isinstance(generator, SessionPreviewGenerator): |
| 223 | preview = await generator.summarize(source) |
| 224 | if not preview: |
| 225 | from nanobot.session.preview import clamp_preview |
| 226 | |
| 227 | preview = clamp_preview(source) |
| 228 | |
| 229 | metadata[_PREVIEW_METADATA_KEY] = preview |
| 230 | metadata[_PREVIEW_SOURCE_METADATA_KEY] = source |
| 231 | session.metadata = metadata |
| 232 | self.save(session) |
| 233 | return preview |
| 234 | |
| 235 | @staticmethod |
| 236 | def safe_key(key: str) -> str: |
| 237 | """Public helper used by HTTP handlers to map an arbitrary key to a stable filename stem.""" |
| 238 | return safe_filename(key.replace(":", "_")) |
| 239 | |
| 240 | def _get_session_path(self, key: str) -> Path: |
| 241 | """Get the file path for a session.""" |
| 242 | return self.sessions_dir / f"{self.safe_key(key)}.jsonl" |
| 243 | |
| 244 | def _get_legacy_session_path(self, key: str) -> Path: |
| 245 | """Legacy global session path (~/.nanobot/sessions/).""" |
| 246 | return self.legacy_sessions_dir / f"{self.safe_key(key)}.jsonl" |
| 247 | |
| 248 | def get_or_create(self, key: str) -> Session: |
| 249 | """ |
| 250 | Get an existing session or create a new one. |
| 251 | |
| 252 | Args: |
| 253 | key: Session key (usually channel:chat_id). |
| 254 | |
| 255 | Returns: |
| 256 | The session. |
| 257 | """ |
| 258 | if key in self._cache: |
| 259 | return self._cache[key] |
| 260 | |
| 261 | session = self._load(key) |
| 262 | if session is None: |
| 263 | session = Session(key=key) |
| 264 | |
| 265 | self._cache[key] = session |
| 266 | return session |
| 267 | |
| 268 | def _load(self, key: str) -> Session | None: |
| 269 | """Load a session from disk.""" |
| 270 | path = self._get_session_path(key) |
| 271 | if not path.exists(): |
| 272 | legacy_path = self._get_legacy_session_path(key) |
| 273 | if legacy_path.exists(): |
| 274 | try: |
| 275 | shutil.move(str(legacy_path), str(path)) |
| 276 | logger.info("Migrated session {} from legacy path", key) |
| 277 | except Exception: |
| 278 | logger.exception("Failed to migrate session {}", key) |
| 279 | |
| 280 | if not path.exists(): |
| 281 | return None |
| 282 | |
| 283 | try: |
| 284 | messages = [] |
| 285 | metadata = {} |
| 286 | created_at = None |
| 287 | updated_at = None |
| 288 | last_consolidated = 0 |
| 289 | |
| 290 | with open(path, encoding="utf-8") as f: |
| 291 | for line in f: |
| 292 | line = line.strip() |
| 293 | if not line: |
| 294 | continue |
| 295 | |
| 296 | data = json.loads(line) |
| 297 | |
| 298 | if data.get("_type") == "metadata": |
| 299 | metadata = data.get("metadata", {}) |
| 300 | created_at = datetime.fromisoformat(data["created_at"]) if data.get("created_at") else None |
| 301 | updated_at = datetime.fromisoformat(data["updated_at"]) if data.get("updated_at") else None |
| 302 | last_consolidated = data.get("last_consolidated", 0) |
| 303 | else: |
| 304 | messages.append(data) |
| 305 | |
| 306 | return Session( |
| 307 | key=key, |
| 308 | messages=messages, |
| 309 | created_at=created_at or datetime.now(), |
| 310 | updated_at=updated_at or datetime.now(), |
| 311 | metadata=metadata, |
| 312 | last_consolidated=last_consolidated |
| 313 | ) |
| 314 | except Exception as e: |
| 315 | logger.warning("Failed to load session {}: {}", key, e) |
| 316 | repaired = self._repair(key) |
| 317 | if repaired is not None: |
| 318 | logger.info("Recovered session {} from corrupt file ({} messages)", key, len(repaired.messages)) |
| 319 | return repaired |
| 320 | |
| 321 | def _repair(self, key: str) -> Session | None: |
| 322 | """Attempt to recover a session from a corrupt JSONL file.""" |
| 323 | path = self._get_session_path(key) |
| 324 | if not path.exists(): |
| 325 | return None |
| 326 | |
| 327 | try: |
| 328 | messages: list[dict[str, Any]] = [] |
| 329 | metadata: dict[str, Any] = {} |
| 330 | created_at: datetime | None = None |
| 331 | updated_at: datetime | None = None |
| 332 | last_consolidated = 0 |
| 333 | skipped = 0 |
| 334 | |
| 335 | with open(path, encoding="utf-8") as f: |
| 336 | for line in f: |
| 337 | line = line.strip() |
| 338 | if not line: |
| 339 | continue |
| 340 | try: |
| 341 | data = json.loads(line) |
| 342 | except json.JSONDecodeError: |
| 343 | skipped += 1 |
| 344 | continue |
| 345 | |
| 346 | if data.get("_type") == "metadata": |
| 347 | metadata = data.get("metadata", {}) |
| 348 | if data.get("created_at"): |
| 349 | try: |
| 350 | created_at = datetime.fromisoformat(data["created_at"]) |
| 351 | except (ValueError, TypeError): |
| 352 | pass |
| 353 | if data.get("updated_at"): |
| 354 | try: |
| 355 | updated_at = datetime.fromisoformat(data["updated_at"]) |
| 356 | except (ValueError, TypeError): |
| 357 | pass |
| 358 | last_consolidated = data.get("last_consolidated", 0) |
| 359 | else: |
| 360 | messages.append(data) |
| 361 | |
| 362 | if skipped: |
| 363 | logger.warning("Skipped {} corrupt lines in session {}", skipped, key) |
| 364 | |
| 365 | if not messages and not metadata: |
| 366 | return None |
| 367 | |
| 368 | return Session( |
| 369 | key=key, |
| 370 | messages=messages, |
| 371 | created_at=created_at or datetime.now(), |
| 372 | updated_at=updated_at or datetime.now(), |
| 373 | metadata=metadata, |
| 374 | last_consolidated=last_consolidated |
| 375 | ) |
| 376 | except Exception as e: |
| 377 | logger.warning("Repair failed for session {}: {}", key, e) |
| 378 | return None |
| 379 | |
| 380 | @staticmethod |
| 381 | def _session_payload(session: Session) -> dict[str, Any]: |
| 382 | return { |
| 383 | "key": session.key, |
| 384 | "created_at": session.created_at.isoformat(), |
| 385 | "updated_at": session.updated_at.isoformat(), |
| 386 | "metadata": session.metadata, |
| 387 | "messages": session.messages, |
| 388 | } |
| 389 | |
| 390 | def save(self, session: Session, *, fsync: bool = False) -> None: |
| 391 | """Save a session to disk atomically. |
| 392 | |
| 393 | When *fsync* is ``True`` the final file and its parent directory are |
| 394 | explicitly flushed to durable storage. This is intentionally off by |
| 395 | default (the OS page-cache is sufficient for normal operation) but |
| 396 | should be enabled during graceful shutdown so that filesystems with |
| 397 | write-back caching (e.g. rclone VFS, NFS, FUSE mounts) do not lose |
| 398 | the most recent writes. |
| 399 | """ |
| 400 | path = self._get_session_path(session.key) |
| 401 | tmp_path = path.with_suffix(".jsonl.tmp") |
| 402 | |
| 403 | try: |
| 404 | with open(tmp_path, "w", encoding="utf-8") as f: |
| 405 | metadata_line = { |
| 406 | "_type": "metadata", |
| 407 | "key": session.key, |
| 408 | "created_at": session.created_at.isoformat(), |
| 409 | "updated_at": session.updated_at.isoformat(), |
| 410 | "metadata": session.metadata, |
| 411 | "last_consolidated": session.last_consolidated |
| 412 | } |
| 413 | f.write(json.dumps(metadata_line, ensure_ascii=False) + "\n") |
| 414 | for msg in session.messages: |
| 415 | f.write(json.dumps(msg, ensure_ascii=False) + "\n") |
| 416 | if fsync: |
| 417 | f.flush() |
| 418 | os.fsync(f.fileno()) |
| 419 | |
| 420 | os.replace(tmp_path, path) |
| 421 | |
| 422 | if fsync: |
| 423 | # fsync the directory so the rename is durable. |
| 424 | # On Windows, opening a directory with O_RDONLY raises |
| 425 | # PermissionError — skip the dir sync there (NTFS |
| 426 | # journals metadata synchronously). |
| 427 | try: |
| 428 | fd = os.open(str(path.parent), os.O_RDONLY) |
| 429 | try: |
| 430 | os.fsync(fd) |
| 431 | finally: |
| 432 | os.close(fd) |
| 433 | except PermissionError: |
| 434 | pass # Windows — directory fsync not supported |
| 435 | except BaseException: |
| 436 | tmp_path.unlink(missing_ok=True) |
| 437 | raise |
| 438 | |
| 439 | self._cache[session.key] = session |
| 440 | |
| 441 | def flush_all(self) -> int: |
| 442 | """Re-save every cached session with fsync for durable shutdown. |
| 443 | |
| 444 | Returns the number of sessions flushed. Errors on individual |
| 445 | sessions are logged but do not prevent other sessions from being |
| 446 | flushed. |
| 447 | """ |
| 448 | flushed = 0 |
| 449 | for key, session in list(self._cache.items()): |
| 450 | try: |
| 451 | self.save(session, fsync=True) |
| 452 | flushed += 1 |
| 453 | except Exception: |
| 454 | logger.warning("Failed to flush session {}", key, exc_info=True) |
| 455 | return flushed |
| 456 | |
| 457 | def invalidate(self, key: str) -> None: |
| 458 | """Remove a session from the in-memory cache.""" |
| 459 | self._cache.pop(key, None) |
| 460 | |
| 461 | def delete_session(self, key: str) -> bool: |
| 462 | """Remove a session from disk and the in-memory cache. |
| 463 | |
| 464 | Returns True if a JSONL file was found and unlinked. |
| 465 | """ |
| 466 | path = self._get_session_path(key) |
| 467 | self.invalidate(key) |
| 468 | if not path.exists(): |
| 469 | return False |
| 470 | try: |
| 471 | path.unlink() |
| 472 | return True |
| 473 | except OSError as e: |
| 474 | logger.warning("Failed to delete session file {}: {}", path, e) |
| 475 | return False |
| 476 | |
| 477 | def record_question_answer( |
| 478 | self, |
| 479 | key: str, |
| 480 | question_batch_id: str, |
| 481 | card_id: str, |
| 482 | value: str, |
| 483 | ) -> bool: |
| 484 | """Persist a user's tap on a question card (survives page refresh).""" |
| 485 | if not question_batch_id.strip() or not card_id.strip(): |
| 486 | return False |
| 487 | answer = value.strip() |
| 488 | if not answer: |
| 489 | return False |
| 490 | |
| 491 | session = self.get_or_create(key) |
| 492 | if self._apply_question_answer(session, question_batch_id.strip(), card_id.strip(), answer): |
| 493 | return True |
| 494 | |
| 495 | # Reload from disk in case another task persisted while this handler waited. |
| 496 | self.invalidate(key) |
| 497 | session = self.get_or_create(key) |
| 498 | return self._apply_question_answer(session, question_batch_id.strip(), card_id.strip(), answer) |
| 499 | |
| 500 | def _apply_question_answer( |
| 501 | self, |
| 502 | session: Session, |
| 503 | question_batch_id: str, |
| 504 | card_id: str, |
| 505 | answer: str, |
| 506 | ) -> bool: |
| 507 | from nanobot.session.question_cards import ( |
| 508 | sync_tool_calls_from_question_snapshot, |
| 509 | update_ask_user_card_answer, |
| 510 | ) |
| 511 | |
| 512 | if update_ask_user_card_answer( |
| 513 | session.messages, |
| 514 | question_batch_id, |
| 515 | card_id, |
| 516 | answer, |
| 517 | ): |
| 518 | session.updated_at = datetime.now() |
| 519 | self.save(session) |
| 520 | return True |
| 521 | |
| 522 | # Legacy native-format rows (questions + question_batch_id on assistant). |
| 523 | for message in reversed(session.messages): |
| 524 | if message.get("question_batch_id") != question_batch_id: |
| 525 | continue |
| 526 | questions = message.get("questions") |
| 527 | if not isinstance(questions, list): |
| 528 | continue |
| 529 | for card in questions: |
| 530 | if not isinstance(card, dict) or card.get("id") != card_id: |
| 531 | continue |
| 532 | card["answered"] = answer |
| 533 | card["status"] = "answered" |
| 534 | sync_tool_calls_from_question_snapshot(message) |
| 535 | session.updated_at = datetime.now() |
| 536 | self.save(session) |
| 537 | return True |
| 538 | return False |
| 539 | |
| 540 | def read_session_file(self, key: str) -> dict[str, Any] | None: |
| 541 | """Load a session from disk without caching; intended for read-only HTTP endpoints. |
| 542 | |
| 543 | Returns ``{"key", "created_at", "updated_at", "metadata", "messages"}`` or |
| 544 | ``None`` when the session file does not exist or fails to parse. |
| 545 | """ |
| 546 | path = self._get_session_path(key) |
| 547 | if not path.exists(): |
| 548 | return None |
| 549 | try: |
| 550 | messages: list[dict[str, Any]] = [] |
| 551 | metadata: dict[str, Any] = {} |
| 552 | created_at: str | None = None |
| 553 | updated_at: str | None = None |
| 554 | stored_key: str | None = None |
| 555 | with open(path, encoding="utf-8") as f: |
| 556 | for line in f: |
| 557 | line = line.strip() |
| 558 | if not line: |
| 559 | continue |
| 560 | data = json.loads(line) |
| 561 | if data.get("_type") == "metadata": |
| 562 | metadata = data.get("metadata", {}) |
| 563 | created_at = data.get("created_at") |
| 564 | updated_at = data.get("updated_at") |
| 565 | stored_key = data.get("key") |
| 566 | else: |
| 567 | messages.append(data) |
| 568 | from nanobot.session.question_cards import ( |
| 569 | apply_following_user_replies_to_question_cards, |
| 570 | normalize_persisted_message, |
| 571 | ) |
| 572 | |
| 573 | normalized = [normalize_persisted_message(dict(row)) for row in messages] |
| 574 | apply_following_user_replies_to_question_cards(normalized) |
| 575 | return { |
| 576 | "key": stored_key or key, |
| 577 | "created_at": created_at, |
| 578 | "updated_at": updated_at, |
| 579 | "metadata": metadata, |
| 580 | "messages": normalized, |
| 581 | } |
| 582 | except Exception as e: |
| 583 | logger.warning("Failed to read session {}: {}", key, e) |
| 584 | repaired = self._repair(key) |
| 585 | if repaired is not None: |
| 586 | logger.info("Recovered read-only session view {} from corrupt file", key) |
| 587 | return self._session_payload(repaired) |
| 588 | return None |
| 589 | |
| 590 | def list_sessions(self) -> list[dict[str, Any]]: |
| 591 | """ |
| 592 | List all sessions. |
| 593 | |
| 594 | Returns: |
| 595 | List of session info dicts. |
| 596 | """ |
| 597 | sessions = [] |
| 598 | |
| 599 | for path in self.sessions_dir.glob("*.jsonl"): |
| 600 | fallback_key = path.stem.replace("_", ":", 1) |
| 601 | try: |
| 602 | # Read just the metadata line |
| 603 | with open(path, encoding="utf-8") as f: |
| 604 | first_line = f.readline().strip() |
| 605 | if first_line: |
| 606 | data = json.loads(first_line) |
| 607 | if data.get("_type") == "metadata": |
| 608 | key = data.get("key") or path.stem.replace("_", ":", 1) |
| 609 | row_metadata = data.get("metadata") |
| 610 | stored_preview = "" |
| 611 | if isinstance(row_metadata, dict): |
| 612 | raw_preview = row_metadata.get(_PREVIEW_METADATA_KEY) |
| 613 | if isinstance(raw_preview, str): |
| 614 | stored_preview = raw_preview.strip() |
| 615 | sessions.append({ |
| 616 | "key": key, |
| 617 | "created_at": data.get("created_at"), |
| 618 | "updated_at": data.get("updated_at"), |
| 619 | "preview": stored_preview |
| 620 | or self._preview_from_session_path(path), |
| 621 | "source": get_source(row_metadata) |
| 622 | if isinstance(row_metadata, dict) |
| 623 | else None, |
| 624 | "autoGenerate": get_auto_generate(row_metadata) |
| 625 | if isinstance(row_metadata, dict) |
| 626 | else False, |
| 627 | "path": str(path), |
| 628 | }) |
| 629 | except Exception: |
| 630 | repaired = self._repair(fallback_key) |
| 631 | if repaired is not None: |
| 632 | preview = "" |
| 633 | for message in repaired.messages: |
| 634 | preview = self._preview_from_message(message) |
| 635 | if preview: |
| 636 | break |
| 637 | row_metadata = repaired.metadata if isinstance(repaired.metadata, dict) else {} |
| 638 | stored_preview = row_metadata.get(_PREVIEW_METADATA_KEY) |
| 639 | sessions.append({ |
| 640 | "key": repaired.key, |
| 641 | "created_at": repaired.created_at.isoformat(), |
| 642 | "updated_at": repaired.updated_at.isoformat(), |
| 643 | "preview": stored_preview.strip() |
| 644 | if isinstance(stored_preview, str) and stored_preview.strip() |
| 645 | else preview, |
| 646 | "source": get_source(row_metadata) |
| 647 | if isinstance(row_metadata, dict) |
| 648 | else None, |
| 649 | "autoGenerate": get_auto_generate(row_metadata) |
| 650 | if isinstance(row_metadata, dict) |
| 651 | else False, |
| 652 | "path": str(path), |
| 653 | }) |
| 654 | continue |
| 655 | |
| 656 | return sorted(sessions, key=lambda x: x.get("updated_at", ""), reverse=True) |
| 657 | |
| 658 |