fix: lock gateway single-instance guard on shared state dir, not just host:port (fixes #6208)

Co-Authored-By: Claude Opus 5.5 <[email protected]>

src/praisonai-bot/praisonai_bot/gateway/port_utils.py

@@ -86,19 +86,21 @@ def _process_create_time(pid: int) -> Optional[float]:
 class GatewayPIDLock:
     """Manages PID lock file for single-instance gateway enforcement."""
     
-    def __init__(self, lock_dir: Optional[Path] = None, host: str = "127.0.0.1", port: int = 8765):
+    def __init__(self, lock_dir: Optional[Path] = None, host: str = "127.0.0.1", port: int = 8765,
+                 lock_name: Optional[str] = None):
         """Initialize PID lock manager.
         
         Args:
             lock_dir: Directory for lock file (default: ~/.praisonai/)
             host: Gateway host
             port: Gateway port
+            lock_name: Override the lock file name (used by the state lock)
         """
         self.lock_dir = lock_dir or Path.home() / ".praisonai"
         # Make lock file specific to host and port to allow multiple instances
         safe_host = host.replace(":", "_").replace(".", "_")
-        self.lock_file = self.lock_dir / f"gateway-{safe_host}-{port}.pid"
-        self.lock_dir.mkdir(exist_ok=True)
+        self.lock_file = self.lock_dir / (lock_name or f"gateway-{safe_host}-{port}.pid")
+        self.lock_dir.mkdir(parents=True, exist_ok=True)
     
     def acquire_lock(self, host: str, port: int) -> bool:
         """Acquire the PID lock for the gateway.
@@ -120,6 +122,9 @@ class GatewayPIDLock:
                     lines = lock_content.split('\n')
                     if lines:
                         existing_pid = int(lines[0])
+                        if existing_pid == current_pid:
+                            # Re-acquire by the same process (in-process restart).
+                            return True
                         existing_create_time = None
                         if len(lines) >= 5 and lines[4]:
                             try:
@@ -253,6 +258,43 @@ class GatewayPIDLock:
             pass
 
 
+def resolve_state_dir() -> Path:
+    """Return the gateway's shared mutable state root (``$PRAISONAI_HOME`` or ``~/.praisonai``).
+
+    The durable outbox / inbound journal / DLQ / pairing stores live here, so it
+    (not the TCP port) is what identifies a gateway instance.
+    """
+    base = os.environ.get("PRAISONAI_HOME")
+    return Path(base).expanduser() if base else Path.home() / ".praisonai"
+
+
+def state_lock(state_dir: Optional[Path] = None, host: str = "127.0.0.1",
+               port: int = 8765) -> GatewayPIDLock:
+    """Single-instance lock keyed on the shared state directory, regardless of port.
+
+    Two gateways sharing one state root would poll the same bot tokens and write
+    the same SQLite stores concurrently (split-brain), so only one may own it.
+    """
+    return GatewayPIDLock(lock_dir=state_dir or resolve_state_dir(), host=host,
+                          port=port, lock_name="gateway.state.pid")
+
+
+def format_state_collision_error(state_dir: Path, lock_info: Optional[dict] = None) -> str:
+    """Error shown when another live gateway already owns the shared state."""
+    holder = ""
+    if lock_info:
+        holder = f" (PID {lock_info['pid']} on {lock_info['host']}:{lock_info['port']})"
+    return "\n".join([
+        f"Error: another gateway already owns the state at {state_dir}{holder}.",
+        "",
+        "  Two gateways sharing state poll the same bot tokens and write the same",
+        "  queues, causing 409 Conflict errors and duplicate delivery.",
+        "  Stop it:  praisonai gateway stop",
+        "  Or run an isolated gateway with its own state and bot tokens:",
+        "    PRAISONAI_HOME=~/.praisonai-b praisonai gateway start",
+    ])
+
+
 def format_collision_error(host: str, port: int, lock_info: Optional[dict] = None) -> str:
     """Format a user-friendly error message for port collisions.
     
@@ -275,10 +317,11 @@ def format_collision_error(host: str, port: int, lock_info: Optional[dict] = Non
         pid = lock_info['pid']
         lines.extend([
             f"  Another gateway may be running (PID {pid}).",
-            f"  Stop it:  praisonai gateway stop",
-            f"  Or use a different port:  GATEWAY_PORT={port + 1} praisonai gateway start",
+            "  Stop it:  praisonai gateway stop",
             "",
-            "  Only ONE gateway process should poll each Telegram bot token.",
+            "  Only ONE gateway process may poll each bot token. To run a second,",
+            "  isolated gateway give it its own state, bot tokens and port:",
+            f"    PRAISONAI_HOME=~/.praisonai-b GATEWAY_PORT={port + 1} praisonai gateway start",
         ])
     else:
         lines.extend([

src/praisonai-bot/praisonai_bot/gateway/server.py

@@ -1412,6 +1412,7 @@ class WebSocketGateway:
         
         # PID lock for single-instance enforcement
         self._pid_lock: Optional[Any] = None
+        self._state_lock: Optional[Any] = None
         
         # Channel supervisor for resilient bot management. Share the gateway's
         # degraded-capability registry so the fleet crash-loop breaker (Issue
@@ -1544,8 +1545,20 @@ class WebSocketGateway:
             logger.error("Gateway startup failed - another instance is running")
             raise RuntimeError(error_msg)
         
-        # Store PID lock for cleanup
+        # The port lock alone admits a second gateway on another port that shares
+        # the same state + bot tokens (split-brain). Also lock the state root.
+        from .port_utils import state_lock, resolve_state_dir, format_state_collision_error
+        state_dir = resolve_state_dir()
+        st_lock = state_lock(state_dir, host=self._host, port=self._port)
+        if not st_lock.acquire_lock(self._host, self._port):
+            lock_info = st_lock.get_lock_info()
+            pid_lock.release_lock()
+            logger.error("Gateway startup failed - another instance owns the shared state")
+            raise RuntimeError(format_state_collision_error(state_dir, lock_info))
+
+        # Store PID locks for cleanup
         self._pid_lock = pid_lock
+        self._state_lock = st_lock
         
         # Validate bind-aware auth configuration before starting
         from .auth import assert_external_bind_safe
@@ -2774,6 +2787,9 @@ class WebSocketGateway:
             if hasattr(self, '_pid_lock') and self._pid_lock:
                 self._pid_lock.release_lock()
                 self._pid_lock = None
+            if getattr(self, '_state_lock', None):
+                self._state_lock.release_lock()
+                self._state_lock = None
             # Re-raise the original exception
             raise
         finally:
@@ -3996,6 +4012,9 @@ class WebSocketGateway:
         # Release PID lock
         if hasattr(self, '_pid_lock') and self._pid_lock:
             self._pid_lock.release_lock()
+        if getattr(self, '_state_lock', None):
+            self._state_lock.release_lock()
+            self._state_lock = None
         
         logger.info("Gateway stopped")
     

src/praisonai-bot/tests/unit/gateway/test_state_scoped_lock.py

@@ -0,0 +1,48 @@
+"""A gateway's single-instance identity is its shared state, not its port (#6208)."""
+
+import os
+
+from praisonai_bot.gateway.port_utils import (
+    format_collision_error,
+    format_state_collision_error,
+    resolve_state_dir,
+    state_lock,
+)
+
+
+def _seed_foreign_holder(lock, pid):
+    lock.lock_file.write_text(f"{pid}\n127.0.0.1\n8765\n0\n\n")
+
+
+def test_second_gateway_on_another_port_is_refused(tmp_path, monkeypatch):
+    first = state_lock(tmp_path, port=8765)
+    _seed_foreign_holder(first, os.getppid())  # a live process other than us
+    second = state_lock(tmp_path, port=8766)
+    assert second.lock_file == first.lock_file
+    assert second.acquire_lock("127.0.0.1", 8766) is False
+    msg = format_state_collision_error(tmp_path, second.get_lock_info())
+    assert str(tmp_path) in msg and "PRAISONAI_HOME" in msg
+
+
+def test_distinct_state_dirs_are_independent(tmp_path):
+    a = state_lock(tmp_path / "a")
+    _seed_foreign_holder(a, os.getppid())
+    assert state_lock(tmp_path / "b").acquire_lock("127.0.0.1", 8766) is True
+
+
+def test_stale_state_lock_is_reclaimed(tmp_path, monkeypatch):
+    lock = state_lock(tmp_path)
+    _seed_foreign_holder(lock, 999999)
+    monkeypatch.setattr(lock, "_is_process_running", lambda pid: False)
+    assert lock.acquire_lock("127.0.0.1", 8765) is True
+
+
+def test_resolve_state_dir_honours_praisonai_home(tmp_path, monkeypatch):
+    monkeypatch.setenv("PRAISONAI_HOME", str(tmp_path))
+    assert resolve_state_dir() == tmp_path
+
+
+def test_collision_error_no_longer_suggests_bare_port_change():
+    msg = format_collision_error("127.0.0.1", 8765, {"pid": 1, "is_running": True})
+    assert "Or use a different port" not in msg
+    assert "PRAISONAI_HOME" in msg