fix(history): render brackets safely, collapse same-text rows

Two bugs in the query history screen:

1. Bracket identifiers like [dbo].[ServiceFeatures] were truncated to
   "[dbo]." in the option list. rich.markup.escape only escapes brackets
   whose tag-name starts with [a-z#/@], so [ServiceFeatures] slipped
   through and Textual's markup parser ate it as an unknown tag. Build
   options (and the preview) as Rich Text instead of markup strings to
   sidestep parsing entirely.

2. Starring one row marked every same-text row as starred, because
   history dedup is per (connection, database, query_hash) but the
   starred store is keyed by (connection, query_text). Collapse rows
   with identical query text in _merge_entries (keeping the most-recent
   timestamp), so the display row count matches the starring model.

Author: Maxteabag

sqlit/domains/query/ui/screens/query_history.py

@@ -7,7 +7,7 @@
 from typing import Any
 
 from textual.app import ComposeResult
-from rich.markup import escape
+from rich.text import Text
 from textual.binding import Binding
 from textual.containers import VerticalScroll
 from textual.screen import ModalScreen
@@ -112,17 +112,32 @@ def __init__(
         self._filtered_entries: list[QueryHistoryEntry] = []
 
     def _merge_entries(self) -> list[QueryHistoryEntry]:
-        """Merge history entries with starred-only queries."""
+        """Merge history entries with starred-only queries.
+
+        Same-text rows collapse to a single most-recent entry so the row
+        count matches the starring model (which is keyed by query text):
+        starring one row never marks two visually-identical rows. In
+        multi-connection (telescope) mode the dedupe key is
+        (connection_name, query); in single-connection mode it's just
+        the query.
+        """
         result: list[QueryHistoryEntry] = []
 
         if self._multi_connection:
-            history_queries_by_connection: dict[str, set[str]] = {}
+            seen: dict[tuple[str, str], QueryHistoryEntry] = {}
             for entry in self.history:
                 starred_queries = self._starred_by_connection.get(entry.connection_name, set())
                 entry.is_starred = entry.query.strip() in starred_queries
                 entry.is_starred_only = False
-                history_queries_by_connection.setdefault(entry.connection_name, set()).add(entry.query.strip())
-                result.append(entry)
+                key = (entry.connection_name, entry.query.strip())
+                existing = seen.get(key)
+                if existing is None or entry.timestamp > existing.timestamp:
+                    seen[key] = entry
+            result.extend(seen.values())
+
+            history_queries_by_connection: dict[str, set[str]] = {}
+            for (conn_name, query_text) in seen:
+                history_queries_by_connection.setdefault(conn_name, set()).add(query_text)
 
             for connection_name, starred_queries in self._starred_by_connection.items():
                 history_queries = history_queries_by_connection.get(connection_name, set())
@@ -137,15 +152,17 @@ def _merge_entries(self) -> list[QueryHistoryEntry]:
                         )
                         result.append(entry)
         else:
-            # Mark history entries that are starred
-            history_queries: set[str] = set()
+            seen_single: dict[str, QueryHistoryEntry] = {}
             for entry in self.history:
                 entry.is_starred = entry.query.strip() in self.starred
                 entry.is_starred_only = False
-                history_queries.add(entry.query.strip())
-                result.append(entry)
+                key = entry.query.strip()
+                existing = seen_single.get(key)
+                if existing is None or entry.timestamp > existing.timestamp:
+                    seen_single[key] = entry
+            result.extend(seen_single.values())
 
-            # Add starred-only entries (queries that were starred but not in history)
+            history_queries = set(seen_single.keys())
             for starred_query in self.starred:
                 if starred_query not in history_queries:
                     entry = QueryHistoryEntry(
@@ -221,7 +238,7 @@ def _update_preview(self, idx: int) -> None:
         entries = self._get_display_entries()
         if idx < len(entries):
             preview = self.query_one("#history-preview", Static)
-            preview.update(escape(entries[idx].query))
+            preview.update(Text(entries[idx].query))
 
     def action_select(self) -> None:
         entries = self._get_display_entries()
@@ -479,19 +496,25 @@ def _build_action_result(self, action: str, entry: QueryHistoryEntry) -> tuple[s
         return action, entry.query
 
     def _build_option(self, entry: QueryHistoryEntry) -> Option:
-        star = "[yellow]*[/] " if entry.is_starred else "  "
-        # Collapse all whitespace to single spaces and strip
+        # Build a Rich Text (not a markup string) so identifiers like
+        # [dbo].[ServiceFeatures] aren't eaten by the markup parser:
+        # rich.markup.escape only escapes "[a-z#/@…]" tags, so uppercase
+        # bracket identifiers slip through and get consumed as fake tags.
         query_single_line = re.sub(r"\s+", " ", entry.query).strip()
         max_len = 40 if self._multi_connection else 55
         if len(query_single_line) > max_len:
             query_preview = query_single_line[:max_len] + "..."
         else:
             query_preview = query_single_line
-        query_preview = escape(query_preview)
+
+        text = Text()
+        if entry.is_starred:
+            text.append("* ", style="yellow")
+        else:
+            text.append("  ")
         if self._multi_connection:
             conn_label = self._truncate_label(self._entry_connection_label(entry), 24)
-            conn_label = escape(conn_label)
-        option_id = self._entry_option_id(entry)
-        if self._multi_connection:
-            return Option(f"{star}[dim]{conn_label}[/]/{query_preview}", id=option_id)
-        return Option(f"{star}{query_preview}", id=option_id)
+            text.append(conn_label, style="dim")
+            text.append("/")
+        text.append(query_preview)
+        return Option(text, id=self._entry_option_id(entry))

tests/ui/test_query_history.py

@@ -296,6 +296,186 @@ def save_query(self, connection_name, query):
             assert all(entry.connection_name == "saved-db" for entry in screen._merged_entries)
 
 
+class TestQueryHistoryStarringDuplicates:
+    """Regression tests for starring behavior when multiple history entries
+    share the same query text (e.g., same SQL run against multiple databases
+    on one connection — file-backed dedup is per-(connection, database))."""
+
+    def _make_screen(
+        self,
+        history: list[QueryHistoryEntry],
+        starred: set[str],
+        connection_name: str = "saved-db",
+    ) -> QueryHistoryScreen:
+        return QueryHistoryScreen(
+            history=history,
+            connection_name=connection_name,
+            starred=starred,
+        )
+
+    def test_same_query_across_databases_collapses_to_single_row(self) -> None:
+        history = [
+            QueryHistoryEntry(
+                query="SELECT 1",
+                timestamp="2026-01-02T00:00:00",
+                connection_name="saved-db",
+                database="db_a",
+            ),
+            QueryHistoryEntry(
+                query="SELECT 1",
+                timestamp="2026-01-01T00:00:00",
+                connection_name="saved-db",
+                database="db_b",
+            ),
+        ]
+        screen = self._make_screen(history, starred={"SELECT 1"})
+
+        merged = screen._merge_entries()
+
+        assert len(merged) == 1, (
+            "Same query text from two databases should collapse to one row "
+            "in the history display (starring is keyed by text, so the row "
+            "count must match)."
+        )
+        assert merged[0].is_starred
+        # Most-recent occurrence wins the row.
+        assert merged[0].database == "db_a"
+
+    @pytest.mark.asyncio
+    async def test_starring_one_marks_duplicate_in_other_database(self) -> None:
+        saved_conn = create_test_connection("saved-db", "sqlite")
+        entries = [
+            QueryHistoryEntry(
+                query="SELECT 1",
+                timestamp="2026-01-02T00:00:00",
+                connection_name="saved-db",
+                database="db_a",
+            ),
+            QueryHistoryEntry(
+                query="SELECT 1",
+                timestamp="2026-01-01T00:00:00",
+                connection_name="saved-db",
+                database="db_b",
+            ),
+        ]
+
+        class StubHistoryStore:
+            def __init__(self, entries):
+                self._entries = entries
+
+            def load_all(self):
+                return list(self._entries)
+
+            def load_for_connection(self, connection_name):
+                return [e for e in self._entries if e.connection_name == connection_name]
+
+            def delete_entry(self, connection_name, timestamp):
+                _ = connection_name
+                _ = timestamp
+                return False
+
+            def save_query(self, connection_name, query, database=""):
+                _ = connection_name
+                _ = query
+                _ = database
+
+        history_store = StubHistoryStore(entries)
+        services = build_test_services(
+            connection_store=MockConnectionStore([saved_conn]),
+            settings_store=MockSettingsStore({"theme": "tokyo-night"}),
+            history_store=history_store,
+        )
+        app = SSMSTUI(services=services)
+
+        async with app.run_test(size=(100, 35)) as pilot:
+            app.connections = [saved_conn]
+            app.current_config = saved_conn
+
+            # Open history, star the first row (which corresponds to db_a),
+            # close — emulates: user opens history, presses '*' on one row.
+            app.action_show_history()
+            await pilot.pause(0.2)
+            screen = next(
+                (s for s in app.screen_stack if isinstance(s, QueryHistoryScreen)),
+                None,
+            )
+            assert screen is not None
+            option_list = screen.query_one("#history-list", OptionList)
+            # The two same-text rows should already be collapsed to one.
+            assert option_list.option_count == 1
+            assert option_list.highlighted == 0
+            screen.action_toggle_star()
+            await pilot.pause(0.2)
+
+            # Re-open: exactly one starred row, not two.
+            app.action_show_history()
+            await pilot.pause(0.2)
+            screen = next(
+                (s for s in app.screen_stack if isinstance(s, QueryHistoryScreen)),
+                None,
+            )
+            assert screen is not None
+            starred_entries = [e for e in screen._merged_entries if e.is_starred]
+            try:
+                assert len(starred_entries) == 1, (
+                    f"Expected exactly one starred row after toggling one; "
+                    f"got {len(starred_entries)}"
+                )
+            finally:
+                # The shared test config dir is module-scoped (see
+                # tests/fixtures/utils.py), so any star we save here would
+                # leak to later tests. Clean up.
+                services.starred_store.clear_for_connection(saved_conn.name)
+
+
+class TestQueryHistoryBracketIdentifierRendering:
+    """Regression tests for MSSQL bracket identifiers like [dbo].[ServiceFeatures]
+    being silently eaten by Textual markup when the bracket contents start with
+    an uppercase letter (rich.markup.escape only escapes [a-z#/@] tags)."""
+
+    @pytest.mark.asyncio
+    async def test_bracket_identifier_with_uppercase_not_truncated(self) -> None:
+        from textual.app import App
+        from textual.widgets.option_list import Option
+
+        # Drive the option directly through a minimal Textual app so we
+        # observe what _build_option's text actually renders to.
+        screen = QueryHistoryScreen(
+            history=[
+                QueryHistoryEntry(
+                    query="SELECT TOP 100 * FROM [dbo].[ServiceFeatures]",
+                    timestamp="2026-01-01T00:00:00",
+                    connection_name="saved-db",
+                ),
+            ],
+            connection_name="saved-db",
+            starred=set(),
+        )
+        screen._merged_entries = screen._merge_entries()
+        built = screen._build_option(screen._merged_entries[0])
+
+        class _ProbeApp(App):
+            def compose(self):
+                yield OptionList(Option(built.prompt, id=built.id or "probe"), id="probe-list")
+
+        app = _ProbeApp()
+        async with app.run_test(size=(120, 10)) as pilot:
+            await pilot.pause()
+            option_list = app.query_one("#probe-list", OptionList)
+            lines = option_list.render_lines(option_list.region)
+            rendered_text = "\n".join(
+                "".join(seg.text for seg in line) for line in lines
+            )
+
+        assert "[ServiceFeatures]" in rendered_text, (
+            "Bracket identifier with uppercase first letter was silently "
+            "swallowed by Textual markup (rich.markup.escape only escapes "
+            "tags starting with [a-z#/@], so '[ServiceFeatures]' is left "
+            "unescaped and the markup parser eats it). Visible text:\n"
+            + rendered_text
+        )
+
+
 class TestQueryHistoryVimNavigation:
     """Tests for j/k vim-style navigation in the history screen."""