#!/usr/bin/env python3
from __future__ import annotations

import argparse
import hashlib
import json
import os
import secrets
import shutil
import sqlite3
import subprocess
import time
import urllib.request
from pathlib import Path

STATE_VERSION=1
RELEASE_MANIFEST_VERSION=1
RELEASE_MANIFEST='release-manifest.json'
RELEASE_ASSETS=('core','ingest','deploy','endpoint','storage','modules','docker-compose.yml','.env.example','install.sh')
UPDATE_BACKUP_KEEP=max(1,int(os.getenv('SURROUND_UPDATE_BACKUP_KEEP','3')))
UPDATE_MIN_FREE_BYTES=max(1,int(os.getenv('SURROUND_UPDATE_MIN_FREE_MB','4096')))*1024*1024


def _host_helper_specs():
    return (
        {'key':'SURROUND_STORAGE_TOKEN','host_env':'/etc/core-storage-helper.env','script':'storage/install-storage-helper.sh','optional':False,'restart':True},
        {'key':'SURROUND_ENGINEERS_TOKEN','host_env':'/etc/core-engineers-helper.env','script':'deploy/install-engineers-helper.sh','optional':False,'restart':True},
        {'key':'SURROUND_REMOTE_TOKEN','host_env':'/etc/core-remote-access-helper.env','script':'deploy/install-remote-access-helper.sh','optional':True,'restart':True},
        {'key':'SURROUND_BLUETOOTH_TOKEN','host_env':'/etc/core-bluetooth-audio-helper.env','script':'deploy/install-bluetooth-audio-helper.sh','optional':True,'restart':True},
        # The update helper can be the parent service that launched this updater.
        # Refresh its files/unit now, but never restart that service from its child.
        {'key':'SURROUND_UPDATE_TOKEN','host_env':'/etc/core-update-helper.env','script':'deploy/install-update-helper.sh','optional':False,'restart':False},
    )


def install_root() -> Path:
    return Path(os.getenv('SURROUNDCORE_INSTALL_ROOT','/opt/surroundcore'))


def data_root() -> Path:
    return Path(os.getenv('SURROUNDCORE_DATA','/var/lib/surroundcore'))


def state_path() -> Path:
    return data_root()/'core-update-state.json'


def _free_bytes(path: Path) -> int:
    """Return free bytes for the nearest existing parent of an update path."""
    probe=path
    while not probe.exists() and probe != probe.parent:
        probe=probe.parent
    return int(shutil.disk_usage(probe).free)


def _require_update_space(path: Path | None=None, required: int | None=None):
    """Fail before DB backup/activation if rollback headroom is not available."""
    target=path or data_root()
    minimum=max(UPDATE_MIN_FREE_BYTES,int(required or 0))
    free=_free_bytes(target)
    if free < minimum:
        raise RuntimeError(
            f'Insufficient free space for Core update: {free//(1024*1024)} MB free; '
            f'{minimum//(1024*1024)} MB required'
        )


def _protected_backup_dirs() -> set[Path]:
    """Return rollback backup directories referenced by persisted update state."""
    protected=set()
    try:
        state=json.loads(state_path().read_text())
    except (FileNotFoundError,json.JSONDecodeError,OSError):
        state={}
    backup=str(state.get('database_backup') or '').strip()
    if backup:
        try:
            protected.add(Path(backup).resolve().parent)
        except OSError:
            pass
    return protected


def _prune_update_backups(keep: int | None=None):
    """Keep recent rollback backups plus any backup still referenced by update state."""
    root=data_root()/'update-backups'
    if not root.is_dir(): return []
    limit=max(1,int(keep or UPDATE_BACKUP_KEEP))
    protected={p.resolve() for p in _protected_backup_dirs()}
    dirs=[p for p in root.iterdir() if p.is_dir()]
    dirs.sort(key=lambda p:p.stat().st_mtime,reverse=True)
    keepers={p.resolve() for p in dirs[:limit]} | protected
    removed=[]
    for path in dirs:
        if path.resolve() in keepers: continue
        shutil.rmtree(path,ignore_errors=True); removed.append(path)
    return removed


