Google Antigravity

#!/usr/bin/env python3
"""adom-antigravity: Google Antigravity in the Adom workspace.

One small CLI + the dock dashboard it serves. What it manages, all inside the user's
container (the WSL2 Adom-Workspace on Windows, the Lima container on a Mac):

  * the Google Antigravity VS Code extension (Open VSX: Google.google-antigravity), installed
    into code-server so it shows up in Hydrogen's editor as the Antigravity side bar;
  * the Antigravity CLI (`agy`, https://antigravity.google/cli/install.sh) in ~/.local/bin,
    which is what a "Antigravity TUI" terminal tab runs;
  * opening the Antigravity panel through adom-vscode's bridge.

    adom-antigravity status              # JSON: extension, cli, editor, container
    adom-antigravity install-extension   # code-server --install-extension Google.google-antigravity
    adom-antigravity install-cli         # the official installer, into ~/.local/bin/agy
    adom-antigravity open                # focus the Antigravity panel in the editor
    adom-antigravity serve --print-url   # the dock dashboard (what a dock click runs)
"""
import argparse, fcntl, glob, json, os, platform, shutil, signal, subprocess, sys, time, urllib.request, urllib.error, urllib.parse
from pathlib import Path
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer

ROOT = Path(__file__).resolve().parent
VERSION = '0.1.9'
EXT_ID = 'Google.google-antigravity'
CLI_INSTALL = 'https://antigravity.google/cli/install.sh'
EXT_DIRS = [Path.home() / '.local/share/code-server/extensions', Path.home() / '.vscode-server/extensions']


def sh(cmd, timeout=20):
    try:
        r = subprocess.run(cmd, shell=isinstance(cmd, str), capture_output=True, text=True, timeout=timeout)
        return r.returncode, (r.stdout or '') + (r.stderr or '')
    except subprocess.TimeoutExpired:
        return 124, f'timed out after {timeout}s'
    except Exception as e:
        return 1, str(e)


def runtime():
    """wsl2 | lima | linux: where this container runs, for the copy the dashboard shows."""
    try:
        pv = Path('/proc/version').read_text().lower()
        if 'microsoft' in pv or 'wsl' in pv:
            return 'wsl2'
    except Exception:
        pass
    if os.environ.get('LIMA_CIDATA_MNT') or Path('/etc/lima').exists() or Path('/mnt/lima-cidata').exists():
        return 'lima'
    return 'linux'


def extension_state():
    hits = []
    for d in EXT_DIRS:
        hits += glob.glob(str(d / (EXT_ID.lower() + '-*')))
    versions = sorted({Path(h).name.split('-', 2)[-1].split('-linux')[0] for h in hits})
    return {'id': EXT_ID, 'installed': bool(hits), 'versions': versions, 'paths': hits}


def _token_state(path):
    """An Antigravity OAuth token file: {token:{access_token,refresh_token,expiry}, id_token}.
    Signed in = a refresh token is there. The email comes from the id_token's payload."""
    try:
        d = json.loads(Path(path).read_text())
    except Exception:
        return {'signedIn': False, 'email': None}
    tok = d.get('token') or {}
    email = None
    try:
        import base64
        seg = d['id_token'].split('.')[1]
        seg += '=' * (-len(seg) % 4)
        email = json.loads(base64.urlsafe_b64decode(seg)).get('email')
    except Exception:
        pass
    return {'signedIn': bool(tok.get('refresh_token') or tok.get('access_token')), 'email': email, 'expiry': tok.get('expiry')}


