kangshifu-board/scripts/workshop-onboarding/upgrade-platform.py

"""Strict known-baseline, documentation-only upgrade of the original workshop Agent."""
import argparse
import importlib.util
import json
import os
from pathlib import Path
import shutil
import sys

HERE = Path(__file__).resolve().parent


def core_at(package):
    spec = importlib.util.spec_from_file_location('workshop_installer_core', package / 'install-core.py')
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def snapshot(core, target):
    values = {}
    for path in sorted(target.rglob('*')):
        core.plain_path(path)
        if path.is_file():
            values[path.relative_to(target).as_posix()] = {
                'sha256': core.digest(path.read_bytes()), 'mode': path.stat().st_mode & 0o777}
        else:
            core.require(path.is_dir(), 'SPECIAL_FILE_NOT_SUPPORTED')
    return values


def write(core, path, value):
    path.write_bytes(core.encode(value))
    path.chmod(0o600)


def load(core, package):
    package = core.plain_path(package)
    for line in (package / 'FILES.sha256').read_text('utf-8').splitlines():
        h, name = line.split('  ', 1)
        core.require(core.safe_name(name), 'INVALID_PACKAGE_PATH')
        p = core.plain_path(package / name)
        core.require(p.is_file() and core.digest(p.read_bytes()) == h, 'PACKAGE_HASH_MISMATCH')
    meta = json.loads((package / 'release.json').read_bytes())
    core.require(meta['schema'] == 'ksf.workshop-platform-upgrade.v1', 'INVALID_RELEASE')
    candidate = meta['candidate']
    core.require(len(candidate) == 43 and candidate['SKILL.md'] ==
        '8757e3b58b461deb70201469f2e3579a7f03e5e6204c94aa327e143aa9b44bb3', 'UNKNOWN_TARGET')
    content = {}
    for name, expected in candidate.items():
        core.require(core.safe_name(name), 'INVALID_PACKAGE_PATH')
        data = core.plain_path(package / 'skill' / name).read_bytes()
        core.require(core.digest(data) == expected, 'CANDIDATE_HASH_MISMATCH')
        content[name] = data
    protected = {n for n in candidate if n.startswith('scripts/') or n == 'runtime-lock.json'}
    core.require(len(protected) == 7 and set(meta['protectedFiles']) == protected, 'PROTECTED_FILE_LAYOUT')
    for baseline in meta['baselines'].values():
        core.require(all(baseline.get(n) == candidate[n] for n in protected), 'PROTECTED_RUNTIME_CHANGED')
    return meta, content


