#!/bin/bash
# macOS bash 3.2; all implementation and dependencies are Python standard library.
set -euo pipefail
exec "${KCO_PYTHON:-python3}" - "$0" "$@" <<'PY'
import argparse
import fcntl
import json
import os
from pathlib import Path
import plistlib
import re
import secrets
import shlex
import signal
import stat
import subprocess
import sys
import tempfile
import time
from urllib.parse import urlsplit
import uuid

HOME = Path(os.environ.get('KCO_HOME', str(Path.home()))).absolute()
FIXTURE = 'KCO_HOME' in os.environ
CONFIG = HOME / '.kapable/kapable.toml'
AGENTS = HOME / 'Library/LaunchAgents'
DAEMONS = Path(os.environ.get('KCO_LAUNCHDAEMONS_DIR', '/Library/LaunchDaemons'))
LABEL = 'dev.kapable.agent'
DOMAIN = 'gui/' + str(os.getuid())
TARGET = DOMAIN + '/' + LABEL
DRY_TRACE = False
JSON_TRACE = False
TOKEN_NOTES = {
    'service-token-file': 'Org owner/admin or keys.manage member: POST <org-url>/auth/v1/auth/service-tokens (token_type=ci); save the returned secret.',
    'v2-key-file': 'Org key manager: POST <org-url>/auth/v1/auth/api-keys; save the returned secret for this org.',
    'tunnel-token-file': 'Tunnel operator: obtain the credential for the existing Burrow tunnel registration.',
    'tunnel-ingress-key-file': 'Tunnel operator: obtain/set the ingress key for that tunnel (client sends ingress_key at registration).',
    'tunnel-id': 'Tunnel operator: obtain the registered UUID tunnel slug; this is not the conductor instance id.',
    'agent-id': 'Conductor operator: obtain the stable platform agent UUID from the org provisioning record.',
}


class Failure(Exception):
    pass


def guard(path):
    if FIXTURE:
        try:
            path.resolve().relative_to(HOME.resolve())
        except ValueError:
            raise Failure('Fixture path escapes KCO_HOME; refused.')
    return path


def fixture_guard():
    if FIXTURE:
        if HOME.resolve() == Path.home().resolve():
            raise Failure('KCO_HOME must not be the real home.')
        for key in ('KCO_LAUNCHCTL', 'KCO_LSOF', 'KCO_CURL', 'KCO_LAUNCHDAEMONS_DIR'):
            if not os.environ.get(key):
                raise Failure('Fixture mode requires ' + key + '; no real-tool fallback.')
            guard(Path(os.environ[key]))
        guard(CONFIG)
        guard(AGENTS)


def run(argv, timeout=10):
    if DRY_TRACE:
        emit('preflight', 'running', 'READ-ONLY CHECK ' + shlex.join([str(x) for x in argv]), JSON_TRACE)
    env = os.environ.copy()
    if FIXTURE:
        env['HOME'] = str(HOME)
    try:
        return subprocess.run([str(x) for x in argv], stdout=subprocess.PIPE,
                              stderr=subprocess.DEVNULL, text=True, timeout=timeout,
                              env=env, cwd=str(HOME))
    except (OSError, subprocess.TimeoutExpired):
        raise Failure('Command failed to execute or timed out; output suppressed.')


def tool(name, default):
    return os.environ.get('KCO_' + name, default)


def launch(*args):
    return run([tool('LAUNCHCTL', '/bin/launchctl')] + list(args))


def command(*args):
    return shlex.join([tool('LAUNCHCTL', '/bin/launchctl')] + list(args))


