fix(bridge): resume persisted session_id after restart when SDK transcript exists

bridge/CLAUDE.md

@@ -122,6 +122,7 @@ utils/chat_logger.py Per-session debug chat logging
 | `CCC_TELEGRAM_SPACING_LINES` | No | Blank lines per vertical gap (paragraph/section/list-item) in the readable renderer, clamped to [1, 3] (default: 2). Set 1 for the compact single-blank layout |
 | `CCC_TELEGRAM_MAX_BUBBLE_CHARS` | No | Max characters per Telegram message; long replies split into multiple bubbles at this size (default: 1200, clamped to [200, 4000]) |
 | `CCC_TELEGRAM_OPTION_BUTTONS` | No | Render multiple-choice questions as tappable inline buttons (default: **false**). Off = options shown as text, user types their choice |
+| `CCC_RESUME_PERSISTED_SESSIONS` | No | Auto-resume a persisted session_id after a bridge restart when its SDK transcript still exists on disk (default: **true**). Set false to restore the old never-resume-across-restart behavior |
 | `OPENAI_API_KEY` | Voice only | API key for Whisper transcription |
 | `OPENAI_BASE_URL` | No | Optional OpenAI-compatible Whisper API base URL |
 | `WHISPER_MODEL` | No | Whisper model name (default: `whisper-1`) |

bridge/core/bot.py

@@ -24,6 +24,7 @@
 from telegram.request import HTTPXRequest  # noqa: F401 - compatibility for tests/patches
 from telegram_bot.utils.config import config
 from telegram_bot.session.manager import session_manager