def signin_state():
    """Both sides of the sign-in. The extension's agy server keeps its Google login in
    ~/.gemini/jetski-standalone-oauth-token; the CLI keeps ~/.gemini/antigravity-cli/
    antigravity-oauth-token. Same OAuth client and JSON (VERIFIED 2026-09-20), and Hydrogen's
    sign-in check copies whichever exists to the missing side, so in a Hydrogen workspace one
    sign-in covers both. Never a live prompt: a 429 quota answer is not "signed out"."""
    g = Path.home() / '.gemini'
    ext = _token_state(g / 'jetski-standalone-oauth-token')
    cli = _token_state(g / 'antigravity-cli' / 'antigravity-oauth-token')
    return {'extension': ext, 'cli': cli, 'signedIn': ext['signedIn'] or cli['signedIn'],
            'email': ext.get('email') or cli.get('email'),
            'sharedByHydrogen': (Path.home() / '.adom' / 'hydrogen-control-url').exists()}


def cli_state():
    agy = shutil.which('agy') or (str(Path.home() / '.local/bin/agy') if (Path.home() / '.local/bin/agy').exists() else None)
    version = None
    if agy:
        code, out = sh([agy, '--version'], timeout=8)
        version = out.strip().splitlines()[0][:80] if out.strip() else None
    cfg = Path.home() / '.gemini/antigravity-cli'
    settings = {}
    try:
        settings = json.loads((cfg / 'settings.json').read_text())
    except Exception:
        pass
    ext_copy = Path.home() / '.gemini' / 'bin' / 'agy'
    real = Path(agy).resolve() if agy else None
    return {'installed': bool(agy), 'path': agy, 'version': version,
            # The VS Code extension downloads its own agy to ~/.gemini/bin. The PATH command
            # should be a link onto it, never a second 214 MB download (no-duplicates rule).
            'extensionCopy': str(ext_copy) if ext_copy.exists() else None,
            'linkedToExtensionCopy': bool(real and ext_copy.exists() and real == ext_copy.resolve()),
            'duplicate': bool(real and ext_copy.exists() and real != ext_copy.resolve() and real.is_file()),
            'configDir': str(cfg), 'configDirExists': cfg.exists(),
            'modelProvider': settings.get('modelProvider'),
            'geminiApiKeySet': bool(os.environ.get('GEMINI_API_KEY'))}


def hub_state():
    """The extension runs `agy --hub --hub-port=N`: Antigravity's own web UI. In Hydrogen's
    editor (code-server) the extension's side-bar webview renders empty, but the hub page
    itself renders fine as a Hydrogen tab, so the dashboard opens THAT (verified 2026-09-06 on
    AdomLapper: sign-in screen, Continue with Google)."""
    port = None
    try:
        for d in Path('/proc').iterdir():
            if not d.name.isdigit():
                continue
            try:
                cmd = (d / 'cmdline').read_bytes().split(b'\0')
            except Exception:
                continue
            if any(c.endswith(b'/agy') or c == b'agy' for c in cmd[:1]) and b'--hub' in cmd:
                for c in cmd:
                    if c.startswith(b'--hub-port='):
                        port = int(c.split(b'=')[1])
    except Exception:
        pass
    up = False
    if port:
        try:
            up = urllib.request.urlopen(f'http://127.0.0.1:{port}/', timeout=1.5).status == 200
        except Exception:
            up = False
    return {'port': port, 'url': f'http://127.0.0.1:{port}/' if port else None, 'up': up}


def proxy_mode():
    """How the editor hands ports to webviews: 'subdomain' (http://<port>.localhost:<proxy>/,
    Hydrogen 1.0.224+, served by Hydrogen's own reverse proxy, the side-bar panel renders) or 'path' (/proxy/<port>/, the panel is
    blank because the hub's assets sit at absolute paths). Read from the extension host env."""
    try:
        for d in Path('/proc').iterdir():
            if not d.name.isdigit():
                continue
            try:
                cmd = (d / 'cmdline').read_bytes()
            except Exception:
                continue
            if b'extensionHost' not in cmd:
                continue
            env = (d / 'environ').read_bytes().split(b'\0')
            for e in env:
                if e.startswith(b'VSCODE_PROXY_URI='):
                    v = e.split(b'=', 1)[1].decode()
                    return {'uri': v, 'mode': 'subdomain' if '{{port}}.' in v else 'path'}
    except Exception:
        pass
    return {'uri': None, 'mode': 'unknown'}