def toml_subset(source):
    """Fail-closed 3.9 parser: tables, arrays of tables, strings, scalar arrays,
    booleans and numbers. Handles comments inside/outside strings and multiline
    arrays. We preserve original TOML bytes; unsupported syntax is never rewritten.
    """
    statements, buf, quote, escaped, depth = [], '', None, False, 0
    comment = False
    for ch in source + '\n':
        if comment:
            if ch != '\n':
                continue
            comment = False
        if quote:
            buf += ch
            if escaped:
                escaped = False
            elif ch == '\\' and quote == '"':
                escaped = True
            elif ch == quote:
                quote = None
            continue
        if ch == '#':
            comment = True
        elif ch in ('"', "'"):
            quote = ch
            buf += ch
        elif ch == '\n' and depth == 0:
            if buf.strip():
                statements.append(buf.strip())
            buf = ''
        else:
            depth += (ch == '[') - (ch == ']')
            buf += ch
    if quote or depth or buf.strip():
        raise Failure('Unsupported or incomplete supervisor TOML.')

    def value(s):
        if s.startswith('[') and s.endswith(']'):
            parts, part, q, esc = [], '', None, False
            for c in s[1:-1] + ',':
                if q:
                    part += c
                    if esc:
                        esc = False
                    elif c == '\\' and q == '"':
                        esc = True
                    elif c == q:
                        q = None
                elif c in ('"', "'"):
                    q = c
                    part += c
                elif c == ',':
                    if part.strip():
                        parts.append(value(part.strip()))
                    part = ''
                else:
                    part += c
            if q:
                raise Failure('Unsupported TOML array.')
            return parts
        if s.startswith('"'):
            return json.loads(s)
        if s.startswith("'") and s.endswith("'") and "'" not in s[1:-1]:
            return s[1:-1]
        if s in ('true', 'false'):
            return s == 'true'
        if re.fullmatch(r'[+-]?\d+', s):
            return int(s)
        if re.fullmatch(r'[+-]?\d+\.\d+', s):
            return float(s)
        raise Failure('Unsupported supervisor TOML value; refused to guess.')

    root, table = {}, None
    for s in statements:
        if s.startswith('['):
            array = s.startswith('[[')
            end = ']]' if array else ']'
            if not s.endswith(end):
                raise Failure('Invalid TOML table.')
            keys = s[2:-2].split('.') if array else s[1:-1].split('.')
            table = root
            for i, key in enumerate(keys):
                key = key.strip()
                if not re.fullmatch(r'[A-Za-z0-9_-]+', key):
                    raise Failure('Unsupported TOML table name.')
                if array and i == len(keys) - 1:
                    table.setdefault(key, []).append({})
                else:
                    table.setdefault(key, {})
                table = table[key]
                if isinstance(table, list):
                    table = table[-1]
        else:
            key, sep, val = s.partition('=')
            key = key.strip()
            dest = root if table is None else table
            if not sep or not re.fullmatch(r'[A-Za-z0-9_-]+', key) or key in dest:
                raise Failure('Unsupported or duplicate TOML key.')
            dest[key] = value(val.strip())
    return root


def read_config():
    if not CONFIG.exists():
        return b'', {}
    raw = guard(CONFIG).read_bytes()
    try:
        data = toml_subset(raw.decode())
        try:
            import tomllib
        except ImportError:
            pass
        else:
            if tomllib.loads(raw.decode()) != data:
                raise Failure('TOML parser disagreement; refused.')
        return raw, data
    except (ValueError, TypeError, KeyError, AttributeError):
        raise Failure('Invalid or unsupported supervisor TOML; content suppressed.')


def plist(path):
    try:
        return plistlib.loads(guard(path).read_bytes())
    except (OSError, ValueError, plistlib.InvalidFileException):
        raise Failure('Cannot read a launchd plist; content suppressed.')


def flags(args):
    return {v: args[i + 1] for i, v in enumerate(args[:-1])
            if isinstance(v, str) and v.startswith('--') and not str(args[i + 1]).startswith('--')}


def inventory(data):
    orgs, ports = [], set()
    for org in data.get('orgs', []):
        workers = org.get('workers', [])
        ports.update(w['port'] for w in workers if isinstance(w.get('port'), int) and w['port'] > 0)
        w = next((w for w in workers if w.get('name') == 'conductor'), None)
        orgs.append({'slug': org['slug'], 'mode': 'supervisor', 'port': w.get('port') if w else None,
                     'flags': flags(w.get('args', [])) if w else {}, 'configured': w is not None,
                     'source': str(CONFIG), 'worker': w})
    for path in sorted(DAEMONS.glob('dev.kapable.conductor-*.plist')):
        d = plist(path)
        f = flags(d.get('ProgramArguments', []))
        listen = f.get('--listen', '')
        port = int(listen.rsplit(':', 1)[-1]) if re.search(r':\d+$', listen) else None
        if port:
            ports.add(port)
        orgs.append({'slug': path.stem[len('dev.kapable.conductor-'):], 'mode': 'isolated',
                     'port': port, 'flags': f, 'configured': True, 'source': str(path),
                     'label': d.get('Label', path.stem)})
    return orgs, ports


