diff --git a/hermes_state.py b/hermes_state.py index 9c1fa6b0548d2..aba3b92992476 100644 --- a/hermes_state.py +++ b/hermes_state.py @@ -176,6 +176,17 @@ def _delegate_from_json(col: str = "model_config") -> str: _MODEL_CONFIG_ROW_MISSING = object() +def _escape_like(text: str) -> str: + """Escape SQL LIKE wildcards so an operator-supplied filter matches + literally. Pair with ``ESCAPE '\\'`` in the clause. + + ``%`` and ``_`` are wildcards to LIKE, and ``_`` in particular is common + in the values these filters run against (branch names, session titles). + A filter documented as a substring match must not silently widen. + """ + return text.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + def _cwd_prefix_clause(cwd_prefix: str) -> Tuple[str, List[str]]: prefix = cwd_prefix.rstrip("/\\") or cwd_prefix return "(s.cwd = ? OR s.cwd LIKE ? OR s.cwd LIKE ?)", [prefix, f"{prefix}/%", f"{prefix}\\%"] @@ -8402,8 +8413,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) clauses.append("s.source = ?") params.append(source) if title_like: - clauses.append("LOWER(COALESCE(s.title, '')) LIKE ?") - params.append(f"%{title_like.lower()}%") + clauses.append("LOWER(COALESCE(s.title, '')) LIKE ? ESCAPE '\\'") + params.append(f"%{_escape_like(title_like.lower())}%") if end_reason: clauses.append("s.end_reason = ?") params.append(end_reason) @@ -8418,8 +8429,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) clauses.append("s.message_count <= ?") params.append(max_messages) if model_like: - clauses.append("LOWER(COALESCE(s.model, '')) LIKE ?") - params.append(f"%{model_like.lower()}%") + clauses.append("LOWER(COALESCE(s.model, '')) LIKE ? ESCAPE '\\'") + params.append(f"%{_escape_like(model_like.lower())}%") if provider: clauses.append("LOWER(COALESCE(s.billing_provider, '')) = ?") params.append(provider.lower()) @@ -8433,8 +8444,8 @@ class SessionDB(SessionSearchMixin, SessionSchemaMixin, SessionPortabilityMixin) clauses.append("s.chat_type = ?") params.append(chat_type) if branch_like: - clauses.append("LOWER(COALESCE(s.git_branch, '')) LIKE ?") - params.append(f"%{branch_like.lower()}%") + clauses.append("LOWER(COALESCE(s.git_branch, '')) LIKE ? ESCAPE '\\'") + params.append(f"%{_escape_like(branch_like.lower())}%") if min_tokens is not None: clauses.append( "(COALESCE(s.input_tokens, 0) + COALESCE(s.output_tokens, 0)) >= ?" diff --git a/tests/test_hermes_state.py b/tests/test_hermes_state.py index 74a028c754713..02d4c874c9d6d 100644 --- a/tests/test_hermes_state.py +++ b/tests/test_hermes_state.py @@ -1068,6 +1068,59 @@ class TestPruneSessionFilters: + def test_title_like_underscore_is_literal_not_a_wildcard(self, db): + """``_`` is a single-character wildcard in SQL LIKE, so an unescaped + filter deletes sessions the operator never selected. The filters are + documented (and shown in the CLI confirmation) as substring matches. + """ + self._mk(db, "target", title="user_auth refactor") + self._mk(db, "bystander1", title="user-auth review") + self._mk(db, "bystander2", title="userXauth notes") + self._mk(db, "bystander3", title="user auth meeting") + + rows = db.list_prune_candidates(title_like="user_auth") + assert {r["id"] for r in rows} == {"target"} + + pruned = db.prune_sessions(older_than_days=None, title_like="user_auth") + assert pruned == 1 + for survivor in ("bystander1", "bystander2", "bystander3"): + assert db.get_session(survivor) is not None + + def test_percent_in_filter_does_not_select_everything(self, db): + """``%`` matches any run of characters — a bare one would delete the + whole table.""" + self._mk(db, "a", title="alpha") + self._mk(db, "b", title="beta") + self._mk(db, "pct", title="100% coverage run") + + # Only the title that really contains a percent sign matches. + assert {r["id"] for r in db.list_prune_candidates(title_like="%")} == {"pct"} + assert {r["id"] for r in db.list_prune_candidates(title_like="100%")} == {"pct"} + + def test_branch_like_underscore_is_literal(self, db): + """Branch names carry underscores routinely.""" + self._mk_rich(db, "want", git_branch="fix/session_prune") + self._mk_rich(db, "other", git_branch="fix/session-prune") + + rows = db.list_prune_candidates(branch_like="session_prune") + assert {r["id"] for r in rows} == {"want"} + + def test_model_like_underscore_is_literal(self, db): + self._mk_rich(db, "want", model="vendor/model_mini") + self._mk_rich(db, "other", model="vendor/model-mini") + + rows = db.list_prune_candidates(model_like="model_mini") + assert {r["id"] for r in rows} == {"want"} + + def test_plain_substring_filters_still_match(self, db): + """Guard against over-escaping: ordinary filters keep working, and a + literal backslash in the needle is matched as itself.""" + self._mk(db, "smoke", title="Codex Smoke Test") + self._mk_rich(db, "winpath", title=r"build C:\tmp artifacts") + + assert {r["id"] for r in db.list_prune_candidates(title_like="smoke")} == {"smoke"} + assert {r["id"] for r in db.list_prune_candidates(title_like=r"c:\tmp")} == {"winpath"} + def test_unknown_filter_rejected(self, db): import pytest as _pytest with _pytest.raises(TypeError):