Something went wrong. Try again.
small gleam coding and persistent agent daemon with a detachable cli and a canvas ui forked from okami.mom/albedo
Something went wrong. Try again.
3.6 kB · 114 lines
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115import albedo/daemon/events as viewimport albedo/daemon/noteimport albedo/daemon/transcriptimport albedo/harness/runtimeimport albedo/openai_api/typesimport gleam/dynamic/decodeimport gleam/jsonimport gleam/listimport gleam/option.{None}import gleam/resultimport gleeunit/should
fn types_of(events: List(String)) -> List(String) { list.filter_map(events, fn(event) { case json.parse(event, decode.field("type", decode.string, decode.success)) { Ok(value) -> Ok(value) Error(_) -> Error(Nil) } })}
fn texts(events: List(String), kind: String) -> List(String) { let decoder = { use name <- decode.field("type", decode.string) use text <- decode.field("text", decode.string) decode.success(#(name, text)) } list.filter_map(events, fn(event) { use #(name, text) <- result.try( json.parse(event, decoder) |> result.map_error(fn(_) { Nil }), ) case name == kind { True -> Ok(text) False -> Error(Nil) } })}
/// Replayed history must show thinking apart from the answer, and never leak/// reasoning (or a provider's raw reasoning_text) into the assistant message.pub fn snapshot_surfaces_saved_reasoning_apart_from_the_answer_test() { let assert Ok(host) = runtime.start(":memory:") let store = runtime.ledger(host) let assert Ok(chat) = json.parse( "{\"role\":\"assistant\",\"content\":\"answer\",\"reasoning_content\":\"think\"}", types.replay_decoder(types.ChatCompletions), ) let chat_events = view.snapshot( store, [transcript.Entry(types.Replay(chat), None, None)], None, ) types_of(chat_events) |> should.equal(["thinking", "message"]) texts(chat_events, "thinking") |> should.equal(["think"]) texts(chat_events, "message") |> should.equal(["answer"])
let assert Ok(reasoning) = json.parse( "{\"type\":\"reasoning\",\"id\":\"rs_1\",\"summary\":[{\"type\":\"summary_text\",\"text\":\"why\"}],\"content\":[{\"type\":\"reasoning_text\",\"text\":\"raw\"}]}", types.replay_decoder(types.Responses), ) let assert Ok(message) = json.parse( "{\"type\":\"message\",\"role\":\"assistant\",\"content\":[{\"type\":\"output_text\",\"text\":\"answer\"}]}", types.replay_decoder(types.Responses), ) let response_events = view.snapshot( store, [ transcript.Entry(types.Replay(reasoning), None, None), transcript.Entry(types.Replay(message), None, None), ], None, ) types_of(response_events) |> should.equal(["thinking", "message"]) texts(response_events, "thinking") |> should.equal(["why"]) texts(response_events, "message") |> should.equal(["answer"]) runtime.stop(host)}
/// A reloaded transcript labels a note by its origin and shows its body, and/// labels a typed message the way the live stream does.pub fn snapshot_labels_notes_by_origin_test() { let assert Ok(host) = runtime.start(":memory:") let decoder = { use source <- decode.field("source", decode.string) use text <- decode.field("text", decode.string) decode.success(#(source, text)) } view.snapshot( runtime.ledger(host), [ types.User("hello"), types.User(note.wrap("daemon restart", "resumed")), types.User("<system-note>legacy</system-note>"), ] |> list.map(transcript.Entry(_, None, None)), None, ) |> list.filter_map(fn(event) { json.parse(event, decoder) |> result.replace_error(Nil) }) |> should.equal([ #("chat", "hello"), #("daemon restart", "resumed"), #("note", "legacy"), ]) runtime.stop(host)}