def _atomic_json(path: Path, value: dict):
    path.parent.mkdir(parents=True,exist_ok=True); tmp=path.with_name('.'+path.name+'.tmp')
    with tmp.open('w') as handle:
        json.dump(value,handle,indent=2,sort_keys=True); handle.write('\n'); handle.flush(); os.fsync(handle.fileno())
    os.replace(tmp,path)


def _write_state(status: str, **fields):
    value={'version':STATE_VERSION,'status':status,**fields}
    _atomic_json(state_path(),value); return value


def _safe_version(value: str) -> str:
    clean=''.join(ch for ch in str(value or '').strip() if ch.isalnum() or ch in '.-_')
    if not clean or clean != str(value or '').strip(): raise ValueError('Invalid release version')
    return clean


def _validate_source(source: Path):
    missing=[name for name in ('core','ingest','deploy','modules/core-cast','docker-compose.yml') if not (source/name).exists()]
    if missing: raise ValueError('Release source is incomplete: '+', '.join(missing))


def _release_files(source: Path):
    files=[]
    for name in RELEASE_ASSETS:
        entry=source/name
        if not entry.exists(): continue
        if entry.is_symlink(): raise ValueError('Release assets may not contain symlinks: '+name)
        if entry.is_file():
            files.append(entry); continue
        if not entry.is_dir(): raise ValueError('Release asset is not a regular file or directory: '+name)
        for path in entry.rglob('*'):
            rel=path.relative_to(source).as_posix()
            if path.is_symlink(): raise ValueError('Release assets may not contain symlinks: '+rel)
            if path.is_dir(): continue
            if not path.is_file(): raise ValueError('Release asset is not a regular file: '+rel)
            files.append(path)
    return sorted(files,key=lambda path:path.relative_to(source).as_posix())


def _sha256_file(path: Path):
    digest=hashlib.sha256()
    with path.open('rb') as handle:
        for chunk in iter(lambda:handle.read(4*1024*1024),b''): digest.update(chunk)
    return digest.hexdigest()


def write_release_manifest(source: Path, version: str) -> Path:
    source=source.resolve(); _validate_source(source); version=_safe_version(version)
    files={}
    for path in _release_files(source):
        rel=path.relative_to(source).as_posix(); files[rel]={'size':path.stat().st_size,'sha256':_sha256_file(path)}
    manifest={'schema':RELEASE_MANIFEST_VERSION,'version':version,'files':files}
    target=source/RELEASE_MANIFEST; _atomic_json(target,manifest); return target


def _verify_release_manifest(source: Path, version: str):
    path=source/RELEASE_MANIFEST
    try: manifest=json.loads(path.read_text())
    except FileNotFoundError as exc: raise ValueError('Release manifest is missing') from exc
    except (OSError,json.JSONDecodeError) as exc: raise ValueError('Release manifest is unreadable') from exc
    if not isinstance(manifest,dict) or int(manifest.get('schema') or 0)!=RELEASE_MANIFEST_VERSION:
        raise ValueError('Release manifest schema is unsupported')
    if str(manifest.get('version') or '') != version: raise ValueError('Release manifest version does not match requested version')
    expected=manifest.get('files')
    if not isinstance(expected,dict) or not expected: raise ValueError('Release manifest file list is empty')
    actual={path.relative_to(source).as_posix():path for path in _release_files(source)}
    if set(actual) != set(expected): raise ValueError('Release manifest file list does not match release assets')
    for rel,path in actual.items():
        row=expected.get(rel)
        if not isinstance(row,dict): raise ValueError('Release manifest entry is invalid: '+rel)
        size=row.get('size')
        if not isinstance(size,int) or size != path.stat().st_size: raise ValueError('Release file size mismatch: '+rel)
        if str(row.get('sha256') or '').lower() != _sha256_file(path): raise ValueError('Release file hash mismatch: '+rel)
    return manifest


