From 409c7c1aca2e591b6d46207b93f36009ef03d83e Mon Sep 17 00:00:00 2001 From: lee Date: Thu, 16 Jul 2026 14:23:35 +0700 Subject: [PATCH] fix: recover persisted live sessions and sync deleted topics --- src/ccgram/handlers/sync_command.py | 23 ++++++++---- src/ccgram/session_monitor.py | 53 ++++++++++++++++++++++++---- tests/ccgram/test_session_monitor.py | 44 ++++++++++++++++++++++- tests/ccgram/test_sync_command.py | 33 +++++++++++++++++ 4 files changed, 139 insertions(+), 14 deletions(-) create mode 100644 tests/ccgram/test_sync_command.py diff --git a/src/ccgram/handlers/sync_command.py b/src/ccgram/handlers/sync_command.py index 001e02a0..7693a307 100644 --- a/src/ccgram/handlers/sync_command.py +++ b/src/ccgram/handlers/sync_command.py @@ -45,6 +45,17 @@ logger = structlog.get_logger() + +def _resolve_bound_group_chat_id(user_id: int, thread_id: int) -> int: + """Resolve a forum group chat ID for a bound thread, with config fallback.""" + chat_id = thread_router.resolve_chat_id(user_id, thread_id) + if chat_id != user_id: + return chat_id + if config.group_id is not None: + return config.group_id + return user_id + + _GHOST_RE = re.compile(r"user:(\d+)\s+thread:(\d+)\s+window:([^\s(]+)") _WINDOW_RE = re.compile(r"([^\s(]+)") @@ -97,7 +108,7 @@ async def _sync_live_topic_names( for user_id, thread_id, window_id in thread_router.iter_thread_bindings(): if window_id not in live_ids: continue - chat_id = thread_router.resolve_chat_id(user_id, thread_id) + chat_id = _resolve_bound_group_chat_id(user_id, thread_id) if chat_id == user_id: continue await sync_topic_name( @@ -228,7 +239,7 @@ async def _close_ghost_topics( current_window_id = thread_router.get_window_for_thread(user_id, thread_id) if current_window_id != window_id: continue - chat_id = thread_router.resolve_chat_id(user_id, thread_id) + chat_id = _resolve_bound_group_chat_id(user_id, thread_id) topic_removed = False if chat_id == user_id: logger.warning( @@ -253,7 +264,7 @@ async def _close_ghost_topics( thread_router.unbind_thread(user_id, thread_id) if topic_removed: closed_count += 1 - except OSError, TelegramError: + except (OSError, TelegramError): logger.exception( "Failed to clean up ghost binding thread=%d window=%s", thread_id, @@ -306,7 +317,7 @@ async def _probe_dead_topics(client: TelegramClient) -> list[AuditIssue]: only ``send_message`` reliably throws "thread not found" for deleted topics. """ bindings = [ - (uid, tid, wid, thread_router.resolve_chat_id(uid, tid)) + (uid, tid, wid, _resolve_bound_group_chat_id(uid, tid)) for uid, tid, wid in thread_router.iter_thread_bindings() ] # Only probe bindings with a group chat (chat_id != user_id) @@ -396,7 +407,7 @@ async def _recreate_dead_topics( # Preserve group_chat_id before unbinding — unbind_thread deletes it, # but _handle_new_window needs it to know which chat to create the topic in. - chat_id = thread_router.resolve_chat_id(user_id, thread_id) + chat_id = _resolve_bound_group_chat_id(user_id, thread_id) # Unbind THEN recreate — must unbind first so _handle_new_window # doesn't skip the window as "already bound". On failure, restore. @@ -412,7 +423,7 @@ async def _recreate_dead_topics( try: await _handle_new_window(event, client) recreated += 1 - except TelegramError, OSError: + except (TelegramError, OSError): logger.exception("Failed to recreate topic for window %s", window_id) # Restore binding so the window isn't orphaned thread_router.bind_thread(user_id, thread_id, window_id, window_name=name) diff --git a/src/ccgram/session_monitor.py b/src/ccgram/session_monitor.py index f0b65577..7d4bd6df 100644 --- a/src/ccgram/session_monitor.py +++ b/src/ccgram/session_monitor.py @@ -60,6 +60,24 @@ logger = structlog.get_logger() +def _persisted_session_map_info(window_id: str) -> dict[str, str] | None: + """Return session-map-like info from persisted window state, if usable.""" + from .window_state_ports.identity_state import get_identity + + identity = get_identity(window_id) + if identity is None or not identity.session_id or identity.transcript_path is None: + return None + if not identity.transcript_path.exists(): + return None + return { + "session_id": identity.session_id, + "cwd": identity.cwd, + "window_name": identity.window_name, + "transcript_path": str(identity.transcript_path), + "provider_name": identity.provider_name, + } + + class SessionMonitor: """Monitors Claude Code sessions for new assistant messages. @@ -245,7 +263,9 @@ async def _read_hook_events(self) -> None: logger.exception("Hook event callback error for %s", event.event_type) async def _load_current_session_map( - self, raw: dict | None = None + self, + raw: dict | None = None, + live_window_ids: set[str] | None = None, ) -> dict[str, dict[str, str]]: """Load current session_map and return window_key -> details mapping. @@ -254,10 +274,24 @@ async def _load_current_session_map( """ if raw is None: raw = await read_session_map_raw() + if live_window_ids is None: + windows = await list_windows_for_reconciliation(tmux_manager) + live_window_ids = {w.window_id for w in windows} if windows else set() if not raw: - return {} + return { + window_id: details + for window_id in live_window_ids + if (details := _persisted_session_map_info(window_id)) is not None + } prefix = session_map_prefix() - return parse_session_map(raw, prefix) + current_map = parse_session_map(raw, prefix) + for window_id in live_window_ids: + if window_id in current_map: + continue + details = _persisted_session_map_info(window_id) + if details is not None: + current_map[window_id] = details + return current_map async def _cleanup_all_stale_sessions(self) -> None: """Clean up all tracked sessions not in current session_map (startup).""" @@ -277,10 +311,12 @@ async def _cleanup_all_stale_sessions(self) -> None: self.state.save_if_dirty() async def _detect_and_cleanup_changes( - self, raw: dict | None = None + self, + raw: dict | None = None, + live_window_ids: set[str] | None = None, ) -> dict[str, dict[str, str]]: """Reconcile session_map; clean up replaced/removed sessions; fire new-window events.""" - current_map = await self._load_current_session_map(raw) + current_map = await self._load_current_session_map(raw, live_window_ids) result = session_lifecycle.reconcile(current_map, self._idle_tracker) for session_id in result.sessions_to_remove: @@ -427,9 +463,12 @@ async def _monitor_loop(self) -> None: raw_session_map = await read_session_map_raw() await session_map_sync.load_session_map(raw_session_map) - current_map = await self._detect_and_cleanup_changes(raw_session_map) - all_windows = await list_windows_for_reconciliation(tmux_manager) + live_window_ids = {w.window_id for w in all_windows} if all_windows else None + current_map = await self._detect_and_cleanup_changes( + raw_session_map, live_window_ids + ) + if all_windows is None: logger.warning( "Multiplexer listing unavailable; skipping window reconciliation" diff --git a/tests/ccgram/test_session_monitor.py b/tests/ccgram/test_session_monitor.py index eba6ccd0..7213e323 100644 --- a/tests/ccgram/test_session_monitor.py +++ b/tests/ccgram/test_session_monitor.py @@ -14,7 +14,7 @@ from ccgram.session import SessionManager from ccgram.session_monitor import NewWindowEvent, SessionMonitor from ccgram.thread_router import thread_router -from ccgram.window_state_store import window_store +from ccgram.window_state_store import WindowState, window_store @pytest.fixture @@ -398,6 +398,48 @@ async def test_callback_error_does_not_crash( assert cb.call_count == 2 +class TestPersistedSessionMapRecovery: + async def test_recovers_live_window_from_persisted_identity( + self, monitor: SessionMonitor, monkeypatch, tmp_path + ) -> None: + transcript = tmp_path / "session.jsonl" + transcript.write_text('{"type":"assistant"}\n') + window_store.window_states["@19"] = WindowState( + provider_name="claude", + session_id="sess-19", + cwd="/repo", + transcript_path=str(transcript), + window_name="agent", + ) + + result = await monitor._load_current_session_map({}, {"@19"}) + + assert result == { + "@19": { + "session_id": "sess-19", + "cwd": "/repo", + "window_name": "agent", + "transcript_path": str(transcript), + "provider_name": "claude", + } + } + + async def test_ignores_missing_transcript_in_persisted_identity( + self, monitor: SessionMonitor + ) -> None: + window_store.window_states["@19"] = WindowState( + provider_name="claude", + session_id="sess-19", + cwd="/repo", + transcript_path="/tmp/does-not-exist.jsonl", + window_name="agent", + ) + + result = await monitor._load_current_session_map({}, {"@19"}) + + assert result == {} + + class TestLoadCurrentSessionMapBackend: """The monitor's session_map reader must honor the active backend prefix. diff --git a/tests/ccgram/test_sync_command.py b/tests/ccgram/test_sync_command.py new file mode 100644 index 00000000..40841043 --- /dev/null +++ b/tests/ccgram/test_sync_command.py @@ -0,0 +1,33 @@ +"""Tests for /sync dead-topic recovery helpers.""" + +from ccgram.config import config +from ccgram.handlers.sync_command import _resolve_bound_group_chat_id + + +class TestResolveBoundGroupChatId: + def test_prefers_bound_group_chat_id(self, monkeypatch) -> None: + monkeypatch.setattr( + "ccgram.handlers.sync_command.thread_router.resolve_chat_id", + lambda user_id, thread_id: -100123, + ) + monkeypatch.setattr(config, "group_id", -100999) + + assert _resolve_bound_group_chat_id(1, 10) == -100123 + + def test_falls_back_to_config_group_id(self, monkeypatch) -> None: + monkeypatch.setattr( + "ccgram.handlers.sync_command.thread_router.resolve_chat_id", + lambda user_id, thread_id: user_id, + ) + monkeypatch.setattr(config, "group_id", -100999) + + assert _resolve_bound_group_chat_id(1, 10) == -100999 + + def test_returns_user_id_without_any_group_context(self, monkeypatch) -> None: + monkeypatch.setattr( + "ccgram.handlers.sync_command.thread_router.resolve_chat_id", + lambda user_id, thread_id: user_id, + ) + monkeypatch.setattr(config, "group_id", None) + + assert _resolve_bound_group_chat_id(1, 10) == 1