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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269"""The model's web_search must use the user's preferred provider and fallback down the order only when one fails, and the /web-search page mustreorder providers and turn them off for every later search.
Exa is the one provider a test can stand in for: its endpoint is a setting,so a local fake answers it. The subscription providers have no sign-in on thetest daemon, so each fails at once without reaching the network.
The order and the Exa key are global settings, so each scenario has its owndaemon."""
import astimport jsonimport threadingimport unittestimport urllib.errorfrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from harness import Albedo, Provider, exclusive, python, text
COLLECTION = "/extensions/web-search/providers"
RESULTS = [ { "url": "https://gleam.run/news/one", "title": "Gleam one", "highlights": ["first highlight", "second"], "publishedDate": "2026-08-01", }, {"url": "https://gleam.run/news/one", "title": "duplicate"}, {"url": "https://example.com/two", "title": None, "highlights": None},]
SEARCH = ( "r = await web_search('latest gleam release', limit=2)\n" "print(repr([r.answer, [(s.title, s.url, s.snippet, s.published)" " for s in r.sources]]))\n" "print(r)")
FAILING = ( "try:\n" " await web_search('latest gleam release')\n" "except Exception as error:\n" " print('raised:', error)")
class FakeExa: """Answers /search with RESULTS and records what each request sent."""
def __init__(self): self.requests = [] owner = self
class Handler(BaseHTTPRequestHandler): def do_POST(self): body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) owner.requests.append((self.path, self.headers["x-api-key"], body)) payload = json.dumps({"results": RESULTS}).encode() self.send_response(200) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(payload))) self.end_headers() self.wfile.write(payload)
def log_message(self, format, *args): pass
self.server = ThreadingHTTPServer(("127.0.0.1", 0), Handler) self.url = f"http://127.0.0.1:{self.server.server_address[1]}" threading.Thread(target=self.server.serve_forever, daemon=True).start()
def close(self): self.server.shutdown() self.server.server_close()
def scripting(cells): """Answers each user turn with the next scripted python cell."""
def reply(request): if request["messages"][-1].get("role") == "user" and cells: return python(cells.pop(0)) return text("done")
return Provider(reply)
def last_output(app, session): results = [ json.loads(part["value"]) for entry in app.history(session)["items"] if entry["kind"] == "tool_result" and entry["tool"]["name"] == "python" for part in entry["content"] if part["kind"] == "json" and part["field"] == "result" ] return results[-1]["output"]
@exclusiveclass WebSearchTests(unittest.TestCase): def setUp(self): self.exa = FakeExa() self.addCleanup(self.exa.close)
def start(self, cells): provider = scripting(cells) self.addCleanup(provider.close)
def prepare(app): app.write_extensions({"exa": {"endpoint": self.exa.url}}) app.store_secrets("exa", {"apiKey": "exa-key"})
app = Albedo(provider, prepare=prepare) app.__enter__() self.addCleanup(app.__exit__, None, None, None) return app
def ask(self, app, prompt): session = app.session() app.prompt(session, prompt).close() app.idle(session) return session, last_output(app, session)
def providers(self, app): with app.api(COLLECTION) as response: page = json.load(response) return [item["value"] for item in page["items"]], page["page"]
def change(self, app, name, change): with app.api(f"{COLLECTION}/{name}", {"change": change}) as response: return json.load(response)["resource"]["value"]
def patch(self, app, route, body): with app.api(route) as response: etag = response.headers["ETag"] with app.api( route, body, method="PATCH", headers={"If-Match": etag} ) as response: return json.load(response)
def test_session_provider_selection_is_loaded_and_isolated(self): app = self.start([FAILING, SEARCH, SEARCH]) disabled = app.session() self.patch( app, f"/sessions/{disabled}?view=configuration", {"selection": {"extensions": {"exa": False}}}, ) with app.api(f"/sessions/{disabled}/reload", {"target": "session"}) as response: self.assertEqual(json.load(response)["session"]["state"], "applied") app.prompt(disabled, "search without exa").close() app.idle(disabled) self.assertIn("raised:", last_output(app, disabled)) self.assertEqual(self.exa.requests, []) self.ask(app, "another session can use exa") self.assertEqual(len(self.exa.requests), 1) self.patch( app, f"/sessions/{disabled}?view=configuration", {"selection": {"extensions": {"exa": True}}}, ) with app.api(f"/sessions/{disabled}/reload", {"target": "session"}) as response: self.assertEqual(json.load(response)["session"]["state"], "applied") app.prompt(disabled, "search after applying exa").close() app.idle(disabled) self.assertEqual(len(self.exa.requests), 2)
def test_global_eligibility_preserves_hidden_provider_preferences(self): app = self.start([]) self.change(app, "exa", "up") self.change(app, "exa", "toggle") before = json.loads((app.home / "extensions.json").read_text())["web-search"] self.patch(app, "/settings?group=extensions", {"defaults": {"exa": False}}) self.assertNotIn("exa", [entry["name"] for entry in self.providers(app)[0]]) self.change(app, "codex", "toggle") saved = json.loads((app.home / "extensions.json").read_text())["web-search"] self.assertEqual(saved["order"], before["order"]) self.assertIn("exa", saved["off"]) self.patch(app, "/settings?group=extensions", {"defaults": {"exa": True}}) exa = next(entry for entry in self.providers(app)[0] if entry["name"] == "exa") self.assertFalse(exa["enabled"]) self.assertEqual(exa["position"], before["order"].index("exa") + 1)
def test_search_passes_over_signed_out_providers_to_exa(self): app = self.start([SEARCH]) session, output = self.ask(app, "search the web") answer, sources = ast.literal_eval(output.splitlines()[0]) self.assertEqual(answer, "") # One source per url, cut to the limit, highlights as the snippet. self.assertEqual( sources, [ ( "Gleam one", "https://gleam.run/news/one", "first highlight … second", "2026-08-01", ), ("https://example.com/two", "https://example.com/two", "", None), ], ) self.assertIn("[1] Gleam one (2026-08-01)", output) # The transcript shows the search's result beside the cell. (activity,) = [ activity for entry in app.history(session)["items"] if entry["kind"] == "tool_result" for part in entry["content"] if part["kind"] == "trace" for activity in part["trace"]["activities"] if activity["kind"] == "web" ] self.assertEqual(activity["target"], "latest gleam release") self.assertIn("- [Gleam one](https://gleam.run/news/one)", activity["detail"]) ((path, key, body),) = self.exa.requests self.assertEqual((path, key), ("/search", "exa-key")) self.assertEqual( (body["query"], body["numResults"]), ("latest gleam release", 2) )
def test_page_reorders_and_turns_off_providers(self): app = self.start([FAILING]) listed, page = self.providers(app) names = [entry["name"] for entry in listed] self.assertEqual( sorted(names), ["antigravity", "claude", "codex", "exa", "perplexity"] ) self.assertEqual([entry["position"] for entry in listed], [1, 2, 3, 4, 5]) self.assertTrue(all(entry["enabled"] for entry in listed)) self.assertEqual( [row["id"] for row in page["rows"]], names, "the page lists the same order" )
index = names.index("exa") moved = self.change(app, "exa", "up") self.assertEqual(moved["position"], index) expected = names[: index - 1] + ["exa", names[index - 1]] + names[index + 1 :] self.assertEqual([entry["name"] for entry in self.providers(app)[0]], expected) settings = json.loads((app.home / "extensions.json").read_text()) self.assertEqual(settings["web-search"]["order"], expected)
# A move past the top changes nothing. top = expected[0] self.assertEqual(self.change(app, top, "up")["position"], 1)
self.assertFalse(self.change(app, "exa", "toggle")["enabled"]) _, output = self.ask(app, "search the web") self.assertIn("raised: every web search provider failed", output) for name in ("codex", "claude", "antigravity"): self.assertIn(name + ":", output) self.assertNotIn("exa:", output) self.assertEqual(self.exa.requests, [], "a provider turned off is never tried")
for name, body, status in ( ("nobody", {"change": "up"}, 404), ("exa", {"change": "sideways"}, 400), ): with self.subTest(name=name, body=body): with self.assertRaises(urllib.error.HTTPError) as refused: app.api(f"{COLLECTION}/{name}", body).close() refused.exception.close() self.assertEqual(refused.exception.code, status)
if __name__ == "__main__": unittest.main()