def editor_state():
    for port in (8821, 8822, 8823):
        try:
            h = json.load(urllib.request.urlopen(f'http://127.0.0.1:{port}/health', timeout=1.5))
            if h.get('extension') == 'adom-vscode':
                return {'reachable': True, 'port': port, 'adomVscode': h.get('version'), 'aiLauncher': 'ai.new' in (h.get('verbs') or [])}
        except Exception:
            continue
    return {'reachable': False, 'port': None, 'adomVscode': None, 'aiLauncher': False}


def context():
    ext, cli, ed, hub, pm = extension_state(), cli_state(), editor_state(), hub_state(), proxy_mode()
    ready = ext['installed'] and cli['installed']
    return {'app': 'adom/antigravity', 'version': VERSION, 'runtime': runtime(),
            'container': {'system': platform.system(), 'kernel': platform.release(), 'hostname': platform.node(), 'arch': platform.machine()},
            'extension': ext, 'cli': cli, 'signin': signin_state(), 'editor': ed, 'hub': hub, 'proxy': pm, 'panelWorks': pm['mode'] == 'subdomain', 'ready': ready,
            'tools': {n: shutil.which(n) for n in ['code-server', 'adom-vscode', 'adom-wiki', 'adom-bridge', 'agy']},
            '_hint': 'extension.installed = the Google Antigravity VS Code extension is in code-server (reload the editor after a fresh install); cli.installed = agy is on PATH (a link onto the extension\'s own ~/.gemini/bin/agy; cli.duplicate flags a second copy). signin = the Google sign-in read from the token files (extension and CLI; Hydrogen keeps them in sync), never a live prompt. POST /api/install-extension, /api/install-cli, /api/open, /api/reload-editor drive the same steps the dashboard buttons do.'}


def install_extension():
    cs = shutil.which('code-server') or '/usr/bin/code-server'
    code, out = sh([cs, '--install-extension', EXT_ID, '--force'], timeout=240)
    ok = code == 0 and ('successfully installed' in out or 'already installed' in out)
    return {'ok': ok, 'exit': code, 'output': out[-1500:], 'extension': extension_state(),
            'next': 'Reload the editor window so the Antigravity side bar appears (POST /api/reload-editor, or the dashboard button).' if ok else 'code-server could not install it; the output above says why (network to open-vsx.org is the usual cause).'}


def install_cli():
    """One copy of agy, ever. The extension downloads its own to ~/.gemini/bin/agy; when it is
    there, ~/.local/bin/agy becomes a link onto it (the same rule Hydrogen's Manage AIs and
    hydrogen-bootstrap apply). The public installer runs only when no copy exists anywhere."""
    bindir = Path.home() / '.local/bin'
    bindir.mkdir(parents=True, exist_ok=True)
    link = bindir / 'agy'
    ext_copy = Path.home() / '.gemini' / 'bin' / 'agy'
    if ext_copy.exists() and os.access(ext_copy, os.X_OK):
        if link.is_symlink() or link.exists():
            link.unlink()
        link.symlink_to(ext_copy)
        code, out = 0, f'linked {link} -> {ext_copy} (the copy the Antigravity extension downloaded; nothing downloaded again)'
    else:
        code, out = sh(f'curl -fsSL {CLI_INSTALL} | bash -s -- -d "{bindir}"', timeout=300)
    st = cli_state()
    si = signin_state()
    nxt = ('Signed in as %s; the CLI shares the extension\'s sign-in.' % si['email']) if si['signedIn'] else 'Sign in inside the Antigravity side bar (Continue with Google); in a Hydrogen workspace the CLI picks that sign-in up on its own. Or run `agy` in a terminal and paste the code it asks for.'
    return {'ok': st['installed'], 'exit': code, 'output': out[-1500:], 'cli': st, 'signin': si,
            'next': nxt if st['installed'] else 'The installer did not leave ~/.local/bin/agy; the output above says why.'}


