kangshifu-board/scripts/run-workshop-quality-update.py

"""Capture dry/apply-once/verify results without shell quoting or blind retries."""
import argparse
import hashlib
import json
import os
from pathlib import Path
import subprocess
import sys


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--out', required=True, help='New absolute result directory; never reuse')
    args = parser.parse_args()
    package = Path(__file__).resolve().parent
    output = Path(args.out).absolute()
    if (not Path(args.out).is_absolute() or output.exists() or
            any(p.is_symlink() for p in (output, *output.parents))):
        raise ValueError('NEW_ABSOLUTE_OUTPUT_DIRECTORY_REQUIRED')
    if not sys.platform.startswith('linux') or os.getuid() != 10000:
        raise ValueError('HERMES_UID_10000_REQUIRED')
    for line in (package / 'FILES.sha256').read_text(encoding='utf-8').splitlines():
        expected, name = line.split('  ', 1)
        parts = Path(name).parts
        if Path(name).is_absolute() or any(p in ('.', '..') for p in parts):
            raise ValueError('INVALID_PACKAGE_PATH')
        if hashlib.sha256((package / name).read_bytes()).hexdigest() != expected:
            raise ValueError('PACKAGE_HASH_MISMATCH')
    output.mkdir(mode=0o700, parents=False)
    steps = {}

    def run(step, *arguments):
        result = subprocess.run(['node', str(package / 'update-skill.mjs'), *arguments],
                                cwd=package, capture_output=True, check=False)
        (output / (step + '.stdout.json')).write_bytes(result.stdout)
        (output / (step + '.stderr.json')).write_bytes(result.stderr)
        (output / (step + '.exit')).write_text(str(result.returncode) + '\n', encoding='utf-8')
        steps[step] = {'exitCode': result.returncode, 'stderrBytes': len(result.stderr)}
        if result.returncode != 0:
            raise ValueError(step.upper() + '_FAILED_SEE_SAVED_OUTPUT')
        payload = json.loads(result.stdout)
        if payload.get('ok') is not True:
            raise ValueError(step.upper() + '_NOT_OK')
        return payload

    receipt = {'schema': 'ksf.workshop-docs-install-run.v1', 'ok': False,
               'releaseId': 'ksf-workshop-skill-update-20261109-r5', 'steps': steps,
               'filesInstalled': False, 'agentLoaded': False,
               'currentReadVerified': False, 'sameGeneration': False,
               'actualBusinessWriteVerified': False, 'formalReportSubmitted': False,
               'databaseWrites': 0, 'paidCalls': 0, 'serviceRestarted': False,
               'credentialsExported': False}
    try:
        dry = run('dry', '--dry')
        expected_plan = {'r2': (17, 14, 3), 'r3': (16, 14, 2), 'r4': (9, 8, 1)}.get(dry.get('baselineId'))
        if dry.get('alreadyApplied') is True:
            if dry.get('changedFiles') != 0:
                raise ValueError('ALREADY_APPLIED_PLAN_MISMATCH')
            receipt.update(ok=True, filesInstalled=True, alreadyApplied=True,
                           note='Full target matched; no apply or new backup performed.')
        else:
            if (expected_plan is None or dry.get('readOnly') is not True or
                    dry.get('changedFiles') != expected_plan[0] or
                    sum(c['action'] == 'update' for c in dry['changes']) != expected_plan[1] or
                    sum(c['action'] == 'create' for c in dry['changes']) != expected_plan[2]):
                raise ValueError('UNEXPECTED_DRY_PLAN')
            applied = run('apply', '--apply')
            receipt['backup'] = applied['backup']
            verified = run('verify', '--verify', '--backup', applied['backup'])
            if verified.get('installedFilesVerified') != 41 or verified.get('changedFilesVerified') != expected_plan[0]:
                raise ValueError('UNEXPECTED_VERIFY_RESULT')
            receipt.update(ok=True, filesInstalled=True, alreadyApplied=False,
                           installedFilesVerified=41, changedFiles=expected_plan[0], baselineId=dry['baselineId'],
                           runtimeAndBindingPreserved=verified['runtimeAndBindingPreserved'],
                           researchFilesUnchanged=verified['researchFilesUnchanged'],
                           legacyEntrypointRetired=verified['legacyEntrypointRetired'])
        receipt['skillSha256'] = dry['skillSha256']
        receipt['manifestSha256'] = dry['manifestSha256']
    except Exception as exc:
        # Store only our fixed error identifiers, not raw exceptions or environment.
        text = str(exc)
        receipt['errorCode'] = text if text and all(c in 'ABCDEFGHIJKLMNOPQRSTUVWXYZ_' for c in text) else 'RUN_FAILED_SEE_SAVED_OUTPUT'
    finally:
        (output / 'install-receipt.json').write_text(json.dumps(receipt, ensure_ascii=False, indent=2) + '\n', encoding='utf-8')
    print(json.dumps(receipt, ensure_ascii=False, indent=2))
    return 0 if receipt['ok'] else 1


if __name__ == '__main__':
    try:
        sys.exit(main())
    except Exception:
        print(json.dumps({'ok': False, 'errorCode': 'PRECHECK_FAILED_NO_INSTALL_STARTED'}))
        sys.exit(1)