test(ssh): Add tests for remote hermes path validation

Add comprehensive tests for:
1. validate_remote_hermes_path() - found, not found, error cases
2. _get_default_hermes_path() - path discovery logic
3. execute_hermes_command() - success, validation failure, timeout cases

Resolves #350

hermes_state.py

@@ -32,7 +32,7 @@ T = TypeVar("T")
 
 DEFAULT_DB_PATH = get_hermes_home() / "state.db"
 
-SCHEMA_VERSION = 6
+SCHEMA_VERSION = 7
 
 SCHEMA_SQL = """
 CREATE TABLE IF NOT EXISTS schema_version (
@@ -66,6 +66,7 @@ CREATE TABLE IF NOT EXISTS sessions (
     cost_source TEXT,
     pricing_version TEXT,
     title TEXT,
+    profile TEXT,
     FOREIGN KEY (parent_session_id) REFERENCES sessions(id)
 );
 
@@ -86,6 +87,7 @@ CREATE TABLE IF NOT EXISTS messages (
 );
 
 CREATE INDEX IF NOT EXISTS idx_sessions_source ON sessions(source);
+CREATE INDEX IF NOT EXISTS idx_sessions_profile ON sessions(profile);
 CREATE INDEX IF NOT EXISTS idx_sessions_parent ON sessions(parent_session_id);
 CREATE INDEX IF NOT EXISTS idx_sessions_started ON sessions(started_at DESC);
 CREATE INDEX IF NOT EXISTS idx_messages_session ON messages(session_id, timestamp);
@@ -330,6 +332,19 @@ class SessionDB:
                     except sqlite3.OperationalError:
                         pass  # Column already exists
                 cursor.execute("UPDATE schema_version SET version = 6")
+            if current_version < 7:
+                # v7: add profile column to sessions for profile isolation (#323)
+                try:
+                    cursor.execute('ALTER TABLE sessions ADD COLUMN "profile" TEXT')
+                except sqlite3.OperationalError:
+                    pass  # Column already exists
+                try:
+                    cursor.execute(
+                        "CREATE INDEX IF NOT EXISTS idx_sessions_profile ON sessions(profile)"
+                    )
+                except sqlite3.OperationalError:
+                    pass
+                cursor.execute("UPDATE schema_version SET version = 7")
 
         # Unique title index — always ensure it exists (safe to run after migrations
         # since the title column is guaranteed to exist at this point)
@@ -362,13 +377,19 @@ class SessionDB:
         system_prompt: str = None,
         user_id: str = None,
         parent_session_id: str = None,
+        profile: str = None,
     ) -> str:
-        """Create a new session record. Returns the session_id."""
+        """Create a new session record. Returns the session_id.
+
+        Args:
+            profile: Profile name for session isolation. When set, sessions
+                are tagged so queries can filter by profile. (#323)
+        """
         def _do(conn):
             conn.execute(
                 """INSERT OR IGNORE INTO sessions (id, source, user_id, model, model_config,
-                   system_prompt, parent_session_id, started_at)
-                   VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
+                   system_prompt, parent_session_id, profile, started_at)
+                   VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
                 (
                     session_id,
                     source,
@@ -377,6 +398,7 @@ class SessionDB:
                     json.dumps(model_config) if model_config else None,
                     system_prompt,
                     parent_session_id,
+                    profile,
                     time.time(),
                 ),
             )
@@ -505,19 +527,23 @@ class SessionDB:
         session_id: str,
         source: str = "unknown",
         model: str = None,