def vscode_command(command, args=None, timeout=20):
    ed = editor_state()
    if not ed['reachable']:
        return {'ok': False, 'error': 'adom-vscode is not answering in the editor (is Hydrogen open?)'}
    body = json.dumps({'command': command, 'args': args or []}).encode()
    req = urllib.request.Request(f'http://127.0.0.1:{ed["port"]}/command', data=body, headers={'Content-Type': 'application/json'}, method='POST')
    try:
        r = urllib.request.urlopen(req, timeout=timeout)
        return {'ok': True, 'command': command, 'result': json.loads(r.read() or b'{}')}
    except urllib.error.HTTPError as e:
        return {'ok': False, 'command': command, 'error': e.read().decode()[:500]}
    except Exception as e:
        return {'ok': False, 'command': command, 'error': str(e)}


def open_panel(new_conversation=False):
    if not extension_state()['installed']:
        return {'ok': False, 'error': 'the Antigravity extension is not installed yet', 'next': 'POST /api/install-extension, then reload the editor'}
    hub = hub_state()
    if hub['up'] and proxy_mode()['mode'] != 'subdomain':
        # the editor is still path-proxying ports (Hydrogen before 1.0.224, or Web Hydrogen):
        # the side-bar panel is blank there, so the dashboard tab navigates to the hub itself.
        return {'ok': True, 'surface': 'hub', 'url': hub['url'], 'hub': hub,
                'note': 'Antigravity\'s own web UI, served by the extension\'s hub inside the container, opened as a Hydrogen tab because this editor path-proxies ports and the side-bar panel is blank there.'}
    r = vscode_command('antigravity.startNewConversation' if new_conversation else 'antigravity.panel.focus')
    if not r.get('ok') and new_conversation:
        r = vscode_command('antigravity.panel.focus')
    r['surface'] = 'side bar'
    r['note'] = 'The hub is not up yet (the extension starts it after an editor reload); focused the side-bar panel instead.'
    return r


