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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188"""Command mutations share conditional edits' checks, persistence, and context reset."""
import jsonimport threadingimport unittest
from harness import Albedo, Provider, ROOT, exclusive, textfrom inspect_support import Inspection
@exclusiveclass SessionConfigureTest(unittest.TestCase): """Isolated global defaults, catalog, and inspection."""
def setUp(self): self.provider = Provider(lambda _request: text("done")) self.addCleanup(self.provider.close)
def prepare(app): app.daemon.env["ALBEDO_INSPECT"] = "1" catalog = { "fixture-cloud": { "api": self.provider.url, "models": { model: { "id": model, "reasoning_options": [ {"type": "effort", "values": efforts} ], } for model, efforts in { "first": ["low", "high"], "second": ["medium", "high"], "third": ["low", "medium"], }.items() }, } } (app.home / "models.json").write_text(json.dumps(catalog)) app.write_extensions({"models": {"refreshHours": 0}})
self.app = Albedo( self.provider, prepare=prepare, providers={ "fixture": { "extension": "openai", "baseUrl": self.provider.url, "apiKey": "fixture-key", "model": "first", "effort": "high", "protocol": "responses", } }, ).__enter__() self.addCleanup(self.app.__exit__, None, None, None) self.session = self.app.session() self.inspection = Inspection( self.app, ROOT / "test/e2e/albedo_session_configure_probe.erl" )
def read(self, path): with self.app.api(path) as response: return json.load(response)
def configuration(self): return self.read(f"/sessions/{self.session}?view=configuration")
def defaults(self): return self.read("/settings?group=providers")
def invoke(self, method, *values): arguments = [self.session, *values] erl = ", ".join( "null" if value is None else f"<<{json.dumps(value)}>>" for value in arguments ) return self.inspection.call_json(method, f"[{erl}]")
def model(self, model, provider=None, effort=None): return self.invoke("model", model, provider, effort)
def turn(self): self.app.prompt(self.session, "prepare a request").close() self.app.idle(self.session)
def context(self): return self.read(f"/sessions/{self.session}/context")
def test_commands_save_defaults_resolve_effort_and_reset_prepared_context(self): self.turn() self.assertEqual(self.context()["state"], "ready") cursor = self.app.stream_page(self.session) selected = self.model("second")["ok"] self.assertEqual((selected["model"], selected["effort"]), ("second", "high")) self.assertEqual(self.context()["state"], "pending") changes = self.app.stream_page(self.session, cursor)["events"] invalidated = { event["data"]["kind"] for event in changes if event["type"] == "invalidate" } self.assertTrue({"session", "context"}.issubset(invalidated), changes) new = self.read(f"/sessions/{self.app.session()}") self.assertEqual(new["model"], "second")
self.turn() defaults = self.defaults() self.assertEqual( self.invoke("effort", " LOW ")["error"], "unsupported reasoning effort: low; available: medium, high", ) self.assertEqual(self.context()["state"], "ready") changed = self.invoke("effort", " MEDIUM ")["ok"] self.assertEqual(changed["effort"], "medium") self.assertEqual(self.defaults(), defaults) # Effort is not part of the prepared request's sections: a session # without a kernel may rebuild its context on demand. self.assertIn(self.context()["state"], ("pending", "ready"))
before = self.configuration() rejected = self.model("third", effort="high") self.assertIn("unsupported reasoning effort", rejected["error"]) self.assertEqual(self.configuration(), before) self.assertEqual(self.defaults(), defaults) self.invoke("effort", "high") self.assertEqual(self.model("third")["ok"]["effort"], "medium") self.assertEqual(self.configuration()["effort"], "medium")
# HTTP edits share mutation validation, but never remember defaults. route = f"/sessions/{self.session}?view=configuration" with self.app.api(route) as response: json.load(response) etag = response.headers["ETag"] defaults = self.defaults() with self.app.api( route, {"model": "second", "effort": "high"}, method="PATCH", headers={"If-Match": etag}, ) as response: json.load(response) self.assertEqual(self.defaults(), defaults) self.turn() self.assertEqual( self.provider.requests[-1]["request"]["reasoning"]["effort"], "high" )
def test_commands_reject_even_unchanged_values_during_a_turn(self): entered, release = threading.Event(), threading.Event() self.addCleanup(release.set)
def script(_request): entered.set() release.wait(20) return text("done")
self.provider.script = script self.app.prompt(self.session, "hold the turn").close() self.assertTrue(entered.wait(10)) before, defaults = self.configuration(), self.defaults() try: self.assertEqual( self.model("first")["error"], "model must be nonempty and session idle" ) self.assertEqual( self.invoke("effort", "high")["error"], "switching effort requires an idle session", ) self.assertEqual(self.configuration(), before) self.assertEqual(self.defaults(), defaults) route = f"/sessions/{self.session}?view=configuration" with self.app.api(route) as response: json.load(response) etag = response.headers["ETag"] with self.app.api( route, {"model": "first", "effort": "high"}, method="PATCH", headers={"If-Match": etag}, ) as response: json.load(response) finally: release.set() self.app.idle(self.session)
if __name__ == "__main__": unittest.main()