"""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)