Something went wrong. Try again.
small gleam coding and (not yet) persistent agent daemon with a detachable cli forked from okami.mom/albedo
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119"""OpenCode Console's side of a sign-in, for opencode_go_login_test.py.
The same RFC 8628 flow as GitHub's but with a different shape: JSON bodies, anorganization to select, and a rotating refresh token in the response theprovider builds its credential from. It exists because `albedo_device_flow` isshared, and the provider that speaks JSON is the one no other test drives."""
import jsonimport threadingimport timefrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServerfrom typing import cast
ACCESS_TOKEN = "opencode-fixture-access"REFRESH_TOKEN = "opencode-fixture-refresh"ORG = {"id": "org-1", "name": "Fixture Workspace"}
class Console(ThreadingHTTPServer): daemon_threads = True
def __init__(self, interval=1): super().__init__(("127.0.0.1", 0), Handler) self.interval = interval self.authorized = False self.seen = [] self.lock = threading.Lock() self.org_requested = threading.Event() self.org_release = threading.Event() self.org_release.set()
@property def url(self): return f"http://127.0.0.1:{self.server_address[1]}"
def start(self): threading.Thread(target=self.serve_forever, daemon=True).start() return self
def close(self) -> None: self.shutdown() self.server_close()
def authorize(self): with self.lock: self.authorized = True
def requests(self, path): with self.lock: return [entry for entry in self.seen if entry["path"] == path]
class Handler(BaseHTTPRequestHandler): protocol_version = "HTTP/1.1"
@property def console(self) -> Console: return cast(Console, self.server)
def log_message(self, format, *args): pass
def record(self, body=None): with self.console.lock: self.console.seen.append( {"path": self.path, "body": body or {}, "at": time.time()} )
def do_GET(self): self.record() if self.path == "/api/user": return self.json(200, {"id": "user-1", "email": "ana@example.com"}) if self.path == "/api/orgs": self.console.org_requested.set() if not self.console.org_release.wait(10): return self.json(500, {"error": "fixture_not_released"}) return self.json(200, [ORG]) self.json(404, {"error": "not_found"})
def do_POST(self): body = self.rfile.read(int(self.headers.get("content-length") or 0)) form = json.loads(body or b"{}") self.record(form) if self.path == "/auth/device/code": return self.json( 200, { "device_code": "opencode-device-code", "user_code": "WXYZ-9876", "verification_uri_complete": f"{self.console.url}/device?code=WXYZ-9876", "expires_in": 600, "interval": self.console.interval, }, ) if self.path == "/auth/device/token": with self.console.lock: authorized = self.console.authorized if not authorized: return self.json(400, {"error": "authorization_pending"}) return self.json( 200, { "access_token": ACCESS_TOKEN, "refresh_token": REFRESH_TOKEN, "expires_in": 3600, "org_id": ORG["id"], }, ) self.json(404, {"error": "not_found"})
def json(self, status, value): body = json.dumps(value).encode() self.send_response(status) self.send_header("content-type", "application/json") self.send_header("content-length", str(len(body))) self.end_headers() self.wfile.write(body)