#include "sqlite/backend_impl.h"

#include "sqlite/agent_common.h"

#include "embedder.h"

#include "config.h"

#include "util/time.h"

#include "util/sqlite.h"

#include "sqlite/text_index.h"

#include <format>

#include "nlohmann_json.hpp"

#include <sqlite3.h>

#include <Eigen/Dense>

#include <algorithm>

#include <ctime>

#include <limits>

#include <map>

#include <optional>

#include <set>

#include <stdexcept>

#include <unordered_map>

#include <unordered_set>

namespace ragger::sqlite {

using json = nlohmann::json;

namespace {

std::string resolve_agent_time_bound(const std::string& s, bool end_of_day) {

}

} // anonymous namespace

std::vector<AgentHit> Backend::Impl::agent_search(const AgentSearchQuery& q) {

// ---- 1. validate ----------------------------------------------------

// PHASE7 input aliases: a layers entry "decision"/"insight" is accepted

// and means knowledge restricted to that kind (existing agents keep

// working; hits always report layer:"knowledge" + a kind).

std::set<std::string> want_layers;

std::set<std::string> kind_filter;

for (const auto& l : q.layers) {

if (l == "decision" || l == "insight") {

want_layers.insert("knowledge");

kind_filter.insert(l);

} else if (l == "summary") {

// PHASE8: the summary layer is gone; notes are knowledge rows.

throw std::invalid_argument(

"layer 'summary' no longer exists — notes are knowledge "

"rows, use layers:[\"knowledge\"] with kinds:[\"note\"]");

} else if (!is_known_agent_layer(l)) {

throw std::invalid_argument("unknown layer: '" + l + "'");

} else want_layers.insert(l);

}

for (const auto& k : q.kinds)

if (!status::find_kind(k))

throw std::invalid_argument("unknown kind: '" + k + "'");

if (!q.kinds.empty()) {

if (kind_filter.empty()) kind_filter.insert(q.kinds.begin(), q.kinds.end());

else {

// explicit kinds intersect the alias set when both are given

std::set<std::string> inter;

for (const auto& k : q.kinds)

if (kind_filter.count(k)) inter.insert(k);

kind_filter = inter;

}

want_layers.insert("knowledge");

}

const std::string since_b = q.since.empty() ? "" : resolve_agent_time_bound(q.since, false);

const std::string until_b = q.until.empty() ? "" : resolve_agent_time_bound(q.until, true);

const int limit = std::clamp(q.limit, 1, 50);

// ---- 2. wide-pool hybrid search --------------------------------------

// Degraded (no usable embedder) mirrors RaggerMemory::search's fallback:

// text-only scoring across the same four corpora.

const int pool = std::max(limit * 5, 50);

std::vector<SearchResult> cands;

if (embedder && embedder->ready())

cands = search(q.query, pool, 0.0f, {}).results;

else

cands = search_text_only(q.query, pool).results;

// Knowledge status comes from the rows, never from cache

// metadata: the hybrid cache only holds active rows and the text-only

// path carries no status at all, so one authoritative batched read per

// structured layer keeps every path consistent (see below).

// 'active' / exact-status needs rows the hybrid cache deliberately hides

// (superseded/retired/pending). SEPARATE query path — ensure_knowledge_cache()

// keeps its narrow allow-list for normal search.

// Lexical-only signal; ids already scored by the hybrid pass are not

// duplicated.

if (q.status != "active") {

text_index_->set_weights({config().fts_w_unigram, config().fts_w_bigram,

config().fts_w_metaphone, config().fts_literal_enabled});

auto supplement = [&](const char* table, const char* pk,

const char* source) {

std::unordered_set<int> seen;

for (const auto& r : cands)

if (r.metadata.value("source", "") == source) seen.insert(r.id);

for (const auto& hit : text_index_->score_query(table, q.query)) {

if (seen.count(hit.id)) continue;

Stmt s(db, std::format(

"SELECT text, status, ifnull(trigger,''), "

"datetime(created_at,'unixepoch','localtime') "

"FROM {} WHERE {} = ?", table, pk));

s.bind(1, hit.id);

if (!s.step()) continue;

const std::string status = s.column_text(1);

if (q.status != "any" && status != q.status) continue;

json meta = json::object();

meta["source"] = source;

meta["status"] = status;

SearchResult sr{hit.id, s.column_text(0), hit.score, std::move(meta), s.column_text(3)};

sr.vec_score = -1.0f;

sr.bm25_score = hit.score;

sr.blended = hit.score;

cands.push_back(std::move(sr));

}

};

supplement("knowledge", "knowledge_id", "knowledge");

}

// ---- session filter helpers ------------------------------------------

// GUID -> sessions.session_id once per filter (unknown GUID -> nullopt:

// `session` then matches nothing, `exclude_session` excludes nothing).

auto guid_to_id = [this](const std::string& guid) -> std::optional<int> {

Stmt s(db, "SELECT session_id FROM sessions WHERE guid = ?");

s.bind(1, guid);

if (!s.step()) return std::nullopt;

return s.column_int(0);

};

std::optional<int> sess_id, excl_id;

if (!q.session.empty()) sess_id = guid_to_id(q.session);

if (!q.exclude_session.empty()) excl_id = guid_to_id(q.exclude_session);

// ---- batched per-layer lookups (review 3A) ----------------------------

// One "WHERE <pk> IN (...)" query per layer per lookup instead of one

// prepared query per candidate hit (pool is up to 250 ids). Ids are

// DB-internal integers, so inlining them into the IN-list adds no

// injection surface and avoids bind-count limits. Rows that vanished

// mid-search simply stay absent from the maps.

std::map<std::string, std::vector<int>> ids_by_layer;

for (const auto& r : cands)

ids_by_layer[r.metadata.value("source", "")].push_back(r.id);

auto in_clause = [](const std::vector<int>& ids) {

std::string out;

for (int id : ids) {

if (!out.empty()) out += ',';

out += std::to_string(id);

}

return out;

};

// kind + status + trigger for the knowledge corpus (the trigger read was

// the same N+1 pattern as the status read).

std::unordered_map<int, std::string> kno_kind, kno_status, kno_trigger;

{

auto it = ids_by_layer.find("knowledge");

if (it != ids_by_layer.end() && !it->second.empty()) {

Stmt s(db, std::format(

"SELECT knowledge_id, kind, status, ifnull(trigger,'') "

"FROM knowledge WHERE knowledge_id IN ({})", in_clause(it->second)));

while (s.step()) {

kno_kind[s.column_int(0)] = s.column_text(1);

kno_status[s.column_int(0)] = s.column_text(2);

kno_trigger[s.column_int(0)] = s.column_text(3);

}

}

}

// Owning session per layer (documents have neither; the turn layer is

// never produced by the search corpora). Only fetched when a session

// filter is active.

std::map<std::string, std::unordered_map<int, int>> session_of;

auto fill_owning = [&](std::map<std::string, std::unordered_map<int, int>>& out,

const char* layer, const char* sql_prefix) {

auto it = ids_by_layer.find(layer);

if (it == ids_by_layer.end() || it->second.empty()) return;

Stmt s(db, std::format("{} IN ({})", sql_prefix, in_clause(it->second)));

while (s.step())

if (!s.is_null(1)) out[layer][s.column_int(0)] = s.column_int(1);

};

if (sess_id || excl_id) {

fill_owning(session_of, "turn_summary",

"SELECT turn_summary_id, session_id FROM turn_summaries WHERE turn_summary_id");

fill_owning(session_of, "knowledge",

"SELECT knowledge_id, session_id FROM knowledge WHERE knowledge_id");

}

// Session GUIDs for turn_summary extras — one query for the whole pool.

std::unordered_map<int, std::string> guid_of_session;

{

std::vector<int> sids;

for (const auto& r : cands)

if (r.metadata.value("source", "") == "turn_summary" &&

r.metadata.contains("session_id"))

sids.push_back(r.metadata["session_id"].get<int>());

if (!sids.empty()) {

Stmt s(db, std::format(

"SELECT session_id, guid FROM sessions WHERE session_id IN ({})",

in_clause(sids)));

while (s.step()) guid_of_session[s.column_int(0)] = s.column_text(1);

}

}

// ---- 3. post-filter + shape ------------------------------------------

std::vector<AgentHit> hits;

for (const auto& r : cands) {

const std::string layer = r.metadata.value("source", "");

if (!want_layers.empty() && !want_layers.count(layer))

continue;

std::string row_status, row_kind;

if (layer == "knowledge") {

auto it = kno_status.find(r.id);

if (it == kno_status.end() || it->second.empty())

continue; // row vanished mid-search

row_status = it->second;

row_kind = kno_kind[r.id];

if (!kind_filter.empty() && !kind_filter.count(row_kind))

continue;

if (q.status == "active") {

// NOTE: deliberately narrower than the search cache's

// allow-list: 'roadmap' is planned work, not an active ruling.

if (!status::is_in(status::kSearchActive, row_status)) continue;

} else if (q.status != "any" && row_status != q.status) {

continue;

}

}

if (!since_b.empty() && r.timestamp < since_b) continue;

if (!until_b.empty() && r.timestamp > until_b) continue;

if (sess_id || excl_id) {

std::optional<int> sid;

auto lit = session_of.find(layer);

if (lit != session_of.end()) {

auto it = lit->second.find(r.id);

if (it != lit->second.end()) sid = it->second;

}

if (sess_id && (!sid || *sid != *sess_id)) continue;

if (excl_id && sid && *sid == *excl_id) continue;

}

AgentHit h;

h.id = layer + ":" + std::to_string(r.id);

h.layer = layer;

h.score = r.blended; // the value that actually ranked this hit

if (r.vec_score > 0.0f && r.bm25_score > 0.0f) h.matched_by = "both";

else if (r.bm25_score > 0.0f) h.matched_by = "lexical";

else h.matched_by = "vector";

h.created = r.timestamp;

h.snippet = agent_snippet(r.text);

if (layer == "knowledge") {

h.extra = json{{"kind", row_kind}, {"status", row_status}};

auto it = kno_trigger.find(r.id);

if (it != kno_trigger.end() && !it->second.empty())

h.extra["trigger"] = it->second;

} else if (layer == "turn_summary") {

h.extra = json::object();

if (r.metadata.contains("turn_id"))

h.extra["turn_id"] = r.metadata["turn_id"];

if (r.metadata.contains("session_id")) {

auto it = guid_of_session.find(r.metadata["session_id"].get<int>());

h.extra["session"] = it != guid_of_session.end()

? it->second : std::string();

}

} else if (layer == "document") {

h.extra = json{{"title", r.metadata.value("title", "")}};

}

hits.push_back(std::move(h));

}

// ---- 4. rank, budget, truncate ----------------------------------------

std::sort(hits.begin(), hits.end(),

[](const AgentHit& a, const AgentHit& b) { return a.score > b.score; });

std::vector<AgentHit> out;

long long token_sum = 0; // running chars/4 over accepted snippets

for (auto& h : hits) {

if (static_cast<int>(out.size()) >= limit) break;

// Budget check happens BEFORE adding, but the first accepted hit is

// always kept ("always return at least 1 hit if any matched").

if (q.max_tokens > 0 && !out.empty() &&

(token_sum + static_cast<long long>(h.snippet.size())) / 4 > q.max_tokens)

break;

token_sum += static_cast<long long>(h.snippet.size());

out.push_back(std::move(h));

}

return out;

}

namespace {

void parse_timeline_cursor(const std::string& cursor, long long& epoch, long long& rowid) {

}

long long bound_to_epoch(const std::string& bound) {

}

} // anonymous namespace

json Backend::Impl::agent_timeline(const TimelineQuery& q) {

static const std::set<std::string> kGranularities = {"turn", "episode", "session"};

if (!kGranularities.count(q.granularity))

throw std::invalid_argument("granularity must be one of: turn, episode, session");

const int limit = std::clamp(q.limit, 1, 100);

long long since_e = 0, until_e = 0;

if (!q.since.empty()) since_e = bound_to_epoch(resolve_agent_time_bound(q.since, false));

if (!q.until.empty()) until_e = bound_to_epoch(resolve_agent_time_bound(q.until, true));

std::optional<long long> cur_epoch, cur_row;

if (!q.cursor.empty()) {

long long e = 0, r = 0;

parse_timeline_cursor(q.cursor, e, r);

cur_epoch = e;

cur_row = r;

}

// Session filter: unknown GUID matches nothing (nullopt).

std::optional<int> sess_id;

if (!q.session.empty()) {

Stmt s(db, "SELECT session_id FROM sessions WHERE guid = ?");

s.bind(1, q.session);

if (s.step()) sess_id = s.column_int(0);

}

// Each granularity builds one SELECT over (created_epoch, rowid, text,

// session_guid) with the same filter/cursor plumbing, ordered newest

// first. The continuation predicate is strict "<" on the (created, rowid)

// pair — never OFFSET. `first_expr` is the created-epoch expression; it

// must match the first projected column so the cursor and the ORDER BY

// agree.

std::string sql;

std::string first_expr, row_expr, since_expr;

std::vector<std::pair<int, long long>> int_binds; // positional binds (1-based)

if (q.granularity == "turn") {

// One entry per turn: its turn_summary text when one exists, else the

// raw user text (turns that haven't been summarized yet). Keyed by

// (turn_datetime, turn_id) — turn_summaries.turn_id is unique, and

// fallback rows are exactly the turns without a summary row.

first_expr =

"COALESCE((SELECT ts.turn_datetime FROM turn_summaries ts "

"WHERE ts.turn_id = t.turn_id AND ts.text != ''), t.created_at)";

row_expr = "t.turn_id";

since_expr = first_expr;

sql =

"SELECT " + first_expr + ", " + row_expr + ", "

" COALESCE((SELECT ts.text FROM turn_summaries ts "

" WHERE ts.turn_id = t.turn_id AND ts.text != ''), t.user_text), "

" COALESCE((SELECT guid FROM sessions WHERE session_id = t.session_id), '') "

"FROM turns t "

"WHERE (t.user_text != '' OR EXISTS "

" (SELECT 1 FROM turn_summaries ts WHERE ts.turn_id = t.turn_id AND ts.text != ''))";

if (sess_id) {

sql += " AND t.session_id = ?";

int_binds.push_back({static_cast<int>(int_binds.size()) + 1, *sess_id});

}

} else if (q.granularity == "episode") {

// PHASE5 C.3: episodes live in their own table now. Rows ordered by

// span_start; a date filter matches on SPAN OVERLAP (span_end >=

// since AND span_start <= until), not on span start alone.

first_expr = "sm.span_start";

row_expr = "sm.episode_id";

since_expr = "sm.span_end";

sql =

"SELECT sm.span_start, sm.episode_id, sm.text, "

" COALESCE((SELECT guid FROM sessions WHERE session_id = sm.session_id), '') "

"FROM episodes sm WHERE sm.text != ''";

if (sess_id) {

sql += " AND sm.session_id = ?";

int_binds.push_back({static_cast<int>(int_binds.size()) + 1, *sess_id});

}

} else { // session

// PHASE5 C.3: one row per session in range — never reads summaries

// (there are no session rollups at 0.17). started_on/ended_on are the

// derived turn span (NULL for a session with no turns). Date filters

// match on span overlap, same rule as episodes.

first_expr = "COALESCE(s.started_on, s.ended_on, 0)";

row_expr = "s.session_id";

since_expr = "COALESCE(s.ended_on, s.started_on, 0)";

sql =

"SELECT " + first_expr + ", " + row_expr + ", "

" s.guid, s.name, "

" datetime(s.started_on,'unixepoch','localtime'), "

" datetime(s.ended_on,'unixepoch','localtime'), "

" (SELECT COUNT(*) FROM episodes e WHERE e.session_id = s.session_id), "

" COALESCE((SELECT e.text FROM episodes e "

" WHERE e.session_id = s.session_id "

" ORDER BY e.span_start DESC, e.episode_id DESC LIMIT 1), '') "

"FROM sessions s WHERE 1=1";

if (sess_id) {

sql += " AND s.session_id = ?";

int_binds.push_back({static_cast<int>(int_binds.size()) + 1, *sess_id});

}

}

// Time bounds and the cursor apply to the first projected column pair

// (created epoch, rowid) — one continuation predicate for all three

// granularities: strictly older in (created, rowid) order. The `since`

// bound uses since_expr (span end for episodes/sessions — overlap).

if (since_e) sql += " AND " + since_expr + " >= ?";

if (until_e) sql += " AND " + first_expr + " <= ?";

if (cur_epoch)

sql += " AND (" + first_expr + " < ? OR (" + first_expr + " = ? AND " + row_expr + " < ?))";

sql += " ORDER BY 1 DESC, 2 DESC LIMIT ?";

Stmt s(db, sql);

int bind = 1;

for (auto& [idx, v] : int_binds) { (void)idx; s.bind(bind++, v); }

if (since_e) s.bind(bind++, since_e);

if (until_e) s.bind(bind++, until_e);

if (cur_epoch) {

s.bind(bind++, *cur_epoch);

s.bind(bind++, *cur_epoch);

s.bind(bind++, *cur_row);

}

s.bind(bind, limit + 1); // one extra row decides next_cursor

json items = json::array();

long long last_epoch = 0, last_row = 0;

bool more = false;

int rows = 0;

while (s.step()) {

++rows;

if (rows > limit) { more = true; break; }

last_epoch = s.column_int64(0);

last_row = s.column_int64(1);

if (q.granularity == "session") {

// PHASE5 C.3 shape: the session itself, its derived span, and a

// snippet of its newest episode ("" when it has none).

items.push_back({

{"session_id", last_row},

{"guid", s.column_text(2)},

{"name", s.is_null(3) ? json(nullptr) : json(s.column_text(3))},

{"started_on", s.is_null(4) ? json(nullptr) : json(s.column_text(4))},

{"ended_on", s.is_null(5) ? json(nullptr) : json(s.column_text(5))},

{"episodes", s.column_int(6)},

{"snippet", agent_snippet(s.column_text(7))},

});

continue;

}

const std::string text = s.column_text(2);

json item = {

// Turn-granularity entries are keyed by turn_id (the summary is

// looked up per turn), so the handle is turn:<id> — resolvable

// by agent_get even when the entry shows summary text.

{"id", q.granularity == "turn"

? std::format("turn:{}", last_row)

: std::format("episode:{}", last_row)},

{"created", db_timestamp(static_cast<std::time_t>(last_epoch))},

{"session", s.column_text(3)},

{"text", agent_snippet(text)},

};

items.push_back(std::move(item));

}

return json{

{"items", std::move(items)},

{"next_cursor", more

? json(std::format("{}:{}", last_epoch, last_row))

: json(nullptr)},

};

}

json Backend::Impl::agent_get(const AgentGetQuery& q) {

const bool has_ids = !q.ids.empty();

const bool has_sess = !q.session.empty();

if (has_ids == has_sess)

throw std::invalid_argument(

"agent_get: exactly one of ids or session must be set");

// ---- mode 1: full rows by handle -------------------------------------

if (has_ids) {

json items = json::array();

for (const auto& handle : q.ids) {

// PHASE7 (TODO Q3 default): typed handles decision:N / insight:N

// are rejected — knowledge:N is the only form. PHASE8: summary:N

// too (notes are knowledge rows).

if (is_legacy_knowledge_handle(handle))

throw std::invalid_argument("use knowledge:<id>");

if (is_legacy_summary_handle(handle))

throw std::invalid_argument(

"use knowledge:<id> (notes are knowledge rows)");

std::string layer;

int rowid = 0;

if (!parse_agent_handle(handle, layer, rowid)) {

items.push_back({{"id", handle}, {"error", "not found"}});

continue;

}

json item = fetch_agent_row(db, layer, rowid);

if (item.is_null()) {

items.push_back({{"id", handle}, {"error", "not found"}});

continue;

}

if (q.expand == "context") {

json expanded = json::array();

if (layer == "turn" || layer == "turn_summary") {

// The 2 turns before and after in the same session. A

// turn_summary anchors on its source turn; a pruned

// (turn_id NULL) summary or an anonymous turn has no

// context window.

int anchor = rowid;

if (layer == "turn_summary") {

Stmt t(db, "SELECT turn_id FROM turn_summaries WHERE turn_summary_id = ?");

t.bind(1, rowid);

if (!t.step() || t.is_null(0)) anchor = -1;

else anchor = t.column_int(0);

}

if (anchor >= 0) {

Stmt c(db, "SELECT turn_id FROM turns "

"WHERE session_id = (SELECT session_id FROM turns WHERE turn_id = ?) "

" AND session_id IS NOT NULL "

" AND turn_id BETWEEN ? AND ? AND turn_id != ? "

"ORDER BY turn_id");

c.bind(1, anchor);

c.bind(2, anchor - 2);

c.bind(3, anchor + 2);

c.bind(4, anchor);

while (c.step())

expanded.push_back(fetch_agent_row(db, "turn", c.column_int(0)));

}

} else if (layer == "document") {

// chunk_index ±1 from the same document_source_id.

Stmt c(db, "SELECT document_id FROM documents "

"WHERE document_source_id = ("

" SELECT document_source_id FROM documents WHERE document_id = ?) "

" AND document_source_id IS NOT NULL "

" AND chunk_index IN ("

" (SELECT chunk_index FROM documents WHERE document_id = ?) - 1, "

" (SELECT chunk_index FROM documents WHERE document_id = ?) + 1) "

"ORDER BY chunk_index");

c.bind(1, rowid);

c.bind(2, rowid);

c.bind(3, rowid);

while (c.step())

expanded.push_back(fetch_agent_row(db, "document", c.column_int(0)));

}

if (!expanded.empty()) item["expanded"] = std::move(expanded);

} else if (q.expand == "sources") {

json expanded = json::array();

if (layer == "knowledge") {

// Its evidence turn_summaries, full rows. The join filter

// keeps done-marker rows (text='') out of expansion even

// if a source turn was later abandoned (RULES 6).

Stmt e(db,

"SELECT de.turn_summary_id FROM knowledge_turn_evidence de "

"JOIN turn_summaries ts ON ts.turn_summary_id = de.turn_summary_id "

"WHERE de.knowledge_id = ? AND ts.text != '' "

"ORDER BY de.turn_summary_id");

e.bind(1, rowid);

while (e.step())

expanded.push_back(fetch_agent_row(db, "turn_summary", e.column_int(0)));

} else if (layer == "episode") {

// Recorded membership (PHASE5 C.2): replaces the old

// turn_datetime BETWEEN span heuristic, which

// mis-assigned turn summaries on same-second boundaries.

Stmt c(db, "SELECT turn_summary_id FROM turn_summaries "

"WHERE episode_id = ? AND text != '' "

"ORDER BY turn_datetime, turn_summary_id");

c.bind(1, rowid);

while (c.step())

expanded.push_back(fetch_agent_row(db, "turn_summary", c.column_int(0)));

}

// PHASE8: the old `summary` expand branch is gone with the

// layer; notes carry their evidence links like any knowledge row.

if (!expanded.empty()) item["expanded"] = std::move(expanded);

} else if (q.expand == "chain" && layer == "knowledge") {

// Full supersede lineage oldest -> newest: walk `supersedes`

// backwards to the root, then forwards from the anchor. The

// visited set makes a corrupt cycle terminate. PHASE7: the

// chain may cross kinds (insight -> decision promotion).

std::vector<int> chain_ids;

std::unordered_set<int> visited{rowid};

{

std::vector<int> up;

int cur = rowid;

while (true) {

Stmt p(db, "SELECT supersedes FROM knowledge WHERE knowledge_id = ?");

p.bind(1, cur);

if (!p.step() || p.is_null(0)) break;

int parent = p.column_int(0);

if (visited.count(parent)) break;

visited.insert(parent);

up.push_back(parent);

cur = parent;

}

std::reverse(up.begin(), up.end());

chain_ids = std::move(up);

chain_ids.push_back(rowid);

}

{

std::vector<int> queue{rowid};

while (!queue.empty()) {

std::vector<int> next;

for (int cur : queue) {

Stmt c(db, "SELECT knowledge_id FROM knowledge WHERE supersedes = ? "

"ORDER BY knowledge_id");

c.bind(1, cur);

while (c.step()) {

int child = c.column_int(0);

if (visited.count(child)) continue;

visited.insert(child);

chain_ids.push_back(child);

next.push_back(child);

}

}

queue = std::move(next);

}

}

json expanded = json::array();

for (int id : chain_ids)

expanded.push_back(fetch_agent_row(db, layer, id));

item["expanded"] = std::move(expanded);

}

items.push_back(std::move(item));

}

return json{{"items", std::move(items)}};

}

// ---- mode 2: paged conversation history for one session --------------

// "current" is NOT resolved here — the HTTP/MCP layer passes a real GUID.

const int limit = std::clamp(q.limit, 1, 100);

std::string sql =

"SELECT t.turn_id, t.user_text, t.assistant_text, "

" datetime(t.created_at,'unixepoch','localtime'), "

" (SELECT text FROM turn_summaries ts "

" WHERE ts.turn_id = t.turn_id AND ts.text != '' LIMIT 1) "

"FROM turns t JOIN sessions s ON s.session_id = t.session_id "

"WHERE s.guid = ?";

if (q.after_turn > 0) sql += " AND t.turn_id > ?";

if (q.before_turn > 0) sql += " AND t.turn_id < ?";

sql += " ORDER BY t.turn_id ASC LIMIT ?";

Stmt s(db, sql);

int bind = 1;

s.bind(bind++, q.session);

if (q.after_turn > 0) s.bind(bind++, q.after_turn);

if (q.before_turn > 0) s.bind(bind++, q.before_turn);

s.bind(bind, limit + 1); // one extra row tells us whether more exist

json turns = json::array();

int last_turn_id = 0;

bool more = false;

int rows = 0;

while (s.step()) {

++rows;

if (rows > limit) { more = true; break; } // the probe row exists

json t = {

{"id", std::format("turn:{}", s.column_int(0))},

{"user", s.column_text(1)},

{"assistant", s.is_null(2) ? json(nullptr) : json(s.column_text(2))},

{"created", s.column_text(3)},

{"summary", s.is_null(4) ? json(nullptr) : json(s.column_text(4))},

};

last_turn_id = s.column_int(0);

turns.push_back(std::move(t));

}

return json{

{"session", q.session},

{"turns", std::move(turns)},

{"next_after", more ? json(last_turn_id) : json(nullptr)},

};

}

} // namespace ragger::sqlite