"""Package the new workshop Agent handoff documents; perform no deployment.""" from pathlib import Path import hashlib import json import os import re import tarfile import zipfile ROOT = Path(__file__).resolve().parents[1] DOCS = Path(os.environ.get('KSF_NEW_AGENT_HANDOFF_DIR', 'F:/文档/新数字人交接')) DEST = ROOT / 'work/delivery-by-recipient-20261102' DOCUMENT_NAMES = [ 'README.md', '00-你需要手动做什么.txt', '01-发给运维001的接入说明.md', '02-新数字人角色说明.txt', '03-新数字人工作手册.md', '04-现有成果与接手范围.md', '05-验收与回报清单.md', '06-安装材料索引.json', ] def digest(data): return hashlib.sha256(data).hexdigest() def require(condition, message): if not condition: raise ValueError(message) def manifest_entries(data): result = {} for line in data.decode('utf-8-sig').splitlines(): if not line.strip(): continue match = re.fullmatch(r'([a-f0-9]{64}) (.+)', line) require(match is not None, 'Malformed manifest') sha, name = match.groups() require(name not in result, 'Duplicate manifest entry') require(not name.startswith('/') and '..' not in name.split('/'), 'Unsafe member') result[name] = sha return result def read_package(path, expected_sha, expected_bytes, manifest_name, expected_count, prefix=''): data = path.read_bytes() require(len(data) == expected_bytes and digest(data) == expected_sha, f'Package mismatch: {path.name}') if path.suffix == '.tgz': with tarfile.open(path) as archive: members = archive.getmembers() require(all(not m.issym() and not m.islnk() for m in members), 'Unexpected archive link') files = {m.name: archive.extractfile(m).read() for m in members if m.isfile()} else: with zipfile.ZipFile(path) as archive: require(archive.testzip() is None, 'Corrupt input ZIP') require(len(archive.namelist()) == len(set(archive.namelist())), 'Duplicate ZIP members') files = {name: archive.read(name) for name in archive.namelist() if not name.endswith('/')} entries = manifest_entries(files[manifest_name]) require(len(entries) == expected_count, 'Unexpected manifest size') for name, sha in entries.items(): require(digest(files[prefix + name]) == sha, f'Member mismatch: {name}') return files, { 'filename': path.name, 'bytes': len(data), 'sha256': digest(data), 'knownLocalPath': path.as_posix(), 'manifestPath': manifest_name, 'manifestVerified': f'{len(entries)}/{expected_count}', 'includedInHandoffZip': False, } def main(): v4, v4_info = read_package( Path('D:/下载/WXWork/1688857450794210/Cache/File/2026-11/ksf-adapter-patch-v4-20261102.tgz'), 'fc1888c99160e8af28a69355e78c91d1f4f8b3f612c1cc0f0fb0d1f4f675f0a5', 65842, 'adapter-patch-v4/MANIFEST.sha256', 21, 'adapter-patch-v4/', ) dedup, dedup_info = read_package( DEST / 'ksf-workshop-dedup-20261102.zip', '96fc5c10557b7601b63767e293bcfc33fcea1058f21961949348c105cd75afde', 339939, 'FILES.sha256', 34, ) hotfix, hotfix_info = read_package( DEST / 'ksf-wecom-tool-result-hotfix-20261102.zip', 'ea1cd7e2663b8d67f0ca12381c131bddd4edd51e45518fb55fb04c4d836538e8', 39426, 'FILES.sha256', 14, ) skill_prefix = 'skills/ksf-workshop-runtime/' skill_entries = manifest_entries(dedup[skill_prefix + 'FILES.sha256']) require(len(skill_entries) == 10, 'Skill manifest count changed') for name, sha in skill_entries.items(): require(digest(dedup[skill_prefix + name]) == sha, 'Skill manifest mismatch') final_files = [] for package_info, files, member, target in [ (v4_info, v4, 'adapter-patch-v4/aibot_identity.py', '/opt/hermes/plugins/platforms/wecom/aibot_identity.py'), (v4_info, v4, 'adapter-patch-v4/patches/adapter.py.patched', '/opt/hermes/plugins/platforms/wecom/adapter.py'), (v4_info, v4, 'adapter-patch-v4/plugin/wecom-identity/plugin.yaml', '/opt/data/plugins/wecom-identity/plugin.yaml'), (v4_info, v4, 'adapter-patch-v4/plugin/wecom-identity/__init__.py', '/opt/data/plugins/wecom-identity/__init__.py'), (hotfix_info, hotfix, 'integrations/wecom-identity/identity_bridge.py', '/opt/hermes/plugins/platforms/wecom/identity_bridge.py'), (hotfix_info, hotfix, 'integrations/wecom-identity/workshop_registration.py', '/opt/hermes/plugins/platforms/wecom/workshop_registration.py'), ]: final_files.append({'package': package_info['filename'], 'member': member, 'targetForVerifiedLayout': target, 'sha256': digest(files[member])}) index = { 'version': 'ksf.new-workshop-agent-handoff.v1', 'preparedDate': '2026-11-02', 'purpose': '新实例接入材料索引；只核对原件与清单，不是首次开通器或部署脚本', 'suggestedNameOnly': '康师傅工作坊助手001', 'newAgentCreated': False, 'newAgentRegistered': False, 'newTargetChanged': False, 'configuredNameTriggersRegistration': False, 'platform': 'Hermes + WeCom aibot WebSocket', 'verifiedLayout': {'hermesVersion': '0.21.3', 'gatewayUser': 'hermes', 'uid': 10000, 'hermesHome': '/opt/data', 'requiresNewTargetVerification': True}, 'packages': [v4_info, dedup_info, hotfix_info], 'adapterRequiredBaselineSha256': digest(v4['adapter-patch-v4/baseline/adapter.py']), 'finalPlatformFiles': final_files, 'skill': {'package': dedup_info['filename'], 'prefix': skill_prefix, 'files': skill_entries, 'manifestVerified': '10/10', 'metadataManifestSha256': digest(dedup[skill_prefix + 'FILES.sha256'])}, 'toolNames': ['wecom_identity_current', 'ksf_register_current_employee'], 'contractMarkers': ['ksf.agent-self.v1', 'ksf.wecom-aibot-sender.v1', 'ksf.ingest-dedup.v1'], 'operatorNotes': [ '原安装包不包含在本交接ZIP；运维可按名称和完整指纹复用已收到的原件。', '旧收据的未部署状态仅指打包时；旧测试实例已验，新实例仍待验。', '现有后台先回读，不运行旧身份部署覆盖当前查重合同。', 'install-workshop-skill.mjs是升级器；新实例运行时/独立凭据须经受控首次开通。', '新bot账号与最终名称需一次受控Agent建档及registration.json绑定；名称本身不触发登记。', '不复制旧私钥、信封、credential-binding、Agent/员工状态、instance或context-K。', '全流程与查重脚本含旧固定任务键；新实例不得原样重放。', '本次核验是档案和指纹核验；不代表新实例功能验收已通过。', ], } (DOCS / '06-安装材料索引.json').write_text(json.dumps(index, ensure_ascii=False, indent=2) + '\n', encoding='utf-8') files = {name: (DOCS / name).read_bytes() for name in DOCUMENT_NAMES} require(all(data for data in files.values()), 'Empty handoff document') for name in ['00-你需要手动做什么.txt', '01-发给运维001的接入说明.md', '02-新数字人角色说明.txt', '03-新数字人工作手册.md']: text = files[name].decode('utf-8') require('ksf_register_current_employee' in text, 'Missing real tool name') require('ksfregistercurrent' not in text, 'Incorrect tool alias') for text in files.values(): require(b'-----BEGIN PRIVATE KEY-----' not in text, 'Unexpected private key') files['FILES.sha256'] = ''.join(f'{digest(data)} {name}\n' for name, data in files.items()).encode('utf-8') (DOCS / 'FILES.sha256').write_bytes(files['FILES.sha256']) archive_path = DOCS / 'ksf-new-workshop-agent-handoff-20261102.zip' with zipfile.ZipFile(archive_path, 'w', compression=zipfile.ZIP_DEFLATED) as archive: for name, data in files.items(): info = zipfile.ZipInfo(name, date_time=(2026, 11, 2, 0, 0, 0)) info.compress_type = zipfile.ZIP_DEFLATED info.external_attr = 0o100644 << 16 archive.writestr(info, data) with zipfile.ZipFile(archive_path) as archive: require(archive.testzip() is None, 'Corrupt output ZIP') require(set(archive.namelist()) == set(files), 'Unexpected output member') entries = manifest_entries(archive.read('FILES.sha256')) require(len(entries) == 8, 'Unexpected handoff document count') for name, sha in entries.items(): require(digest(archive.read(name)) == sha, 'Handoff manifest mismatch') receipt = { 'file': archive_path.name, 'bytes': archive_path.stat().st_size, 'sha256': digest(archive_path.read_bytes()), 'archiveIntegrity': True, 'manifestVerified': '8/8', 'zipEntries': 9, 'originalPackageManifestsVerified': ['21/21', '34/34', '14/14'], 'skillManifestVerified': '10/10', 'contents': 'handoff-documents-only', 'includesInstallationArchives': False, 'includesCredentials': False, 'newAgentCreated': False, 'newAgentRegistered': False, 'newTargetChanged': False, } (DOCS / 'ksf-new-workshop-agent-handoff-20261102.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)) if __name__ == '__main__': main()