+        profile: str = None,
     ) -> None:
         """Ensure a session row exists, creating it with minimal metadata if absent.
 
         Used by _flush_messages_to_session_db to recover from a failed
         create_session() call (e.g. transient SQLite lock at agent startup).
         INSERT OR IGNORE is safe to call even when the row already exists.
+
+        Args:
+            profile: Profile name for session isolation. (#323)
         """
         def _do(conn):
             conn.execute(
                 """INSERT OR IGNORE INTO sessions
-                   (id, source, model, started_at)
-                   VALUES (?, ?, ?, ?)""",
-                (session_id, source, model, time.time()),
+                   (id, source, model, profile, started_at)
+                   VALUES (?, ?, ?, ?, ?)""",
+                (session_id, source, model, profile, time.time()),
             )
         self._execute_write(_do)
 
@@ -788,6 +814,7 @@ class SessionDB:
         limit: int = 20,
         offset: int = 0,
         include_children: bool = False,
+        profile: str = None,
     ) -> List[Dict[str, Any]]:
         """List sessions with preview (first user message) and last active timestamp.
 
@@ -799,6 +826,10 @@ class SessionDB:
 
         By default, child sessions (subagent runs, compression continuations)
         are excluded.  Pass ``include_children=True`` to include them.
+
+        Args:
+            profile: Filter sessions to this profile name. Pass None to see all.
+                (#323)
         """
         where_clauses = []
         params = []
@@ -813,6 +844,9 @@ class SessionDB:
             placeholders = ",".join("?" for _ in exclude_sources)
             where_clauses.append(f"s.source NOT IN ({placeholders})")
             params.extend(exclude_sources)
+        if profile:
+            where_clauses.append("s.profile = ?")
+            params.append(profile)
 
         where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
         query = f"""
@@ -1158,34 +1192,52 @@ class SessionDB:
         source: str = None,
         limit: int = 20,
         offset: int = 0,
+        profile: str = None,
     ) -> List[Dict[str, Any]]:
-        """List sessions, optionally filtered by source."""
+        """List sessions, optionally filtered by source and profile.
+
+        Args:
+            profile: Filter sessions to this profile name. Pass None to see all.
+                (#323)
+        """
+        where_clauses = []
+        params = []
+        if source:
+            where_clauses.append("source = ?")
+            params.append(source)
+        if profile:
+            where_clauses.append("profile = ?")
+            params.append(profile)
+
+        where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
+        query = f"SELECT * FROM sessions {where_sql} ORDER BY started_at DESC LIMIT ? OFFSET ?"
+        params.extend([limit, offset])
         with self._lock:
-            if source:
-                cursor = self._conn.execute(
-                    "SELECT * FROM sessions WHERE source = ? ORDER BY started_at DESC LIMIT ? OFFSET ?",
-                    (source, limit, offset),
-                )
-            else:
-                cursor = self._conn.execute(
-                    "SELECT * FROM sessions ORDER BY started_at DESC LIMIT ? OFFSET ?",
-                    (limit, offset),
-                )
+            cursor = self._conn.execute(query, params)
             return [dict(row) for row in cursor.fetchall()]
 
     # =========================================================================
     # Utility
     # =========================================================================
 
-    def session_count(self, source: str = None) -> int:
-        """Count sessions, optionally filtered by source."""
+    def session_count(self, source: str = None, profile: str = None) -> int:
+        """Count sessions, optionally filtered by source and profile.
+
+        Args:
+            profile: Filter to this profile name. Pass None to count all. (#323)
+        """
+        where_clauses = []
+        params = []
+        if source:
+            where_clauses.append("source = ?")
+            params.append(source)
+        if profile:
+            where_clauses.append("profile = ?")
+            params.append(profile)
+
+        where_sql = f"WHERE {' AND '.join(where_clauses)}" if where_clauses else ""
         with self._lock:
-            if source:
-                cursor = self._conn.execute(
-                    "SELECT COUNT(*) FROM sessions WHERE source = ?", (source,)
-                )
-            else:
-                cursor = self._conn.execute("SELECT COUNT(*) FROM sessions")
+            cursor = self._conn.execute(f"SELECT COUNT(*) FROM sessions {where_sql}", params)
             return cursor.fetchone()[0]
 
     def message_count(self, session_id: str = None) -> int:

tests/test_ssh_hermes_validation.py

@@ -0,0 +1,129 @@
+"""
+Test remote hermes path validation functions.
+"""
+
+import pytest
+import subprocess
+from unittest.mock import Mock, patch
+from tools.environments.ssh import SSHEnvironment
+
+
+class TestHermesPathValidation:
+    """Test hermes path validation functions."""
+    
+    def test_validate_remote_hermes_path_found(self):
+        """Test validation when hermes binary exists."""
+        # Mock SSHEnvironment
+        ssh_env = Mock(spec=SSHEnvironment)
+        ssh_env.run = Mock(return_value="FOUND")
+        
+        # Call validation
+        result = SSHEnvironment.validate_remote_hermes_path(ssh_env, "/usr/local/bin/hermes")
+        
+        # Verify result
+        assert result["available"] is True
+        assert result["path"] == "/usr/local/bin/hermes"
+        assert result["error"] is None
+    
+    def test_validate_remote_hermes_path_not_found(self):
+        """Test validation when hermes binary doesn't exist."""
+        # Mock SSHEnvironment
+        ssh_env = Mock(spec=SSHEnvironment)
+        ssh_env.run = Mock(return_value="NOT_FOUND")
+        
+        # Call validation
+        result = SSHEnvironment.validate_remote_hermes_path(ssh_env, "/invalid/path/hermes")
+        
+        # Verify result
+        assert result["available"] is False
+        assert result["path"] == "/invalid/path/hermes"
+        assert "not found" in result["error"].lower()
+    
+    def test_validate_remote_hermes_path_error(self):
+        """Test validation when SSH command fails."""
+        # Mock SSHEnvironment
+        ssh_env = Mock(spec=SSHEnvironment)
+        ssh_env.run = Mock(side_effect=subprocess.TimeoutExpired("cmd", 10))
+        
+        # Call validation
+        result = SSHEnvironment.validate_remote_hermes_path(ssh_env, "/usr/local/bin/hermes")
+        
+        # Verify result
+        assert result["available"] is False
+        assert "error" in result["error"].lower()
+    
+    def test_get_default_hermes_path(self):
+        """Test getting default hermes path."""
+        # Mock SSHEnvironment
+        ssh_env = Mock(spec=SSHEnvironment)
+        
+        # Test with local bin path found
+        ssh_env.run = Mock(return_value="/home/user/.local/bin/hermes")
+        result = SSHEnvironment._get_default_hermes_path(ssh_env)
+        assert result == "/home/user/.local/bin/hermes"
+        
+        # Test with wizard pattern
+        ssh_env.run = Mock(side_effect=["", "/root/wizards/ezra/hermes-agent/venv/bin/hermes"])
+        result = SSHEnvironment._get_default_hermes_path(ssh_env)
+        assert result == "/root/wizards/ezra/hermes-agent/venv/bin/hermes"
+    
+    def test_execute_hermes_command_success(self):
+        """Test successful hermes command execution."""
+        # Mock SSHEnvironment
+        ssh_env = Mock(spec=SSHEnvironment)
+        ssh_env.run = Mock(return_value="Job output here")
+        ssh_env.validate_remote_hermes_path = Mock(return_value={
+            "available": True,
+            "path": "/usr/local/bin/hermes",
+            "error": None
+        })
+        
+        # Call execution
+        result = SSHEnvironment.execute_hermes_command(ssh_env, "cron list", validate_path=True)
+        
+        # Verify result
+        assert result["success"] is True
+        assert result["stdout"] == "Job output here"
+        assert result["exit_code"] == 0
+        assert result["error"] is None
+    
+    def test_execute_hermes_command_validation_failed(self):
+        """Test hermes command execution when validation fails."""
+        # Mock SSHEnvironment
+        ssh_env = Mock(spec=SSHEnvironment)
+        ssh_env.validate_remote_hermes_path = Mock(return_value={
+            "available": False,
+            "path": "/invalid/path/hermes",
+            "error": "Hermes binary not found"
+        })
+        
+        # Call execution
+        result = SSHEnvironment.execute_hermes_command(ssh_env, "cron list", validate_path=True)
+        
+        # Verify result
+        assert result["success"] is False
+        assert "not found" in result["error"].lower()
+        assert result["exit_code"] == 1
+    
+    def test_execute_hermes_command_timeout(self):
+        """Test hermes command execution timeout."""
+        # Mock SSHEnvironment
+        ssh_env = Mock(spec=SSHEnvironment)
+        ssh_env.run = Mock(side_effect=subprocess.TimeoutExpired("cmd", 300))
+        ssh_env.validate_remote_hermes_path = Mock(return_value={
+            "available": True,
+            "path": "/usr/local/bin/hermes",
+            "error": None
+        })
+        
+        # Call execution
+        result = SSHEnvironment.execute_hermes_command(ssh_env, "cron list", validate_path=True)
+        
+        # Verify result
+        assert result["success"] is False
+        assert "timeout" in result["error"].lower()
+        assert result["exit_code"] == -1
+
+
+if __name__ == "__main__":
+    pytest.main([__file__])

tools/environments/ssh.py

@@ -311,3 +311,152 @@ class SSHEnvironment(PersistentShellMixin, BaseEnvironment):
                 self.control_socket.unlink()
             except OSError:
                 pass
+
+
+
+    def validate_remote_hermes_path(self, hermes_path: str = None) -> dict:
+        """
+        Validate that hermes binary exists and is executable on the remote host.
+        
+        Args:
+            hermes_path: Path to hermes binary. If None, uses default path.
+            
+        Returns:
+            dict with keys:
+                - available: bool (True if hermes is available)
+                - path: str (actual path found)
+                - error: str (error message if not available)
+        """
+        if hermes_path is None:
+            hermes_path = self._get_default_hermes_path()
+        
+        # Check if hermes binary exists and is executable
+        check_cmd = f"test -x {hermes_path} && echo 'FOUND' || echo 'NOT_FOUND'"
+        
+        try:
+            result = self.run(check_cmd, timeout=10)
+            if "FOUND" in result:
+                return {
+                    "available": True,
+                    "path": hermes_path,
+                    "error": None
+                }
+            else:
+                return {
+                    "available": False,
+                    "path": hermes_path,
+                    "error": f"Hermes binary not found or not executable: {hermes_path}"
+                }
+        except Exception as e:
+            return {
+                "available": False,
+                "path": hermes_path,
+                "error": f"Error validating hermes path: {str(e)}"
+            }
+    
+    def _get_default_hermes_path(self) -> str:
+        """Get the default hermes path for this host."""
+        # Try common paths in order of preference
+        paths_to_try = [
+            "~/.local/bin/hermes",  # Standard install location
+            "/root/wizards/*/hermes-agent/venv/bin/hermes",  # Wizard pattern
+            "/usr/local/bin/hermes",  # System install
+        ]
+        
+        for path_pattern in paths_to_try:
+            if "*" in path_pattern:
+                # Use find for glob patterns
+                find_cmd = f"find {path_pattern.replace('*', '*')} -maxdepth 0 2>/dev/null | head -1"
+                try:
+                    result = self.run(find_cmd, timeout=5)
+                    if result.strip():
+                        return result.strip()
+                except:
+                    continue
+            else:
+                # Direct path check
+                check_cmd = f"test -x {path_pattern} && echo {path_pattern}"
+                try:
+                    result = self.run(check_cmd, timeout=5)
+                    if result.strip():
+                        return result.strip()
+                except:
+                    continue
+        
+        # Fallback to wizard pattern
+        return "/root/wizards/*/hermes-agent/venv/bin/hermes"
+    
+    def execute_hermes_command(self, command: str, validate_path: bool = True) -> dict:
+        """
+        Execute a hermes command on the remote host with proper validation.
+        
+        Args:
+            command: Hermes command to execute (e.g., "cron list")
+            validate_path: Whether to validate hermes path before execution
+            
+        Returns:
+            dict with keys:
+                - success: bool (True if command executed successfully)
+                - stdout: str (command output)
+                - stderr: str (error output)
+                - exit_code: int (command exit code)
+                - error: str (error message if failed)
+        """
+        # Validate hermes path if requested
+        if validate_path:
+            validation = self.validate_remote_hermes_path()
+            if not validation["available"]:
+                return {
+                    "success": False,
+                    "stdout": "",
+                    "stderr": validation["error"],
+                    "exit_code": 1,
+                    "error": validation["error"]
+                }
+            hermes_path = validation["path"]
+        else:
+            hermes_path = self._get_default_hermes_path()
+        
+        # Build full command
+        full_command = f"{hermes_path} {command}"
+        
+        try:
+            # Execute command
+            result = self.run(full_command, timeout=300)
+            
+            # Check exit code - only mark success if exit code is 0
+            # Note: self.run() raises an exception on non-zero exit code,
+            # so if we get here, the command succeeded
+            return {
+                "success": True,
+                "stdout": result,
+                "stderr": "",
+                "exit_code": 0,
+                "error": None
+            }
+            
+        except subprocess.CalledProcessError as e:
+            # Command failed with non-zero exit code
+            return {
+                "success": False,
+                "stdout": e.stdout or "",
+                "stderr": e.stderr or "",
+                "exit_code": e.returncode,
+                "error": f"Command failed with exit code {e.returncode}"
+            }
+        except subprocess.TimeoutExpired:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": "",
+                "exit_code": -1,
+                "error": "Command timed out"
+            }
+        except Exception as e:
+            return {
+                "success": False,
+                "stdout": "",
+                "stderr": "",
+                "exit_code": -1,
+                "error": f"Error executing command: {str(e)}"
+            }