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

"""Stage/publish r5 with exact r2/r3/r4 baseline reconstruction; never overwrite a ZIP."""
from pathlib import Path
import argparse
import hashlib
import json
import re
import zipfile

ROOT = Path(__file__).resolve().parents[1]
OUT = ROOT / 'docs/workshop-skill-release-20261109-r5'
NAME = 'ksf-workshop-skill-update-20261109-r5'
sha = lambda b: hashlib.sha256(b).hexdigest()
encode = lambda v: (json.dumps(v, ensure_ascii=False, indent=2) + '\n').encode()
ARCHIVES = {
    'r2': ('docs/workshop-skill-release-20261108-r2/ksf-workshop-unified-20261108-r2.zip', '65660a366589d1b7a6c93b333e9ab1017a487e4b3f5c7399f225c38aedb9e844'),
    'r3': ('docs/workshop-skill-release-20261109-r3/ksf-workshop-skill-update-20261109-r3.zip', '83c44c8a2f16c682faa5f91f41f513c7d638839cedf43516f96a6cb9bec14828'),
    'r4': ('docs/workshop-skill-release-20261109-r4/ksf-workshop-skill-update-20261109-r4.zip', 'c9b60ce30fb5d3bed2b2a73ac7bc33f6c6b639dc100afd7cef4fe82a7e988308'),
}