def stage_release(source: Path, version: str, root: Path | None=None) -> Path:
    source=source.resolve(); _validate_source(source); version=_safe_version(version); _verify_release_manifest(source,version)
    root=root or install_root(); releases=root/'releases'; releases.mkdir(parents=True,exist_ok=True)
    target=releases/version; temp=releases/('.'+version+'.staging')
    if temp.exists(): shutil.rmtree(temp)
    temp.mkdir(parents=True)
    for name in RELEASE_ASSETS:
        src=source/name
        if not src.exists(): continue
        dst=temp/name
        if src.is_dir(): shutil.copytree(src,dst)
        else: shutil.copy2(src,dst)
    shutil.copy2(source/RELEASE_MANIFEST,temp/RELEASE_MANIFEST)
    _verify_release_manifest(temp,version)
    if target.exists(): shutil.rmtree(target)
    os.replace(temp,target)
    return target


def _current_release(root: Path | None=None) -> Path | None:
    root=root or install_root(); current=root/'current'
    if not current.is_symlink(): return None
    try: return current.resolve(strict=True)
    except FileNotFoundError: return None


def _atomic_symlink(link: Path, target: Path):
    link.parent.mkdir(parents=True,exist_ok=True); temp=link.with_name('.'+link.name+'.new')
    try: temp.unlink()
    except FileNotFoundError: pass
    os.symlink(str(target),str(temp)); os.replace(temp,link)


def _backup_sqlite(source: Path, destination: Path):
    destination.parent.mkdir(parents=True,exist_ok=True); temp=destination.with_suffix(destination.suffix+'.tmp')
    try: temp.unlink()
    except FileNotFoundError: pass
    src=sqlite3.connect(source); dst=sqlite3.connect(temp)
    try: src.backup(dst); dst.commit()
    finally: dst.close(); src.close()
    os.replace(temp,destination)


def _restore_file(source: Path, destination: Path):
    temp=destination.with_suffix(destination.suffix+'.restore')
    shutil.copy2(source,temp)
    with temp.open('rb') as handle: os.fsync(handle.fileno())
    os.replace(temp,destination)


def _read_env_value(path: Path, key: str) -> str:
    if not path.exists(): return ''
    for line in path.read_text().splitlines():
        if line.startswith(key+'='): return line.split('=',1)[1]
    return ''


def _write_env_value(path: Path, key: str, value: str):
    lines=path.read_text().splitlines() if path.exists() else []
    out=[]; found=False
    for line in lines:
        if line.startswith(key+'='):
            out.append(f'{key}={value}'); found=True
        else: out.append(line)
    if not found: out.append(f'{key}={value}')
    temp=path.with_name('.'+path.name+'.tmp'); temp.write_text('\n'.join(out)+'\n'); os.replace(temp,path)


def _sync_existing_helper_tokens(root: Path):
    """Prefer already-provisioned host helper credentials before starting a candidate."""
    env_file=root/'.env'; synced=[]
    for spec in _host_helper_specs():
        host_env=Path(spec['host_env'])
        token=_read_env_value(host_env,spec['key'])
        if not token: continue
        if _read_env_value(env_file,spec['key']) != token:
            _write_env_value(env_file,spec['key'],token)
        synced.append(spec['key'])
    return synced


