def search_messages(
connection: sqlite3.Connection,
query: str,
*,
embedder: Embedder | None = None,
project: str | None = None,
after: str | None = None,
before: str | None = None,
role: str | None = None,
files: list[str] | None = None,
limit: int = 8,
) -> SearchResponse:
if role is not None and role not in {"user", "assistant", "tool"}:
raise ValueError("role must be user, assistant, or tool")
limit = max(1, min(limit, MAX_RESULTS))
if not query.strip():
raise ValueError("query must not be empty")
if embedder is None and not re.search(r"\w+", query, flags=re.UNICODE):
raise ValueError("query must contain at least one searchable term")
filters, parameters = _filters(
project=project, after=after, before=before, role=role, files=files
)
candidate_limit = limit * 4
keyword_ranking = _keyword_ranking(
connection, query, filters, parameters, candidate_limit
)
semantic_ranking = (
_semantic_ranking(
connection, query, embedder, filters, parameters, candidate_limit
)
if embedder is not None
else []
)
fused = _fuse_rankings(keyword_ranking, semantic_ranking)
if not fused:
rows = []
else:
placeholders = ",".join("?" for _ in fused)
fetched = connection.execute(
"SELECT m.id, m.session_id, s.project, m.timestamp, m.role, m.text "
"FROM messages m JOIN sessions s ON s.id = m.session_id "
f"WHERE m.id IN ({placeholders})",
[message_id for message_id, _ in fused],
).fetchall()
by_id = {row["id"]: row for row in fetched}
rows = [(by_id[message_id], score) for message_id, score in fused]
results: list[SearchResult] = []
seen_text: set[str] = set()
for row, score in rows:
normalized_text = " ".join(row["text"].split()).casefold()
if normalized_text in seen_text:
continue
seen_text.add(normalized_text)
paths = [
item[0]
for item in connection.execute(
"SELECT path FROM file_operations WHERE message_id = ? ORDER BY path",
(row["id"],),
)
]
results.append(
{
"session_id": row["session_id"],
"message_id": row["id"],
"project": row["project"],
"timestamp": row["timestamp"],
"role": row["role"],
"score": round(score, 6),
"excerpt": row["text"][:EXCERPT_LENGTH],
"files": paths,
}
)
if len(results) == limit:
break
refresh = connection.execute(
"SELECT max(updated_at), count(*) FILTER (WHERE last_error IS NOT NULL) FROM import_files"
).fetchone()
return {
"results": results,
"index": {"last_refresh": refresh[0], "refresh_errors": refresh[1]},
}