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.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495"""GitHub device authorization and account-scoped rosters on the shared provider server."""
import jsonimport threadingfrom urllib.parse import parse_qs
from harness import Provider, text
class CopilotProvider(Provider): def __init__(self, rosters): super().__init__(lambda _: text("copilot ready")) self.rosters = rosters self.auto_sessions = {} self.seen = [] self.authorised = threading.Event() self.polled = threading.Event() self.oauth_token = "fixture-copilot-device-token"
def record(self, handler, body=None): self.seen.append( { "path": handler.path.removeprefix(f"/t/{self.route}"), "headers": { name.lower(): value for name, value in handler.headers.items() }, "body": body, } )
def respond(self, handler, status, payload): body = json.dumps(payload).encode() handler.send_response(status) handler.send_header("Content-Type", "application/json") handler.send_header("Content-Length", str(len(body))) handler.end_headers() handler.wfile.write(body)
def _get(owner, self): owner.record(self) path = self.path.removeprefix(f"/t/{owner.route}") if path == "/user": return owner.respond(self, 200, {"id": 42, "login": "fixture-copilot-user"}) if path == "/copilot_internal/user": return owner.respond(self, 200, {"endpoints": {"api": owner.url}}) if path == "/models": token = self.headers.get("Authorization", "").removeprefix("Bearer ") if token not in owner.rosters: return owner.respond( self, 401, {"error": {"message": "unknown fixture account"}} ) return owner.respond(self, 200, {"data": owner.rosters[token]}) return owner.respond(self, 404, {"error": "not_found"})
def _post(owner, self): path = self.path.removeprefix(f"/t/{owner.route}") if path == "/models/session": body = json.loads(self.rfile.read(int(self.headers["Content-Length"]))) owner.record(self, body) token = self.headers.get("Authorization", "").removeprefix("Bearer ") grant = owner.auto_sessions.get(token) if grant is None: return owner.respond(self, 404, {"error": "auto_not_available"}) status, payload = grant() if callable(grant) else grant return owner.respond(self, status, payload) if path.startswith("/login/"): body = self.rfile.read(int(self.headers["Content-Length"])) form = {key: values[0] for key, values in parse_qs(body.decode()).items()} owner.record(self, form) if path == "/login/device/code": return owner.respond( self, 200, { "device_code": "fixture-device-code", "user_code": "FIXT-1234", "verification_uri": owner.url + "/login/device", "expires_in": 60, "interval": 1, }, ) if path == "/login/oauth/access_token": authorised = owner.authorised.is_set() owner.polled.set() if not authorised: return owner.respond(self, 200, {"error": "authorization_pending"}) return owner.respond( self, 200, {"access_token": owner.oauth_token, "token_type": "bearer"}, ) return owner.respond(self, 404, {"error": "not_found"}) owner.record(self) return super()._post(self)