def _refresh_host_helpers(release: Path, root: Path):
    """Refresh bounded host helper code without turning a module update into apt/host update."""
    if os.geteuid()!=0:
        return []
    refreshed=[]; env_file=root/'.env'

    safety_installer=Path(release)/'deploy/install-tmp-guard.sh'
    if safety_installer.is_file():
        result=subprocess.run([str(safety_installer)],text=True,capture_output=True,check=False)
        if result.returncode:
            detail=(result.stderr or result.stdout or 'tmpfs safety refresh failed').strip()[:4000]
            raise RuntimeError(f"Host safety refresh failed for deploy/install-tmp-guard.sh: {detail}")
        refreshed.append('deploy/install-tmp-guard.sh')
    for spec in _host_helper_specs():
        script=Path(release)/spec['script']
        if not script.is_file():
            continue
        host_env=Path(spec['host_env'])
        if spec.get('optional') and not host_env.is_file():
            continue
        token=_read_env_value(env_file,spec['key']) or _read_env_value(host_env,spec['key'])
        if not token:
            raise RuntimeError(f"Host helper credential is missing: {spec['key']}")
        helper_env=os.environ.copy()
        helper_env.update({
            spec['key']:token,
            'SURROUNDCORE_BUILD_VERSION':_read_env_value(env_file,'SURROUNDCORE_BUILD_VERSION') or '1.0.0',
            'SURROUND_HELPER_REFRESH_ONLY':'1',
            'SURROUND_HELPER_NO_RESTART':'0' if spec.get('restart') else '1',
        })
        result=subprocess.run([str(script)],text=True,capture_output=True,env=helper_env,check=False)
        if result.returncode:
            detail=(result.stderr or result.stdout or 'host helper refresh failed').strip()[:4000]
            raise RuntimeError(f"Host helper refresh failed for {spec['script']}: {detail}")
        returned=next((line.strip() for line in reversed((result.stdout or '').splitlines()) if line.strip()),'')
        if returned and returned != token:
            _write_env_value(env_file,spec['key'],returned)
        refreshed.append(spec['script'])
    return refreshed


def _ensure_database_env(env_file: Path):
    password=_read_env_value(env_file,'CORE_DB_PASSWORD')
    if not password or password == 'change-me-db':
        password=secrets.token_urlsafe(36)
        _write_env_value(env_file,'CORE_DB_PASSWORD',password)
    return password


def _ensure_release_secrets(release: Path, env_file: Path):
    """Backfill machine secrets introduced by newer releases without rotating existing ones."""
    example=Path(release)/'.env.example'
    if not example.is_file(): return []
    generated=[]
    for raw in example.read_text().splitlines():
        line=raw.strip()
        if not line or line.startswith('#') or '=' not in line: continue
        key,placeholder=line.split('=',1)
        key=key.strip(); placeholder=placeholder.strip()
        if not key or not placeholder.startswith('change-me'): continue
        current=_read_env_value(env_file,key)
        if current and current != placeholder: continue
        _write_env_value(env_file,key,secrets.token_urlsafe(36)); generated.append(key)
    return generated


def _container_virtualization() -> str:
    explicit=os.getenv('container','').strip().lower()
    if explicit: return explicit
    marker=Path('/run/systemd/container')
    try: return marker.read_text().strip().lower()
    except OSError: return ''


def _ensure_runtime_compatibility(env_file: Path) -> dict:
    """Apply bounded host-runtime fallbacks without weakening bare-metal defaults."""
    virt=_container_virtualization()
    changed={}
    if virt in {'lxc','lxc-libvirt','systemd-nspawn','nspawn'}:
        # Docker inside these container hosts cannot grant CAP_SYS_RAWIO. Keep
        # SYS_ADMIN and duplicate it in the optional slot so Compose remains
        # structurally identical; physical bare-metal installs retain SYS_RAWIO.
        if _read_env_value(env_file,'SURROUNDCORE_INGEST_RAWIO_CAP') != 'SYS_ADMIN':
            _write_env_value(env_file,'SURROUNDCORE_INGEST_RAWIO_CAP','SYS_ADMIN')
            changed['SURROUNDCORE_INGEST_RAWIO_CAP']='SYS_ADMIN'
    return {'virtualization':virt or 'bare-metal','changed':changed}


def _compose_base(release: Path, root: Path):
    env_file=root/'.env'
    _ensure_database_env(env_file)
    _ensure_release_secrets(release,env_file)
    _ensure_runtime_compatibility(env_file)
    cmd=['docker','compose','-p','surroundcore']
    if env_file.exists(): cmd += ['--env-file',str(env_file)]
    return cmd + ['--project-directory',str(release),'-f',str(release/'docker-compose.yml')]


def _compose(release: Path, root: Path, action: str):
    cmd=_compose_base(release,root)
    if action=='up': cmd += ['up','-d','--build']
    elif action=='down': cmd += ['down']
    else: raise ValueError('Unsupported compose action')
    subprocess.run(cmd,check=True)


