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

"""Build r6 from verified local instructions and reconstruct the sealed r5 baseline."""
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-r6'
NAME = 'ksf-workshop-skill-update-20261109-r6'
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'),
    ('r5', 'docs/workshop-skill-release-20261109-r5/ksf-workshop-skill-update-20261109-r5.zip', 'e72bdd9c08817ae6e017808d3f75ad00e4a15dc044605c70678018fab95b25d4'),
]
DELTA = ['SKILL.md', 'references/workshop-category-scan.md', 'references/workshop-course-flow.md',
         'references/workshop-feedback.md', 'references/workshop-intake.md',
         'references/workshop-research-loop.md', 'FILES.sha256']
sha = lambda b: hashlib.sha256(b).hexdigest()
encode = lambda v: (json.dumps(v, ensure_ascii=False, indent=2) + '\n').encode('utf-8')


def stage():
    bodies = {}
    for version, relative, digest in ARCHIVES:
        archive = ROOT / relative
        assert sha(archive.read_bytes()) == digest, version
        with zipfile.ZipFile(archive) as z:
            assert z.testzip() is None
            meta = json.loads(z.read('release.json'))
            prefix = 'skill/' if version == 'r2' else 'payload/'
            bodies = {n: z.read(prefix + n) if prefix + n in z.namelist() else bodies[n]
                      for n in meta['candidate']}
            for n, h in meta['candidate'].items():
                assert sha(bodies[n]) == h, (version, n)
    baseline = meta['candidate']
    assert len(baseline) == 41 and baseline['SKILL.md'] == '16b428f4f4fa0819a764505269116f3f91e503342d9813d3855bc5c2cf80feb6'
    for n, raw in bodies.items():
        p = OUT / 'rehearsal-baseline-r5' / n
        p.parent.mkdir(parents=True, exist_ok=True)
        p.write_bytes(raw)
    skill = ROOT / 'skills/ksf-workshop-runtime'
    candidate = {}
    for line in (skill / 'FILES.sha256').read_text(encoding='utf-8').splitlines():
        digest, n = line.split('  ', 1)
        assert re.fullmatch(r'[a-f0-9]{64}', digest)
        assert re.fullmatch(r'[A-Za-z0-9_./-]+', n) and '..' not in Path(n).parts
        assert not Path(n).is_absolute() and n not in candidate
        assert sha((skill / n).read_bytes()) == digest, n
        candidate[n] = digest
    candidate['FILES.sha256'] = sha((skill / 'FILES.sha256').read_bytes())
    assert len(candidate) == 42
    assert candidate['SKILL.md'] == '3cb64f381937411b0a2b853772bb19d37f44528a2902a89ef475b707472d3473'
    assert candidate['FILES.sha256'] == 'd1ba798caa776f1691436db885365927e4d77378327578a6c1f12d3fc8564373'
    assert not set(baseline) - set(candidate)
    assert {n for n in candidate if baseline.get(n) != candidate[n]} == set(DELTA)
    assert set(candidate) - set(baseline) == {'references/workshop-intake.md'}
    runtime = [n for n in baseline if n == 'runtime-lock.json' or n.startswith('scripts/')]
    assert len(runtime) == 7 and all(candidate[n] == baseline[n] for n in runtime)
    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
    plan = [{'path': n, 'action': 'update' if n in baseline else 'create',
             'beforeSha256': baseline.get(n), 'afterSha256': candidate[n]} for n in DELTA]
    release = {'schema': 'ksf.workshop-research-update.v1', 'releaseId': NAME,
               'baselines': {'r5': baseline}, 'candidate': candidate,
               'baselineArchiveSha256': {'r5': ARCHIVES[-1][2]},
               'researchEntryNames': meta['researchEntryNames'],
               'researchRuntimePins': meta['researchRuntimePins'],
               'plans': {'r5': plan}, 'runtimeReplacementAllowed': False,
               'businessWritesDuringUpgrade': False, 'serviceRestartAllowed': False,
               'credentialsIncluded': False}
    integrity = {'ok': True, 'skillSha256': candidate['SKILL.md'], 'manifestSha256': candidate['FILES.sha256'],
                 'manifestEntries': 41, 'publicFiles': 42, 'relativeLinks': links,
                 'references': {'all': 33, 'md': 31, 'json': 2},
                 'runtimeFilesUnchanged': 7, 'courseModulesPreserved': 13}
    review = json.loads((ROOT / 'docs/workshop-intake-improvement-20261109/integrity-review.json').read_bytes())
    assert review['currentLocalSkillSha256'] == candidate['SKILL.md'] and review['runtimeFilesModified'] == 0
    local = {'schema': 'ksf.intake-local-review.v1', 'ok': True,
             'skillSha256': candidate['SKILL.md'], 'manifestSha256': candidate['FILES.sha256'],
             'reviewSourceSha256': sha(encode(review)),
             'behaviorReview': review['behaviorReview'],
             'skillStructureValidation': review['skillStructureValidation'],
             'scope': 'manual local synthetic scenarios, not live-instance acceptance',
             'remoteInstalled': False, 'productionDatabaseWrites': 0, 'paidCalls': 0}
    files = {'payload/' + n: (skill / n).read_bytes() for n in DELTA}
    files.update({'release.json': encode(release),
                  'update-skill.mjs': (ROOT / 'scripts/update-workshop-intake.mjs').read_bytes(),
                  'run-update.py': (ROOT / 'scripts/run-workshop-intake-update.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)})
    validation = OUT / 'installer-validation.json'
    if validation.exists():
        value = json.loads(validation.read_bytes())
        assert value['ok'] and value['updaterSha256'] == sha(files['update-skill.mjs'])
        assert value['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)
    (OUT / 'skill-integrity.json').write_bytes(encode(integrity))
    (OUT / 'local-acceptance.json').write_bytes(encode(local))
    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, 'stagedFiles': len(files), 'baseline': 'r5', 'changes': 7}))
        return
    assert 'installer-validation.json' in files, 'Installer validation required before packaging'
    target = OUT / (NAME + '.zip')
    assert not target.exists(), 'Never overwrite a released ZIP'
    with zipfile.ZipFile(target, '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(target) 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': target.name, 'bytes': target.stat().st_size,
               'sha256': sha(target.read_bytes()), 'packageEntries': len(files),
               'outerManifestEntries': len(files) - 1, 'payloadFiles': 7, 'targetPublicFiles': 42,
               'plans': {'r5': {'updates': 6, 'creates': 1}},
               'candidateSkillSha256': release['candidate']['SKILL.md'],
               'candidateManifestSha256': release['candidate']['FILES.sha256'],
               'runtimeFilesChanged': 0, 'credentialsIncluded': False,
               'remoteInstalled': False, 'productionDatabaseWrites': 0, 'paidCalls': 0}
    (OUT / 'package-receipt.json').write_bytes(encode(receipt))
    print(json.dumps(receipt, ensure_ascii=False, indent=2))


if __name__ == '__main__':
    main()