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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348"""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, sotests that change those layers are exclusive, on a home of their own.Read-only default lookups and authentication share the daemon without overrides.
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 jsonfrom datetime import datetimeimport osimport timeimport unittestimport urllib.errorimport urllib.request
from harness import Albedo, Provider, exclusive, text
def milliseconds(timestamp): return round( datetime.fromisoformat(timestamp.replace("Z", "+00:00")).timestamp() * 1000 )
# `/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)
# exclusive: configures daemon-wide TTL/catalog layers and writes cache files@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("/models?view=cache-policy&limit=200") as response: return json.load(response)
def resolved(self, query): with self.app.api("/models?view=cache-policy" + query) as response: return json.load(response)["matched"]
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_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_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"]) failure = self.layers(table)["local"]["error"] self.assertEqual(set(failure), {"code", "detail"}) self.assertEqual(failure["code"], "cache_policy_layer_failed") self.assertIn("duplicate id: deepseek", failure["detail"]) self.assertLessEqual(len(failure["detail"]), 4096) 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"]["detail"] ) 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}/reload", {"target": "both"}): pass path = self.app.home / "cache-ttl-remote.json" before = path.read_bytes() self.provider.catalog = {"entries": [REMOTE_TABLE["entries"][0]] * 2} with self.app.api( f"/sessions/{session}/reload", {"target": "both"} ) as response: outcome = json.load(response)["cache_policy"] self.assertEqual(outcome["state"], "failed") self.assertIn("duplicate id", outcome["failure"]["detail"]) 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}/reload", {"target": "both"}): pass os.utime(path, (timestamp, timestamp)) self.assertEqual( self.entries(self.table())["deepseek"]["note"], "remote revised!" )
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.assertEqual(set(local["error"]), {"code", "detail"}) self.assertEqual(local["error"]["code"], "cache_policy_layer_failed") self.assertIn("not valid JSON", local["error"]["detail"]) self.assertLessEqual(len(local["error"]["detail"]), 4096) 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}/reload", {"target": "both"} ) as response: outcome = json.load(response) self.assertEqual(outcome["session"]["state"], "applied") self.assertEqual(outcome["cache_policy"]["state"], "refreshed") 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}/context?view=requests") as response: finished = milliseconds(json.load(response)["items"][0]["ended_at"]) fading = [(finished + 600_000, None), (finished + 3_600_000, 0)]
def latest_fade(): with self.app.api(f"/sessions/{session}?tail=0") as response: fade = json.load(response)["usage"]["cache_fade"] return [(milliseconds(step["at"]), step["cached_tokens"]) for step in fade]
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)
class DefaultCacheTtlTests(unittest.TestCase): def setUp(self): self.app = Albedo() self.app.__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def test_route_requires_authentication(self): request = urllib.request.Request(self.app.base + "/models?view=cache-policy") with self.assertRaises(urllib.error.HTTPError) as refused: urllib.request.urlopen(request, timeout=20) self.assertEqual(refused.exception.code, 401)