import albedo/daemon/events
import albedo/daemon/image_fit
import albedo/daemon/images
import albedo/daemon/mail
import albedo/daemon/migrations/conversation_columns
import albedo/daemon/note
import albedo/daemon/notice
import albedo/daemon/operations
import albedo/daemon/requests
import albedo/daemon/store
import albedo/daemon/transcript
import albedo/daemon/usage
import albedo/harness/cache_fade
import albedo/harness/extensions/python/cells as journal
import albedo/openai_api/types
import gleam/bool
import gleam/dynamic/decode
import gleam/int
import gleam/json
import gleam/list
import gleam/option.{type Option, None, Some}
import gleam/result
import gleam/string
import sqlight
pub type Info {
Info(
id: String,
title: String,
cwd: String,
provider: String,
model: String,
protocol: types.Protocol,
stage: Stage,
last_assistant_at: Option(Int),
effort: Option(String),
)
}
/// Where a session's turn stood at its last commit. A daemon restart resumes
/// a session left at `Model` or `Tool`; `Interrupted` is a turn that ended
/// cancelled or failed and is not resumed.
pub type Stage {
Idle
Model
Tool
Interrupted
}
/// The stored and reported spelling of a stage.
pub fn stage_name(stage: Stage) -> String {
case stage {
Idle -> "idle"
Model -> "model"
Tool -> "tool"
Interrupted -> "interrupted"
}
}
/// An unrecognised stored stage is treated as a turn that did not finish.
fn parse_stage(name: String) -> Stage {
case name {
"idle" -> Idle
"model" -> Model
"tool" -> Tool
_ -> Interrupted
}
}
/// A restart picks up a turn that was waiting on the model or a tool.
pub fn resumable(stage: Stage) -> Bool {
stage == Model || stage == Tool
}
pub fn initialise(store: store.Store) -> Result(Nil, String) {
store.query(store, fn(db) {
use _ <- result.try(store.exec(
db,
"CREATE TABLE IF NOT EXISTS sessions(id TEXT PRIMARY KEY,title TEXT NOT NULL DEFAULT 'new session',cwd TEXT NOT NULL,model TEXT NOT NULL,protocol TEXT NOT NULL,stage TEXT NOT NULL DEFAULT 'idle',provider TEXT,activity_seq INTEGER,last_assistant_at INTEGER,usage_model TEXT,usage_recorded_at INTEGER,usage_prompt_tokens INTEGER,usage_completion_tokens INTEGER,usage_cached_prompt_tokens INTEGER,usage_cache_creation_tokens INTEGER,effort TEXT); CREATE TABLE IF NOT EXISTS transcript(seq INTEGER PRIMARY KEY AUTOINCREMENT,session TEXT NOT NULL REFERENCES sessions(id),payload BLOB NOT NULL,timestamp INTEGER,provider TEXT,thought_ms INTEGER,row_class TEXT CHECK(row_class IN ('user','image_fit','other'))); CREATE INDEX IF NOT EXISTS transcript_session ON transcript(session,seq);"
<> images.schema
<> requests.schema
<> operations.schema,
))
use _ <- result.try(conversation_columns.apply(db))
use _ <- result.try(recover_sessions(db))
store.exec(
db,
"CREATE INDEX IF NOT EXISTS sessions_activity ON sessions(activity_seq DESC)",
)
})
}
fn recovery_decoder() -> decode.Decoder(#(String, String, Int)) {
use id <- decode.field(0, decode.string)
use title <- decode.field(1, decode.string)
use activity_seq <- decode.field(2, decode.int)
decode.success(#(id, title, activity_seq))
}
fn recover_sessions(db: sqlight.Connection) -> Result(Nil, String) {
use sessions <- result.try(store.rows(
db,
"SELECT id,COALESCE(title,''),COALESCE(activity_seq,-1) FROM sessions WHERE activity_seq IS NULL OR title IS NULL OR title=''",
[],
recovery_decoder(),
))
list.try_each(sessions, fn(session) {
let #(id, saved_title, activity_seq) = session
use last_seq <- result.try(store.one(
db,
"SELECT COALESCE(MAX(seq),0) FROM transcript WHERE session=?",
[sqlight.text(id)],
decode.field(0, decode.int, decode.success),
"could not recover session activity",
))
use recovered_title <- result.try(
case
saved_title == ""
|| { saved_title == "new session" && activity_seq < 0 }
{
False -> Ok(saved_title)
True ->
store.rows(
db,
"SELECT payload FROM transcript WHERE session=? ORDER BY seq",
[sqlight.text(id)],
decode.field(0, decode.bit_array, decode.success),
)
|> result.map(fn(rows) {
rows
|> list.filter_map(unpack(_, unread))
|> latest_user
|> title_or_default
})
},
)
store.run(
db,
"UPDATE sessions SET title=?,activity_seq=CASE WHEN activity_seq IS NULL THEN ? ELSE activity_seq END WHERE id=?",
[
sqlight.text(recovered_title),
sqlight.int(last_seq),
sqlight.text(id),
],
)
})
}
pub fn assign_provider(
store: store.Store,
provider: String,
) -> Result(Nil, String) {
store.write(
store,
"UPDATE sessions SET provider=? WHERE provider IS NULL OR provider=''",
[sqlight.text(provider)],
)
}
pub fn assign_session_provider(
store: store.Store,
id: String,
provider: String,
) -> Result(Nil, String) {
store.write(
store,
"UPDATE sessions SET provider=? WHERE id=? AND (provider IS NULL OR provider='')",
[sqlight.text(provider), sqlight.text(id)],
)
}
/// What `info_decoder` reads. A name someone gave the session outranks the
/// title its latest message suggests.
pub const info_columns = "id,COALESCE(NULLIF(name,''),NULLIF(title,''),'new session'),cwd,COALESCE(provider,''),model,protocol,stage,last_assistant_at,effort"
/// Decodes `info_columns`.
pub fn info_decoder() -> decode.Decoder(Info) {
use id <- decode.field(0, decode.string)
use title <- decode.field(1, decode.string)
use cwd <- decode.field(2, decode.string)
use provider <- decode.field(3, decode.string)
use model <- decode.field(4, decode.string)
use protocol <- decode.field(5, decode.string)
use stage <- decode.field(6, decode.string)
use last_assistant_at <- decode.field(7, decode.optional(decode.int))
use effort <- decode.field(8, decode.optional(decode.string))
decode.success(Info(
id,
title,
cwd,
provider,
model,
case protocol {
"responses" -> types.Responses
_ -> types.ChatCompletions
},
parse_stage(stage),
last_assistant_at,
effort,
))
}
pub fn list(store: store.Store) -> Result(List(Info), String) {
store.read(
store,
"SELECT "
<> info_columns
<> " FROM sessions ORDER BY activity_seq DESC,rowid DESC",
[],
info_decoder(),
)
}
pub fn get(store: store.Store, id: String) -> Result(Info, String) {
store.query(store, read_info(_, id))
}
/// `get` inside a query the caller already holds.
pub fn read_info(db: sqlight.Connection, id: String) -> Result(Info, String) {
store.one(
db,
"SELECT " <> info_columns <> " FROM sessions WHERE id=?",
[sqlight.text(id)],
info_decoder(),
"session not found",
)
}
/// Give a session a name that its messages no longer retitle. A blank name
/// hands the title back to the latest message. Activity order is untouched.
pub fn rename(
store: store.Store,
id: String,
name: String,
) -> Result(Info, String) {
let name = case excerpt(name, 80) {
"" -> None
clean -> Some(clean)
}
store.query(store, fn(db) {
use _ <- result.try(
store.run(db, "UPDATE sessions SET name=? WHERE id=?", [
sqlight.nullable(sqlight.text, name),
sqlight.text(id),
]),
)
read_info(db, id)
})
}
/// The name someone gave `id`, if any.
pub fn given_name(store: store.Store, id: String) -> Option(String) {
store.read(
store,
"SELECT name FROM sessions WHERE id=? AND name<>''",
[sqlight.text(id)],
decode.field(0, decode.string, decode.success),
)
|> result.unwrap([])
|> list.first
|> option.from_result
}
/// Permanently remove a session and its dependent records in one transaction.
/// The tables here are the daemon's own; each extension's `cleaners` delete
/// what it keeps for the session, before the session row goes.
pub fn delete(
store: store.Store,
id: String,
cleaners: List(fn(sqlight.Connection, String) -> Result(Nil, String)),
) -> Result(Nil, String) {
store.query(store, fn(db) {
store.transaction(db, fn() {
use _ <- result.try(operations.cancel_in(db, id))
use hashes <- result.try(images.session_hashes(db, id))
use tables <- result.try(store.rows(
db,
"SELECT name FROM sqlite_master WHERE type='table'",
[],
decode.field(0, decode.string, decode.success),
))
use cell_hashes <- result.try(case list.contains(tables, "cells") {
True -> journal.session_hashes(db, id)
False -> Ok([])
})
// Cell rows stay here: deleting them feeds the image reference count
// released below, in this transaction.
use _ <- result.try(
list.try_each(
[
"transcript",
"submission_events",
"continuation_markers",
"provider_requests",
"session_extensions",
"cells",
],
fn(table) {
case list.contains(tables, table) {
False -> Ok(Nil)
True -> {
use _ <- result.try(case table {
"cells" ->
store.run(
db,
"DELETE FROM cell_traces WHERE id IN (SELECT id FROM cells WHERE session=?)",
[sqlight.text(id)],
)
_ -> Ok(Nil)
})
store.run(db, "DELETE FROM " <> table <> " WHERE session=?", [
sqlight.text(id),
])
}
}
},
),
)
use _ <- result.try(list.try_each(cleaners, fn(clean) { clean(db, id) }))
use _ <- result.try(
store.run(db, "DELETE FROM sessions WHERE id=?", [sqlight.text(id)]),
)
images.release(db, list.append(hashes, cell_hashes) |> list.unique)
})
})
}
pub fn create(store: store.Store, info: Info) -> Result(Nil, String) {
store.query(store, create_in(_, info))
}
pub fn create_operation(
ledger: store.Store,
info: Info,
request: operations.Request,
http_status: Int,
response: String,
) -> Result(operations.Receipt, String) {
operations.admit(
ledger,
operations.Request(..request, target: info.id),
http_status,
response,
None,
create_in(_, info),
)
}
fn create_in(db: sqlight.Connection, info: Info) -> Result(Nil, String) {
store.run(
db,
"INSERT INTO sessions(id,title,cwd,provider,model,protocol,activity_seq,last_assistant_at,effort) SELECT ?,?,?,?,?,?,COALESCE(MAX(activity_seq),0)+1,?,? FROM sessions",
[
sqlight.text(info.id),
sqlight.text(info.title),
sqlight.text(info.cwd),
sqlight.text(info.provider),
sqlight.text(info.model),
sqlight.text(protocol(info.protocol)),
sqlight.nullable(sqlight.int, info.last_assistant_at),
sqlight.nullable(sqlight.text, info.effort),
],
)
}
pub fn protocol(protocol: types.Protocol) -> String {
types.protocol_name(protocol)
}
pub fn title(text: String) -> String {
case excerpt(text, 80) {
"" -> "new session"
clean -> clean
}
}
/// One line of display-safe text: control and invisible characters become
/// spaces, runs of whitespace collapse, and anything past `limit` graphemes is
/// cut with an ellipsis.
pub fn excerpt(text: String, limit: Int) -> String {
let assert Ok(space) = string.utf_codepoint(32)
let clean =
text
|> string.to_utf_codepoints
|> list.map(fn(codepoint) {
let value = string.utf_codepoint_to_int(codepoint)
case
value <= 31
|| { value >= 127 && value <= 159 }
|| value == 173
|| value == 8203
|| { value >= 8206 && value <= 8207 }
|| { value >= 8232 && value <= 8238 }
|| { value >= 8288 && value <= 8297 }
|| value == 65_279
{
True -> space
False -> codepoint
}
})
|> string.from_utf_codepoints
|> string.trim
|> string.split(" ")
|> list.filter(fn(part) { part != "" })
|> string.join(" ")
case string.length(clean) > limit {
True -> string.slice(clean, 0, limit - 1) <> "…"
False -> clean
}
}
fn has_visible_assistant(inputs: List(types.Input)) -> Bool {
list.any(inputs, fn(input) {
option.is_some(events.visible_assistant_text(input))
})
}
/// Whether one committed input is a row the model itself produced: an
/// assistant message or a replayed provider item, including a tool call. The
/// first such row of a commit is the transcript row that commit links to.
fn is_assistant_row(input: types.Input) -> Bool {
case input {
types.Replay(_) | types.Assistant(_) -> True
_ -> False
}
}
/// The newest message a person wrote. Notes are the daemon talking and mail
/// comes from other agents or outside, so neither names a session.
pub fn latest_user(inputs: List(types.Input)) -> Option(String) {
list.fold(inputs, None, fn(latest, input) {
case input {
types.User(text) | types.UserImage(text, _) ->
case notice.is_notice(text), note.parse(text), mail.is_mail(text) {
False, None, False -> Some(text)
_, _, _ -> latest
}
_ -> latest
}
})
}
/// The title `latest_user` suggests: its text, or the default title.
pub fn title_or_default(user: Option(String)) -> String {
case user {
Some(text) -> title(text)
None -> "new session"
}
}
pub fn load_entries(
store: store.Store,
id: String,
) -> Result(List(transcript.Entry), String) {
load_sources(store, id)
|> result.map(fn(rows) { list.map(rows, fn(row) { row.entry }) })
}
/// Captured append boundary shared by metadata and range reads.
pub type Snapshot {
Snapshot(session: String, upper: Int)
}
pub fn snapshot(
ledger: store.Store,
session: String,
) -> Result(Snapshot, String) {
last_seq(ledger, session)
|> result.map(fn(upper) { Snapshot(session, upper) })
}
pub type SourceStats {
SourceStats(latest_user: Option(Int), uncovered: Bool, uncovered_users: Int)
}
pub fn source_stats(
ledger: store.Store,
snapshot: Snapshot,
covered: Int,
) -> Result(SourceStats, String) {
store.query(ledger, fn(db) {
use latest <- result.try(store.one(
db,
"SELECT MAX(seq) FROM transcript WHERE session=? AND seq<=? AND row_class IN ('user','image_fit')",
[sqlight.text(snapshot.session), sqlight.int(snapshot.upper)],
decode.field(0, decode.optional(decode.int), decode.success),
"transcript user boundary",
))
use count <- result.try(store.one(
db,
"SELECT COUNT(*) FROM transcript WHERE session=? AND seq>? AND seq<=? AND row_class IN ('user','image_fit')",
[
sqlight.text(snapshot.session),
sqlight.int(covered),
sqlight.int(snapshot.upper),
],
decode.field(0, decode.int, decode.success),
"transcript user count",
))
use uncovered <- result.try(store.one(
db,
"SELECT EXISTS(SELECT 1 FROM transcript WHERE session=? AND seq>? AND seq<=?)",
[
sqlight.text(snapshot.session),
sqlight.int(covered),
sqlight.int(snapshot.upper),
],
decode.field(0, decode.int, decode.success),
"transcript uncovered rows",
))
Ok(SourceStats(latest, uncovered == 1, count))
})
}
pub type FoldStep(a) {
Continue(a)
Stop(a)
}
type FitSchedule =
List(#(Int, image_fit.Replacements))
/// Chronological range fold. Stop avoids reading another page; later image
/// fits up to the captured boundary still apply to earlier selected images.
pub fn fold_sources(
ledger: store.Store,
snapshot: Snapshot,
first: Int,
last: Int,
acc: a,
step: fn(a, transcript.SourcedEntry) -> FoldStep(a),
) -> Result(a, String) {
use <- bool.guard(first > int.min(last, snapshot.upper), Ok(acc))
let read = images.reader(ledger)
use fits <- result.try(
store.read(
ledger,
"SELECT seq,payload FROM transcript WHERE session=? AND seq>=? AND seq<=? AND row_class='image_fit' ORDER BY seq DESC",
[
sqlight.text(snapshot.session),
sqlight.int(first),
sqlight.int(snapshot.upper),
],
{
use seq <- decode.field(0, decode.int)
use payload <- decode.field(1, decode.bit_array)
decode.success(#(seq, payload))
},
),
)
use #(schedule, replacements) <- result.try(
list.try_fold(fits, #([], image_fit.replacements()), fn(acc, row) {
use fit <- result.try(
unpack_fit(row.1, read) |> result.replace_error("corrupt image fit row"),
)
let replacements = image_fit.add(acc.1, fit)
Ok(#([#(row.0, acc.1), ..acc.0], replacements))
}),
)
fold_source_pages(
ledger,
snapshot,
first - 1,
int.min(last, snapshot.upper),
acc,
step,
schedule,
replacements,
)
}
fn fold_source_pages(
ledger: store.Store,
snapshot: Snapshot,
after: Int,
upper: Int,
acc: a,
step: fn(a, transcript.SourcedEntry) -> FoldStep(a),
schedule: FitSchedule,
replacements: image_fit.Replacements,
) -> Result(a, String) {
let read = images.reader(ledger)
use rows <- result.try(
store.query(ledger, fn(db) {
use rows <- result.try(store.rows(
db,
"SELECT seq,payload,timestamp,provider,thought_ms FROM transcript WHERE session=? AND seq>? AND seq<=? ORDER BY seq LIMIT 128",
[sqlight.text(snapshot.session), sqlight.int(after), sqlight.int(upper)],
source_row(),
))
list.try_map(rows, sourced_entry(snapshot.session, _, read))
}),
)
case list.reverse(rows) {
[] -> Ok(acc)
[last, ..] ->
case fold_source_rows(rows, acc, step, schedule, replacements) {
Stop(#(acc, _, _)) -> Ok(acc)
Continue(#(acc, schedule, replacements)) -> {
fold_source_pages(
ledger,
snapshot,
last.source.seq,
upper,
acc,
step,
schedule,
replacements,
)
}
}
}
}
fn fold_source_rows(
rows: List(transcript.SourcedEntry),
acc: a,
step: fn(a, transcript.SourcedEntry) -> FoldStep(a),
schedule: FitSchedule,
replacements: image_fit.Replacements,
) -> FoldStep(#(a, FitSchedule, image_fit.Replacements)) {
case rows {
[] -> Continue(#(acc, schedule, replacements))
[row, ..rest] -> {
let #(schedule, replacements) =
advance_fits(schedule, replacements, row.source.seq)
let entry = row.entry
let row =
transcript.SourcedEntry(
..row,
entry: transcript.Entry(
..entry,
input: image_fit.apply_replacements(entry.input, replacements),
),
)
case step(acc, row) {
Stop(acc) -> Stop(#(acc, schedule, replacements))
Continue(acc) ->
fold_source_rows(rest, acc, step, schedule, replacements)
}
}
}
}
fn advance_fits(
schedule: FitSchedule,
replacements: image_fit.Replacements,
seq: Int,
) -> #(FitSchedule, image_fit.Replacements) {
case schedule {
[#(at, next), ..rest] if at <= seq -> advance_fits(rest, next, seq)
_ -> #(schedule, replacements)
}
}
/// Read durable entries with their SQLite identities in chronological order.
pub fn load_sources(
store: store.Store,
id: String,
) -> Result(List(transcript.SourcedEntry), String) {
use upper <- result.try(
store.query(store, fn(db) {
store.rows(
db,
"SELECT COALESCE(MAX(seq),-1) FROM transcript WHERE session=?",
[sqlight.text(id)],
decode.field(0, decode.int, decode.success),
)
|> result.map(fn(rows) { list.first(rows) |> result.unwrap(-1) })
}),
)
load_source_pages(store, id, -1, upper, [])
}
/// A loaded row, and the image fit it is when it is one.
type LoadedRow =
#(transcript.SourcedEntry, Result(transcript.ImageFit, Nil))
/// Rows per store round trip. A transcript is read a page at a time so the
/// raw rows and their decoded entries are only ever live for one page, not
/// the whole transcript at once.
const load_page_rows = 128
fn load_source_pages(
store: store.Store,
id: String,
after: Int,
upper: Int,
pages: List(List(LoadedRow)),
) -> Result(List(transcript.SourcedEntry), String) {
let read = images.reader(store)
let page =
store.query(store, fn(db) {
use rows <- result.try(store.rows(
db,
"SELECT seq,payload,timestamp,provider,thought_ms FROM transcript WHERE session=? AND seq>? AND seq<=? ORDER BY seq LIMIT ?",
[
sqlight.text(id),
sqlight.int(after),
sqlight.int(upper),
sqlight.int(load_page_rows),
],
source_row(),
))
use entries <- result.try(
list.try_map(rows, fn(row) {
sourced_entry(id, row, read)
|> result.map(fn(entry) { #(entry, unpack_fit(row.1, read)) })
}),
)
Ok(#(entries, list.last(rows) |> result.map(fn(row) { row.0 })))
})
// New rows may land between pages; the captured upper bound keeps one view.
case page {
Error(error) -> Error(error)
Ok(#(entries, Ok(last))) ->
case list.length(entries) == load_page_rows {
True -> load_source_pages(store, id, last, upper, [entries, ..pages])
False -> Ok(fitted(list.flatten(list.reverse([entries, ..pages]))))
}
Ok(#(entries, Error(_))) ->
Ok(fitted(list.flatten(list.reverse([entries, ..pages]))))
}
}
/// The entries with each image fit applied to every row before it, so a
/// history read up to any row is what the model was sent at that point.
fn fitted(rows: List(LoadedRow)) -> List(transcript.SourcedEntry) {
let #(entries, _) =
list.fold(list.reverse(rows), #([], image_fit.replacements()), fn(acc, row) {
let #(entries, replacements) = acc
let #(sourced, fit) = row
let entry = sourced.entry
let sourced =
transcript.SourcedEntry(
..sourced,
entry: transcript.Entry(
..entry,
input: image_fit.apply_replacements(entry.input, replacements),
),
)
let replacements = case fit {
Ok(fit) -> image_fit.add(replacements, fit)
Error(_) -> replacements
}
#([sourced, ..entries], replacements)
})
entries
}
/// A transcript row as the reads select it: seq, payload, timestamp,
/// provider, thought_ms.
fn source_row() -> decode.Decoder(
#(Int, BitArray, Option(Int), Option(String), Option(Int)),
) {
use seq <- decode.field(0, decode.int)
use payload <- decode.field(1, decode.bit_array)
use timestamp <- decode.field(2, decode.optional(decode.int))
use provider <- decode.field(3, decode.optional(decode.string))
use thought_ms <- decode.field(4, decode.optional(decode.int))
decode.success(#(seq, payload, timestamp, provider, thought_ms))
}
/// One selected transcript row as its entry, keeping the row's source
/// reference and metadata. An undecodable payload is an invalid saved item.
fn sourced_entry(
session: String,
row: #(Int, BitArray, Option(Int), Option(String), Option(Int)),
read: fn(String) -> Result(String, Nil),
) -> Result(transcript.SourcedEntry, String) {
use input <- result.try(
unpack(row.1, read)
|> result.replace_error("invalid saved transcript item"),
)
Ok(transcript.SourcedEntry(
transcript.SourceRef(session, row.0),
transcript.Entry(
input,
row.2,
row.3,
row.4,
Some(transcript.SourceRef(session, row.0)),
),
))
}
/// Decode a packed transcript row with lazy image payload reads.
pub fn read_input(
store: store.Store,
payload: BitArray,
) -> Result(types.Input, Nil) {
unpack(payload, images.reader(store))
}
/// Rows read per step when a tail that starts on a tool result is widened.
const tail_step_rows = 32
/// At most this many extra rows widen a tail.
const tail_step_limit = 512
/// The newest `rows` transcript rows before `before` (exclusive; None reads
/// from the end), chronological. A page never starts on a tool result, whose
/// call would be on the page before: it is widened back to the nearest row
/// that is not one. The flag says whether older rows remain. Clients page
/// history with this instead of loading it whole.
pub fn load_tail(
store: store.Store,
id: String,
before: Option(Int),
rows: Int,
) -> Result(#(List(transcript.SourcedEntry), Bool), String) {
let upper = option.unwrap(before, 9_223_372_036_854_775_807)
use #(newest, full) <- result.try(read_before(
store,
id,
upper,
int.max(1, rows),
))
widen(store, id, newest, full, 0)
}
fn widen(
store: store.Store,
id: String,
entries: List(transcript.SourcedEntry),
full: Bool,
extra: Int,
) -> Result(#(List(transcript.SourcedEntry), Bool), String) {
case entries {
[] -> Ok(#([], False))
[first, ..] ->
case
is_tool_result(first.entry.input) && full && extra < tail_step_limit
{
False -> Ok(#(entries, full))
True -> {
use #(older, older_full) <- result.try(read_before(
store,
id,
first.source.seq,
tail_step_rows,
))
// Keep only what is needed: from the newest row that is not a
// tool result. Older rows left out mean more remain.
let skipped =
older
|> list.reverse
|> list.take_while(fn(item) { is_tool_result(item.entry.input) })
|> list.length
case list.length(older) - skipped - 1 {
head if head >= 0 ->
Ok(#(
list.append(list.drop(older, head), entries),
head > 0 || older_full,
))
_ ->
widen(
store,
id,
list.append(older, entries),
older_full,
extra + tail_step_rows,
)
}
}
}
}
}
fn is_tool_result(input: types.Input) -> Bool {
case input {
types.ToolOutput(..) -> True
_ -> False
}
}
/// Up to `limit` rows before `upper`, chronological, and whether the read
/// was full (so older rows may remain).
fn read_before(
store: store.Store,
id: String,
upper: Int,
limit: Int,
) -> Result(#(List(transcript.SourcedEntry), Bool), String) {
let read = images.reader(store)
store.query(store, fn(db) {
use rows <- result.try(store.rows(
db,
"SELECT seq,payload,timestamp,provider,thought_ms FROM transcript WHERE session=? AND seq<? ORDER BY seq DESC LIMIT ?",
[sqlight.text(id), sqlight.int(upper), sqlight.int(limit)],
source_row(),
))
use entries <- result.try(
list.try_map(list.reverse(rows), sourced_entry(id, _, read)),
)
Ok(#(entries, list.length(rows) == limit))
})
}
/// The session's newest transcript sequence, or 0 when it has none.
pub fn last_seq(store: store.Store, id: String) -> Result(Int, String) {
store.query(store, fn(db) {
store.one(
db,
"SELECT COALESCE(MAX(seq),0) FROM transcript WHERE session=?",
[sqlight.text(id)],
decode.field(0, decode.int, decode.success),
"session not found",
)
})
}
pub fn load(
store: store.Store,
id: String,
) -> Result(List(types.Input), String) {
load_entries(store, id)
|> result.map(fn(entries) { list.map(entries, fn(entry) { entry.input }) })
}
/// A prompt prefix a live session keeps while it is cached: the system
/// instructions and extension context blocks included in the system prompt.
pub type PinnedPrompt {
PinnedPrompt(instructions: String, context: List(types.Input))
}
/// Atomically retain the old prompt prefix and append a short capability
/// notice to history. An existing pin wins across subsequent reloads.
pub fn append_capability_update(
store: store.Store,
id: String,
pinned: PinnedPrompt,
head: Int,
update: String,
) -> Result(Int, String) {
let timestamp = usage.now()
store.query(store, fn(db) {
store.transaction(db, fn() {
use _ <- result.try(
store.run(
db,
"UPDATE sessions SET "
<> "pinned_context=CASE WHEN pinned_instructions IS NULL THEN ? ELSE pinned_context END, "
<> "pinned_head=CASE WHEN pinned_instructions IS NULL THEN ? ELSE pinned_head END, "
<> "pinned_instructions=COALESCE(pinned_instructions, ?) WHERE id=?",
[
sqlight.blob(pack_list(pinned.context)),
sqlight.int(head),
sqlight.text(pinned.instructions),
sqlight.text(id),
],
),
)
store.run(
db,
"INSERT INTO transcript(session,payload,timestamp,row_class) VALUES(?,?,?,'user')",
[
sqlight.text(id),
sqlight.blob(pack(types.User(update))),
sqlight.int(timestamp),
],
)
|> result.replace(timestamp)
})
})
}
/// The session's pinned prompt, if a capability change left one. A pin whose
/// context cannot be decoded is ignored, so the session uses its current prompt.
pub fn prompt_pin(
store: store.Store,
id: String,
) -> Result(Option(#(PinnedPrompt, Int)), String) {
store.query(store, fn(db) {
use rows <- result.try(
store.rows(
db,
"SELECT pinned_instructions, pinned_context, COALESCE(pinned_head, 0) FROM sessions WHERE id=?",
[sqlight.text(id)],
{
use instructions <- decode.field(0, decode.optional(decode.string))
use context <- decode.field(1, decode.optional(decode.bit_array))
use head <- decode.field(2, decode.int)
decode.success(#(instructions, context, head))
},
),
)
case rows {
[#(Some(instructions), Some(context), head)] ->
Ok(
unpack_list(context, images.reader(store))
|> result.map(fn(context) {
#(PinnedPrompt(instructions, context), head)
})
|> option.from_result,
)
[_] -> Ok(None)
_ -> Error("session not found")
}
})
}
pub fn clear_prompt_pin(store: store.Store, id: String) -> Result(Nil, String) {
store.write(
store,
"UPDATE sessions SET pinned_instructions=NULL, pinned_context=NULL, pinned_head=NULL WHERE id=?",
[sqlight.text(id)],
)
}
/// The entries inputs committed together become. A response's thinking time
/// goes on its first input that shows the thinking, so it is counted once.
pub fn entries(
inputs: List(types.Input),
timestamp: Option(Int),
provider: Option(String),
thought_ms: Option(Int),
) -> List(transcript.Entry) {
let #(_, entries) =
list.map_fold(inputs, thought_ms, fn(thought_ms, input) {
let entry = transcript.Entry(input, timestamp, provider, _, None)
case input {
types.Replay(item) if thought_ms != None ->
case events.thinking_text(item) {
"" -> #(thought_ms, entry(None))
_ -> #(None, entry(thought_ms))
}
_ -> #(thought_ms, entry(None))
}
})
entries
}
/// Appends image fit rows (see transcript.ImageFit), storing each fitted
/// payload, and returns their shared daemon time.
pub fn commit_fits(
store: store.Store,
id: String,
fits: List(transcript.ImageFit),
provider: String,
) -> Result(Int, String) {
let timestamp = usage.now()
store.query(store, fn(db) {
store.transaction(db, fn() {
list.try_each(fits, fn(fit) {
use stored <- result.try(images.store_images(db, [fit.image]))
let assert [image] = stored
store.run(
db,
"INSERT INTO transcript(session,payload,timestamp,provider,row_class) VALUES(?,?,?,?,'image_fit')",
[
sqlight.text(id),
sqlight.blob(pack_fit(fit.note, fit.source, image)),
sqlight.int(timestamp),
sqlight.text(provider),
],
)
})
|> result.replace(timestamp)
})
})
}
/// Atomically appends transcript inputs and returns their shared daemon time.
/// Empty commits still update the session stage and return the operation time.
pub fn commit(
store: store.Store,
id: String,
inputs: List(types.Input),
stage: Stage,
) -> Result(Int, String) {
commit_from(store, id, inputs, stage, None)
}
/// Appends inputs with the provider that produced or accepted them. Provider
/// provenance lets model preparation distinguish lossless replay from transfer.
pub fn commit_from(
store: store.Store,
id: String,
inputs: List(types.Input),
stage: Stage,
provider: Option(String),
) -> Result(Int, String) {
commit_with_letters(store, id, inputs, stage, provider, None, [], [])
|> result.map(fn(committed) { committed.0 })
}
/// Appends a model response's inputs; `thought_ms`, how long the model
/// thought before it, is kept on the first that shows the thinking. Answers
/// the shared daemon time and the seq of the response's first assistant row,
/// so a recorded provider call can link to the transcript row it produced;
/// a commit with no assistant row links nothing.
pub fn commit_response(
store: store.Store,
id: String,
inputs: List(types.Input),
stage: Stage,
provider: Option(String),
thought_ms: Option(Int),
) -> Result(#(Int, Option(Int)), String) {
commit_with_letters(store, id, inputs, stage, provider, thought_ms, [], [])
}
/// Letters and the inputs that carry them commit together, so a letter retried
/// after a crash cannot land in the transcript twice.
pub fn commit_letters(
store: store.Store,
id: String,
inputs: List(types.Input),
stage: Stage,
provider: Option(String),
letters: List(String),
) -> Result(Int, String) {
commit_with_letters(store, id, inputs, stage, provider, None, letters, [])
|> result.map(fn(committed) { committed.0 })
}
/// Admission receipts retire in the same transaction that writes model input.
pub fn commit_operations(
ledger: store.Store,
id: String,
inputs: List(types.Input),
stage: Stage,
provider: Option(String),
letters: List(String),
commits: List(operations.Commit),
) -> Result(Int, String) {
commit_with_letters(
ledger,
id,
inputs,
stage,
provider,
None,
letters,
commits,
)
|> result.map(fn(committed) { committed.0 })
}
fn commit_with_letters(
store: store.Store,
id: String,
inputs: List(types.Input),
stage: Stage,
provider: Option(String),
thought_ms: Option(Int),
letters: List(String),
commits: List(operations.Commit),
) -> Result(#(Int, Option(Int)), String) {
let timestamp = usage.now()
let read = images.reader(store)
store.query(store, fn(db) {
store.transaction(db, fn() {
use _ <- result.try(mail.receive(db, id, letters))
use previous_seq <- result.try(store.one(
db,
"SELECT COALESCE((SELECT seq FROM sqlite_sequence WHERE name='transcript'),0)",
[],
decode.field(0, decode.int, decode.success),
"transcript boundary",
))
let advances_assistant = case has_visible_assistant(inputs) {
True -> 1
False -> 0
}
// The first assistant row of the commit, so a caller that recorded the
// provider call behind it can link to the transcript row it produced.
use linked <- result.try(
entries(inputs, Some(timestamp), provider, thought_ms)
|> list.try_fold(None, fn(linked, entry) {
use input <- result.try(images.externalize(db, entry.input, read))
use _ <- result.try(
store.run(
db,
"INSERT INTO transcript(session,payload,timestamp,provider,thought_ms,row_class) VALUES(?,?,?,?,?,?)",
[
sqlight.text(id),
sqlight.blob(pack(input)),
sqlight.int(timestamp),
sqlight.nullable(sqlight.text, provider),
sqlight.nullable(sqlight.int, entry.thought_ms),
sqlight.text(transcript.row_class(input)),
],
),
)
case linked, is_assistant_row(input) {
None, True ->
store.one(
db,
"SELECT last_insert_rowid()",
[],
decode.field(0, decode.int, decode.success),
"committed row sequence",
)
|> result.map(Some)
_, _ -> Ok(linked)
}
}),
)
use seq <- result.try(store.one(
db,
"SELECT COALESCE(MAX(seq),0) FROM transcript WHERE session=?",
[sqlight.text(id)],
decode.field(0, decode.int, decode.success),
"committed input sequence",
))
use _ <- result.try(operations.committed(
db,
id,
commits,
previous_seq + 1,
seq,
timestamp,
))
// COALESCE keeps the old title when no new user message suggests one.
store.run(
db,
"UPDATE sessions SET stage=?,title=COALESCE(?,title),activity_seq=(SELECT COALESCE(MAX(activity_seq),0)+1 FROM sessions),last_assistant_at=CASE WHEN ?=1 THEN unixepoch() ELSE last_assistant_at END WHERE id=?",
[
sqlight.text(stage_name(stage)),
sqlight.nullable(sqlight.text, option.map(latest_user(inputs), title)),
sqlight.int(advances_assistant),
sqlight.text(id),
],
)
|> result.replace(linked)
})
|> result.map(fn(linked) { #(timestamp, linked) })
})
}
fn usage_decoder() -> decode.Decoder(Option(usage.Metadata)) {
use model <- decode.field(0, decode.optional(decode.string))
use recorded_at <- decode.field(1, decode.optional(decode.int))
use prompt <- decode.field(2, decode.optional(decode.int))
use completion <- decode.field(3, decode.optional(decode.int))
use cached <- decode.field(4, decode.optional(decode.int))
use creation <- decode.field(5, decode.optional(decode.int))
use write_5m <- decode.field(6, decode.optional(decode.int))
use write_1h <- decode.field(7, decode.optional(decode.int))
use reasoning <- decode.field(8, decode.optional(decode.int))
use cache <- decode.field(9, decode.optional(decode.string))
// A fading that no longer parses only loses the footer's countdown.
let cache =
option.then(cache, fn(stored) {
json.parse(stored, cache_fade.decoder()) |> option.from_result
})
case model, recorded_at, prompt, completion {
None, None, None, None -> decode.success(None)
Some(model), Some(recorded_at), None, None ->
decode.success(Some(usage.Metadata(model, recorded_at, None, cache)))
Some(model), Some(recorded_at), Some(prompt), Some(completion) ->
decode.success(
Some(usage.Metadata(
model,
recorded_at,
Some(usage.Tokens(
prompt,
completion,
cached,
creation,
write_5m,
write_1h,
reasoning,
)),
cache,
)),
)
_, _, _, _ -> decode.failure(None, "consistent saved usage metadata")
}
}
pub fn load_usage(
store: store.Store,
id: String,
) -> Result(Option(usage.Metadata), String) {
store.query(store, fn(db) {
store.one(
db,
"SELECT usage_model,usage_recorded_at,usage_prompt_tokens,usage_completion_tokens,usage_cached_prompt_tokens,usage_cache_creation_tokens,usage_cache_write_5m_tokens,usage_cache_write_1h_tokens,usage_reasoning_tokens,usage_cache FROM sessions WHERE id=?",
[sqlight.text(id)],
usage_decoder(),
"session not found",
)
})
}
pub fn clear_usage(store: store.Store, id: String) -> Result(Nil, String) {
store.write(
store,
"UPDATE sessions SET usage_model=NULL,usage_recorded_at=NULL,usage_prompt_tokens=NULL,usage_completion_tokens=NULL,usage_cached_prompt_tokens=NULL,usage_cache_creation_tokens=NULL,usage_cache_write_5m_tokens=NULL,usage_cache_write_1h_tokens=NULL,usage_reasoning_tokens=NULL,usage_cache=NULL WHERE id=?",
[sqlight.text(id)],
)
}
pub fn record_usage(
store: store.Store,
id: String,
metadata: usage.Metadata,
) -> Result(Nil, String) {
let usage.Metadata(model, recorded_at, tokens, cache) = metadata
let #(prompt, completion, cached, creation, write_5m, write_1h, reasoning) = case
tokens
{
Some(usage.Tokens(
prompt,
completion,
cached,
creation,
write_5m,
write_1h,
reasoning,
)) -> #(
Some(prompt),
Some(completion),
cached,
creation,
write_5m,
write_1h,
reasoning,
)
None -> #(None, None, None, None, None, None, None)
}
let cache =
option.map(cache, fn(fade) { json.to_string(cache_fade.to_json(fade)) })
store.write(
store,
"UPDATE sessions SET usage_model=?,usage_recorded_at=?,usage_prompt_tokens=?,usage_completion_tokens=?,usage_cached_prompt_tokens=?,usage_cache_creation_tokens=?,usage_cache_write_5m_tokens=?,usage_cache_write_1h_tokens=?,usage_reasoning_tokens=?,usage_cache=? WHERE id=?",
[
sqlight.text(model),
sqlight.int(recorded_at),
sqlight.nullable(sqlight.int, prompt),
sqlight.nullable(sqlight.int, completion),
sqlight.nullable(sqlight.int, cached),
sqlight.nullable(sqlight.int, creation),
sqlight.nullable(sqlight.int, write_5m),
sqlight.nullable(sqlight.int, write_1h),
sqlight.nullable(sqlight.int, reasoning),
sqlight.nullable(sqlight.text, cache),
sqlight.text(id),
],
)
}
@external(erlang, "albedo_conversation", "pack")
fn pack(input: types.Input) -> BitArray
@external(erlang, "albedo_conversation", "pack_fit")
fn pack_fit(note: String, source: String, image: types.Image) -> BitArray
@external(erlang, "albedo_conversation", "unpack_fit")
fn unpack_fit(
bytes: BitArray,
read: fn(String) -> Result(String, Nil),
) -> Result(transcript.ImageFit, Nil)
@external(erlang, "albedo_conversation", "unpack")
fn unpack(
bytes: BitArray,
read: fn(String) -> Result(String, Nil),
) -> Result(types.Input, Nil)
/// For rows read only for their text, inside the store (which a real reader
/// would call back into).
fn unread(_hash: String) -> Result(String, Nil) {
Error(Nil)
}
@external(erlang, "albedo_conversation", "pack_list")
fn pack_list(inputs: List(types.Input)) -> BitArray
@external(erlang, "albedo_conversation", "unpack_list")
fn unpack_list(
bytes: BitArray,
read: fn(String) -> Result(String, Nil),
) -> Result(List(types.Input), Nil)
pub fn set_effort(
store: store.Store,
id: String,
effort: Option(String),
) -> Result(Nil, String) {
store.write(store, "UPDATE sessions SET effort=? WHERE id=?", [
sqlight.nullable(sqlight.text, effort),
sqlight.text(id),
])
}
pub fn set_configuration(
store: store.Store,
id: String,
provider: String,
model: String,
selected_protocol: types.Protocol,
effort: Option(String),
) -> Result(Nil, String) {
store.query(store, fn(db) {
store.transaction(db, fn() {
use _ <- result.try(
store.run(
db,
"UPDATE transcript SET provider=(SELECT provider FROM sessions WHERE id=?) WHERE session=? AND provider IS NULL",
[sqlight.text(id), sqlight.text(id)],
),
)
store.run(
db,
"UPDATE sessions SET provider=?,model=?,protocol=?,effort=? WHERE id=?",
[
sqlight.text(provider),
sqlight.text(model),
sqlight.text(protocol(selected_protocol)),
sqlight.nullable(sqlight.text, effort),
sqlight.text(id),
],
)
})
})
}
/// Workspace changes are serialized by the session actor with submissions.
pub fn set_workspace(
store: store.Store,
id: String,
cwd: String,
) -> Result(Nil, String) {
store.write(store, "UPDATE sessions SET cwd=? WHERE id=?", [
sqlight.text(cwd),
sqlight.text(id),
])
}