def _compose_service_up(release: Path, root: Path, service: str):
    subprocess.run(_compose_base(release,root)+['up','-d',str(service)],check=True)


def _database_backend(root: Path) -> str:
    env_file=root/'.env'
    configured=_read_env_value(env_file,'CORE_DB_BACKEND').strip().lower()
    if configured in {'postgres','postgresql'}: return 'postgres'
    if configured=='sqlite': return 'sqlite'
    if _read_env_value(env_file,'CORE_DATABASE_URL').strip() or _read_env_value(env_file,'CORE_DB_PASSWORD').strip():
        return 'postgres'
    return 'sqlite' if (data_root()/'surroundcore.sqlite3').exists() else 'postgres'


def _postgres_config(root: Path):
    env_file=root/'.env'
    return {
        'container':_read_env_value(env_file,'CORE_DB_CONTAINER').strip() or 'core-postgres',
        'host':'127.0.0.1',
        'port':_read_env_value(env_file,'CORE_DB_PORT').strip() or '5432',
        'database':_read_env_value(env_file,'CORE_DB_NAME').strip() or 'core',
        'user':_read_env_value(env_file,'CORE_DB_USER').strip() or 'core',
        'password':_read_env_value(env_file,'CORE_DB_PASSWORD'),
    }


def _postgres_command(root: Path, args, *, stdin=None, stdout=None, check=True):
    cfg=_postgres_config(root)
    if not cfg['password']: raise RuntimeError('PostgreSQL update backup requires CORE_DB_PASSWORD')
    env=os.environ.copy(); env['PGPASSWORD']=cfg['password']
    cmd=['docker','exec']
    if stdin is not None: cmd.append('-i')
    cmd += ['-e','PGPASSWORD',cfg['container'],*list(args)]
    result=subprocess.run(cmd,stdin=stdin,stdout=(stdout if stdout is not None else subprocess.PIPE),stderr=subprocess.PIPE,env=env,check=False)
    if check and result.returncode:
        detail=(result.stderr or b'PostgreSQL command failed').decode(errors='replace').strip()[:4000]
        raise RuntimeError(detail)
    return result


def _backup_postgres(root: Path, destination: Path):
    cfg=_postgres_config(root); destination.parent.mkdir(parents=True,exist_ok=True)
    temp=destination.with_suffix(destination.suffix+'.tmp')
    try: temp.unlink()
    except FileNotFoundError: pass
    with temp.open('wb') as handle:
        _postgres_command(root,['pg_dump','-h',cfg['host'],'-p',cfg['port'],'-U',cfg['user'],
                                '--format=custom','--create',cfg['database']],stdout=handle)
        handle.flush(); os.fsync(handle.fileno())
    if not temp.stat().st_size: raise RuntimeError('PostgreSQL update backup is empty')
    os.replace(temp,destination)


def _wait_postgres(root: Path, timeout: int=60):
    cfg=_postgres_config(root); deadline=time.time()+max(5,int(timeout)); last='not ready'
    while time.time()<deadline:
        result=_postgres_command(root,['pg_isready','-h',cfg['host'],'-p',cfg['port'],'-U',cfg['user'],
                                       '-d','postgres'],check=False)
        if result.returncode==0: return
        last=(result.stderr or b'not ready').decode(errors='replace').strip()
        time.sleep(1)
    raise RuntimeError('PostgreSQL did not become ready for rollback: '+last[:1000])


def _restore_postgres(root: Path, backup: Path):
    cfg=_postgres_config(root)
    with backup.open('rb') as handle:
        _postgres_command(root,['pg_restore','-h',cfg['host'],'-p',cfg['port'],'-U',cfg['user'],
                                '--clean','--if-exists','--create','--exit-on-error','-d','postgres'],stdin=handle)


def _health(url: str):
    with urllib.request.urlopen(url,timeout=5) as response:
        return json.loads(response.read().decode() or '{}')