class Handler(SimpleHTTPRequestHandler):
    def __init__(self, *a, **kw):
        super().__init__(*a, directory=str(ROOT), **kw)

    def log_message(self, *a):
        pass

    def json(self, obj, status=200):
        b = json.dumps(obj).encode()
        self.send_response(status); self.send_header('Content-Type', 'application/json'); self.send_header('Content-Length', str(len(b))); self.end_headers(); self.wfile.write(b)

    def proxy(self):
        path = Path.home() / '.adom/hydrogen-control-url'
        base = path.read_text().strip() if path.exists() else 'http://127.0.0.1:47084'
        data = self.rfile.read(int(self.headers.get('Content-Length', '0'))) if self.command == 'POST' else None
        req = urllib.request.Request(base.rstrip('/') + self.path, data=data, method=self.command, headers={'Content-Type': 'application/json'})
        try:
            r = urllib.request.urlopen(req, timeout=120)
        except urllib.error.HTTPError as e:
            r = e
        except Exception as e:
            return self.json({'ok': False, 'error': 'Hydrogen control API unavailable', 'detail': str(e)}, 502)
        b = r.read(); self.send_response(r.status); self.send_header('Content-Type', r.headers.get('Content-Type', 'application/json')); self.send_header('Content-Length', str(len(b))); self.end_headers(); self.wfile.write(b)

    def do_GET(self):
        if self.path == '/api/health':
            # Dock readiness must not wait for agy or editor port discovery.
            return self.json({'app': 'adom/antigravity', 'version': VERSION,
                              'pid': os.getpid(), 'state': 'healthy'})
        if self.path in ['/api/status', '/state']:
            return self.json(context())
        if self.path == '/events':
            b = ('data: ' + json.dumps(context()) + '\n\n').encode()
            self.send_response(200); self.send_header('Content-Type', 'text/event-stream'); self.send_header('Content-Length', str(len(b))); self.end_headers(); self.wfile.write(b); return
        if self.path == '/updater/status' or self.path.startswith('/ai-threads/run-prompt/status?'):
            return self.proxy()
        if self.path == '/':
            self.path = '/dashboard/index.html'
        if not self.path.startswith(('/dashboard/', '/docs/')):
            return self.send_error(404)
        return super().do_GET()

    def do_POST(self):
        origin = self.headers.get('Origin')
        if origin and urllib.parse.urlsplit(origin).netloc != self.headers.get('Host'):
            return self.json({'error': 'Origin refused'}, 403)
        if self.headers.get('Content-Type', '').split(';')[0] != 'application/json':
            return self.json({'error': 'JSON required'}, 415)
        if self.path == '/ai-threads/run-prompt':
            return self.proxy()
        try:
            body = json.loads(self.rfile.read(int(self.headers.get('Content-Length', '0'))) or b'{}')
        except Exception:
            body = {}
        if self.path == '/api/install-extension':
            return self.json(install_extension())
        if self.path == '/api/install-cli':
            return self.json(install_cli())
        if self.path == '/api/open':
            if body.get('panel'):
                r = vscode_command('antigravity.panel.focus'); r['surface'] = 'side bar'; return self.json(r)
            return self.json(open_panel(bool(body.get('new'))))
        if self.path == '/api/reload-editor':
            return self.json(vscode_command('workbench.action.reloadWindow'))
        if self.path == '/api/tui':
            ed = editor_state()
            if not ed['reachable']:
                return self.json({'ok': False, 'error': 'adom-vscode is not answering in the editor'})
            req = urllib.request.Request(f'http://127.0.0.1:{ed["port"]}/terminal/tui', data=json.dumps({'name': 'Antigravity TUI', 'command': 'export ADOM_AI_THREAD=tui ADOM_AI_THREAD_AI=antigravity ADOM_AI_THREAD_LABEL=Antigravity_TUI; clear; exec agy', 'viewColumn': 'active', 'icon': 'adom:antigravity', 'color': 'terminal.ansiBlue'}).encode(), headers={'Content-Type': 'application/json'}, method='POST')
            try:
                return self.json(json.loads(urllib.request.urlopen(req, timeout=20).read()))
            except Exception as e:
                return self.json({'ok': False, 'error': str(e)})
        return self.send_error(404)


def server_identity(url):
    """Cheap readiness probe; allow older servers time for their full diagnostics."""
    try:
        return json.load(urllib.request.urlopen(url + 'api/health', timeout=1))
    except urllib.error.HTTPError as e:
        if e.code != 404:
            raise
    # 0.1.5's status can take 8s for agy, 4.5s for editor ports, and 1.5s
    # for the hub. A short timeout falsely declares that live server dead.
    return json.load(urllib.request.urlopen(url + 'api/status', timeout=20))