def stage():
    OUT.mkdir(exist_ok=True)
    baselines, bodies = {}, {}
    previous = {}
    for version, (relative, digest) in ARCHIVES.items():
        source = ROOT / relative
        assert sha(source.read_bytes()) == digest
        with zipfile.ZipFile(source) as z:
            assert z.testzip() is None
            meta = json.loads(z.read('release.json'))
            baselines[version] = meta['candidate']
            prefix = 'skill/' if version == 'r2' else 'payload/'
            current = {n: z.read(prefix + n) if prefix + n in z.namelist() else previous[n] for n in meta['candidate']}
            for n, h in baselines[version].items():
                assert sha(current[n]) == h, (version, n)
                p = OUT / ('rehearsal-baseline-' + version) / n
                p.parent.mkdir(parents=True, exist_ok=True)
                p.write_bytes(current[n])
            bodies[version] = current
            previous = current
            if version == 'r2':
                policy = meta['legacyManifestPolicy']
            if version == 'r4':
                expected = json.loads(z.read('expected-agent-read.json'))
    skill = ROOT / 'skills/ksf-workshop-runtime'
    candidate = {}
    for line in (skill / 'FILES.sha256').read_text(encoding='utf-8').splitlines():
        h, n = line.split('  ', 1)
        assert '..' not in Path(n).parts and not Path(n).is_absolute() and n not in candidate
        assert sha((skill / n).read_bytes()) == h
        candidate[n] = h
    assert len(candidate) == 40
    candidate['FILES.sha256'] = sha((skill / 'FILES.sha256').read_bytes())
    assert candidate['SKILL.md'] == '16b428f4f4fa0819a764505269116f3f91e503342d9813d3855bc5c2cf80feb6'
    assert candidate['FILES.sha256'] == '21ae0a245de46e9acbf97ed31718aeb1cdb4ea8ff1ca38259a2b3a9b0c9795e1'
    deltas = {v: {n: h for n, h in candidate.items() if base.get(n) != h} for v, base in baselines.items()}
    for v, counts in {'r2': (17, 3), 'r3': (16, 2), 'r4': (9, 1)}.items():
        assert len(deltas[v]) == counts[0] and len(set(candidate) - set(baselines[v])) == counts[1]
        assert not set(baselines[v]) - set(candidate)
        assert all(candidate[n] == h for n, h in baselines[v].items() if n.startswith('scripts/') or n == 'runtime-lock.json')
    expected['skillFingerprint'] = candidate['SKILL.md']
    release = {
        'schema': 'ksf.workshop-research-update.v1', 'releaseId': NAME,
        'baselines': baselines, 'candidate': candidate,
        'baselineArchiveSha256': {v: x[1] for v, x in ARCHIVES.items()},
        'researchEntryNames': [n for n in policy['entryNames'] if n != 'SKILL.md'],
        'researchRuntimePins': policy['verifiedRuntimeEntries'],
        'plans': {v: [{'path': n, 'action': 'update' if n in baselines[v] else 'create', 'beforeSha256': baselines[v].get(n), 'afterSha256': h} for n, h in delta.items()] for v, delta in deltas.items()},
        'runtimeReplacementAllowed': False, 'businessWritesDuringUpgrade': False,
        'serviceRestartAllowed': False, 'credentialsIncluded': False,
    }
    links = 0
    for n in candidate:
        if not n.endswith('.md'):
            continue
        p = skill / n
        for target in re.findall(r'\]\(([^)]+)\)', p.read_text(encoding='utf-8')):
            if '://' in target or target.startswith('#'):
                continue
            resolved = (p.parent / target.split('#', 1)[0]).resolve()
            assert resolved.is_relative_to(skill.resolve()) and resolved.is_file(), (n, target)
            links += 1
    refs = [n for n in candidate if n.startswith('references/')]
    integrity = {'ok': True, 'skillSha256': candidate['SKILL.md'], 'manifestSha256': candidate['FILES.sha256'],
                 'manifestEntries': 40, 'publicFiles': 41, 'relativeLinks': links,
                 'references': {'all': len(refs), 'md': sum(n.endswith('.md') for n in refs), 'json': sum(n.endswith('.json') for n in refs)},
                 'runtimeFilesUnchanged': 7, 'courseModulesPreserved': 13}
    assert integrity['references'] == {'all': 32, 'md': 30, 'json': 2}
    (OUT / 'skill-integrity.json').write_bytes(encode(integrity))
    local = json.loads((ROOT / 'docs/workshop-decision-quality-20261109/local-revision.json').read_bytes())
    assert local['skillSha256'] == candidate['SKILL.md'] and local['manifestSha256'] == candidate['FILES.sha256']
    local_receipt = {'schema': 'ksf.quality-revision-local-acceptance.v1', 'ok': True,
                     'skillSha256': candidate['SKILL.md'], 'manifestSha256': candidate['FILES.sha256'],
                     'sourceReceiptSha256': sha(encode(local)), 'manualSyntheticProbes': 5,
                     'method': local['manualContentReview'], 'sourceAndArithmeticChecks': local['sourceAndArithmeticChecks'],
                     'previewRenderChecks': 30, 'screenshotsManuallyReviewed': True,
                     'researchModulesPreserved': 13, 'runtimeFilesUnchanged': 7,
                     'actualDifferentModelReview': False, 'realVOCInvocation': False,
                     'remoteInstalled': False, 'productionDatabaseWrites': 0, 'paidCalls': 0}
    (OUT / 'local-acceptance.json').write_bytes(encode(local_receipt))
    files = {'payload/' + n: (skill / n).read_bytes() for n in deltas['r2']}
    files.update({'release.json': encode(release), 'expected-agent-read.json': encode(expected),
                  'update-skill.mjs': (ROOT / 'scripts/update-workshop-quality.mjs').read_bytes(),
                  'run-update.py': (ROOT / 'scripts/run-workshop-quality-update.py').read_bytes(),
                  'probe-agent-library-read.py': (ROOT / 'scripts/probe-agent-library-read.py').read_bytes(),
                  'README.md': (OUT / 'README.md').read_bytes(), 'CHANGES.md': (OUT / 'CHANGES.md').read_bytes(),
                  'skill-integrity.json': encode(integrity), 'local-acceptance.json': encode(local_receipt)})
    validation = OUT / 'installer-validation.json'
    if validation.exists():
        v = json.loads(validation.read_bytes())
        assert v['ok'] and v['updaterSha256'] == sha(files['update-skill.mjs']) and v['runnerSha256'] == sha(files['run-update.py'])
        files[validation.name] = validation.read_bytes()
    files['FILES.sha256'] = ''.join(sha(raw) + '  ' + n + '\n' for n, raw in sorted(files.items())).encode()
    for n, raw in files.items():
        assert not any(t in n.lower() for t in ('credential-binding', 'private.pem', 'envelope', 'channel-config'))
        p = OUT / 'package' / n
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_bytes(raw)
    return files, release


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument('--pack', action='store_true')
    args = parser.parse_args()
    files, release = stage()
    if not args.pack:
        print(json.dumps({'ok': True, 'staged': len(files), 'plans': {v: len(p) for v, p in release['plans'].items()}}))
        return
    assert 'installer-validation.json' in files
    destination = OUT / (NAME + '.zip')
    assert not destination.exists(), 'Choose another revision; never overwrite a released ZIP'
    with zipfile.ZipFile(destination, 'x', zipfile.ZIP_DEFLATED) as z:
        for n, raw in sorted(files.items()):
            info = zipfile.ZipInfo(n, (2026, 11, 9, 0, 0, 0))
            info.external_attr = 0o100600 << 16
            z.writestr(info, raw, compress_type=zipfile.ZIP_DEFLATED)
    with zipfile.ZipFile(destination) as z:
        assert z.testzip() is None and set(z.namelist()) == set(files)
        for n, raw in files.items():
            assert z.read(n) == raw
    receipt = {'ok': True, 'releaseId': NAME, 'filename': destination.name,
               'bytes': destination.stat().st_size, 'sha256': sha(destination.read_bytes()),
               'packageEntries': len(files), 'outerManifestEntries': len(files) - 1,
               'payloadFiles': len([n for n in files if n.startswith('payload/')]), 'targetPublicFiles': 41,
               'plans': {v: {a + 's': sum(p['action'] == a for p in plans) for a in ['update', 'create']} for v, plans in release['plans'].items()},
               'candidateSkillSha256': release['candidate']['SKILL.md'], 'candidateManifestSha256': release['candidate']['FILES.sha256'],
               'runtimeFilesChanged': 0, 'credentialsIncluded': False, 'remoteInstalled': False, 'paidCalls': 0}
    (OUT / 'package-receipt.json').write_bytes(encode(receipt))
    print(json.dumps(receipt, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()