def _wait_for_confirmation(version: str, url: str, timeout: int=90):
    deadline=time.time()+max(5,int(timeout)); last='Core health did not respond'
    while time.time() < deadline:
        try:
            health=_health(url); update=health.get('update') or {}; schema=health.get('database_schema') or {}
            if (health.get('ready') and health.get('version')==version and schema.get('current_for_binary') and
                    update.get('status')=='confirmed' and update.get('target_version')==version):
                return health
            last=(f"ready={health.get('ready')} version={health.get('version')} "
                  f"schema={schema.get('current_for_binary')} update={update.get('status')}")
        except Exception as exc: last=str(exc)
        time.sleep(2)
    raise RuntimeError('Updated Core did not confirm healthy startup: '+last)


def _transaction_context(version: str, target: Path, previous: Path, root: Path):
    old_version=_read_env_value(root/'.env','SURROUNDCORE_BUILD_VERSION')
    _prune_update_backups()
    _require_update_space(data_root())
    tx=f'{int(time.time())}-{version}'; backup_dir=data_root()/'update-backups'/tx; backup=''
    backend=_database_backend(root)
    try:
        if backend=='postgres':
            backup_path=backup_dir/'core.pgcustom'; _backup_postgres(root,backup_path); backup=str(backup_path)
        else:
            database=data_root()/'surroundcore.sqlite3'
            if database.exists():
                backup_path=backup_dir/'surroundcore.sqlite3'; _backup_sqlite(database,backup_path); backup=str(backup_path)
    except Exception:
        shutil.rmtree(backup_dir,ignore_errors=True)
        raise
    return {'target_version':version,'previous_version':old_version,'target_release':str(target),
            'previous_release':str(previous),'database_backend':backend,
            'database_backup':backup,'started_at':time.time()}


def _validated_transaction_context(context: dict, root: Path):
    if not isinstance(context,dict): raise RuntimeError('Update transaction is invalid')
    releases=(root/'releases').resolve()
    target=Path(str(context.get('target_release') or '')).resolve()
    previous=Path(str(context.get('previous_release') or '')).resolve()
    if not target.is_relative_to(releases) or not previous.is_relative_to(releases):
        raise RuntimeError('Update transaction release path is outside the release root')
    if not target.is_dir() or not previous.is_dir():
        raise RuntimeError('Update transaction release path is missing')
    _safe_version(str(context.get('target_version') or ''))
    backup=str(context.get('database_backup') or '')
    if backup:
        backup_path=Path(backup).resolve(); allowed=(data_root()/'update-backups').resolve()
        if not backup_path.is_relative_to(allowed):
            raise RuntimeError('Update transaction database backup is outside the backup root')
        if not backup_path.is_file():
            raise RuntimeError('Update transaction database backup is missing')
    clean=dict(context); clean.pop('status',None); clean.pop('version',None)
    return clean


def _rollback_context(context: dict, reason: str, root: Path):
    context=_validated_transaction_context(context,root)
    current=root/'current'; target=Path(context['target_release']); previous=Path(context['previous_release'])
    try: _compose(target,root,'down')
    except Exception: pass
    backup=str(context.get('database_backup') or '')
    backend=str(context.get('database_backend') or 'sqlite').lower()
    if backup and backend=='postgres':
        _compose_service_up(previous,root,'core-postgres'); _wait_postgres(root); _restore_postgres(root,Path(backup))
    elif backup:
        _restore_file(Path(backup),data_root()/'surroundcore.sqlite3')
    _write_env_value(root/'.env','SURROUNDCORE_BUILD_VERSION',str(context.get('previous_version') or ''))
    _atomic_symlink(current,previous)
    _sync_existing_helper_tokens(root)
    helper_restore_error=''
    try: _refresh_host_helpers(previous,root)
    except Exception as exc: helper_restore_error=str(exc)[:2000]
    _compose(previous,root,'up')
    _schedule_update_helper_restart()
    fields={'rolled_back_at':time.time(),'rollback_reason':str(reason)[:2000]}
    if helper_restore_error: fields['helper_restore_error']=helper_restore_error
    return _write_state('rolled_back',**context,**fields)