+from telegram_bot.core import session_resume
 from telegram_bot.core.push_notifier import PushNotifier
 from telegram_bot.core.task_queue import UserTaskQueue
 from telegram_bot.core.project_chat import (
@@ -152,16 +153,37 @@ async def _save_session_id(self, session_key: Any, response: ChatResponse):
             self._runtime_active_sessions.add(session_key)
 
     def _effective_session_id(self, session_key: Any, session: dict) -> Optional[str]:
-        """Prevent cross-process auto-resume from persisted session data."""
+        """Return a session_id that is safe to auto-resume.
+
+        Sessions touched in the current runtime always resume. After a bridge
+        restart the runtime set is empty; to avoid conversation memory loss,
+        a persisted session_id is still resumed when its SDK transcript exists
+        on disk (opt-out: CCC_RESUME_PERSISTED_SESSIONS=false). A persisted id
+        without a transcript is still ignored (stale/foreign session data).
+        """
         session_id = session.get("session_id")
         if not session_id:
             return None
-        if session_key not in self._runtime_active_sessions:
+        if session_key in self._runtime_active_sessions:
+            return session_id
+        if session_resume.resume_persisted_enabled() and session_resume.persisted_transcript_exists(
+            self._sdk_conversations_dir(), session_id
+        ):
             logger.info(
-                f"Ignoring persisted session_id for conversation {session_key} (not active in current runtime)"
+                f"Resuming persisted session_id for conversation {session_key} after restart"
             )
-            return None
-        return session_id
+            self._runtime_active_sessions.add(session_key)
+            return session_id
+        logger.info(
+            f"Ignoring persisted session_id for conversation {session_key} (not active in current runtime)"
+        )
+        return None
+
+    @staticmethod
+    def _sdk_conversations_dir():
+        """Resolve CONVERSATIONS_DIR from the live project_chat module (test-stub safe)."""
+        module = sys.modules.get("telegram_bot.core.project_chat")
+        return getattr(module, "CONVERSATIONS_DIR", None)
 
     @staticmethod
     def _message_timestamp_utc(message: Message) -> datetime:

bridge/core/session_resume.py

@@ -0,0 +1,44 @@
+"""Persisted-session resume policy after a bridge restart.
+
+After a bridge restart the in-memory ``_runtime_active_sessions`` set is
+empty, so historically every persisted session_id was rejected and each
+conversation started blank ("memory loss", diagnosed as PATH1). This module
+holds the pure decision helpers: a persisted session_id may be auto-resumed
+when its SDK conversation transcript still exists on disk. An environment
+off-switch (``CCC_RESUME_PERSISTED_SESSIONS=false``) restores the old
+never-resume behavior.
+"""
+
+import os
+from pathlib import Path
+from typing import Mapping, Optional
+
+from telegram_bot.core.conversation_paths import resolve_conversation_file
+
+_FALSE_VALUES = {"false", "0", "no", "off"}
+
+ENV_FLAG = "CCC_RESUME_PERSISTED_SESSIONS"
+
+
+def resume_persisted_enabled(environ: Optional[Mapping[str, str]] = None) -> bool:
+    """True unless CCC_RESUME_PERSISTED_SESSIONS is set to a false-y value."""
+    env = os.environ if environ is None else environ
+    raw = str(env.get(ENV_FLAG, "true")).strip().lower()
+    return raw not in _FALSE_VALUES
+
+
+def persisted_transcript_exists(
+    conversations_dir: Optional[Path], session_id: Optional[str]
+) -> bool:
+    """True when the SDK JSONL transcript for session_id exists under conversations_dir.
+
+    Uses resolve_conversation_file so a malicious/corrupt session_id cannot
+    escape the conversations directory. Any filesystem error means "no".
+    """
+    if not conversations_dir or not session_id:
+        return False
+    try:
+        filepath = resolve_conversation_file(conversations_dir, str(session_id))
+        return bool(filepath and filepath.is_file())
+    except OSError:
+        return False

bridge/tests/test_session_resume.py

@@ -0,0 +1,67 @@
+# ruff: noqa: E402
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
+
+from telegram_bot.core.session_resume import (
+    persisted_transcript_exists,
+    resume_persisted_enabled,
+)
+
+
+class ResumePersistedEnabledTests(unittest.TestCase):
+    def test_default_is_enabled(self):
+        self.assertTrue(resume_persisted_enabled({}))
+
+    def test_explicit_true_values(self):
+        for value in ("true", "1", "yes", "on", "TRUE", " anything "):
+            with self.subTest(value=value):
+                self.assertTrue(
+                    resume_persisted_enabled({"CCC_RESUME_PERSISTED_SESSIONS": value})
+                )
+
+    def test_false_values_disable(self):
+        for value in ("false", "0", "no", "off", "FALSE", " Off "):
+            with self.subTest(value=value):
+                self.assertFalse(
+                    resume_persisted_enabled({"CCC_RESUME_PERSISTED_SESSIONS": value})
+                )
+
+
+class PersistedTranscriptExistsTests(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.conv_dir = Path(self._tmp.name)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_existing_transcript_is_resumable(self):
+        (self.conv_dir / "abc-123.jsonl").write_text("{}\n")
+        self.assertTrue(persisted_transcript_exists(self.conv_dir, "abc-123"))
+
+    def test_missing_transcript_is_not_resumable(self):
+        self.assertFalse(persisted_transcript_exists(self.conv_dir, "abc-123"))
+
+    def test_none_or_empty_inputs(self):
+        self.assertFalse(persisted_transcript_exists(None, "abc-123"))
+        self.assertFalse(persisted_transcript_exists(self.conv_dir, None))
+        self.assertFalse(persisted_transcript_exists(self.conv_dir, ""))
+
+    def test_path_traversal_session_id_rejected(self):
+        outside = Path(self._tmp.name) / "outside.jsonl"
+        outside.write_text("{}\n")
+        nested = self.conv_dir / "nested"
+        nested.mkdir()
+        self.assertFalse(persisted_transcript_exists(nested, "../outside"))
+
+    def test_directory_named_like_transcript_is_not_a_file(self):
+        (self.conv_dir / "dir-id.jsonl").mkdir()
+        self.assertFalse(persisted_transcript_exists(self.conv_dir, "dir-id"))
+
+
+if __name__ == "__main__":
+    unittest.main()