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.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114"""Spawning a child asks endpoints that never answer once, together.
A bare model the parent's provider does not list is looked up on every otherprofile, and a generic profile asks its endpoint's `/models`. An endpoint thataccepts connections and never answers costs one fetch timeout: asked one afteranother, three of them held a spawn for three timeouts, and two spawns thatarrived together each opened their own connection and waited their owntimeout. The wait is the daemon's, so this runs through the real daemon."""
import jsonimport socketimport threadingimport timeimport unittest
from harness import Albedo, Provider, operation_id, text
STALLED = 3# One fetch timeout is 5 s; asked in sequence, three took at least 15.TOGETHER_SECONDS = 12
class StalledEndpointsTests(unittest.TestCase): def setUp(self): self.provider = Provider(lambda _request: text("ok")) self.addCleanup(self.provider.close) self.held = [] profiles = {} for name in ("alpha", "beta"): profiles[f"{name}-{self.provider.route}"] = self.profile( self.provider.url + f"/{name}/v1", f"fixture-{name}-{self.provider.route}", ) for number in range(STALLED): profiles[f"stalled{number}-{self.provider.route}"] = self.profile( self.stall(), f"fixture-stalled{number}-{self.provider.route}" ) self.alpha = f"alpha-{self.provider.route}" self.beta = f"beta-{self.provider.route}" self.model = f"fixture-beta-{self.provider.route}" self.app = Albedo(self.provider, providers=profiles).__enter__() self.addCleanup(self.app.__exit__, None, None, None)
def profile(self, url, model): return { "extension": "openai", "baseUrl": url, "apiKey": "key", "model": model, "protocol": "chat_completions", }
def stall(self): """An endpoint url whose server accepts connections and never answers.""" server = socket.create_server(("127.0.0.1", 0)) connections = []
def hold(): while True: try: connections.append(server.accept()[0]) except OSError: return
threading.Thread(target=hold, daemon=True).start() self.held.append(connections) self.addCleanup(lambda: [c.close() for c in connections]) self.addCleanup(server.close) return f"http://127.0.0.1:{server.getsockname()[1]}/v1"
def spawn(self, parent, name): session = operation_id() body = { "kind": "child", "parent_id": parent, "address": name, "name": name, "initial_input_id": operation_id(), "task": "scan", "model": self.model, } with self.app.api( f"/sessions/{session}", body, method="PUT", headers={"If-None-Match": "*"} ) as response: return json.load(response)
def test_stalled_endpoints_are_asked_together_not_in_sequence(self): parent = self.app.session() began = time.monotonic() made = self.spawn(parent, "scout") elapsed = time.monotonic() - began self.assertEqual(made["provider_profile"], self.beta) self.assertLess(elapsed, TOGETHER_SECONDS) self.assertEqual([len(c) for c in self.held], [1] * STALLED)
def test_spawns_that_arrive_together_share_one_connection_per_endpoint(self): parent = self.app.session() made = [] threads = [ threading.Thread(target=lambda n=n: made.append(self.spawn(parent, n))) for n in ("scout-1", "scout-2") ] for thread in threads: thread.start() for thread in threads: thread.join() self.assertEqual(len(made), 2) self.assertEqual([len(c) for c in self.held], [1] * STALLED)
if __name__ == "__main__": unittest.main()