def listening(port):
    if not port:
        return False
    result = run([tool('LSOF', '/usr/sbin/lsof'), '-nP', '-iTCP:' + str(port), '-sTCP:LISTEN', '-t'])
    if result.returncode not in (0, 1):
        raise Failure('Cannot determine listening ports (lsof failed).')
    return result.returncode == 0


def next_port(ports):
    for port in range(3115, 65536):
        if port not in ports and not listening(port):
            return port
    raise Failure('No free port at or above 3115.')


def health(url):
    result = run([tool('CURL', '/usr/bin/curl'), '-q', '--noproxy', '*', '--silent', '--show-error',
                  '--connect-timeout', '2', '--max-time', '3', '--write-out', '\n%{http_code}', url], 5)
    body, _, code = result.stdout.rstrip('\r\n').rpartition('\n')
    return result.returncode == 0 and code.strip() == '200'


def process_running(target):
    result = launch('print', target)
    if result.returncode:
        return False
    pid = re.search(r'^\s*pid = (\d+)\s*$', result.stdout, re.M)
    if not pid or int(pid[1]) <= 0:
        return False
    try:
        os.kill(int(pid[1]), 0)
        return True
    except PermissionError:
        return True
    except ProcessLookupError:
        return False


def tunnels():
    return [(path, plist(path)) for path in sorted(AGENTS.glob('dev.kapable.conductor-tunnel*.plist'))]


def tunnel_for(org, known):
    expected = 'dev.kapable.conductor-tunnel' + ('' if org['slug'] == 'kapable' else '-' + org['slug'])
    matches = [(p, d) for p, d in known if d.get('Label') == expected]
    if not matches:
        tid = org['flags'].get('--platform-tunnel-id')
        matches = [(p, d) for p, d in known if
                   (tid and d.get('EnvironmentVariables', {}).get('TUNNEL_SLUG') == tid) or
                   (org['port'] and d.get('EnvironmentVariables', {}).get('LOCAL_PORT') == str(org['port']))]
    if len(matches) > 1:
        raise Failure('Ambiguous tunnel identity; refused to select a plist.')
    if not matches:
        return {'state': 'missing', 'label': expected, 'plist': str(AGENTS / (expected + '.plist')), 'log': None}
    path, d = matches[0]
    label = d.get('Label', path.stem)
    return {'state': 'running' if process_running(DOMAIN + '/' + label) else 'stopped',
            'label': label, 'plist': str(path), 'log': d.get('StandardOutPath')}


def status(slug=None):
    _, data = read_config()
    orgs, ports = inventory(data)
    known = tunnels()
    result = {'supervisor': {'loaded': launch('print', TARGET).returncode == 0, 'label': LABEL},
              'next_port': next_port(ports), 'orgs': []}
    if slug and not any(o['slug'] == slug for o in orgs):
        orgs.append({'slug': slug, 'mode': 'supervisor', 'port': None, 'flags': {},
                     'configured': False, 'source': None})
    for org in orgs:
        if slug and org['slug'] != slug:
            continue
        port, f = org['port'], org['flags']
        url = 'http://127.0.0.1:%s/health' % port if port else None
        state = 'missing' if not org['configured'] else (
            'running' if url and health(url) else 'unhealthy' if listening(port) else 'stopped')
        org_url = f.get('--v2-platform-api-url', 'https://' + org['slug'] + '.kapable.ai').removesuffix('/auth')
        result['orgs'].append({'slug': org['slug'], 'org_url': org_url, 'mode': org['mode'], 'port': port,
            'conductor': {'state': state, 'health_url': url, 'open_url': None, 'db': f.get('--db'),
                          'config': f.get('--config-file'), 'supervisor_config': org['source'] if org['mode'] == 'supervisor' else None,
                          'plist': org['source'] if org['mode'] == 'isolated' else None},
            'tunnel': tunnel_for(org, known)})
    return result


