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

"""Public r4 payload; supports the two explicitly delivered full baseline states."""
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-r4'
NAME = 'ksf-workshop-skill-update-20261109-r4'
sha = lambda b: hashlib.sha256(b).hexdigest()
encode = lambda x: (json.dumps(x, ensure_ascii=False, indent=2) + '\n').encode()


def stage():
    old_path = ROOT / 'docs/workshop-skill-release-20261108-r2/ksf-workshop-unified-20261108-r2.zip'
    third_path = ROOT / 'docs/workshop-skill-release-20261109-r3/ksf-workshop-skill-update-20261109-r3.zip'
    assert sha(old_path.read_bytes()) == '65660a366589d1b7a6c93b333e9ab1017a487e4b3f5c7399f225c38aedb9e844'
    assert sha(third_path.read_bytes()) == '83c44c8a2f16c682faa5f91f41f513c7d638839cedf43516f96a6cb9bec14828'
    with zipfile.ZipFile(old_path) as z:
        old = json.loads(z.read('release.json'))
        r2 = old['candidate']
        r2bytes = {n: z.read('skill/' + n) for n in r2}
        expected = json.loads(z.read('expected-agent-read.json'))
    with zipfile.ZipFile(third_path) as z:
        third = json.loads(z.read('release.json'))
        r3 = third['candidate']
        r3bytes = {n: z.read('payload/' + n) if r2.get(n) != h else r2bytes[n] for n, h in r3.items()}
    for version, mapping, bodies in [('r2', r2, r2bytes), ('r3', r3, r3bytes)]:
        for n, h in mapping.items():
            assert sha(bodies[n]) == h
            p = OUT / ('rehearsal-baseline-' + version) / n
            p.parent.mkdir(parents=True, exist_ok=True)
            p.write_bytes(bodies[n])
    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) == 39
    candidate['FILES.sha256'] = sha((skill / 'FILES.sha256').read_bytes())
    assert candidate['SKILL.md'] == '85e44939bdbb3b55d512a4f5eb6daf9eac216634d1165ac3b576e9cc470176e8'
    deltas = {version: {n: h for n, h in candidate.items() if base.get(n) != h} for version, base in [('r2', r2), ('r3', r3)]}
    assert len(deltas['r2']) == 14 and len(deltas['r3']) == 12
    assert len(set(candidate) - set(r2)) == 2 and len(set(candidate) - set(r3)) == 1
    for base in (r2, r3):
        assert not set(base) - set(candidate)
        assert all(candidate[n] == h for n, h in base.items() if n.startswith('scripts/') or n == 'runtime-lock.json')
    expected['skillFingerprint'] = candidate['SKILL.md']
    expected['generationPolicy'] = 'Pin current contract generation; compare every read and board to it.'
    release = {'schema': 'ksf.workshop-research-update.v1', 'releaseId': NAME,
               'baselines': {'r2': r2, 'r3': r3}, 'candidate': candidate,
               'baselineArchiveSha256': {'r2': sha(old_path.read_bytes()), 'r3': sha(third_path.read_bytes())},
               'researchEntryNames': [n for n in old['legacyManifestPolicy']['entryNames'] if n != 'SKILL.md'],
               'researchRuntimePins': old['legacyManifestPolicy']['verifiedRuntimeEntries'],
               'plans': {v: [{'path': n, 'action': 'update' if n in base else 'create', 'beforeSha256': base.get(n), 'afterSha256': h} for n, h in deltas[v].items()] for v, base in [('r2', r2), ('r3', r3)]},
               'runtimeReplacementAllowed': False, 'businessWritesDuringUpgrade': False,
               'serviceRestartAllowed': False, 'credentialsIncluded': False}
    # Link QA is offline and limited to relative file targets, including each anchor's file.
    links = 0
    for p in [skill / n for n in candidate if n.endswith('.md')]:
        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(), (p, 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': 39, 'publicFiles': 40, '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, 'generationPolicy': 'same current research generation; external originals listed separately'}
    (OUT / 'skill-integrity.json').write_bytes(encode(integrity))
    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-research.mjs').read_bytes(),
                  'run-update.py': (ROOT / 'scripts/run-workshop-research-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)})
    for n in ('local-acceptance.json', 'installer-validation.json'):
        p = OUT / n
        if p.exists():
            v = json.loads(p.read_bytes())
            assert v['ok']
            if n == 'local-acceptance.json':
                assert v['skillSha256'] == candidate['SKILL.md'] and v['manifestSha256'] == candidate['FILES.sha256']
            else:
                assert v['updaterSha256'] == sha(files['update-skill.mjs']) and v['runnerSha256'] == sha(files['run-update.py'])
            files[n] = p.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', 'local-acceptance.json'} <= set(files)
    dest = OUT / (NAME + '.zip')
    assert not dest.exists(), 'Choose a new revision; never overwrite a released ZIP'
    with zipfile.ZipFile(dest, '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(dest) 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': dest.name, 'bytes': dest.stat().st_size,
               'sha256': sha(dest.read_bytes()), 'packageEntries': len(files), 'outerManifestEntries': len(files) - 1,
               'payloadFiles': 14, 'targetPublicFiles': 40, 'plans': {'r2': {'updates': 12, 'creates': 2}, 'r3': {'updates': 11, 'creates': 1}},
               'candidateSkillSha256': release['candidate']['SKILL.md'], 'candidateManifestSha256': release['candidate']['FILES.sha256'],
               'runtimeFilesChanged': 0, 'credentialsIncluded': False, 'remoteInstalled': False, 'externalCollectionPerformed': False, 'paidCalls': 0}
    (OUT / 'package-receipt.json').write_bytes(encode(receipt))
    print(json.dumps(receipt, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()