fix(state): honor AND/NOT operators in the CJK LIKE fallback

hermes_state.py

@@ -8958,22 +8958,58 @@ def _search_messages_impl(
             if not _trigram_succeeded:
                 # Short / mixed CJK query, trigram unavailable, or trigram
                 # <3 CJK chars. Fall back to LIKE substring search.
-                # For multi-token OR queries (e.g. "广西 OR 桂林 OR 漓江"),
-                # build one LIKE condition per non-operator token so each term
-                # is matched independently (#20494).
-                non_op_tokens = [
-                    t for t in raw_query.split()
-                    if t.upper() not in {"AND", "OR", "NOT"}
-                ] or [raw_query]
-                token_clauses = []
+                # Honor the same boolean semantics FTS5 applies on the trigram
+                # path: whitespace/AND joins terms conjunctively, OR splits
+                # alternatives (#20494), and NOT excludes the following run.
+                # FTS5 precedence (verified against SQLite): implicit AND
+                # binds tighter than NOT ("a NOT b c" == "a NOT (b AND c)"),
+                # while explicit AND binds looser ("a NOT b AND c" ==
+                # "(a NOT b) AND c"), and OR binds loosest. Group tokens into
+                # OR-separated (positives, negated-runs) buckets accordingly:
+                # NOT opens a conjunctive run that implicit adjacency extends
+                # and explicit AND/OR terminates.
+                groups: list = [([], [])]  # (positives, list of negated runs)
+                current: list = groups[-1][0]
+                for tok in raw_query.split():
+                    upper = tok.upper()
+                    if upper == "OR":
+                        if groups[-1][0] or any(groups[-1][1]):
+                            groups.append(([], []))
+                        current = groups[-1][0]
+                    elif upper == "AND":
+                        current = groups[-1][0]
+                    elif upper == "NOT":
+                        groups[-1][1].append([])
+                        current = groups[-1][1][-1]
+                    else:
+                        current.append(tok)
+                groups = [(p, [r for r in n if r]) for p, n in groups]
+                groups = [g for g in groups if g[0] or g[1]]
+                if not groups:
+                    groups = [([raw_query], [])]
+
                 like_params: list = []
-                for tok in non_op_tokens:
+
+                def _like_term(tok: str) -> str:
                     esc = tok.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
-                    token_clauses.append(
-                        "(m.content LIKE ? ESCAPE '\\' OR m.tool_name LIKE ? ESCAPE '\\' OR m.tool_calls LIKE ? ESCAPE '\\')"
+                    like_params.extend([f"%{esc}%"] * 3)
+                    # COALESCE keeps NULL tool columns from turning a negated
+                    # term into NULL (which WHERE would treat as non-matching).
+                    return (
+                        "(m.content LIKE ? ESCAPE '\\'"
+                        " OR COALESCE(m.tool_name, '') LIKE ? ESCAPE '\\'"
+                        " OR COALESCE(m.tool_calls, '') LIKE ? ESCAPE '\\')"
                     )
-                    like_params += [f"%{esc}%", f"%{esc}%", f"%{esc}%"]
-                like_where = [f"({' OR '.join(token_clauses)})"]
+
+                group_clauses = []
+                for positives, negated_runs in groups:
+                    parts = [_like_term(t) for t in positives]
+                    parts += [
+                        "NOT (" + " AND ".join(_like_term(t) for t in run) + ")"
+                        for run in negated_runs
+                    ]
+                    group_clauses.append("(" + " AND ".join(parts) + ")")
+                like_where = [f"({' OR '.join(group_clauses)})"]
                 if not include_inactive:
                     # Same visibility rule as the FTS5 paths: live rows and
                     # compaction-archived rows are discoverable; rewind/undo
@@ -9003,7 +9039,10 @@ def _search_messages_impl(
                 """
                 like_params.extend([limit, offset])
                 # instr() for snippet uses first search token
-                like_params = [non_op_tokens[0]] + like_params
+                _snippet_token = next(
+                    (t for positives, _ in groups for t in positives), raw_query
+                )
+                like_params = [_snippet_token] + like_params
                 with self._read_ctx() as conn:
                     like_cursor = conn.execute(like_sql, like_params)
                     matches = [dict(row) for row in like_cursor.fetchall()]

tests/test_hermes_state.py

@@ -2617,6 +2617,100 @@ def test_cjk_short_token_or_query_preserves_filters(self, db):
         assert len(results) == 1
         assert results[0]["source"] == "telegram"
 
+    def test_cjk_like_fallback_honors_not_operator(self, db):
+        """NOT must exclude, not match: "报告 NOT 草稿" previously returned
+        the drafts too, because every non-operator token was OR-joined."""
+        # Force the LIKE fallback: with the bigram index available
+        # (PR #65544) short-CJK queries route to messages_fts_cjk, which
+        # already honors operators. The LIKE path still serves DBs where
+        # the index is absent or its backfill is pending.
+        db._fts_cjk_available = False
+        db.create_session(session_id="s1", source="cli")
+        db.create_session(session_id="s2", source="cli")
+        db.append_message("s1", role="user", content="季度报告已经定稿")
+        db.append_message("s2", role="user", content="报告还是草稿状态")
+
+        results = db.search_messages("报告 NOT 草稿")
+        session_ids = {r["session_id"] for r in results}
+        assert session_ids == {"s1"}, "NOT term must exclude drafts"
+
+    def test_cjk_like_fallback_honors_and_operator(self, db):
+        """AND must require both terms; the OR-join returned either."""
+        db._fts_cjk_available = False  # exercise the LIKE fallback
+        db.create_session(session_id="s1", source="cli")
+        db.create_session(session_id="s2", source="cli")
+        db.append_message("s1", role="user", content="广西的桂林很有名")
+        db.append_message("s2", role="user", content="广西南宁出差记录")
+
+        results = db.search_messages("广西 AND 桂林")
+        session_ids = {r["session_id"] for r in results}
+        assert session_ids == {"s1"}, "AND must require both terms"
+
+    def test_cjk_like_fallback_implicit_adjacency_is_conjunctive(self, db):
+        """Whitespace adjacency means AND in FTS5; the LIKE fallback must
+        match the trigram path's semantics."""
+        db._fts_cjk_available = False  # exercise the LIKE fallback
+        db.create_session(session_id="s1", source="cli")
+        db.create_session(session_id="s2", source="cli")
+        db.append_message("s1", role="user", content="广西的桂林很有名")
+        db.append_message("s2", role="user", content="广西南宁出差记录")
+
+        results = db.search_messages("广西 桂林")
+        session_ids = {r["session_id"] for r in results}
+        assert session_ids == {"s1"}
+
+    def test_cjk_like_fallback_not_binds_looser_than_implicit_and(self, db):
+        """FTS5 gives implicit AND tighter binding than NOT: "报告 NOT 草稿 初版"
+        means 报告 NOT (草稿 AND 初版) — a message is excluded only when it
+        contains the whole negated run, not any single term of it. The naive
+        emit 报告 AND NOT 草稿 AND 初版 would wrongly drop s3 (has 草稿 but
+        not the full run) and wrongly require 初版 as a positive term."""
+        db._fts_cjk_available = False  # exercise the LIKE fallback
+        db.create_session(session_id="s1", source="cli")
+        db.create_session(session_id="s2", source="cli")
+        db.create_session(session_id="s3", source="cli")
+        db.append_message("s1", role="user", content="季度报告初版完成")
+        db.append_message("s2", role="user", content="报告草稿初版都在这里")
+        db.append_message("s3", role="user", content="报告还是草稿状态")
+
+        results = db.search_messages("报告 NOT 草稿 初版")
+        session_ids = {r["session_id"] for r in results}
+        assert session_ids == {"s1", "s3"}, (
+            "only the full negated run (草稿 AND 初版) may exclude a message"
+        )
+
+    def test_cjk_like_fallback_explicit_and_terminates_not_run(self, db):
+        """Explicit AND binds looser than NOT in FTS5: "报告 NOT 草稿 AND 初版"
+        means (报告 NOT 草稿) AND 初版 — 初版 is a required positive term
+        again, not part of the negated run."""
+        db._fts_cjk_available = False  # exercise the LIKE fallback
+        db.create_session(session_id="s1", source="cli")
+        db.create_session(session_id="s2", source="cli")
+        db.create_session(session_id="s3", source="cli")
+        db.append_message("s1", role="user", content="季度报告初版完成")
+        db.append_message("s2", role="user", content="报告草稿初版都在这里")
+        db.append_message("s3", role="user", content="报告最终版")
+
+        results = db.search_messages("报告 NOT 草稿 AND 初版")
+        session_ids = {r["session_id"] for r in results}
+        assert session_ids == {"s1"}, (
+            "初版 must be required and 草稿 excluded independently"
+        )
+
+    def test_cjk_like_fallback_or_of_and_groups(self, db):
+        """OR binds loosest: "广西 AND 桂林 OR 漓江" = (广西 AND 桂林) OR 漓江."""
+        db._fts_cjk_available = False  # exercise the LIKE fallback
+        db.create_session(session_id="s1", source="cli")
+        db.create_session(session_id="s2", source="cli")
+        db.create_session(session_id="s3", source="cli")
+        db.append_message("s1", role="user", content="广西的桂林很有名")
+        db.append_message("s2", role="user", content="漓江风景很美")
+        db.append_message("s3", role="user", content="广西南宁出差记录")
+
+        results = db.search_messages("广西 AND 桂林 OR 漓江")
+        session_ids = {r["session_id"] for r in results}
+        assert session_ids == {"s1", "s2"}
+
 
 # =========================================================================
 # Session search and listing