def _install_runtime_updater(release: Path) -> bool:
    """Atomically promote the updater shipped by a confirmed release.

    This lets a release extend the manifest/runtime layout (for example bundled
    modules) without leaving the appliance pinned to an older verifier.
    """
    if os.geteuid()!=0:
        return False
    source=Path(release)/'deploy'/'surroundcore-update'
    if not source.is_file():
        raise RuntimeError('Confirmed release does not contain the update runtime')
    destination=Path(os.getenv('SURROUNDCORE_UPDATER_PATH','/usr/local/bin/surroundcore-update'))
    destination.parent.mkdir(parents=True,exist_ok=True)
    temp=destination.with_name('.'+destination.name+'.new')
    try:
        shutil.copy2(source,temp)
        os.chmod(temp,0o755)
        os.replace(temp,destination)
    finally:
        try:
            if temp.exists(): temp.unlink()
        except OSError:
            pass
    return True


def _schedule_update_helper_restart(delay_seconds: int=3) -> bool:
    """Restart the parent update-helper only after this updater has had time to exit.

    A transient systemd timer is outside the helper service cgroup, so refreshing
    the update-helper no longer requires a reboot/manual restart and cannot kill
    the updater child that is completing the transaction.
    """
    if os.geteuid()!=0:
        return False
    systemd_run=shutil.which('systemd-run'); systemctl=shutil.which('systemctl')
    if not systemd_run or not systemctl:
        return False
    delay=max(2,min(int(delay_seconds),30))
    unit=f'core-update-helper-refresh-{os.getpid()}-{int(time.time()*1000)}'
    try:
        result=subprocess.run([
            systemd_run,'--quiet','--collect',f'--unit={unit}',f'--on-active={delay}s',
            systemctl,'restart','core-update-helper.service',
        ],text=True,capture_output=True,check=False)
    except OSError:
        return False
    return result.returncode==0


def _confirm_already_current(version: str, current: Path, health_url: str, root: Path) -> dict | None:
    """Return healthy same-version state without creating a rollback-to-self transaction."""
    try:
        health=_health(health_url)
    except Exception:
        return None
    schema=health.get('database_schema') or {}
    if not (health.get('ready') and str(health.get('version') or '')==version and
            schema.get('current_for_binary')):
        return None
    try: state=json.loads(state_path().read_text())
    except FileNotFoundError: state={'version':STATE_VERSION,'status':'idle'}
    except Exception: state={}
    if state.get('status')=='pending' and str(state.get('target_version') or '')==version:
        fields={k:v for k,v in state.items() if k not in {'status','version','confirmed_at','confirmed_schema'}}
        state=_write_state('confirmed',**fields,confirmed_at=time.time(),
                           confirmed_schema=int(schema.get('current') or 0))
        health=dict(health); health['update']=state
    _sync_existing_helper_tokens(root)
    _refresh_host_helpers(current,root)
    _install_runtime_updater(current)
    _schedule_update_helper_restart()
    return health


def activate_release(source: Path, version: str, *, root: Path | None=None,
                     health_url: str='http://127.0.0.1:8080/api/v1/health', timeout: int=90):
    root=root or install_root(); version=_safe_version(version)
    source=Path(source).resolve(); _validate_source(source); _verify_release_manifest(source,version)
    previous=_current_release(root)
    if previous is None: raise RuntimeError('Versioned Core layout is not initialized')
    expected=(root/'releases'/version).resolve()
    if previous.resolve()==expected:
        health=_confirm_already_current(version,previous,health_url,root)
        if health is not None:
            return health
        raise RuntimeError('Release is already current but is not healthy; refusing rollback-to-self activation')
    target=stage_release(source,version,root)
    _sync_existing_helper_tokens(root)
    context=_transaction_context(version,target,previous,root)
    _write_state('pending',**context)
    _write_env_value(root/'.env','SURROUNDCORE_BUILD_VERSION',version)
    _atomic_symlink(root/'current',target)
    try:
        _compose(target,root,'up')
        health=_wait_for_confirmation(version,health_url,timeout)
        _refresh_host_helpers(target,root)
        _install_runtime_updater(target)
        _prune_update_backups()
        _schedule_update_helper_restart()
        return health
    except Exception as exc:
        _rollback_context(context,str(exc),root)
        raise