def run(command, target, output, package=HERE):
    package = Path(package).resolve()
    core = core_at(package)
    core.require(Path(target).is_absolute() and Path(output).is_absolute(), 'ABSOLUTE_PATH_REQUIRED')
    target, output = core.plain_path(target), core.plain_path(output)
    core.require(target.is_dir() and target.name == 'ksf-workshop-runtime', 'EXISTING_WORKSHOP_TARGET_REQUIRED')
    core.require(not output.is_relative_to(target.parent) and not package.is_relative_to(target)
        and not output.is_relative_to(package), 'RESULT_TARGET_OVERLAP')
    core.plain_path(target / 'credential-binding.json')
    core.require((target / 'credential-binding.json').is_file(), 'EXISTING_BINDING_REQUIRED')
    if os.name != 'nt':
        core.require(os.geteuid() == target.stat().st_uid, 'TARGET_OWNER_REQUIRED')
        core.require((target / 'credential-binding.json').stat().st_mode & 0o777 == 0o600,
                     'PRIVATE_BINDING_PERMISSIONS')
    meta, content = load(core, package)
    files = {'ksf-workshop-runtime': content}

    if command == 'inventory':
        public = {n: core.digest(core.plain_path(target / n).read_bytes())
                  if core.plain_path(target / n).is_file() else None for n in content}
        return {'ok': True, 'readOnly': True, 'publicFilesOnly': True,
                'files': public, 'bindingIncluded': False, 'businessWrites': 0}

    if command == 'plan':
        core.require(not output.exists(), 'NEW_RESULT_DIRECTORY_REQUIRED')
        actions = core.plan(target.parent, files, meta['baselines'])
        output.mkdir(parents=True, mode=0o700)
        output.chmod(0o700)
        record = {'schema': 'ksf.workshop-platform-upgrade-plan.v1', 'releaseId': meta['releaseId'],
                  'target': str(target), 'candidate': meta['candidate'], 'actions': actions,
                  'before': snapshot(core, target)}
        write(core, output / 'plan.json', record)
        action = actions[0]
        baseline = action.get('before', meta['candidate'])
        changed = [n for n in content if baseline.get(n) != meta['candidate'][n]]
        result = {'ok': True, 'readOnly': True, 'action': action['action'],
                  'baselineId': action.get('fromVersion', 'r10'), 'changes': len(changed),
                  'creates': len(set(content) - set(baseline)), 'changedFiles': changed,
                  'protectedFilesModified': 0, 'vocModified': False, 'businessWrites': 0}
        write(core, output / 'plan-summary.json', result)
        return result

    record = json.loads((output / 'plan.json').read_bytes())
    core.require(record['target'] == str(target) and record['candidate'] == meta['candidate'], 'PLAN_MISMATCH')

    def verify():
        core.verify_component(target, content)
        after = snapshot(core, target)
        for name, old in record['before'].items():
            if name not in content or name in meta['protectedFiles']:
                core.require(after.get(name) == old, 'PRESERVED_FILE_CHANGED')
        return {'ok': True, 'workshopVersion': 'r10', 'publicFilesVerified': 43,
                'protectedFilesModified': 0, 'localFilesPreserved': True,
                'agentLoaded': False, 'vocModified': False, 'businessWrites': 0}

    if command == 'verify':
        result = verify()
        write(core, output / 'verify.json', result)
        return result

    core.require(command == 'apply', 'UNKNOWN_COMMAND')
    if (output / 'apply.json').exists():
        old = json.loads((output / 'apply.json').read_bytes())
        core.require(old.get('ok') is True, 'PRIOR_APPLY_REQUIRES_REVIEW')
        return {**verify(), 'alreadyApplied': True}
    core.require(snapshot(core, target) == record['before'], 'TARGET_CHANGED_SINCE_PLAN')
    actions = core.plan(target.parent, files, meta['baselines'])
    core.require(actions == record['actions'], 'PLAN_CHANGED')
    if actions[0]['action'] == 'verify-only':
        return {**verify(), 'alreadyApplied': True}
    backup = output / 'before-full'
    core.require(not backup.exists(), 'PRIOR_BACKUP_REQUIRES_REVIEW')
    shutil.copytree(target, backup, copy_function=shutil.copy2)
    backup.chmod(0o700)
    core.require(snapshot(core, backup) == record['before'], 'BACKUP_VERIFY_FAILED')
    core.require(snapshot(core, target) == record['before'], 'TARGET_CHANGED_SINCE_PLAN')
    try:
        core.install_files(target.parent, files, actions, output / 'before-changed')
        result = {**verify(), 'applied': True, 'fullBackupVerified': True}
        write(core, output / 'apply.json', result)
        return result
    except Exception:
        restored = snapshot(core, target) == record['before']
        write(core, output / 'apply.json', {'ok': False, 'originalRestored': restored})
        raise


if __name__ == '__main__':
    p = argparse.ArgumentParser(description=__doc__)
    p.add_argument('command', choices=['plan', 'apply', 'verify', 'inventory'])
    p.add_argument('--target-dir', required=True)
    p.add_argument('--out', required=True)
    args = p.parse_args()
    try:
        print(json.dumps(run(args.command, args.target_dir, args.out), ensure_ascii=False, indent=2))
    except Exception as exc:
        allowed = str(exc)
        code = allowed if allowed and all(c.isupper() or c.isdigit() or c == '_' for c in allowed) else 'PLATFORM_UPGRADE_STOPPED'
        print(json.dumps({'ok': False, 'errorCode': code}), file=sys.stderr)
        sys.exit(1)