Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli
Something went wrong. Try again.
Python
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206"""The OpenAI-compatible proxy on a real daemon against fixture providers. No live model."""import http.serverimport jsonimport osfrom pathlib import Pathimport socketimport subprocessimport tempfileimport threadingimport urllib.errorimport urllib.request
ROOT = Path(__file__).resolve().parents[2]
class Provider(http.server.BaseHTTPRequestHandler): requests = []
def log_message(self, *_): pass
def do_POST(self): request = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) self.requests.append((self.path, request)) self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.end_headers() send = lambda value: self.wfile.write(("data: " + json.dumps(value) + "\n\n").encode()) if self.path.endswith("/chat/completions"): last = request["messages"][-1] if last["role"] == "tool": send({"id": "c", "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ran it"}, "finish_reason": None}]}) send({"id": "c", "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}) else: send({"id": "c", "choices": [{"index": 0, "delta": {"role": "assistant", "reasoning_content": "hmm"}, "finish_reason": None}]}) send({"id": "c", "choices": [{"index": 0, "delta": {"content": "hel"}, "finish_reason": None}]}) send({"id": "c", "choices": [{"index": 0, "delta": {"tool_calls": [{"index": 0, "id": "call-1", "type": "function", "function": {"name": "bash", "arguments": "{\"command\":"}}]}, "finish_reason": None}]}) send({"id": "c", "choices": [{"index": 0, "delta": {"tool_calls": [{"index": 0, "function": {"arguments": "\"ls\"}"}}]}, "finish_reason": None}]}) send({"id": "c", "choices": [{"index": 0, "delta": {}, "finish_reason": "tool_calls"}], "usage": {"prompt_tokens": 7, "completion_tokens": 3}}) self.wfile.write(b"data: [DONE]\n\n") else: last = request["input"][-1] if last.get("role") == "user" and last.get("content") == "call a tool": output = [ {"type": "reasoning", "id": "rs_1", "summary": [], "encrypted_content": "sealed-reasoning"}, {"type": "function_call", "id": "fc_1", "call_id": "call_r1", "name": "bash", "arguments": "{\"command\":\"ls\"}", "status": "completed"}, ] else: output = [{"type": "message", "role": "assistant", "status": "completed", "content": [{"type": "output_text", "text": "done", "annotations": []}]}] send({"type": "response.completed", "response": {"id": "r", "status": "completed", "output": output, "usage": {"input_tokens": 5, "output_tokens": 1}}})
def free_port(): with socket.socket() as probe: probe.bind(("127.0.0.1", 0)) return probe.getsockname()[1]
def post(url, body, headers=None): request = urllib.request.Request(url, json.dumps(body).encode(), {"Content-Type": "application/json", **(headers or {})}) return urllib.request.urlopen(request, timeout=30)
def status(call): try: with call() as response: return response.status except urllib.error.HTTPError as error: return error.code
def chunks(response): events = [line[6:] for line in response.read().decode().splitlines() if line.startswith("data: ")] assert events[-1] == "[DONE]", events return [json.loads(event) for event in events[:-1]]
def main(): server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Provider) threading.Thread(target=server.serve_forever, daemon=True).start() endpoint = f"http://127.0.0.1:{server.server_address[1]}/v1" with tempfile.TemporaryDirectory(prefix="albedo-proxy-") as directory: home, user_home = Path(directory)/"state", Path(directory)/"user" home.mkdir(mode=0o700), user_home.mkdir(mode=0o700) (home/"config.json").write_text(json.dumps({"active": "chat", "providers": { "chat": {"baseUrl": endpoint, "apiKey": "k", "model": "m-chat", "protocol": "chat_completions"}, "resp": {"baseUrl": endpoint, "apiKey": "k", "model": "m-resp", "protocol": "responses"}, "cx": {"extension": "codex", "model": "gpt-saved", "protocol": "responses"}, "broken": {"model": 3}, }})) # A cached models.dev catalog that also knows the fixture endpoint. (home/"models.json").write_text(json.dumps({ "openai": {"api": "https://api.openai.com/v1", "models": {"gpt-a": {"id": "gpt-a"}, "gpt-b": {"id": "gpt-b"}}}, "local": {"api": endpoint, "models": {"m-chat": {"id": "m-chat"}, "catalog-only": {"id": "catalog-only"}}}, })) settings = home/"extensions.json" settings.write_text(json.dumps({"models": {"refreshHours": 0}})) port = free_port() env = dict(os.environ, HOME=str(user_home), ALBEDO_HOME=str(home), ALBEDO_PORT=str(port), ALBEDO_PARENT_PID=str(os.getpid())) result = subprocess.run([str(ROOT / "cli/bin/albedo"), "sessions"], cwd=ROOT, env=env, capture_output=True, text=True, timeout=60) assert result.returncode == 0, result.stdout + result.stderr assert json.loads((home/"daemon.json").read_text())["port"] == port base = f"http://127.0.0.1:{port}/proxy/v1" models = lambda: urllib.request.urlopen(base + "/models", timeout=10)
# Disabled by default: the path falls through to the token-guarded daemon. assert status(models) == 403 settings.write_text(json.dumps({"models": {"refreshHours": 0}, "enabled": {"proxy": True}})) with models() as response: listing = json.load(response) ids = {model["id"] for model in listing["data"]} # Endpoint profiles list their saved model; catalog profiles list the catalog. # A malformed profile is reported without hiding the usable ones. assert ids == {"chat/m-chat", "resp/m-resp", "cx/gpt-a", "cx/gpt-b"}, ids assert [error["profile"] for error in listing["errors"]] == ["broken"], listing # Listing is only a hint: an unlisted model id is requested as given. with post(base + "/chat/completions", {"model": "chat/unlisted", "messages": [{"role": "user", "content": "hi"}]}) as response: assert json.load(response)["model"] == "chat/unlisted" assert Provider.requests[-1][1]["model"] == "unlisted", Provider.requests[-1] # Broken and signed-out profiles fail only their own requests. for model in ["broken", "cx/gpt-a"]: try: post(base + "/chat/completions", {"model": model, "messages": [{"role": "user", "content": "hi"}]}) raise AssertionError(model + " should fail") except urllib.error.HTTPError as error: body = json.load(error) assert error.code == 400 and "/login" in body["error"]["message"], (model, error.code, body) assert status(lambda: post(base + "/chat/completions", {}, {"Origin": "https://example.com"})) == 403
tools = [{"type": "function", "function": {"name": "bash", "parameters": {"type": "object"}}}] opening = [{"role": "system", "content": "be brief"}, {"role": "user", "content": "list files"}] with post(base + "/chat/completions", {"model": "chat", "messages": opening, "tools": tools, "stream": True, "stream_options": {"include_usage": True}}) as response: assert response.headers["content-type"].startswith("text/event-stream") stream = chunks(response) deltas = [c["choices"][0]["delta"] for c in stream if c["choices"]] assert deltas[0]["role"] == "assistant" assert "".join(d.get("content", "") for d in deltas) == "hel" assert "".join(d.get("reasoning_content", "") for d in deltas) == "hmm" calls = [call for d in deltas for call in d.get("tool_calls", [])] [streamed_call] = calls assert streamed_call["id"].startswith("call-1__albedo__"), streamed_call assert streamed_call["function"] == {"name": "bash", "arguments": "{\"command\":\"ls\"}"}, streamed_call assert [c["choices"][0]["finish_reason"] for c in stream if c["choices"]][-1] == "tool_calls" assert stream[-1]["usage"]["prompt_tokens"] == 7 and stream[-1]["choices"] == [] assert all(c["model"] == "chat" for c in stream) path, sent = Provider.requests[-1] assert path == "/v1/chat/completions" and sent["model"] == "m-chat", (path, sent) assert sent["messages"][0] == {"role": "system", "content": "be brief"}, sent["messages"]
history = opening + [ {"role": "assistant", "content": "hel", "tool_calls": [{"id": "call-1", "type": "function", "function": {"name": "bash", "arguments": "{\"command\":\"ls\"}"}}]}, {"role": "tool", "tool_call_id": "call-1", "content": "a.txt"}, ] for model, check in (("chat/m-chat", "chat"), ("resp/m-resp", "responses")): with post(base + "/chat/completions", {"model": model, "messages": history, "tools": tools}) as response: completion = json.load(response) assert completion["object"] == "chat.completion", completion assert completion["choices"][0]["finish_reason"] == "stop", completion path, sent = Provider.requests[-1] if check == "chat": assert [m["role"] for m in sent["messages"]] == ["system", "user", "assistant", "tool"], sent assert sent["messages"][2]["tool_calls"][0]["id"] == "call-1" else: assert path == "/v1/responses" and sent["instructions"] == "be brief", sent kinds = [item.get("type", item.get("role")) for item in sent["input"]] assert kinds == ["user", "assistant", "function_call", "function_call_output"], sent["input"] assert sent["input"][3] == {"type": "function_call_output", "call_id": "call-1", "output": "a.txt"}
# Provider state rides back through the client in the tool-call id. ask = [{"role": "user", "content": "call a tool"}] with post(base + "/chat/completions", {"model": "resp", "messages": ask, "tools": tools}) as response: first = json.load(response)["choices"][0]["message"] [call] = first["tool_calls"] assert call["id"].startswith("call_r1__albedo__"), call echoed = ask + [{"role": "assistant", "content": None, "tool_calls": [call]}, {"role": "tool", "tool_call_id": call["id"], "content": "a.txt"}] with post(base + "/chat/completions", {"model": "resp", "messages": echoed, "tools": tools}): pass path, sent = Provider.requests[-1] assert sent["input"][1] == {"type": "reasoning", "id": "rs_1", "summary": [], "encrypted_content": "sealed-reasoning"}, sent["input"] assert sent["input"][2]["call_id"] == "call_r1" and sent["input"][3]["call_id"] == "call_r1", sent["input"] # Another profile never sees that state; it gets the portable call. with post(base + "/chat/completions", {"model": "chat", "messages": echoed, "tools": tools}): pass path, sent = Provider.requests[-1] assert path == "/v1/chat/completions", path assert "sealed-reasoning" not in json.dumps(sent), sent assert sent["messages"][1]["tool_calls"][0]["id"] == "call_r1", sent["messages"] assert sent["messages"][2]["tool_call_id"] == "call_r1", sent["messages"]
bad = lambda: post(base + "/chat/completions", {"model": "missing/x", "messages": opening}) assert status(bad) == 400 print("proxy: enablement, models, streaming chunks, projection, carried provider state, origin refusal passed")
if __name__ == "__main__": main()