// session_search.mbt — ported from proto/src/search/{truncate,format-conv,search}.ts
// ── helpers ──
///|
fn srch_max(a : Int, b : Int) -> Int {
if a > b { a } else { b }
}
///|
fn srch_min(a : Int, b : Int) -> Int {
if a < b { a } else { b }
}
///|
fn srch_abs(a : Int) -> Int {
if a < 0 { -a } else { a }
}
// ── Pure MoonBit helpers ──
///| Find every starting index of `pattern` inside `text` (both
///| case-folded to lowercase). Returns empty when pattern is empty.
///| Substring match — no regex needed, cross-target.
fn _find_all_positions(text : String, pattern : String) -> Array[Int] {
let out : Array[Int] = []
if pattern.is_empty() {
return out
}
let t = text.to_lower()
let p = pattern.to_lower()
let tv = t.view()
let pv = p.view()
let tl = tv.length()
let pl = pv.length()
if pl > tl {
return out
}
let mut i = 0
while i <= tl - pl {
let mut j = 0
let mut ok = true
while j < pl {
if tv.get_char(i + j) != pv.get_char(j) {
ok = false
break
}
j = j + 1
}
if ok {
out.push(i)
}
i = i + 1
}
out
}
///| Tokenize on ASCII whitespace, lowercase, and drop empties.
fn _split_words(s : String) -> Array[String] {
let tokens : Array[String] = []
let buf = StringBuilder::new()
let view = s.to_lower().view()
let flush = fn() {
let t = buf.to_string()
if !t.is_empty() {
tokens.push(t)
buf.reset()
}
}
let mut i = 0
while i < view.length() {
let ch = match view.get_char(i) {
Some(c) => c
None => ' '
}
if ch == ' ' || ch == '\t' || ch == '\n' || ch == '\r' || ch == '\u{000C}' || ch == '\u{000B}' {
flush()
} else {
buf.write_char(ch)
}
i = i + 1
}
flush()
tokens
}
///| Extract `name` field (or `function.name`) from each entry of a
///| JSON-serialized tool_calls array. Returns an empty array on parse
///| failure or non-array input. Uses moonbitlang/core/json so it stays
///| cross-target.
fn _parse_tool_call_names(json_str : String) -> Array[String] {
let out : Array[String] = []
let root = try {
@json.parse(json_str)
} catch {
_ => return out
}
match root {
Json::Array(items) =>
for item in items {
match item {
Json::Object(obj) => {
// Prefer top-level "name" then "function.name".
let top = match obj.get("name") {
Some(Json::String(s)) => Some(s)
_ => None
}
let name = match top {
Some(s) => s
None =>
match obj.get("function") {
Some(Json::Object(f)) =>
match f.get("name") {
Some(Json::String(s)) => s
_ => "?"
}
_ => "?"
}
}
out.push(name)
}
_ => ()
}
}
_ => ()
}
out
}
///|
fn _get_summarizer_model() -> String {
@ffi.get_env("MNEMO_SUMMARIZER_MODEL")
}
///|
fn _get_summarizer_api_key() -> String {
let key = @ffi.get_env("OPENROUTER_API_KEY")
if key != "" {
key
} else {
@ffi.get_env("MNEMO_SUMMARIZER_API_KEY")
}
}
///|
let _summarize_system_prompt : String =
"You summarize conversations concisely."
///|
let _summarize_prompt_prefix : String =
"Summarize this conversation in 3-5 sentences focusing on:\n1. What the user was trying to accomplish\n2. Key decisions or actions taken\n3. Any important outcomes or unresolved issues\n\n\n"
///|
let _summarize_prompt_suffix : String = "\n"
///|
fn _summarize_with_llm(conversation_text : String) -> String {
let model = _get_summarizer_model()
let api_key = _get_summarizer_api_key()
if model == "" || api_key == "" {
return ""
}
let provider = @openai.OpenAIProvider::new(
api_key,
endpoint=@openai.OpenAIEndpoint::OpenRouter,
model=model,
max_tokens=512,
system_prompt=_summarize_system_prompt,
)
let prompt = _summarize_prompt_prefix + conversation_text + _summarize_prompt_suffix
let messages = [@llm.Message::user(prompt)]
@llm.collect_text(provider, messages)
}
///| JS-only async variant. Returns a Promise that resolves to the LLM summary
///| (or "" when env is missing / network fails). Enables fan-out via Promise.all.
fn _summarize_with_llm_async(
conversation_text : String
) -> @ffi.Promise[String] {
let model = _get_summarizer_model()
let api_key = _get_summarizer_api_key()
if model == "" || api_key == "" {
return _resolved_string("")
}
let provider = @openai.OpenAIProvider::new(
api_key,
endpoint=@openai.OpenAIEndpoint::OpenRouter,
model=model,
max_tokens=512,
system_prompt=_summarize_system_prompt,
)
let prompt = _summarize_prompt_prefix +
conversation_text +
_summarize_prompt_suffix
let messages = [@llm.Message::user(prompt)]
provider.collect_text_async(messages)
}
///| Promise.resolve(s) — helper for sync-path early returns.
extern "js" fn _resolved_string(s : String) -> @ffi.Promise[String] =
#| (s) => Promise.resolve(s)
///| Promise.all for an array of string-producing promises. Resolves to the
///| array of resolved values in input order. Rejects if any promise rejects
///| — caller should ensure the per-request path already swallows errors into "".
extern "js" fn _promise_all_strings(
promises : Array[@ffi.Promise[String]]
) -> @ffi.Promise[Array[String]] =
#| (promises) => Promise.all(promises)
// ── truncate_around_matches ──
pub let max_session_chars : Int = 100000
///|
pub fn truncate_around_matches(
full_text : String,
query : String,
max_chars~ : Int = max_session_chars
) -> String {
if full_text.length() <= max_chars {
return full_text
}
let query_trimmed = query.to_lower().trim().to_string()
// 1. Full phrase search
let phrase_positions : Array[Int] = _find_all_positions(full_text, query_trimmed)
let match_positions : Array[Int] = []
if phrase_positions.length() > 0 {
for p in phrase_positions {
match_positions.push(p)
}
}
// 2. Proximity co-occurrence (all terms within 200 chars of rarest)
if match_positions.length() == 0 {
let terms = _split_words(query)
if terms.length() > 1 {
// Collect positions per term
let term_pos : Array[(String, Array[Int])] = terms.map(fn(t) {
(t, _find_all_positions(full_text, t))
})
// Find rarest term (fewest matches)
let mut rarest_idx = 0
let mut rarest_len = term_pos[0].1.length()
let mut ti = 1
while ti < term_pos.length() {
let cur_len = term_pos[ti].1.length()
if cur_len < rarest_len {
rarest_len = cur_len
rarest_idx = ti
}
ti = ti + 1
}
let (rarest_term, rarest_positions) = term_pos[rarest_idx]
for pos in rarest_positions {
// Check that all other terms appear within 200 chars of pos
let mut all_near = true
let mut j = 0
while j < term_pos.length() {
let (t, positions) = term_pos[j]
if t != rarest_term {
let mut found_near = false
let mut k = 0
while k < positions.length() {
if srch_abs(positions[k] - pos) < 200 {
found_near = true
break
}
k = k + 1
}
if !found_near {
all_near = false
break
}
}
j = j + 1
}
if all_near {
match_positions.push(pos)
}
}
}
}
// 3. Individual term positions (last resort)
if match_positions.length() == 0 {
let terms = _split_words(query)
for t in terms {
let pos = _find_all_positions(full_text, t)
for p in pos {
match_positions.push(p)
}
}
}
// No hits — take head
if match_positions.length() == 0 {
let truncated = full_text[0:max_chars].to_string()
return truncated + "\n\n...[later conversation truncated]..."
}
// Sort positions ascending
let sorted = match_positions.copy()
sorted.sort_by(fn(a, b) { a - b })
// Pick window covering the most positions (25% bias before match)
let mut best_start = 0
let mut best_count = 0
for candidate in sorted {
let ws = srch_max(0, candidate - max_chars / 4)
let we = ws + max_chars
let mut count = 0
for p in sorted {
if p >= ws && p < we {
count = count + 1
}
}
if count > best_count {
best_count = count
best_start = ws
}
}
// Clamp to end of text
if best_start + max_chars > full_text.length() {
best_start = srch_max(0, full_text.length() - max_chars)
}
let start = best_start
let end = srch_min(full_text.length(), start + max_chars)
let truncated = full_text[start:end].to_string()
let prefix = if start > 0 { "...[earlier conversation truncated]...\n\n" } else { "" }
let suffix = if end < full_text.length() { "\n\n...[later conversation truncated]..." } else { "" }
prefix + truncated + suffix
}
// ── format_conversation ──
///|
pub struct MsgForFormat {
role : String
content : String
tool_name : String
tool_calls : String
}
///|
pub fn format_conversation(messages : Array[MsgForFormat]) -> String {
let parts : Array[String] = []
for msg in messages {
let role = msg.role.to_upper()
let content = msg.content
if role == "TOOL" && msg.tool_name != "" {
let c = if content.length() > 500 {
content[0:250].to_string() +
"\n...[truncated]...\n" +
content[(content.length() - 250):content.length()].to_string()
} else {
content
}
parts.push("[TOOL:\{msg.tool_name}]: \{c}")
} else if role == "ASSISTANT" {
if msg.tool_calls != "" {
let names = _parse_tool_call_names(msg.tool_calls)
if names.length() > 0 {
let joined = names.join(", ")
parts.push("[ASSISTANT]: [Called: \{joined}]")
}
}
if content != "" {
parts.push("[ASSISTANT]: \{content}")
}
} else {
parts.push("[\{role}]: \{content}")
}
}
parts.join("\n\n")
}
// ── session_search orchestration ──
///|
pub struct SearchResult {
session_id : String
when : String
source : String
model : String
summary : String
}
///|
pub struct SessionSearchResponse {
ok : Bool
query : String
results : Array[SearchResult]
count : Int
sessions_searched : Int
}
///| Render an epoch-ms timestamp as "YYYY-MM-DD HH:MM UTC". Previously
///| used JS Date.toLocaleString; swapped to moonbitlang/x/time so the
///| formatter works on native too. Returns "unknown" if parsing fails.
fn _format_timestamp(ts_str : String) -> String {
let ms = (try? @string.parse_int64(ts_str.view())).unwrap_or(0L)
if ms == 0L {
return "unknown"
}
let secs = ms / 1000L
let zdt_res = (try? @time.unix(secs))
match zdt_res {
Ok(zdt) => {
let d = zdt.to_plain_date_time()
// ISO-like, UTC-based. Stable for tests, readable for humans.
let pad2 = fn(n : Int) -> String {
if n < 10 {
"0" + n.to_string()
} else {
n.to_string()
}
}
d.year().to_string() +
"-" +
pad2(d.month()) +
"-" +
pad2(d.day()) +
" " +
pad2(d.hour()) +
":" +
pad2(d.minute()) +
" UTC"
}
Err(_) => ts_str
}
}
///| Return metadata (id, source, model, started_at, parent_session_id)
///| for a single session as positional strings. Empty array on miss.
fn _sess_get_session_meta(
db : SqliteDb,
session_id : String
) -> Array[String] {
let stmt = prepare(
db,
"SELECT id, source, model, started_at, parent_session_id FROM sessions WHERE id = ?",
)
_stmt_bind_text(stmt, 1, session_id)
let rows = _stmt_all(stmt)
if _rows_length(rows) == 0 {
return []
}
let row = _row_at(rows, 0)
let source = if _row_is_null(row, "source") {
"unknown"
} else {
_row_get_text(row, "source")
}
[
_row_get_text(row, "id"),
source,
_row_get_text(row, "model"),
_row_get_int64(row, "started_at").to_string(),
_row_get_text(row, "parent_session_id"),
]
}
///|
fn srch_resolve_to_root(sdb : SessionDb, session_id : String) -> String {
let mut current = session_id
let visited : Array[String] = []
let mut limit = 0
while limit < 100 {
// Check if already visited
let mut already = false
for v in visited {
if v == current {
already = true
break
}
}
if already { break }
visited.push(current)
let meta = _sess_get_session_meta(sdb.db, current)
if meta.length() == 0 { break }
// meta[4] is parent_session_id
let parent = meta[4]
if parent == "" { break }
current = parent
limit = limit + 1
}
current
}
///| Parse an integer from `s`. Returns -1 if `s` is not a valid decimal.
///| Pure MoonBit via @string.parse_int, cross-target.
fn _parse_concurrency_js(s : String) -> Int {
(try? @string.parse_int(s.view())).unwrap_or(-1)
}
///| Max concurrent LLM summarizations in flight. Parity with proto's
///| MNEMO_SEARCH_CONCURRENCY default (3, clamped to 5).
fn _get_search_concurrency() -> Int {
let raw = @ffi.get_env("MNEMO_SEARCH_CONCURRENCY")
if raw == "" {
return 3
}
let n = _parse_concurrency_js(raw)
if n < 1 { 1 } else if n > 5 { 5 } else { n }
}
///| Carrier for per-session work that must survive the parallel await step.
priv struct SearchWork {
session_id : String
conversation_text : String
meta : Array[String]
}
///|
pub async fn session_search(
sdb : SessionDb,
query : String,
limit~ : Int = 3,
role_filter~ : String? = None
) -> SessionSearchResponse {
let fts_limit = srch_max(1, srch_min(limit, 5))
let q = query.trim().to_string()
let raw_results = search_messages(sdb, q, limit=50, role_filter~)
if raw_results.length() == 0 {
return { ok: true, query: q, results: [], count: 0, sessions_searched: 0 }
}
// Group by resolved root session, deduplicate, up to fts_limit
let seen_ids : Array[String] = []
let mut ri = 0
while ri < raw_results.length() && seen_ids.length() < fts_limit {
let result = raw_results[ri]
let resolved = srch_resolve_to_root(sdb, result.session_id)
let mut already = false
for s in seen_ids {
if s == resolved {
already = true
break
}
}
if !already {
seen_ids.push(resolved)
}
ri = ri + 1
}
// ── Gather per-session work (sync DB reads) ─────────────────────────────
let work : Array[SearchWork] = []
for sid in seen_ids {
let messages = get_messages(sdb, sid)
if messages.length() > 0 {
let meta = _sess_get_session_meta(sdb.db, sid)
let fmt_msgs : Array[MsgForFormat] = messages.map(fn(m) {
{ role: m.role, content: m.content, tool_name: m.tool_name, tool_calls: "" }
})
let conversation_text = truncate_around_matches(
format_conversation(fmt_msgs),
q,
)
work.push({ session_id: sid, conversation_text, meta })
}
}
// ── Fan out LLM summarization in chunks (parity with proto Semaphore) ───
let concurrency = _get_search_concurrency()
let summaries : Array[String] = []
let mut wi = 0
while wi < work.length() {
let end = srch_min(wi + concurrency, work.length())
let chunk : Array[@ffi.Promise[String]] = []
for j in wi.. 0 {
_format_timestamp(w.meta[3])
} else {
"unknown"
}
let source = if w.meta.length() > 0 { w.meta[1] } else { "unknown" }
let model = if w.meta.length() > 0 { w.meta[2] } else { "" }
results.push({ session_id: w.session_id, when, source, model, summary })
}
{ ok: true, query: q, results, count: results.length(), sessions_searched: seen_ids.length() }
}
// ── inline tests ──
///|
test "truncate: returns full text when short" {
let text = "hello world"
let result = truncate_around_matches(text, "hello")
assert_eq(result, text)
}
///|
test "truncate: adds later marker when no match and long text" {
let text = "x".repeat(200)
let result = truncate_around_matches(text, "notfound", max_chars=50)
assert_eq(result.contains("[later conversation truncated]"), true)
}
///|
test "truncate: phrase fallback finds phrase" {
let prefix = "x".repeat(1000)
let suffix = "y".repeat(1000)
let text = prefix + "target phrase" + suffix
let result = truncate_around_matches(text, "target phrase", max_chars=200)
assert_eq(result.contains("target phrase"), true)
}
///|
test "truncate: proximity fallback finds terms near each other" {
// Place "alpha" and "beta" close together, far from the start
let far = "z".repeat(2000)
let text = far + "alpha blah beta" + "z".repeat(2000)
let result = truncate_around_matches(text, "alpha beta", max_chars=500)
assert_eq(result.contains("alpha"), true)
assert_eq(result.contains("beta"), true)
}
///|
test "truncate: individual term fallback finds at least one term" {
// "foo" appears, "bar" does not
let far = "z".repeat(2000)
let text = far + "foo" + "z".repeat(2000)
let result = truncate_around_matches(text, "foo bar", max_chars=200)
assert_eq(result.contains("foo"), true)
}
///|
test "format_conversation: user message" {
let msgs : Array[MsgForFormat] = [
{ role: "user", content: "hello", tool_name: "", tool_calls: "" },
]
let result = format_conversation(msgs)
assert_eq(result, "[USER]: hello")
}
///|
test "format_conversation: assistant with content" {
let msgs : Array[MsgForFormat] = [
{ role: "assistant", content: "I can help", tool_name: "", tool_calls: "" },
]
let result = format_conversation(msgs)
assert_eq(result, "[ASSISTANT]: I can help")
}
///|
test "format_conversation: tool truncates long content" {
let long_content = "x".repeat(600)
let msgs : Array[MsgForFormat] = [
{ role: "tool", content: long_content, tool_name: "bash", tool_calls: "" },
]
let result = format_conversation(msgs)
assert_eq(result.contains("[truncated]"), true)
assert_eq(result.length() < long_content.length(), true)
}
///|
test "format_conversation: multiple messages joined" {
let msgs : Array[MsgForFormat] = [
{ role: "user", content: "hi", tool_name: "", tool_calls: "" },
{ role: "assistant", content: "hello", tool_name: "", tool_calls: "" },
]
let result = format_conversation(msgs)
assert_eq(result.contains("[USER]: hi"), true)
assert_eq(result.contains("[ASSISTANT]: hello"), true)
}
///|
async test "session_search: no results when db empty" {
let dir = _mkdtemp_sess()
let sdb = open_session_db(_path_join_2(dir, "s.db"))
let resp = session_search(sdb, "anything")
assert_eq(resp.ok, true)
assert_eq(resp.count, 0)
assert_eq(resp.sessions_searched, 0)
_rmrf_sess(dir)
}
///|
async test "session_search: finds session with matching message" {
let dir = _mkdtemp_sess()
let sdb = open_session_db(_path_join_2(dir, "s.db"))
let sid = create_session(sdb, "cli", "test system")
let _ = append_message(sdb, sid, "user", 1000L, content=Some("unique_srch_term_xyz"))
let resp = session_search(sdb, "unique_srch_term_xyz")
assert_eq(resp.ok, true)
assert_eq(resp.count > 0, true)
_rmrf_sess(dir)
}
///|
test "summarize falls back when env unset" {
// With no env vars set, _summarize_with_llm returns ""
// and the search result falls back to raw preview.
// We verify _summarize_with_llm returns "" when model/key are absent.
// (In CI neither MNEMO_SUMMARIZER_MODEL nor OPENROUTER_API_KEY are set.)
let model = @ffi.get_env("MNEMO_SUMMARIZER_MODEL")
let api_key = @ffi.get_env("OPENROUTER_API_KEY")
if model == "" || api_key == "" {
// Verify the fallback path: empty result from _summarize_with_llm
let result = _summarize_with_llm("test transcript")
assert_eq(result, "")
}
}
///|
test "summarize uses LLM when env set" {
if @ffi.get_env("MNEMO_LIVE_LLM_TEST") == "" {
return
}
// Live test: requires OPENROUTER_API_KEY + MNEMO_SUMMARIZER_MODEL
let model = @ffi.get_env("MNEMO_SUMMARIZER_MODEL")
let api_key = @ffi.get_env("OPENROUTER_API_KEY")
assert_eq(model != "", true)
assert_eq(api_key != "", true)
let transcript = "[USER]: How do I reverse a list in Python?\n[ASSISTANT]: You can use list.reverse() or reversed()."
let summary = _summarize_with_llm(transcript)
assert_eq(summary != "", true)
assert_eq(summary != "", true)
}