fix(sessions): escape LIKE wildcards in prune/archive substring filters
_prune_filter_where documents title_like / model_like / branch_like as "case-insensitive substring matches", and the CLI confirmation renders them as "title contains 'X'". They were bound straight into a bare LIKE, so `_` matched any single character and `%` any run. The builder backs prune_sessions(), which deletes session rows and their on-disk transcripts, so the over-match is unrecoverable: pruning title_like="user_auth" also destroys "user-auth", "userXauth" and "user auth". `_` is not exotic here -- git branch names and session titles carry it routinely. Escape the operator's needle and add ESCAPE '\' to the three clauses, the same convention the rest of this file already uses for LIKE queries. Match direction is unchanged for needles without wildcards. Left alone: _cwd_prefix_clause has the same unescaped shape but is shared by four call sites beyond prune, so it is a separate change.
This commit is contained in:
parent
9ea01979dc
commit
1d2dabce56
|
|
@ -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)) >= ?"
|
||||
|
|
|
|||
|
|
@ -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):
|
||||
|
|
|
|||
Loading…
Reference in New Issue