def recover_pending(*, root: Path | None=None, health_url: str='http://127.0.0.1:8080/api/v1/health', timeout: int=30):
    root=root or install_root()
    try: state=json.loads(state_path().read_text())
    except FileNotFoundError: return {'action':'none','state':{'version':STATE_VERSION,'status':'idle'}}
    except Exception as exc: raise RuntimeError('Update transaction state is unreadable') from exc
    if state.get('status') != 'pending': return {'action':'none','state':state}
    context=_validated_transaction_context(state,root)
    version=str(context.get('target_version') or '')
    try:
        health=_wait_for_confirmation(version,health_url,timeout)
        target=Path(context['target_release'])
        _sync_existing_helper_tokens(root)
        _refresh_host_helpers(target,root)
        _install_runtime_updater(target)
        _prune_update_backups()
        _schedule_update_helper_restart()
        return {'action':'confirmed','health':health,'state':json.loads(state_path().read_text())}
    except Exception as exc:
        rolled=_rollback_context(context,'interrupted update recovery: '+str(exc),root)
        return {'action':'rolled_back','state':rolled}


def rollback_pending(*, root: Path | None=None, reason: str='manual rollback'):
    root=root or install_root()
    try: state=json.loads(state_path().read_text())
    except Exception as exc: raise RuntimeError('No readable update transaction is available') from exc
    if state.get('status') not in {'pending','confirmed'}: raise RuntimeError('No active update can be rolled back')
    if not state.get('previous_release'): raise RuntimeError('Update transaction has no previous release')
    return _rollback_context(state,reason,root)


def _parser():
    parser=argparse.ArgumentParser(description='SurroundCore atomic release updater')
    sub=parser.add_subparsers(dest='command',required=True)
    manifest=sub.add_parser('manifest'); manifest.add_argument('version'); manifest.add_argument('source')
    stage=sub.add_parser('stage'); stage.add_argument('version'); stage.add_argument('source')
    activate=sub.add_parser('activate'); activate.add_argument('version'); activate.add_argument('source')
    activate.add_argument('--health-url',default='http://127.0.0.1:8080/api/v1/health'); activate.add_argument('--timeout',type=int,default=90)
    rollback=sub.add_parser('rollback'); rollback.add_argument('--reason',default='manual rollback')
    recover=sub.add_parser('recover'); recover.add_argument('--health-url',default='http://127.0.0.1:8080/api/v1/health'); recover.add_argument('--timeout',type=int,default=30)
    sub.add_parser('status')
    return parser


def main(argv=None):
    args=_parser().parse_args(argv); root=install_root()
    if args.command=='manifest':
        print(write_release_manifest(Path(args.source),args.version)); return 0
    if args.command=='stage':
        print(stage_release(Path(args.source),args.version,root)); return 0
    if args.command=='activate':
        if os.geteuid()!=0: raise SystemExit('activate must run as root')
        health=activate_release(Path(args.source),args.version,root=root,health_url=args.health_url,timeout=args.timeout)
        print(json.dumps(health,indent=2,sort_keys=True)); return 0
    if args.command=='rollback':
        if os.geteuid()!=0: raise SystemExit('rollback must run as root')
        print(json.dumps(rollback_pending(root=root,reason=args.reason),indent=2,sort_keys=True)); return 0
    if args.command=='recover':
        if os.geteuid()!=0: raise SystemExit('recover must run as root')
        print(json.dumps(recover_pending(root=root,health_url=args.health_url,timeout=args.timeout),indent=2,sort_keys=True)); return 0
    if args.command=='status':
        try: value=json.loads(state_path().read_text())
        except FileNotFoundError: value={'version':STATE_VERSION,'status':'idle'}
        print(json.dumps(value,indent=2,sort_keys=True)); return 0
    return 2


if __name__=='__main__':
    raise SystemExit(main())
