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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403"""The reloadable prompt-cache TTL table: layers, merge precedence, live reload.
The table is daemon-wide state read from files, so this is where its behaviouris visible: the shipped default served with every entry, lookups resolvingthrough match order, a local override replacing an entry by id and winningwith a new more-specific one — without a restart — and a malformed filekeeping the last good table. A turn's usage carries the steps its cachedcount fades through by that table, which the chat footer counts down from andwhich must survive a restart. It writes extensions.json and cache-ttl.json, soevery test is exclusive, on a home of its own.
The remote layer's fetch is background (like the models catalog's), so thefixture provider serves the remote table on a loopback URL and the test waitsfor it to land; `/reload` fetches it synchronously. No live network is reached."""
from typing import Anyimport jsonimport osimport timeimport unittestimport urllib.errorimport urllib.request
from harness import ROOT, Albedo, Provider, exclusive, text
SHIPPED = json.loads((ROOT / "priv" / "cache-ttl.json").read_text())
# `/reload` fetches the models catalog before the TTL table, so it needs a# fixture url too — the default would reach the live models.dev.MODELS_CATALOG = { "fixture": { "id": "fixture", "name": "Fixture", "models": { "fixture-model": {"id": "fixture-model", "limit": {"context": 8000}} }, }}
REMOTE_TABLE: dict[str, Any] = { "version": 1, "entries": [ # Replaces the shipped deepseek entry by id, in place. { "id": "deepseek", "match": {"host": "api.deepseek.com"}, "policy": "fixed", "clock": "response", "evidence": "measured", "source": "fixture", "note": "remote override", }, # A new id, which goes before every default entry. { "id": "fixture-gateway", "match": {"host": "gateway.fixture"}, "policy": "evict", "survival": {"typical": 600}, "evidence": "measured", "source": "fixture", "note": "new remote entry", }, ],}
LOCAL_OVERRIDE: dict[str, Any] = { "version": 1, "entries": [ # Replaces the remote or default deepseek entry by id, in place. { "id": "deepseek", "match": {"host": "api.deepseek.com"}, "policy": "fixed", "clock": "request", "evidence": "measured", "source": "local", "note": "local override", }, # A new id, more specific than anything shipped, which must win. { "id": "laptop-gateway", "match": {"host": "gateway.laptop"}, "policy": "refresh", "tiers": [{"seconds": 600, "write": 1.5}], "evidence": "measured", "source": "local", "note": "new local entry", }, ],}
def write_atomic(path, payload): temporary = path.with_suffix(".tmp") temporary.write_text(json.dumps(payload)) temporary.replace(path)
@exclusiveclass CacheTtlTests(unittest.TestCase): def setUp(self): self.provider = Provider(lambda _request: text("ok"), catalog=REMOTE_TABLE) self.addCleanup(self.provider.close) self.catalog = Provider(lambda _request: text("ok"), catalog=MODELS_CATALOG) self.addCleanup(self.catalog.close)
def prepare(app): app.write_extensions( { "models": {"url": self.catalog.url + "/models.json"}, "cacheTtl": { "url": self.provider.url + "/cache-ttl.json", "refreshHours": 24, }, } )
self.app = Albedo( self.provider, prepare=prepare, providers={ "fixture": { "baseUrl": self.provider.url, "apiKey": "fixture-key", "model": "fixture-model", "protocol": "chat_completions", }, }, ) self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def table(self): with self.app.api("/cache-ttl") as response: return json.load(response)
def resolved(self, query): with self.app.api("/cache-ttl" + query) as response: return json.load(response)
def layers(self, table): return {layer["name"]: layer for layer in table["layers"]}
def entries(self, table): return {entry["id"]: entry for entry in table["entries"]}
def test_default_table_is_served_and_lookups_resolve(self): table = self.table() self.assertTrue(self.layers(table)["default"]["loaded"]) self.assertIn("priv/cache-ttl.json", self.layers(table)["default"]["path"]) self.assertFalse(self.layers(table)["local"]["loaded"]) # Every shipped entry, each tagged with the layer it came from. self.assertEqual( set(self.entries(table)), {e["id"] for e in SHIPPED["entries"]} ) for entry in table["entries"]: self.assertEqual(entry["layer"], "default") # First match in table order: the subscription entry shadows nothing, # but a specific model glob beats the general host entry after it. claude = self.resolved("?extension=claude") self.assertEqual(claude["id"], "claude-subscription") self.assertEqual(claude["policy"], "refresh") self.assertEqual(claude["clock"], "request") self.assertEqual([t["seconds"] for t in claude["tiers"]], [300, 3600]) self.assertEqual(claude["read"], 0.1) deepseek = self.resolved("?host=api.deepseek.com") self.assertEqual(deepseek["id"], "deepseek") self.assertEqual(deepseek["policy"], "evict") self.assertEqual(deepseek["survival"]["typical"], 14400) # Matching is case-insensitive. self.assertEqual(self.resolved("?host=API.DEEPSEEK.COM")["id"], "deepseek") # A model glob within a list of patterns, ahead of the plain host rule. self.assertEqual( self.resolved("?host=api.openai.com&model=gpt-5.6-turbo")["id"], "openai-5.6", ) self.assertEqual( self.resolved("?host=api.openai.com&model=gpt-4o")["id"], "openai" ) self.assertIsNone(self.resolved("?extension=never-heard-of-it"))
def test_local_override_replaces_and_adds_without_restart(self): write_atomic(self.app.home / "cache-ttl.json", LOCAL_OVERRIDE) table = self.table() local = self.layers(table)["local"] self.assertTrue(local["loaded"]) self.assertEqual(local["path"], str(self.app.home / "cache-ttl.json")) entries = self.entries(table) # Replaced by id: same entry, now from the local layer. self.assertEqual(entries["deepseek"]["layer"], "local") self.assertEqual(entries["deepseek"]["note"], "local override") self.assertEqual(entries["deepseek"]["clock"], "request") # New ids go before all earlier-layer entries and win the lookup. self.assertEqual(table["entries"][0]["id"], "laptop-gateway") self.assertEqual(self.resolved("?host=gateway.laptop")["id"], "laptop-gateway") # The replaced default keeps answering, with the override's values. self.assertEqual( self.resolved("?host=api.deepseek.com")["note"], "local override" ) # Untouched defaults still resolve. self.assertEqual( self.resolved("?extension=claude")["id"], "claude-subscription" )
def test_three_layers_keep_replacements_in_place_and_new_ids_first(self): write_atomic(self.app.home / "cache-ttl-remote.json", REMOTE_TABLE) gateway = { **REMOTE_TABLE["entries"][1], "match": {"host": "gateway.*"}, "note": "local gateway replacement", } local = {"entries": [*LOCAL_OVERRIDE["entries"], gateway]} write_atomic(self.app.home / "cache-ttl.json", local) table = self.table() self.assertEqual( [entry["id"] for entry in table["entries"]], ["laptop-gateway", "fixture-gateway"] + [entry["id"] for entry in SHIPPED["entries"]], ) self.assertTrue(self.layers(table)["remote"]["loaded"]) self.assertEqual(self.entries(table)["fixture-gateway"]["layer"], "local") self.assertEqual(self.entries(table)["deepseek"]["note"], "local override") # A fresh local rule wins before the replaced, more general gateway rule. self.assertEqual(self.resolved("?host=gateway.laptop")["id"], "laptop-gateway") self.assertEqual( self.resolved("?host=gateway.fixture")["note"], "local gateway replacement", ) self.assertEqual(self.resolved("?host=api.deepseek.com")["clock"], "request")
def test_duplicate_ids_reject_the_layer_and_retain_last_good_entries(self): duplicate = {"entries": [LOCAL_OVERRIDE["entries"][0]] * 2} write_atomic(self.app.home / "cache-ttl.json", duplicate) table = self.table() self.assertFalse(self.layers(table)["local"]["loaded"]) self.assertIn("duplicate id: deepseek", self.layers(table)["local"]["error"]) self.assertFalse(any(entry["layer"] == "local" for entry in table["entries"])) write_atomic(self.app.home / "cache-ttl.json", LOCAL_OVERRIDE) self.assertTrue(self.layers(self.table())["local"]["loaded"]) write_atomic(self.app.home / "cache-ttl.json", duplicate) for _ in range(3): table = self.table() self.assertFalse(self.layers(table)["local"]["loaded"]) self.assertIn( "duplicate id: deepseek", self.layers(table)["local"]["error"] ) self.assertEqual(self.entries(table)["deepseek"]["note"], "local override") write_atomic(self.app.home / "cache-ttl.json", LOCAL_OVERRIDE) self.assertTrue(self.layers(self.table())["local"]["loaded"])
def test_duplicate_remote_ids_are_rejected_before_replacement(self): session = self.app.session() with self.app.api( f"/sessions/{session}/commands", {"name": "/reload", "args": {}} ): pass path = self.app.home / "cache-ttl-remote.json" before = path.read_bytes() self.provider.catalog = {"entries": [REMOTE_TABLE["entries"][0]] * 2} with self.assertRaises(urllib.error.HTTPError) as rejected: self.app.api( f"/sessions/{session}/commands", {"name": "/reload", "args": {}} ) self.assertIn("duplicate id", rejected.exception.read().decode()) self.assertEqual(path.read_bytes(), before) self.assertEqual( self.entries(self.table())["deepseek"]["note"], "remote override" )
def test_same_revision_remote_replacement_invalidates_parsed_layer(self): session = self.app.session() path = self.app.home / "cache-ttl-remote.json" write_atomic(path, REMOTE_TABLE) # A future mtime prevents background refresh. Restoring it after reload # keeps the cached revision unchanged without depending on scheduling. replacement = json.loads(json.dumps(REMOTE_TABLE)) replacement["entries"][0]["note"] = "remote revised!" self.assertEqual(len(json.dumps(replacement)), len(json.dumps(REMOTE_TABLE))) self.provider.catalog = replacement
timestamp = int(time.time()) + 60 os.utime(path, (timestamp, timestamp)) self.assertEqual( self.entries(self.table())["deepseek"]["note"], "remote override" ) with self.app.api( f"/sessions/{session}/commands", {"name": "/reload", "args": {}} ): pass os.utime(path, (timestamp, timestamp)) self.assertEqual( self.entries(self.table())["deepseek"]["note"], "remote revised!" )
def test_invalid_override_shadows_previous_entry_before_decoding(self): write_atomic( self.app.home / "cache-ttl.json", {"entries": [{**LOCAL_OVERRIDE["entries"][0], "policy": "invalid"}]}, ) table = self.table() self.assertTrue(self.layers(table)["local"]["loaded"]) self.assertNotIn("deepseek", self.entries(table)) self.assertIsNone(self.resolved("?host=api.deepseek.com")) self.assertEqual( self.resolved("?extension=claude")["id"], "claude-subscription" )
def test_malformed_local_file_keeps_the_last_good_table(self): write_atomic(self.app.home / "cache-ttl.json", LOCAL_OVERRIDE) self.assertTrue(self.layers(self.table())["local"]["loaded"]) (self.app.home / "cache-ttl.json").write_text("{ this is not json") table = self.table() local = self.layers(table)["local"] self.assertFalse(local["loaded"]) self.assertIn("not valid JSON", local["error"]) entries = self.entries(table) self.assertEqual(entries["deepseek"]["layer"], "local") self.assertEqual(entries["deepseek"]["note"], "local override") self.assertEqual(self.resolved("?host=gateway.laptop")["id"], "laptop-gateway") # Fixing the file is picked up live, without a restart. write_atomic(self.app.home / "cache-ttl.json", LOCAL_OVERRIDE) self.assertTrue(self.layers(self.table())["local"]["loaded"])
def test_remote_url_is_fetched_and_merged(self): deadline = time.monotonic() + 20 table = self.table() while ( not self.layers(table)["remote"]["loaded"] and time.monotonic() < deadline ): time.sleep(0.1) table = self.table() remote = self.layers(table)["remote"] self.assertTrue(remote["loaded"]) self.assertEqual(remote["path"], str(self.app.home / "cache-ttl-remote.json")) entries = self.entries(table) # Replaced by id, in the default entry's place, ahead of general rules. ids = [entry["id"] for entry in table["entries"]] self.assertEqual(entries["deepseek"]["layer"], "remote") self.assertEqual(entries["deepseek"]["note"], "remote override") self.assertEqual(entries["deepseek"]["policy"], "fixed") self.assertLess(ids.index("claude-subscription"), ids.index("deepseek")) # The remote layer's new id sits in front of the default entries. self.assertEqual(ids[0], "fixture-gateway") self.assertEqual( self.resolved("?host=gateway.fixture")["id"], "fixture-gateway" )
def test_session_reload_refetches_the_remote_copy(self): session = self.app.session() self.assertFalse((self.app.home / "cache-ttl-remote.json").exists()) with self.app.api( f"/sessions/{session}/commands", {"name": "/reload", "args": {}} ) as response: outcome = json.load(response)["result"] self.assertEqual(outcome["reloaded"], "models+session") table = self.table() self.assertTrue(self.layers(table)["remote"]["loaded"]) self.assertEqual(self.entries(table)["deepseek"]["note"], "remote override")
def test_a_turns_usage_says_when_its_cache_fades_across_a_restart(self): # A provider that only evicts: the count is unknown past typical # survival and gone past the bound, counted from the response's end. write_atomic( self.app.home / "cache-ttl.json", { "version": 1, "entries": [ { "id": "fixture-evict", "match": {"host": "127.0.0.1"}, "policy": "evict", "survival": {"typical": 600, "max": 3600}, "evidence": "measured", } ], }, ) session = self.app.session() self.app.prompt(session, "hello").close() self.app.idle(session) with self.app.api(f"/sessions/{session}/requests") as response: finished = json.load(response)["rows"][0]["finishedMs"] fading = [{"at": finished + 600_000}, {"at": finished + 3_600_000, "cached": 0}]
def latest_fade(): usages = [ event for event in self.app.events(session) if event["type"] == "usage" ] return usages[-1].get("cacheFade")
self.assertEqual(latest_fade(), fading) # The steps are stored with the usage, so a client attaching to a # restarted daemon still counts down from the same call. self.app.restart() self.assertEqual(latest_fade(), fading)
def test_route_requires_authentication(self): request = urllib.request.Request(self.app.base + "/cache-ttl") with self.assertRaises(urllib.error.HTTPError) as refused: urllib.request.urlopen(request, timeout=20) self.assertEqual(refused.exception.code, 403)