"""Fast mode stays opt-in when a provider cannot verify capability. These offline real-daemon scenarios cover adapters without fast mode, conditional state updates, and the account-specific Codex tier cache without live requests. """ import json import time import threading import unittest import 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"]) @exclusive class 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) @exclusive class 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, [])