feat(state): add same-pane session deduplication

Handle edge case where Claude's --resume creates an orphan session file
before resuming the original session, leaving two session files pointing
to the same Zellij pane.

The deduplication algorithm (_dedupe_same_pane_sessions) resolves
conflicts by preferring:
1. Sessions with context_usage (indicates actual conversation occurred)
2. Higher conversation_mtime_ns (more recent file activity)

When an orphan is identified, its session file is deleted from disk to
prevent re-discovery on subsequent state collection cycles.

Test coverage includes:
- Keeping session with context_usage over one without
- Keeping higher mtime when both have context_usage
- Keeping higher mtime when neither has context_usage
- Preserving sessions on different panes (no false positives)
- Single session per pane unchanged
- Sessions without pane info unchanged
- Handling non-numeric mtime values defensively

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

amc_server/mixins/state.py

@@ -2,18 +2,18 @@ import hashlib
 import json
 import subprocess
 import time
+from collections import defaultdict
 from datetime import datetime, timezone
 from pathlib import Path
 
-from amc_server.context import (
+from amc_server.config import (
     EVENTS_DIR,
     SESSIONS_DIR,
     STALE_EVENT_AGE,
     STALE_STARTING_AGE,
-    ZELLIJ_BIN,
     _state_lock,
-    _zellij_cache,
 )
+from amc_server.zellij import ZELLIJ_BIN, _zellij_cache
 from amc_server.logging_utils import LOGGER
 
 
@@ -158,6 +158,9 @@ class StateMixin:
             # Sort by session_id for stable, deterministic ordering (no visual jumping)
             sessions.sort(key=lambda s: s.get("session_id", ""))
 
+            # Dedupe same-pane sessions (handles --resume creating orphan + real session)
+            sessions = self._dedupe_same_pane_sessions(sessions)
+
             # Clean orphan event logs (sessions persist until manually dismissed or SessionEnd)
             self._cleanup_stale(sessions)
 
@@ -236,7 +239,7 @@ class StateMixin:
         Returns:
             set: Absolute paths of transcript files with active processes.
         """
-        from amc_server.context import CODEX_SESSIONS_DIR
+        from amc_server.agents import CODEX_SESSIONS_DIR
 
         if not CODEX_SESSIONS_DIR.exists():
             return set()
@@ -381,3 +384,56 @@ class StateMixin:
                         f.unlink()
             except (json.JSONDecodeError, OSError):
                 pass
+
+    def _dedupe_same_pane_sessions(self, sessions):
+        """Remove orphan sessions when multiple sessions share the same Zellij pane.
+
+        This handles the --resume edge case where Claude creates a new session file
+        before resuming the old one, leaving an orphan with no context_usage.
+
+        When multiple sessions share (zellij_session, zellij_pane), keep the one with:
+        1. context_usage (has actual conversation data)
+        2. Higher conversation_mtime_ns (more recent activity)
+        """
+
+        def session_score(s):
+            """Score a session for dedup ranking: (has_context, mtime)."""
+            has_context = 1 if s.get("context_usage") else 0
+            mtime = s.get("conversation_mtime_ns") or 0
+            # Defensive: ensure mtime is numeric
+            if not isinstance(mtime, (int, float)):
+                mtime = 0
+            return (has_context, mtime)
+
+        # Group sessions by pane
+        pane_groups = defaultdict(list)
+        for s in sessions:
+            zs = s.get("zellij_session", "")
+            zp = s.get("zellij_pane", "")
+            if zs and zp:
+                pane_groups[(zs, zp)].append(s)
+
+        # Find orphans to remove
+        orphan_ids = set()
+        for group in pane_groups.values():
+            if len(group) <= 1:
+                continue
+
+            # Pick the best session: prefer context_usage, then highest mtime
+            group_sorted = sorted(group, key=session_score, reverse=True)
+
+            # Mark all but the best as orphans
+            for s in group_sorted[1:]:
+                session_id = s.get("session_id")
+                if not session_id:
+                    continue  # Skip sessions without valid IDs
+                orphan_ids.add(session_id)
+                # Also delete the orphan session file
+                try:
+                    orphan_file = SESSIONS_DIR / f"{session_id}.json"
+                    orphan_file.unlink(missing_ok=True)
+                except OSError:
+                    pass
+
+        # Return filtered list
+        return [s for s in sessions if s.get("session_id") not in orphan_ids]

tests/test_state.py

@@ -416,5 +416,233 @@ class TestCollectSessions(unittest.TestCase):
             self.assertEqual(ids, ["alpha", "middle", "zebra"])
 
 
+class TestDedupeSamePaneSessions(unittest.TestCase):
+    """Tests for _dedupe_same_pane_sessions (--resume orphan handling)."""
+
+    def setUp(self):
+        self.handler = DummyStateHandler()
+
+    def test_keeps_session_with_context_usage(self):
+        """When two sessions share a pane, keep the one with context_usage."""
+        sessions = [
+            {
+                "session_id": "orphan",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": None,
+                "conversation_mtime_ns": 1000,
+            },
+            {
+                "session_id": "real",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": {"current_tokens": 5000},
+                "conversation_mtime_ns": 900,  # older but has context
+            },
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sessions_dir = Path(tmpdir)
+            # Create dummy session files
+            for s in sessions:
+                (sessions_dir / f"{s['session_id']}.json").write_text("{}")
+
+            with patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
+                result = self.handler._dedupe_same_pane_sessions(sessions)
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["session_id"], "real")
+
+    def test_keeps_higher_mtime_when_both_have_context(self):
+        """When both have context_usage, keep the one with higher mtime."""
+        sessions = [
+            {
+                "session_id": "older",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": {"current_tokens": 5000},
+                "conversation_mtime_ns": 1000,
+            },
+            {
+                "session_id": "newer",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": {"current_tokens": 6000},
+                "conversation_mtime_ns": 2000,
+            },
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sessions_dir = Path(tmpdir)
+            for s in sessions:
+                (sessions_dir / f"{s['session_id']}.json").write_text("{}")
+
+            with patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
+                result = self.handler._dedupe_same_pane_sessions(sessions)
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["session_id"], "newer")
+
+    def test_no_dedup_when_different_panes(self):
+        """Sessions in different panes should not be deduped."""
+        sessions = [
+            {
+                "session_id": "session1",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+            },
+            {
+                "session_id": "session2",
+                "zellij_session": "infra",
+                "zellij_pane": "43",  # different pane
+            },
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            with patch.object(state_mod, "SESSIONS_DIR", Path(tmpdir)):
+                result = self.handler._dedupe_same_pane_sessions(sessions)
+
+        self.assertEqual(len(result), 2)
+
+    def test_no_dedup_when_empty_pane_info(self):
+        """Sessions without pane info should not be deduped."""
+        sessions = [
+            {"session_id": "session1", "zellij_session": "", "zellij_pane": ""},
+            {"session_id": "session2", "zellij_session": "", "zellij_pane": ""},
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            with patch.object(state_mod, "SESSIONS_DIR", Path(tmpdir)):
+                result = self.handler._dedupe_same_pane_sessions(sessions)
+
+        self.assertEqual(len(result), 2)
+
+    def test_deletes_orphan_session_file(self):
+        """Orphan session file should be deleted from disk."""
+        sessions = [
+            {
+                "session_id": "orphan",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": None,
+            },
+            {
+                "session_id": "real",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": {"current_tokens": 5000},
+            },
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sessions_dir = Path(tmpdir)
+            orphan_file = sessions_dir / "orphan.json"
+            real_file = sessions_dir / "real.json"
+            orphan_file.write_text("{}")
+            real_file.write_text("{}")
+
+            with patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
+                self.handler._dedupe_same_pane_sessions(sessions)
+
+            self.assertFalse(orphan_file.exists())
+            self.assertTrue(real_file.exists())
+
+    def test_handles_session_without_id(self):
+        """Sessions without session_id should be skipped, not cause errors."""
+        sessions = [
+            {
+                "session_id": None,  # Missing ID
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": None,
+            },
+            {
+                "session_id": "real",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": {"current_tokens": 5000},
+            },
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sessions_dir = Path(tmpdir)
+            (sessions_dir / "real.json").write_text("{}")
+
+            with patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
+                result = self.handler._dedupe_same_pane_sessions(sessions)
+
+        # Should keep real, skip the None-id session (not filter it out)
+        self.assertEqual(len(result), 2)  # Both still in list
+        self.assertIn("real", [s.get("session_id") for s in result])
+
+    def test_keeps_only_best_with_three_sessions(self):
+        """When 3+ sessions share a pane, keep only the best one."""
+        sessions = [
+            {
+                "session_id": "worst",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": None,
+                "conversation_mtime_ns": 1000,
+            },
+            {
+                "session_id": "middle",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": None,
+                "conversation_mtime_ns": 2000,
+            },
+            {
+                "session_id": "best",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": {"current_tokens": 100},
+                "conversation_mtime_ns": 500,  # Oldest but has context
+            },
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sessions_dir = Path(tmpdir)
+            for s in sessions:
+                (sessions_dir / f"{s['session_id']}.json").write_text("{}")
+
+            with patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
+                result = self.handler._dedupe_same_pane_sessions(sessions)
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["session_id"], "best")
+
+    def test_handles_non_numeric_mtime(self):
+        """Non-numeric mtime values should be treated as 0."""
+        sessions = [
+            {
+                "session_id": "bad_mtime",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": None,
+                "conversation_mtime_ns": "not a number",  # Invalid
+            },
+            {
+                "session_id": "good_mtime",
+                "zellij_session": "infra",
+                "zellij_pane": "42",
+                "context_usage": None,
+                "conversation_mtime_ns": 1000,
+            },
+        ]
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            sessions_dir = Path(tmpdir)
+            for s in sessions:
+                (sessions_dir / f"{s['session_id']}.json").write_text("{}")
+
+            with patch.object(state_mod, "SESSIONS_DIR", sessions_dir):
+                # Should not raise, should keep session with valid mtime
+                result = self.handler._dedupe_same_pane_sessions(sessions)
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0]["session_id"], "good_mtime")
+
+
 if __name__ == "__main__":
     unittest.main()