Codex

#!/usr/bin/env python3
"""Package-owned container integration. No marker means no modifications."""
import argparse
import fcntl
import json
import os
from pathlib import Path
import re
import shlex
import shutil
import subprocess
import sys
import time

from settings import change, entries
from walkthrough import patch_tour, restore_tour

ROOT = Path(__file__).resolve().parent
KEY = 'chatgpt.cliExecutable'
CRON_TAG = '# adom/codex runtime'
JS_BEGIN = '/* adom/codex renderer:begin */'
JS_END = '/* adom/codex renderer:end */'


def atomic(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + '.adom-codex.tmp')
    temporary.write_text(data)
    if path.exists(): temporary.chmod(path.stat().st_mode & 0o777)
    temporary.replace(path)


class Integration:
    def __init__(self, home):
        self.home = home.resolve()
        self.directory = self.home / '.local/share/adom/codex/runtime'
        self.state = self.home / '.local/state/adom/codex-integration'
        self.marker = self.state / 'installed.json'
        self.settings = self.home / '.local/share/code-server/User/settings.json'
        self.extensions = self.home / '.local/share/code-server/extensions'
        self.launcher = self.home / '.local/bin/adom-codex-runtime'

    def write_json(self, path, value):
        atomic(path, json.dumps(value, indent=2) + '\n')

    def install(self, isolated=False):
        if sys.platform != 'linux':
            return {'state': 'unsupported', 'reason': 'Runtime integration requires Linux code-server'}
        text = self.settings.read_text() if self.settings.exists() else '{}\n'
        current = {e[0]: e[1] for e in entries(text)[0]}
        previous = current.get(KEY)
        if previous and previous != str(self.launcher):
            return {'state': 'custom_cli_preserved', 'reason': 'Existing custom Codex CLI retained'}
        if self.launcher.exists() and not any(s in self.launcher.read_text() for s in
                ('adom/codex', '.local/share/adom/codex-runtime/runtime.py')):
            raise RuntimeError('Existing launcher is not owned by adom/codex')
        self.state.mkdir(parents=True, exist_ok=True, mode=0o700)
        self.state.chmod(0o700)
        if not self.marker.exists():
            backup = {'had_key': KEY in current, 'previous': previous}
            # Adopt the earlier prototype without restoring its own launcher
            # on uninstall. Its original setting remains the restore target.
            legacy = self.home / '.local/share/adom/codex-runtime/settings-before-install.json'
            if previous == str(self.launcher) and legacy.exists():
                old = json.loads(legacy.read_text())
                backup = {'had_key': old['had_cli_executable'], 'previous': old['previous_cli_executable']}
            self.write_json(self.marker, {'owner': 'adom/codex', 'installed_at': time.time(), **backup})
        if ROOT != self.directory:
            shutil.copytree(ROOT, self.directory, dirs_exist_ok=True,
                            ignore=shutil.ignore_patterns('__pycache__'))
        launcher = '#!/bin/sh\n# adom/codex runtime\nexec python3 ' + shlex.quote(str(self.directory / 'runtime.py')) + ' "$@"\n'
        atomic(self.launcher, launcher)
        self.launcher.chmod(0o755)
        atomic(self.settings, change(text, KEY, str(self.launcher)))
        report = self.reconcile()
        if not isolated:
            self.cron(True)
            self.ensure_watch()
        return {'state': 'installed', 'extensions': report}

    def reconcile(self):
        if not self.marker.exists(): return []
        result = []
        for extension in sorted(self.extensions.glob('openai.chatgpt-*')):
            manifest = extension / 'package.json'
            if not manifest.is_file(): continue
            try:
                package = json.loads(manifest.read_text())
                if package.get('publisher') != 'openai' or package.get('name') != 'chatgpt': continue
                result.append(self.patch(extension, package))
            except (OSError, ValueError) as error:
                result.append({'path': str(extension), 'state': 'pending', 'reason': str(error)})
        self.write_json(self.state / 'status.json', {'checked_at': time.time(), 'extensions': result})
        return result

    def patch(self, extension, package):
        tour = patch_tour(extension, atomic)
        main = (extension / package.get('main', 'out/extension.js')).resolve()
        if not main.is_relative_to(extension.resolve()): raise ValueError('Extension entry point escapes directory')
        text = main.read_text()
        # Match the semantic call site and capture minifier-generated names.
        pattern = r'(?P<w>[\w$]+)\.html=await this\.getWebviewContent\((?P=w)\),(?P<g>[\w$]+)\.start\(\)(?=\}onPanelReady\()'
        matches = list(re.finditer(pattern, text))
        if JS_BEGIN not in text:
            if len(matches) != 1:
                return {'version': package['version'], 'state': 'unsupported',
                        'reason': 'Renderer startup changed upstream; renderer left intact', 'introductory_tour': tour}
            m = matches[0]
            # Get role and disposal callback from initializeWebview's arguments.
            declarations = list(re.finditer(r'async initializeWebview\(([\w$]+),([\w$]+),([\w$]+),[\w$]+\)', text[:m.start()]))
            if not declarations: raise ValueError('Renderer initializer signature changed')
            w, role, disposal = declarations[-1].groups()
            if w != m['w']: raise ValueError('Renderer initializer does not match startup call')
            replacement = (f'{w}.html=await this.getWebviewContent({w}),' + JS_BEGIN +
                f'require({json.dumps(str(self.directory / "webview.cjs"))}).watchRenderer({w},{role},this,{m["g"]},{disposal})' + JS_END)
            patched = text[:m.start()] + replacement + text[m.end():]
            if main.read_text() != text: raise ValueError('Extension is changing; retrying later')
            atomic(main, patched)
            # Store just the reversible fragment, never restore a whole old VSIX.
            self.write_json(extension / '.adom-codex-patch.json',
                            {'owner': 'adom/codex', 'original': m[0], 'replacement': replacement,
                             'main': str(main.relative_to(extension.resolve()))})
        css = (self.directory / 'composer.css').read_text()
        styles = []
        for path in (extension / 'webview/assets').glob('*.css'):
            original = path.read_text()
            if '.vertical-scroll-fade-mask{' not in original: continue
            if '/* adom/codex composer fade: begin */' not in original:
                if path.read_text() != original: raise ValueError('Stylesheet changed during update')
                atomic(path, original + '\n' + css)
            styles.append(path.name)
        return {'version': package['version'], 'state': 'patched', 'composer_styles': styles, 'introductory_tour': tour}

    def cron(self, install):
        if not shutil.which('crontab'):
            raise RuntimeError('crontab is required to reconcile future VSIX updates')
        old = subprocess.run(['crontab', '-l'], capture_output=True, text=True)
        if old.returncode and 'no crontab' not in old.stderr.lower():
            raise RuntimeError('Could not read existing crontab: ' + old.stderr.strip())
        lines = [s for s in old.stdout.splitlines() if not s.endswith(CRON_TAG)]
        if install:
            cmd = 'python3 ' + shlex.quote(str(self.directory / 'manage.py')) + ' ensure-watch'
            # Cron treats percent specially even inside shell quotes.
            cmd = cmd.replace('%', r'\%')
            lines += [f'@reboot {cmd} >/dev/null 2>&1 {CRON_TAG}',
                      f'* * * * * {cmd} >/dev/null 2>&1 {CRON_TAG}']
        subprocess.run(['crontab', '-'], input='\n'.join(lines) + '\n', text=True, check=True)

    def ensure_watch(self):
        if not self.marker.exists(): return
        with (self.state / 'watch.lock').open('a') as guard:
            try: fcntl.flock(guard, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError: return
        with (self.state / 'watch.log').open('ab') as log:
            subprocess.Popen([sys.executable, str(self.directory / 'manage.py'), 'watch'],
                             stdin=subprocess.DEVNULL, stdout=log, stderr=log,
                             start_new_session=True, close_fds=True)

    def watch(self):
        if not self.marker.exists(): return
        with (self.state / 'watch.lock').open('a') as guard:
            try: fcntl.flock(guard, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError: return
            revision = (ROOT / 'manage.py').stat().st_mtime_ns
            while self.marker.exists() and (ROOT / 'manage.py').stat().st_mtime_ns == revision:
                try:
                    with (self.state / 'mutation.lock').open('a') as mutation:
                        fcntl.flock(mutation, fcntl.LOCK_EX)
                        self.reconcile()
                except Exception as error: print(str(error), flush=True)
                time.sleep(5)

    def uninstall(self, isolated=False):
        if not self.marker.exists(): return {'state': 'not_installed'}
        backup = json.loads(self.marker.read_text())
        with (self.state / 'mutation.lock').open('a') as mutation:
            fcntl.flock(mutation, fcntl.LOCK_EX)
            self.marker.unlink()
            if not isolated: self.cron(False)
            for extension in self.extensions.glob('openai.chatgpt-*'):
                restore_tour(extension, atomic)
            for record in self.extensions.glob('openai.chatgpt-*/.adom-codex-patch.json'):
                data = json.loads(record.read_text())
                main = (record.parent / data['main']).resolve()
                if data.get('owner') != 'adom/codex' or not main.is_relative_to(record.parent.resolve()): continue
                if main.exists():
                    text = main.read_text()
                    if data['replacement'] in text:
                        atomic(main, text.replace(data['replacement'], data['original'], 1))
                    elif JS_BEGIN in text:
                        raise RuntimeError('Managed renderer fragment was edited; retained for review')
                for css in (record.parent / 'webview/assets').glob('*.css'):
                    text = css.read_text()
                    restored = re.sub(r'\n/\* adom/codex composer fade: begin \*/.*?/\* adom/codex composer fade: end \*/\n?', '', text, flags=re.S)
                    if restored != text: atomic(css, restored)
                record.unlink()
            text = self.settings.read_text() if self.settings.exists() else '{}'
            current = {e[0]: e[1] for e in entries(text)[0]}
            if current.get(KEY) == str(self.launcher):
                atomic(self.settings, change(text, KEY, backup['previous'], remove=not backup['had_key']))
        # Retain the launcher and code until running work finishes. With the
        # setting restored, new extension hosts use upstream. Never kill turns.
        return {'state': 'uninstalled', 'running_work': 'preserved; close old Codex views after work finishes'}


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('command', choices=['install', 'uninstall', 'reconcile', 'ensure-watch', 'watch', 'status'])
    parser.add_argument('--home', type=Path)
    args = parser.parse_args()
    manager = Integration(args.home or Path.home())
    if args.command in ('install', 'uninstall'):
        result = getattr(manager, args.command)(isolated=args.home is not None)
    elif args.command == 'status':
        result = {'installed': manager.marker.exists(),
                  'detail': json.loads((manager.state / 'status.json').read_text()) if (manager.state / 'status.json').exists() else None}
    else: result = getattr(manager, args.command.replace('-', '_'))()
    if result is not None: print(json.dumps(result, indent=2))


if __name__ == '__main__': main()