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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382// Account pins and 429 handoff depend on OAuth identities and provider responses unavailable in E2E.import albedo/clockimport albedo/harness/extensions/codex/extension as codeximport albedo/harness/rotationimport albedo/openai_apiimport albedo/openai_api/typesimport gleam/erlang/processimport gleam/intimport gleam/jsonimport gleam/listimport gleam/option.{None, Some}import gleam/stringimport gleeunit/should
const credentials = "{\"accounts\":{\"openai-codex\":[{\"type\":\"oauth\",\"access\":\"h.eyJodHRwczovL2FwaS5vcGVuYWkuY29tL2F1dGgiOnsiY2hhdGdwdF9hY2NvdW50X2lkIjoiYWNjb3VudC0xIiwiY2hhdGdwdF9hY2NvdW50X3VzZXJfaWQiOiJzZWF0LTEifX0.s\",\"refresh\":\"refresh-1\",\"expires\":9999999999999,\"accountId\":\"account-1\",\"accountUserId\":\"seat-1\"},{\"type\":\"oauth\",\"access\":\"h.eyJodHRwczovL2FwaS5vcGVuYWkuY29tL2F1dGgiOnsiY2hhdGdwdF9hY2NvdW50X2lkIjoiYWNjb3VudC0yIiwiY2hhdGdwdF9hY2NvdW50X3VzZXJfaWQiOiJzZWF0LTIifX0.s\",\"refresh\":\"refresh-2\",\"expires\":9999999999999,\"accountId\":\"account-2\",\"accountUserId\":\"seat-2\"}]}}"
const reversed_credentials = "{\"accounts\":{\"openai-codex\":[{\"type\":\"oauth\",\"access\":\"h.eyJodHRwczovL2FwaS5vcGVuYWkuY29tL2F1dGgiOnsiY2hhdGdwdF9hY2NvdW50X2lkIjoiYWNjb3VudC0yIiwiY2hhdGdwdF9hY2NvdW50X3VzZXJfaWQiOiJzZWF0LTIifX0.s\",\"refresh\":\"refresh-2\",\"expires\":9999999999999,\"accountId\":\"account-2\",\"accountUserId\":\"seat-2\"},{\"type\":\"oauth\",\"access\":\"h.eyJodHRwczovL2FwaS5vcGVuYWkuY29tL2F1dGgiOnsiY2hhdGdwdF9hY2NvdW50X2lkIjoiYWNjb3VudC0xIiwiY2hhdGdwdF9hY2NvdW50X3VzZXJfaWQiOiJzZWF0LTEifX0.s\",\"refresh\":\"refresh-1\",\"expires\":9999999999999,\"accountId\":\"account-1\",\"accountUserId\":\"seat-1\"}]}}"
pub fn sessions_stick_to_and_distribute_across_codex_accounts_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let first = access(home, "session-1") access(home, "session-1") |> should.equal(first) let accounts = [access(home, "session-1"), access(home, "session-2")] |> list.unique list.length(accounts) |> should.equal(2) cleanup(root)}
pub fn session_pin_survives_credential_reordering_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let selected = access(home, "durable-session") let _ = write(home, "creds.json", reversed_credentials) access(home, "durable-session") |> should.equal(selected) cleanup(root)}
pub fn revoked_codex_account_is_removed_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let #(revoked, client) = codex_client(home, "revoked-session") let assert Some(message) = codex.account_failure(home, client, types.HttpError(401, "token_revoked")) string.ends_with(message, "run /login") |> should.be_true codex.account_failure(home, client, types.HttpError(500, "")) |> should.equal(None) let remaining = access(home, "revoked-session") { remaining == revoked } |> should.be_false access(home, "another-session") |> should.equal(remaining) let #(_, last) = codex_client(home, "revoked-session") let assert Some(_) = codex.account_failure(home, last, types.HttpError(401, "token_revoked")) let assert Error(_) = native_access(home, "revoked-session") cleanup(root)}
pub fn usage_limit_moves_the_session_to_a_sibling_account_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let #(limited, client) = codex_client(home, "busy-session")
// Ordinary rate limiting is transient and must not move the session. codex.account_failure(home, client, types.HttpError(429, "slow down")) |> should.equal(None) access(home, "busy-session") |> should.equal(limited)
let body = "{\"error\":{\"type\":\"usage_limit_reached\",\"resets_in_seconds\":3600}}" let assert Some(message) = codex.account_failure(home, client, types.HttpError(429, body)) string.contains(message, "usage limit reached") |> should.be_true string.contains(message, "the next turn will use") |> should.be_true let next = access(home, "busy-session") { next == limited } |> should.be_false // Every session avoids the limited account while a sibling has usage. access(home, "fresh-session") |> should.equal(next)
// With every account limited the message says so instead of naming one. let #(_, sibling) = codex_client(home, "busy-session") let assert Some(exhausted) = codex.account_failure(home, sibling, types.HttpError(429, body)) string.contains(exhausted, "no other ChatGPT account") |> should.be_true cleanup(root)}
/// Replays a stream where the named tokens answer 429 with `body` and every/// other account fails plainly, recording which accounts were tried.fn limited_stream( limited: List(String), body: String, tried: process.Subject(String),) -> fn(types.Client, a, b) -> Result(c, types.Error) { fn(client: types.Client, _request, _on_event) { process.send(tried, client.api_key) case list.contains(limited, client.api_key) { True -> Error(types.HttpError(429, body)) False -> Error(types.HttpError(500, "second account answered")) } }}
fn rotate( home: String, session: String, first: types.Client, request: types.Request, stream: fn(types.Client, types.Request, fn(types.Event) -> types.Control) -> Result(types.Turn, types.Error), pause: fn(Int) -> Nil,) -> Result(types.Turn, types.Error) { rotation.stream( codex.pool(home, session, False, stream), first, request, fn(_) { types.Continue }, rotation.Budget(8, 0, pause, False), fn(_) { Nil }, )}
fn drain(tried: process.Subject(String), found: List(String)) -> List(String) { case process.receive(tried, 0) { Ok(token) -> drain(tried, [token, ..found]) Error(_) -> list.reverse(found) }}
pub fn fast_mode_is_resolved_again_for_each_rotated_account_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let session = "fast-rotation-session" let #(first_account, first) = codex_client(home, session) let second_account = case first_account { "account-1" -> "account-2" _ -> "account-1" } let now = clock.os_system_ms() let _ = write( home, "codex-models.json", json.to_string( json.object([ #("clientVersion", json.string("1.2.3")), #("versionCheckedAt", json.int(now)), #( "accounts", json.object([ #("account-1", cached_fast_model(now, "priority")), #("account-2", cached_fast_model(now, "fast")), ]), ), ]), ), ) let tried = process.new_subject() let usage_limit = "{\"error\":{\"type\":\"usage_limit_reached\",\"resets_in_seconds\":3600}}" let request = openai_api.request("gpt-5.5", [types.User("hi")]) rotation.stream( codex.pool(home, session, True, fn(client, request, _) { let assert types.Codex(account, _) = client.policy let tier = case request.options.service_tier { Some(tier) -> tier None -> "missing" } process.send(tried, account <> ":" <> tier) case account == first_account { True -> Error(types.HttpError(429, usage_limit)) False -> Error(types.HttpError(500, "second account answered")) } }), first, request, fn(_) { types.Continue }, rotation.Budget(8, 0, fn(_) { Nil }, False), fn(_) { Nil }, ) |> should.equal(Error(types.HttpError(500, "second account answered"))) let assert [first_tier, second_tier] = drain(tried, []) first_tier |> should.equal( first_account <> ":" <> case first_account { "account-1" -> "priority" _ -> "fast" }, ) second_tier |> should.equal( second_account <> ":" <> case second_account { "account-1" -> "priority" _ -> "fast" }, ) cleanup(root)}
fn cached_fast_model(now: Int, tier_id: String) -> json.Json { json.object([ #("fetchedAt", json.int(now)), #("clientVersion", json.string("1.2.3")), #( "models", json.array( [ json.object([ #("slug", json.string("gpt-5.5")), #( "service_tiers", json.array( [ json.object([ #("id", json.string(tier_id)), #("name", json.string("fast")), ]), ], fn(value) { value }, ), ), ]), ], fn(value) { value }, ), ), ])}
pub fn a_limited_account_hands_the_same_request_to_the_next_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let #(_, first) = codex_client(home, "swarm-session") let tried = process.new_subject() let body = "{\"error\":{\"type\":\"usage_limit_reached\",\"resets_in_seconds\":3600}}" let request = openai_api.request("gpt-5.5", [types.User("hi")]) rotate( home, "swarm-session", first, request, limited_stream([first.api_key], body, tried), fn(_) { panic as "a sibling had room; nothing should wait" }, ) |> should.equal(Error(types.HttpError(500, "second account answered"))) let attempts = drain(tried, []) list.length(attempts) |> should.equal(2) list.first(attempts) |> should.equal(Ok(first.api_key)) cleanup(root)}
pub fn a_burst_rate_limit_body_rotates_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let #(_, first) = codex_client(home, "burst-session") let tried = process.new_subject() let request = openai_api.request("gpt-6-luna", [types.User("hi")]) rotate( home, "burst-session", first, request, limited_stream( [first.api_key], "{\"detail\":\"Rate limit exceeded\"}", tried, ), fn(_) { panic as "a sibling had room; nothing should wait" }, ) |> should.equal(Error(types.HttpError(500, "second account answered"))) list.length(drain(tried, [])) |> should.equal(2) cleanup(root)}
pub fn a_rate_limit_rotates_too_and_stops_when_all_are_busy_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let #(_, first) = codex_client(home, "busy-swarm") let tried = process.new_subject() let body = "{\"error\":{\"type\":\"rate_limit_exceeded\",\"resets_in_seconds\":20}}" let #(_, second) = codex_client(home, "other-busy-swarm") let both = list.unique([first.api_key, second.api_key]) let request = openai_api.request("gpt-5.5", [types.User("hi")]) let paused = process.new_subject() rotate( home, "busy-swarm", first, request, limited_stream(both, body, tried), fn(delay) { process.send(paused, int.to_string(delay)) }, ) |> should.equal(Error(types.HttpError(429, body))) // Each account once, then four waits with a retry after each, then it stops. list.length(drain(tried, [])) |> should.equal(6) drain(paused, []) |> should.equal(["4000", "8000", "15000", "30000"]) cleanup(root)}
pub fn a_usage_limit_on_every_account_does_not_wait_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let #(_, first) = codex_client(home, "spent-swarm") let #(_, second) = codex_client(home, "other-spent-swarm") let tried = process.new_subject() let body = "{\"error\":{\"type\":\"usage_limit_reached\",\"resets_in_seconds\":3600}}" let request = openai_api.request("gpt-6-luna", [types.User("hi")]) rotate( home, "spent-swarm", first, request, limited_stream(list.unique([first.api_key, second.api_key]), body, tried), fn(_) { panic as "a usage limit lasts hours; it must not be waited on" }, ) |> should.equal(Error(types.HttpError(429, body))) list.length(drain(tried, [])) |> should.equal(2) cleanup(root)}
pub fn selected_account_overrides_the_session_pin_test() -> Nil { let #(root, _, home) = fixture() let _ = write(home, "creds.json", credentials) let pinned = access(home, "pinned-session") let other = case pinned { "account-1" -> "account-2" _ -> "account-1" } let _ = write( home, "creds.json", string.replace( credentials, "\"accountId\":\"" <> other <> "\"", "\"accountId\":\"" <> other <> "\",\"selected\":true", ), ) access(home, "pinned-session") |> should.equal(other) access(home, "any-session") |> should.equal(other) cleanup(root)}
fn codex_client(home: String, session: String) -> #(String, types.Client) { let assert Ok(access) = native_access(home, session) #( access.account_id, openai_api.codex_client( "http://127.0.0.1", access.token, access.account_id, session, ), )}
fn access(home: String, session: String) -> String { let assert Ok(access) = native_access(home, session) access.account_id}
@external(erlang, "albedo_openai_auth", "codex_access")fn native_access(home: String, session: String) -> Result(codex.Access, String)
@external(erlang, "albedo_skills_test_support", "fixture")fn fixture() -> #(String, String, String)
@external(erlang, "albedo_skills_test_support", "write")fn write(base: String, relative: String, content: String) -> String
@external(erlang, "albedo_skills_test_support", "cleanup")fn cleanup(root: String) -> Nil