def emit(step, state, detail, as_json):
    # No child output, credential values, config contents, or server response bodies.
    if as_json:
        print(json.dumps({'step': step, 'state': state, 'detail': detail}), flush=True)
    else:
        print('%s: %s — %s' % (step, state, detail), flush=True)


def atomic(path, raw, mode=0o600, exclusive=False):
    guard(path)
    if exclusive:
        fd = os.open(str(path), os.O_WRONLY | os.O_CREAT | os.O_EXCL, mode)
        try:
            with os.fdopen(fd, 'wb') as stream:
                stream.write(raw)
                stream.flush()
                os.fsync(stream.fileno())
        except BaseException:
            path.unlink(missing_ok=True)
            raise
        return
    fd, tmp = tempfile.mkstemp(prefix='.kco-', dir=str(path.parent))
    try:
        with os.fdopen(fd, 'wb') as stream:
            os.fchmod(stream.fileno(), mode)
            stream.write(raw)
            stream.flush()
            os.fsync(stream.fileno())
        os.replace(tmp, path)
    finally:
        if os.path.exists(tmp):
            os.unlink(tmp)


def poll(check, detail):
    deadline = time.monotonic() + 60
    while True:
        if check():
            return
        if time.monotonic() >= deadline:
            raise Failure(detail + ' (60-second timeout).')
        time.sleep(min(1, max(0, deadline - time.monotonic())))


def connected(log, port):
    if not log.exists():
        return False
    with guard(log).open('rb') as stream:
        stream.seek(max(0, log.stat().st_size - 65536))
        lines = stream.read().decode(errors='replace').splitlines()
    ready = False
    for line in lines:
        if 'Tunnel ready — proxying to localhost:' + str(port) in line or '[heartbeat] connected=true' in line:
            ready = True
        elif '[heartbeat] connected=false' in line or 'Tunnel closed' in line:
            ready = False
    return ready


