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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547"""Fast mode stays opt-in when a provider cannot verify capability.
These offline real-daemon scenarios cover adapters without fast mode, conditionalstate updates, and the account-specific Codex tier cache without live requests."""
import jsonimport timeimport threadingimport unittestimport urllib.error
from harness import Albedo, Provider, exclusive, latest_user, operation_id, python, text
class FastModeTests(unittest.TestCase): def setUp(self): self.provider = Provider(lambda _request: text("ok")) self.addCleanup(self.provider.close)
def test_unverified_compatible_endpoint_rejects_enable_without_mutating_configuration( self, ): with Albedo(self.provider) as app: session = app.session() route = f"/sessions/{session}/fast-mode" with app.api(route) as response: initial = json.load(response) revision = response.headers["ETag"] self.assertFalse(initial["fast"]) self.assertEqual(initial["support"]["status"], "unsupported") self.assertTrue(initial["message"])
for state in ("on", "toggle"): with self.assertRaises(urllib.error.HTTPError) as caught: app.api( route, {"state": state}, method="PATCH", headers={"If-Match": revision}, ) self.assertEqual(caught.exception.code, 400) self.assertEqual( json.load(caught.exception)["code"], "fast_unsupported" ) with app.api(route) as response: after = json.load(response) unchanged_revision = response.headers["ETag"] self.assertFalse(after["fast"]) self.assertEqual(unchanged_revision, revision) with app.api( route, {"state": "off"}, method="PATCH", headers={"If-Match": revision} ) as response: self.assertFalse(json.load(response)["fast"])
def test_fast_mode_endpoint_checks_etag_and_state_validation(self): with Albedo(self.provider) as app: session = app.session() route = f"/sessions/{session}/fast-mode" with app.api(route) as response: json.load(response) etag = response.headers["ETag"] for state in ("enabled", "true", ""): with ( self.subTest(state=state), self.assertRaises(urllib.error.HTTPError) as caught, ): app.api( route, {"state": state}, method="PATCH", headers={"If-Match": etag}, ) self.assertEqual(caught.exception.code, 400) with self.assertRaises(urllib.error.HTTPError) as caught: app.api(route, {"state": "off"}, method="PATCH") self.assertEqual(caught.exception.code, 428) with self.assertRaises(urllib.error.HTTPError) as caught: app.api( route, {"state": "off"}, method="PATCH", headers={"If-Match": '"stale"'}, ) self.assertEqual(caught.exception.code, 412) self.assertEqual(json.load(caught.exception)["code"], "precondition_failed") with app.api(route) as response: self.assertFalse(json.load(response)["fast"])
def test_generic_configuration_patch_and_fast_endpoint_share_state(self): with Albedo(self.provider) as app: session = app.session() route = f"/sessions/{session}?view=configuration" fast_route = f"/sessions/{session}/fast-mode" with app.api(route) as response: self.assertFalse(json.load(response)["fast"]) etag = response.headers["ETag"] with app.api( route, {"fast": False}, method="PATCH", headers={"If-Match": etag} ) as response: saved = json.load(response) etag = response.headers["ETag"] self.assertFalse(saved["resource"]["value"]["fast"]) with app.api(fast_route) as response: self.assertFalse(json.load(response)["fast"]) self.assertEqual(response.headers["ETag"], etag) with self.assertRaises(urllib.error.HTTPError) as caught: app.api( route, {"fast": True}, method="PATCH", headers={"If-Match": etag} ) self.assertEqual(caught.exception.code, 400) self.assertEqual(json.load(caught.exception)["code"], "fast_unsupported") with app.api(route) as response: self.assertFalse(json.load(response)["fast"])
@exclusiveclass CodexFastModeTests(unittest.TestCase): """Cached account metadata enables fast mode without a Codex inference call."""
def setUp(self): self.child_started = threading.Event() self.release_child = threading.Event()
def script(request): if "fast-mode-child-marker" in latest_user(request): self.child_started.set() if not self.release_child.wait(30): return text("child gate timed out") return text("done")
self.provider = Provider(script) self.addCleanup(self.provider.close) self.addCleanup(self.release_child.set)
def prepare_codex(self, app): now = int(time.time() * 1000) app.store_secrets( "accounts", { "openai-codex": [ { "type": "oauth", "_albedo_account_id": "one", "access": "dummy-access", "refresh": "dummy-refresh", "accountId": "acct-one", "expires": 9999999999999, "selected": True, } ] }, ) (app.home / "codex-models.json").write_text( json.dumps( { "clientVersion": "1.2.3", "versionCheckedAt": now, "accounts": { "acct-one": { "fetchedAt": now, "clientVersion": "1.2.3", "models": [ { "slug": "fixture-codex", "name": "Fixture Codex", "efforts": [], "service_tiers": [ {"id": "priority", "name": "Fast"} ], } ], } }, } ) ) app.write_extensions({"models": {"refreshHours": 0}})
def app_with_codex(self, *, prepare=None): app = Albedo( self.provider, prepare=self.prepare_codex if prepare is None else prepare, providers={ "fixture": { "extension": "openai", "baseUrl": self.provider.url, "apiKey": "fixture-key", "model": "fixture-model", "protocol": "responses", }, "codex-fixture": { "extension": "codex", "model": "fixture-codex", "protocol": "responses", "accountId": "one", }, }, ) app.__enter__() self.addCleanup(app.__exit__, None, None, None) return app
def patch_configuration(self, app, session, body): route = f"/sessions/{session}?view=configuration" with app.api(route) as response: json.load(response) etag = response.headers["ETag"] with app.api( route, body, method="PATCH", headers={"If-Match": etag} ) as response: return json.load(response)
def test_implemented_check_keeps_legacy_metadata_unknown(self): def prepare_legacy(app): self.prepare_codex(app) path = app.home / "codex-models.json" catalog = json.loads(path.read_text()) del catalog["accounts"]["acct-one"]["models"][0]["service_tiers"] path.write_text(json.dumps(catalog))
app = self.app_with_codex(prepare=prepare_legacy) session = app.session() self.patch_configuration( app, session, {"provider_profile": "codex-fixture", "model": "fixture-codex"}, ) route = f"/sessions/{session}/fast-mode" with app.api(route) as response: initial = json.load(response) etag = response.headers["ETag"] self.assertEqual(initial["support"]["status"], "unknown") self.assertFalse(initial["fast"]) with self.assertRaises(urllib.error.HTTPError) as caught: app.api(route, {"state": "on"}, method="PATCH", headers={"If-Match": etag}) self.assertEqual(caught.exception.code, 400) self.assertEqual(json.load(caught.exception)["code"], "fast_unknown") with app.api(route) as response: self.assertFalse(json.load(response)["fast"]) self.assertEqual(response.headers["ETag"], etag)
def test_bare_fast_command_toggles_and_stale_replays_cannot_toggle_back(self): app = self.app_with_codex() session = app.session() self.patch_configuration( app, session, {"provider_profile": "codex-fixture", "model": "fixture-codex"}, ) with app.api(f"/sessions/{session}/catalog?kind=commands") as response: catalog = json.load(response) commands = catalog.get("native_commands", []) + catalog["loaded"]["commands"] fast = [command for command in commands if command["slash_name"] == "/fast"] self.assertEqual(len(fast), 1) command = fast[0] self.assertEqual(command["arguments"], []) self.assertEqual(command["delivery"], "mutation") self.assertEqual(command["caller_permissions"], ["human"]) operation = command["operation"] self.assertEqual( operation["body"]["/state"], {"source": "literal", "value": "toggle"} ) route = operation["path_template"].replace("{session_id}", session) self.assertEqual(route, f"/sessions/{session}/fast-mode") with app.api(route) as response: self.assertFalse(json.load(response)["fast"]) etag = response.headers["ETag"] body = {"state": operation["body"]["/state"]["value"]} with app.api( route, body, method=operation["method"], headers={"If-Match": etag} ) as response: self.assertTrue(json.load(response)["fast"]) enabled_etag = response.headers["ETag"] with self.assertRaises(urllib.error.HTTPError) as caught: app.api(route, body, method="PATCH", headers={"If-Match": etag}) self.assertEqual(caught.exception.code, 412) with app.api(route) as response: self.assertTrue(json.load(response)["fast"]) self.assertEqual(response.headers["ETag"], enabled_etag)
# Losing provider support must not trap a user in the enabled state. cache_path = app.home / "codex-models.json" cache = json.loads(cache_path.read_text()) cache["accounts"]["acct-one"]["models"][0]["service_tiers"] = [] cache_path.write_text(json.dumps(cache)) with app.api( route, body, method="PATCH", headers={"If-Match": enabled_etag} ) as response: saved = json.load(response) self.assertFalse(saved["fast"]) self.assertEqual(saved["support"]["status"], "unsupported")
def test_codex_advertised_tier_enables_persists_and_model_switch_resets(self): app = self.app_with_codex() session = app.session() app.prompt(session, "create a local fixture checkpoint").close() app.idle(session) checkpoint = [ item["checkpoint_id"] for item in app.history(session)["items"] if item["kind"] == "assistant" ][-1] changed = self.patch_configuration( app, session, { "provider_profile": "codex-fixture", "model": "fixture-codex", }, ) self.assertEqual(changed["session"]["provider_profile"], "codex-fixture") route = f"/sessions/{session}/fast-mode" with app.api(route) as response: support = json.load(response) etag = response.headers["ETag"] self.assertEqual(support["support"]["status"], "supported") self.assertEqual(support["support"]["mode"], "priority") self.assertTrue(support["support"]["source"]) with app.api( route, {"state": "on"}, method="PATCH", headers={"If-Match": etag} ) as response: self.assertTrue(json.load(response)["fast"]) app.restart() with app.api(route) as response: self.assertTrue(json.load(response)["fast"])
fork_id = operation_id() with app.api( f"/sessions/{fork_id}", { "kind": "fork", "source_session_id": session, "checkpoint_id": checkpoint, }, method="PUT", headers={"If-None-Match": "*"}, ): pass with app.api(f"/sessions/{fork_id}/fast-mode") as response: self.assertTrue(json.load(response)["fast"])
child_id = operation_id() with app.api( f"/sessions/{child_id}", { "kind": "child", "parent_id": session, "address": "fast-mode-child", "name": "fast-mode-child", "initial_input_id": operation_id(), "task": "fast-mode-child-marker: Use only the local fake provider.", "model": "fixture/fixture-model", }, method="PUT", headers={"If-None-Match": "*"}, ): pass self.assertTrue( self.child_started.wait(15), "fake child did not reach its gated provider" ) with app.api(f"/sessions/{child_id}/fast-mode") as response: self.assertFalse(json.load(response)["fast"]) with app.api(route) as response: self.assertTrue(json.load(response)["fast"])
try: changed = self.patch_configuration( app, session, {"provider_profile": "fixture", "model": "fixture-model"} ) self.assertEqual(changed["session"]["provider_profile"], "fixture") with app.api(route) as response: self.assertFalse(json.load(response)["fast"]) finally: self.release_child.set()
app.idle(child_id) app.idle(session) self.assertTrue(self.provider.requests) self.assertTrue( all(item["model"] == "fixture-model" for item in self.provider.requests), self.provider.requests, )
class FastModeCommandAuthorizationTests(unittest.TestCase): def setUp(self): def script(request): items = request.get("input", []) if items and items[-1].get("type") == "function_call_output": return text("done") return python( "assert not hasattr(commands, 'fast'), 'fast must be user-only'\n" "try:\n" " await commands.invoke('/fast')\n" " assert False, 'model must not toggle fast mode'\n" "except CommandsError as error:\n" " assert 'not callable from the model' in str(error), error\n" "print('FAST_COMMAND_AUTH_OK')" )
self.provider = Provider(script) self.addCleanup(self.provider.close)
def test_model_cannot_invoke_the_user_only_fast_toggle(self): with Albedo(self.provider, protocol="responses") as app: session = app.session() app.prompt(session, "try the fast toggle").close() app.idle(session) requests = [item["request"] for item in self.provider.requests] self.assertIn("FAST_COMMAND_AUTH_OK", json.dumps(requests)) self.assertEqual(len(requests), 2)
@exclusiveclass DocumentedProviderFastModeTests(unittest.TestCase): """Native callbacks are checked without auth probes or inference on those hosts."""
def test_documented_profile_modes_toggle_without_native_inference(self): profiles = { "native-openai": { "extension": "openai", "baseUrl": "https://api.openai.com/v1", "apiKey": "dummy-key", "protocol": "responses", "model": "gpt-6.1-sol", }, "console-claude": { "extension": "claude", "apiKey": "dummy-key", "model": "claude-opus-5-5", }, "bedrock-optimized": { "extension": "bedrock", "apiKey": "dummy-key", "api": "converse", "baseUrl": "https://bedrock-runtime.us-east-2.amazonaws.com", "model": "us.anthropic.claude-3-5-haiku-20241022-v1:0", }, "claude-oauth-unverified": { "extension": "claude", "model": "claude-opus-5-5", }, "vertex-regional": { "extension": "vertex", "project": "fixture-project", "location": "us-central1", "model": "gemini-2.5-pro", }, "bedrock-messages": { "extension": "bedrock", "api": "messages", "baseUrl": "https://bedrock-runtime.us-east-2.amazonaws.com", "model": "us.anthropic.claude-3-5-haiku-20241022-v1:0", }, "no-fast-adapter": { "extension": "opencode-zen", "apiKey": "dummy-key", "model": "fixture-model", "protocol": "responses", }, "vertex-priority": { "extension": "vertex", "project": "fixture-project", "location": "global", "model": "gemini-2.5-pro", }, }
for profile in profiles.values(): profile.setdefault("protocol", "chat_completions")
def offline(app): app.write_extensions({"models": {"refreshHours": 0}}) (app.home / "models.json").write_text( json.dumps( { catalog: { "id": catalog, "models": { model: { "id": model, "limit": {"context": 32000, "output": 4096}, } }, } for catalog, model in [ ("openai", "gpt-6.1-sol"), ("anthropic", "claude-opus-5-5"), ("amazon-bedrock", profiles["bedrock-optimized"]["model"]), ("google-vertex", "gemini-2.5-pro"), ] } ) )
provider = Provider(lambda _request: text("unexpected native request")) self.addCleanup(provider.close) with Albedo(provider, providers=profiles, prepare=offline) as app: for profile, mode in [ ("native-openai", "fast"), ("console-claude", "fast"), ("bedrock-optimized", "optimized"), ("vertex-priority", "priority"), ("claude-oauth-unverified", None), ("vertex-regional", None), ("bedrock-messages", None), ("no-fast-adapter", None), ]: with self.subTest(profile=profile): app.profile = profile session = app.session() route = f"/sessions/{session}/fast-mode" with app.api(route) as response: initial = json.load(response) revision = response.headers["ETag"] if mode is None: self.assertEqual(initial["support"]["status"], "unsupported") with self.assertRaises(urllib.error.HTTPError) as caught: app.api( route, {"state": "on"}, method="PATCH", headers={"If-Match": revision}, ) self.assertEqual(caught.exception.code, 400) self.assertEqual( json.load(caught.exception)["code"], "fast_unsupported" ) with app.api(route) as response: self.assertFalse(json.load(response)["fast"]) self.assertEqual(response.headers["ETag"], revision) continue self.assertEqual(initial["support"]["status"], "supported") self.assertEqual(initial["support"]["mode"], mode) for wanted in (True, False): with app.api( route, {"state": "toggle"}, method="PATCH", headers={"If-Match": revision}, ) as response: self.assertEqual(json.load(response)["fast"], wanted) revision = response.headers["ETag"] self.assertEqual(app.history(session)["items"], []) self.assertEqual(provider.requests, [])