def retire_server(port):
    """Retire only our same-user Python server owning this loopback listener."""
    inodes = set()
    for line in Path('/proc/net/tcp').read_text().splitlines()[1:]:
        fields = line.split()
        if fields[1] == f'0100007F:{port:04X}' and fields[3] == '0A':
            inodes.add(fields[9])
    owners = []
    for proc in Path('/proc').iterdir():
        if not proc.name.isdigit():
            continue
        try:
            if proc.stat().st_uid != os.getuid():
                continue
            links = {os.readlink(fd) for fd in (proc / 'fd').iterdir()}
            if not any(f'socket:[{inode}]' in links for inode in inodes):
                continue
            argv = (proc / 'cmdline').read_bytes().rstrip(b'\0').split(b'\0')
            if (len(argv) != 5 or Path(os.fsdecode(argv[1])).name != Path(__file__).name
                    or argv[2:] != [b'_serve', b'--port', str(port).encode()]):
                raise RuntimeError('Refusing to stop an unrecognized port owner')
            owners.append(int(proc.name))
        except (FileNotFoundError, PermissionError, ProcessLookupError):
            continue
    if len(owners) != 1:
        raise RuntimeError('Cannot identify exactly one Antigravity server owning the port')
    pid = owners[0]
    # A pidfd binds the signal to this process, even if Linux later reuses its PID.
    with os.fdopen(os.pidfd_open(pid), 'rb') as process_handle:
        argv = Path(f'/proc/{pid}/cmdline').read_bytes().rstrip(b'\0').split(b'\0')
        if (len(argv) != 5 or Path(os.fsdecode(argv[1])).name != Path(__file__).name
                or argv[2:] != [b'_serve', b'--port', str(port).encode()]):
            raise RuntimeError('Port owner changed before retirement')
        signal.pidfd_send_signal(process_handle.fileno(), signal.SIGTERM)
    for _ in range(50):
        try:
            alive = bool(Path(f'/proc/{pid}/cmdline').read_bytes())
        except FileNotFoundError:
            alive = False
        if not alive:
            return
        time.sleep(.1)
    raise RuntimeError('Previous dashboard did not stop; refusing to force-kill it')


def ensure_server(port, log):
    url = f'http://127.0.0.1:{port}/'
    try:
        running = server_identity(url)
    except (OSError, ValueError):
        running = None
    if running:
        if running.get('app') != 'adom/antigravity':
            raise RuntimeError('Port is occupied by a different service')
        if running.get('version') == VERSION:
            return url
        old_version = tuple(int(v) for v in running['version'].split('.'))
        if old_version > tuple(int(v) for v in VERSION.split('.')):
            raise RuntimeError('Refusing to replace a newer Antigravity dashboard')
        retire_server(port)
    with (log / 'dashboard.log').open('ab') as out:
        process = subprocess.Popen([sys.executable, str(Path(__file__).resolve()), '_serve', '--port', str(port)], stdin=subprocess.DEVNULL, stdout=out, stderr=out, start_new_session=True)
    for _ in range(50):
        if process.poll() is not None:
            raise RuntimeError('Dashboard process exited; inspect ' + str(log / 'dashboard.log'))
        try:
            ready = server_identity(url)
            if ready.get('app') == 'adom/antigravity' and ready.get('version') == VERSION:
                return url
        except (OSError, ValueError):
            pass
        time.sleep(.1)
    raise RuntimeError('Dashboard failed to start; inspect ' + str(log / 'dashboard.log'))


def main():
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument('command', choices=['status', 'context', 'install-extension', 'install-cli', 'open', 'serve', '_serve'])
    p.add_argument('--print-url', action='store_true'); p.add_argument('--port', type=int, default=47867); p.add_argument('--new', action='store_true')
    a = p.parse_args()
    if a.command in ('status', 'context'):
        print(json.dumps(context(), indent=2)); return
    if a.command == 'install-extension':
        r = install_extension(); print(json.dumps(r, indent=2)); sys.exit(0 if r['ok'] else 1)
    if a.command == 'install-cli':
        r = install_cli(); print(json.dumps(r, indent=2)); sys.exit(0 if r['ok'] else 1)
    if a.command == 'open':
        r = open_panel(a.new); print(json.dumps(r, indent=2)); sys.exit(0 if r.get('ok') else 1)
    if a.command == '_serve':
        ThreadingHTTPServer(('127.0.0.1', a.port), Handler).serve_forever(); return
    log = Path.home() / '.local/state/adom/antigravity'; log.mkdir(parents=True, exist_ok=True)
    with (log / f'launch-{a.port}.lock').open('a') as lock:
        fcntl.flock(lock, fcntl.LOCK_EX)
        url = ensure_server(a.port, log)
    print(url, flush=True)


if __name__ == '__main__':
    main()