def setup(args):
    global DRY_TRACE, JSON_TRACE
    DRY_TRACE, JSON_TRACE = args.dry_run, args.json
    step, old, appended, backup = 'preflight', None, None, None
    owned, dirs = [], []
    config_written = False
    bootstrap_attempted = False
    restart_attempted = False
    lockfd = None
    port = None
    tunnel_target = None

    def say(state, detail):
        emit(step, state, detail, args.json)

    def mkdir(path):
        guard(path)
        missing = []
        p = path
        while not p.exists():
            missing.append(p)
            p = p.parent
        for p in reversed(missing):
            p.mkdir(mode=0o700)
            dirs.append(p)

    try:
        say('running', 'Checking supervisor, binary, org, port, credential files and /auth/health.')
        if not re.fullmatch(r'[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?', args.slug):
            raise Failure('Slug must be a lowercase DNS label.')
        parsed = urlsplit(args.org_url)
        if (parsed.scheme != 'https' or parsed.hostname != args.slug + '.kapable.ai' or
                parsed.username or parsed.password or parsed.port or parsed.path not in ('', '/') or parsed.query or parsed.fragment):
            raise Failure('Use the HTTPS org origin https://<slug>.kapable.ai (no credentials, port, path or query).')
        org_url = args.org_url.rstrip('/')
        old, data = read_config()
        if not old:
            raise Failure('Supervisor config is missing.')
        # Lock the stable parent directory, not the inode replaced by atomic writes.
        lockfd = os.open(str(guard(CONFIG.parent)), os.O_RDONLY)
        try:
            fcntl.flock(lockfd, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise Failure('Another setup is active; retry after it finishes.')
        old, data = read_config()
        orgs, ports = inventory(data)
        if any(o['slug'] == args.slug for o in orgs):
            raise Failure('Slug already configured; existing orgs are never overwritten.')
        if launch('print', TARGET).returncode:
            raise Failure('Supervisor is not loaded: ' + TARGET)
        template = next((o for o in orgs if o['slug'] == 'kapable' and o['mode'] == 'supervisor' and o['configured']), None)
        if template is None:
            raise Failure('Kapable supervisor conductor template is missing.')
        binary = guard(Path(template['worker']['binary']))
        if not binary.is_file() or run([binary, '--help']).returncode:
            raise Failure('Conductor binary is missing or --help failed.')
        port = args.port if args.port is not None else next_port(ports)
        if port < 3115 or port > 65535 or port in ports or listening(port):
            raise Failure('Port is taken or outside 3115..65535.')
        credentials, missing = {}, []
        for key, note in TOKEN_NOTES.items():
            supplied = getattr(args, key.replace('-', '_'))
            if not supplied:
                missing.append('--' + key + ': ' + note)
                continue
            if key.endswith('-file'):
                try:
                    value = guard(Path(supplied).absolute()).read_text().strip()
                except OSError:
                    value = ''
                if not value or any(c.isspace() for c in value):
                    missing.append('--' + key + ': file must contain one nonempty credential. ' + note)
                else:
                    credentials[key] = value
            else:
                try:
                    credentials[key] = str(uuid.UUID(supplied))
                except ValueError:
                    raise Failure('--' + key + ' must be a provisioned UUID.')
        if missing:
            note = ' Session-token-only onboarding cannot provision the complete legacy tunnel bundle; supply the files below.' if args.session_token_file else ''
            raise Failure('Missing required values.' + note + '\n' + '\n'.join(missing))
        if not credentials['service-token-file'].startswith('st_ci_'):
            raise Failure('--service-token-file must contain an st_ci_ service token.')
        tid, aid = credentials['tunnel-id'], credentials['agent-id']
        if any(o['flags'].get('--platform-tunnel-id') == tid or o['flags'].get('--platform-agent-id') == aid for o in orgs):
            raise Failure('Tunnel or agent UUID is already used by a configured org.')
        for _, d in tunnels():
            if d.get('EnvironmentVariables', {}).get('TUNNEL_SLUG') == tid:
                raise Failure('Tunnel UUID already has a LaunchAgent.')
        cfg = HOME / ('.kapable-agent/conductor-' + args.slug + '-config.json')
        db = HOME / ('.kapable-agent/konductor-' + args.slug + '.db')
        label = 'dev.kapable.conductor-tunnel-' + args.slug
        tunnel_target = DOMAIN + '/' + label
        tunnel_path = AGENTS / (label + '.plist')
        logdir = HOME / ('.kapable-tunnel/' + args.slug)
        log = logdir / 'tunnel.log'
        db_files = [db, Path(str(db) + '-wal'), Path(str(db) + '-shm'), Path(str(db) + '-journal')]
        for path in [cfg, tunnel_path, logdir] + db_files:
            if guard(path).exists() or path.is_symlink():
                raise Failure('Destination already exists; refusing to overwrite org files.')
        if launch('print', tunnel_target).returncode == 0:
            raise Failure('Tunnel LaunchAgent is already loaded.')
        bun = guard(HOME / '.bun/bin/bun')
        client = guard(HOME / 'WebstormProjects/kapable/konductor-tunnel-client.ts')
        if not bun.is_file() or not os.access(bun, os.X_OK) or not client.is_file():
            raise Failure('Bun or konductor-tunnel-client.ts is missing.')
        template_path = guard(Path(template['flags']['--config-file']))
        config_data = json.loads(template_path.read_text())
        if not isinstance(config_data, dict):
            raise Failure('Conductor config template must be a JSON object.')
        replacements = {template['flags'].get('--platform-tunnel-id', ''): tid,
                        template['flags'].get('--platform-agent-id', ''): aid,
                        'https://kapable.kapable.ai': org_url,
                        'mac-studio-kapable': 'mac-studio-' + args.slug,
                        str(HOME / '.kapable-agent/conductor-kapable-config.json'): str(cfg),
                        str(HOME / '.kapable-agent/konductor-kapable.db'): str(db)}

        def substitute(v):
            if isinstance(v, dict):
                return {k: substitute(x) for k, x in v.items()}
            if isinstance(v, list):
                return [substitute(x) for x in v]
            if isinstance(v, str):
                if v == 'kapable':
                    return args.slug
                for before, after in replacements.items():
                    if before:
                        v = v.replace(before, after)
            return v

        config_bytes = (json.dumps(substitute(config_data), indent=2) + '\n').encode()
        worker_args = ['--listen', '127.0.0.1:' + str(port), '--db', str(db),
            '--platform-api-url', template['flags'].get('--platform-api-url', 'https://api.kapable.dev'),
            '--platform-api-key', credentials['service-token-file'], '--v2-platform-api-url', org_url + '/auth',
            '--v2-platform-api-key', credentials['v2-key-file'], '--platform-instance-name', data.get('agent', {}).get('name', 'mac-studio') + '-' + args.slug,
            '--platform-tunnel-id', tid, '--platform-agent-id', aid, '--config-file', str(cfg),
            '--cors-permissive', '--no-relay', '--auth-secret', secrets.token_hex(32)]
        block = '\n\n[[orgs]]\nslug = ' + json.dumps(args.slug) + '\n\n[[orgs.workers]]\n'
        fields = {'name': 'conductor', 'binary': str(binary), 'port': port, 'health_path': '/health',
                  'restart_policy': 'always', 'max_restarts': 5, 'health_interval_secs': 5,
                  'health_failures_before_restart': 3, 'args': worker_args}
        block += ''.join(k + ' = ' + json.dumps(v) + '\n' for k, v in fields.items())
        # Carry only launcher environment, never the kapable org's credentials.
        env = {'HOME': str(HOME), 'PATH': str(HOME / '.local/bin') + ':' + str(HOME / '.bun/bin') + ':/opt/homebrew/bin:/usr/local/bin:/usr/bin:/bin',
               'V2_TUNNEL_BASE_DOMAIN': 'tunnel.kapable.ai', 'BOARD_URL': org_url}
        block += '\n[orgs.workers.env]\n' + ''.join(k + ' = ' + json.dumps(v) + '\n' for k, v in env.items())
        appended = old + block.encode()
        toml_subset(appended.decode())
        tunnel_data = {'Label': label, 'ProgramArguments': [str(bun), str(client)],
            'EnvironmentVariables': {'AGENT_ID': tid, 'LOCAL_PORT': str(port), 'TUNNEL_BASE_DOMAIN': 'tunnel.kapable.ai',
                'TUNNEL_INGRESS_KEY': credentials['tunnel-ingress-key-file'], 'TUNNEL_SLUG': tid, 'TUNNEL_TOKEN': credentials['tunnel-token-file']},
            'KeepAlive': True, 'RunAtLoad': True, 'StandardOutPath': str(log), 'StandardErrorPath': str(logdir / 'tunnel.err')}
        backup = CONFIG.with_name('kapable.toml.bak-' + time.strftime('%Y%m%dT%H%M%S') + '-' + str(time.time_ns()))
        health_url = 'http://127.0.0.1:%d/health' % port
        say('running', 'GET ' + org_url + '/auth/health; binary --help; launchctl print supervisor/tunnel; lsof port ' + str(port))
        if not health(org_url + '/auth/health'):
            raise Failure('Org health check failed: GET ' + org_url + '/auth/health (expected HTTP 200).')
        say('done', 'Preflight passed; port ' + str(port) + '. Credentials loaded without displaying them.')
        if args.dry_run:
            for step, detail in [
                ('write-config', 'WOULD WRITE ' + str(backup) + ' (backup), ' + str(cfg) + ', ' + str(CONFIG) + ' (append one org); secret-bearing contents masked.'),
                ('start-conductor', 'WOULD RUN ' + command('kickstart', '-k', TARGET) + '; poll GET ' + health_url + ' up to 60s. Worker may create ' + ', '.join(map(str, db_files))),
                ('write-tunnel', 'WOULD WRITE ' + str(tunnel_path) + ', ' + str(log) + ', ' + str(logdir / 'tunnel.err') + '; six environment keys, secrets masked. WOULD RUN ' + command('bootstrap', DOMAIN, str(tunnel_path)) + '; poll launchctl print, process alive and fresh Tunnel ready log up to 60s.'),
                ('verify', 'WOULD GET ' + health_url + '; check tunnel connection and status --json. On failure WOULD RUN ' + command('bootout', tunnel_target) + ', restore TOML, remove created files and RUN ' + command('kickstart', '-k', TARGET))]:
                say('skipped', detail)
            return 0
        step = 'write-config'
        say('running', 'Backing up supervisor TOML and writing the new conductor config.')
        if CONFIG.read_bytes() != old:
            raise Failure('Supervisor config changed during preflight; retry.')
        mode = stat.S_IMODE(CONFIG.stat().st_mode)
        atomic(backup, old, exclusive=True)
        mkdir(cfg.parent)
        atomic(cfg, config_bytes, exclusive=True)
        owned.append(cfg)
        config_written = True
        atomic(CONFIG, appended, mode)
        say('done', 'Appended one org; backup: ' + str(backup))
        step = 'start-conductor'
        say('running', 'Restarting the shared supervisor (all supervised workers): ' + command('kickstart', '-k', TARGET))
        restart_attempted = True
        if launch('kickstart', '-k', TARGET).returncode:
            raise Failure('Supervisor kickstart failed.')
        poll(lambda: health(health_url), 'Conductor did not become healthy')
        say('done', 'Conductor /health returned HTTP 200.')
        step = 'write-tunnel'
        say('running', 'Writing tunnel LaunchAgent and waiting for its connection.')
        mkdir(AGENTS)
        mkdir(logdir)
        for path, raw in [(log, b''), (logdir / 'tunnel.err', b''), (tunnel_path, plistlib.dumps(tunnel_data))]:
            atomic(path, raw, exclusive=True)
            owned.append(path)
        bootstrap_attempted = True
        if launch('bootstrap', DOMAIN, str(tunnel_path)).returncode:
            raise Failure('Tunnel bootstrap failed.')
        poll(lambda: process_running(tunnel_target) and connected(log, port), 'Tunnel did not connect')
        say('done', 'Tunnel process is alive and its new log reports connected.')
        step = 'verify'
        say('running', 'Checking health, connection and status --json.')
        result = status(args.slug)
        row = result['orgs'][0]
        if row['conductor']['state'] != 'running' or row['tunnel']['state'] != 'running' or not connected(log, port):
            raise Failure('Final status verification failed.')
        say('done', 'Org conductor and tunnel are running; open_url is unavailable until an instance id is independently known.')
        return 0
    except (Exception, KeyboardInterrupt) as error:
        detail = str(error) if isinstance(error, Failure) else 'Setup interrupted or failed; diagnostic content suppressed to protect secrets.'
        say('failed', detail)
        problems, actions = [], []
        if bootstrap_attempted:
            try:
                rc = launch('bootout', tunnel_target).returncode
                if rc and launch('print', tunnel_target).returncode == 0:
                    raise Failure('Tunnel still loaded')
                actions.append('tunnel booted out')
            except Exception:
                problems.append('could not boot out tunnel; run ' + command('bootout', tunnel_target))
        if config_written:
            try:
                current = CONFIG.read_bytes()
                if current == appended:
                    atomic(CONFIG, old, mode)
                elif current != old:
                    if current.count(block.encode()) != 1:
                        raise Failure('Concurrent config edit overlaps new org')
                    atomic(CONFIG, current.replace(block.encode(), b'', 1), mode)
                actions.append('new config block removed; original TOML bytes restored')
            except Exception:
                problems.append('could not restore TOML; inspect backup ' + str(backup))
        if restart_attempted:
            try:
                if launch('kickstart', '-k', TARGET).returncode:
                    raise Failure('Restore kickstart failed')
                actions.append('supervisor restarted with restored config')
            except Exception:
                problems.append('could not restart restored supervisor; run ' + command('kickstart', '-k', TARGET))
        if config_written:
            owned += db_files
        for path in reversed(owned):
            try:
                guard(path).unlink(missing_ok=True)
            except OSError:
                problems.append('could not remove created file ' + str(path))
        for path in reversed(dirs):
            try:
                path.rmdir()
            except OSError:
                problems.append('created directory not empty: ' + str(path))
        if owned:
            actions.append('created org config, database and tunnel files removed where present')
        emit('rollback', 'failed' if problems else 'done', '; '.join(actions + problems) or 'No changes were made.', args.json)
        return 1
    finally:
        if lockfd is not None:
            os.close(lockfd)


def control(action, slug, as_json):
    _, data = read_config()
    orgs, _ = inventory(data)
    matches = [o for o in orgs if o['slug'] == slug]
    if len(matches) != 1 or not matches[0]['configured']:
        raise Failure('Org is missing or ambiguous.')
    org = matches[0]
    if org['mode'] == 'isolated':
        cmd = {'start': 'sudo launchctl bootstrap system ' + shlex.quote(org['source']),
               'stop': 'sudo launchctl bootout system/' + org['label'],
               'restart': 'sudo launchctl kickstart -k system/' + org['label']}[action]
        emit(action, 'skipped', cmd, as_json)
        return 0
    if action == 'stop':
        emit(action, 'skipped', 'The supervisor restarts workers; it has no per-worker stop control. To stop only this conductor, an operator must back up kapable.toml, remove that conductor worker from its org, then restart the shared supervisor. This command edits nothing.', as_json)
        return 0
    emit(action, 'running', 'No per-worker control: restarting the shared supervisor affects all supervised workers. ' + command('kickstart', '-k', TARGET), as_json)
    if launch('kickstart', '-k', TARGET).returncode:
        raise Failure('Shared supervisor kickstart failed.')
    port = org['port']
    poll(lambda: health('http://127.0.0.1:%d/health' % port), 'Conductor did not become healthy')
    emit(action, 'done', 'Conductor /health returned HTTP 200.', as_json)
    return 0


class Parser(argparse.ArgumentParser):
    def error(self, message):
        raise Failure('Invalid arguments; use --help (argument values suppressed).')


def main():
    parser = Parser(prog='kapable-conductor-onboard', description='Set up, inspect and drive a per-org conductor on this Mac.',
        epilog='Runbook: preflight → write-config (backup first) → start-conductor → write-tunnel → verify. Failed steps roll back. New orgs use the existing user supervisor; no sudo. Setup restarts ALL supervised workers. Credentials are supplied by file only; session-only auto-provisioning is not implemented. A service-token mint exists at POST <org-url>/auth/v1/auth/service-tokens (keys.manage required), but no complete legacy tunnel credential bundle mint was found. Ask the tunnel operator for token, ingress key and registered IDs. open prints null when no verified heartbeat instance ID is available.')
    sub = parser.add_subparsers(dest='command', required=True, parser_class=Parser)
    st = sub.add_parser('status', help='Read conductor and tunnel state.')
    st.add_argument('--json', action='store_true')
    st.add_argument('--org')
    se = sub.add_parser('setup', help='Run the five-step supervisor onboarding runbook.')
    se.add_argument('slug')
    se.add_argument('--org-url', required=True)
    se.add_argument('--port', type=int)
    se.add_argument('--dry-run', action='store_true')
    se.add_argument('--json', action='store_true')
    for key, note in TOKEN_NOTES.items():
        se.add_argument('--' + key, help=note)
    se.add_argument('--session-token-file', help='Informational only: session-only provisioning is unsupported; all six provisioned inputs remain required.')
    for name in ('start', 'stop', 'restart', 'open'):
        sp = sub.add_parser(name)
        sp.add_argument('slug')
        sp.add_argument('--json', action='store_true')
    args = parser.parse_args(sys.argv[2:])
    fixture_guard()
    if args.command == 'setup':
        return setup(args)
    if args.command in ('start', 'stop', 'restart'):
        return control(args.command, args.slug, args.json)
    result = status(args.slug if args.command == 'open' else args.org)
    if args.command == 'open':
        print(result['orgs'][0]['conductor']['open_url'] or 'null')
    elif args.json:
        print(json.dumps(result))
    else:
        print('supervisor=%s label=%s next_port=%s' % (result['supervisor']['loaded'], LABEL, result['next_port']))
        for o in result['orgs']:
            c, t = o['conductor'], o['tunnel']
            print('%s mode=%s port=%s conductor=%s tunnel=%s org_url=%s health_url=%s open_url=%s db=%s config=%s supervisor_config=%s conductor_plist=%s tunnel_label=%s tunnel_plist=%s tunnel_log=%s' %
                (o['slug'], o['mode'], o['port'], c['state'], t['state'], o['org_url'], c['health_url'], c['open_url'], c['db'], c['config'], c['supervisor_config'], c['plist'], t['label'], t['plist'], t['log']))
    return 0


def interrupted(signum, frame):
    raise KeyboardInterrupt()


signal.signal(signal.SIGTERM, interrupted)
try:
    sys.exit(main())
except (Exception, KeyboardInterrupt) as error:
    detail = str(error) if isinstance(error, Failure) else 'Operation failed; diagnostic content suppressed to protect secrets.'
    emit('error', 'failed', detail, '--json' in sys.argv[2:])
    sys.exit(1)
PY
