ape2e/tools/promote_instructors_v2.py

"""Promote the three validated v2 bundles without replacing platform identities."""
import importlib.util
import json
import shlex
import shutil
import subprocess
from pathlib import Path

import paramiko
from retrain_instructors_v2 import ROOT, REPORT, BUNDLES, IDS, FILES, api_session, API, HTTP, SERVICE, save

RELEASE = '/home/digital-human-service/releases/instructors-v2-20260914'
HOST_SOURCE = '/home/ai_person/ai_person_service/builtin_avatars'
CONTAINER = 'digital-human-service'


def connect():
    spec = importlib.util.spec_from_file_location('connection', ROOT / 'ai_person/ai_person_api/.run/test_file_service_tunnel.py')
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    host, port, user, password = module._read_connection()
    client = paramiko.SSHClient()
    client.load_system_host_keys()
    client.set_missing_host_key_policy(paramiko.RejectPolicy())
    client.connect(host, port=port, username=user, password=password, look_for_keys=False, allow_agent=False)
    return client


def run(client, command):
    _, stdout, stderr = client.exec_command(command, timeout=300)
    output, error = stdout.read().decode('utf-8', 'replace'), stderr.read().decode('utf-8', 'replace')
    if stdout.channel.recv_exit_status() != 0:
        raise RuntimeError(output + error)
    return output.strip()


def remote_promote(client, jobs, checks):
    # Use the service's own metadata operation. No SQL is issued against the platform database.
    script = r'''
import hashlib,json,shutil
from pathlib import Path
from app.config import settings
from app import database
rows=PAYLOAD
backup=settings.data_root/'backups'/'instructors-v2-20260914'
if backup.exists():raise RuntimeError('Backup exists; refuse a second promotion')
source=settings.project_root/'builtin_avatars'
catalog={x['id']:x for x in json.loads((source/'catalog.json').read_text())}
for row in rows:
    assert row['stable_id'] in catalog
    assets=settings.avatars_root/row['avatar_id']/'assets'
    assert database.get_avatar(row['avatar_id'])['status']=='ready'
    for name,digest in row['sha256'].items():assert hashlib.sha256((assets/name).read_bytes()).hexdigest()==digest
backup.mkdir(parents=True)
before=[]
for row in rows:
    ident=row['stable_id']
    before.append(database.get_avatar(ident))
    shutil.copytree(source/ident,backup/'builtin'/ident)
    shutil.copytree(settings.avatars_root/ident,backup/'runtime'/ident)
(backup/'before-metadata.json').write_text(json.dumps(before,ensure_ascii=False,indent=2))
for row in rows:
    ident=row['stable_id'];assets=settings.avatars_root/row['avatar_id']/'assets'
    for root in (source,settings.avatars_root):
        stage=root/(ident+'.v2-stage')
        if stage.exists():raise RuntimeError('Unexpected stale staging directory')
        shutil.copytree(root/ident,stage)
        for name in row['sha256']:shutil.copy2(assets/name,stage/'assets'/name)
        previous=root/(ident+'.v1-previous')
        if previous.exists():raise RuntimeError('Unexpected previous directory')
        (root/ident).rename(previous)
        try:stage.rename(root/ident)
        except BaseException:previous.rename(root/ident);raise
        # Keep the renamed directories until all replacement checks finish.
    manifest=json.loads((assets/'manifest.json').read_text())
    spec=catalog[ident]
    database.upsert_builtin_avatar(avatar_id=ident,name=spec['name'],source_filename='built_in:'+spec['code'],
        frame_count=manifest['frame_count'],width=manifest['width'],height=manifest['height'],duration=manifest['duration'])
print(json.dumps({'backup':str(backup),'updated':[row['stable_id'] for row in rows]}))
'''.replace('PAYLOAD', repr([{**job, 'sha256': check['sha256']} for job, check in zip(jobs, checks)]))
    return run(client, 'docker exec -w /app ' + CONTAINER + ' python -c ' + shlex.quote(script))


def main():
    if (REPORT / 'promotion.json').exists():
        raise RuntimeError('Already promoted; do not run twice')
    jobs = json.loads((REPORT / 'jobs.json').read_text('utf-8'))
    checks = json.loads((REPORT / 'generated-validation.json').read_text('utf-8'))
    playback = json.loads((REPORT / 'generated-viewer-validation.json').read_text('utf-8'))
    assert len(jobs) == len(checks) == len(playback) == 3
    assert all(not row['errors'] and row['driver']['active'] > 30 for row in playback)
    client = connect()
    try:
        base_image = run(client, "docker inspect digital-human-service --format '{{.Image}}'")
        run(client, f'mkdir -p {RELEASE}/backup-host {RELEASE}/builtin_avatars')
        for ident in IDS:
            run(client, f'test -d {HOST_SOURCE}/{ident} && cp -a {HOST_SOURCE}/{ident} {RELEASE}/backup-host/{ident}')
        save('promotion-start.json', {'baseImage': base_image, 'remoteRelease': RELEASE})
        result = json.loads(remote_promote(client, jobs, checks))
        save('promotion-runtime.json', result)
        for ident in IDS:
            run(client, f'docker cp {CONTAINER}:/app/builtin_avatars/{ident} {RELEASE}/builtin_avatars/{ident}')
            for filename in FILES:
                run(client, f'cp {RELEASE}/builtin_avatars/{ident}/assets/{filename} {HOST_SOURCE}/{ident}/assets/{filename}')
                shutil.copy2(REPORT / 'generated' / ident / 'assets' / filename, BUNDLES / ident / 'assets' / filename)
        # Persist bundles in the deployment image as well as the running container.
        # The base is the exact current image; no application source is upgraded.
        dockerfile = f'FROM {base_image}\nCOPY builtin_avatars/ /app/builtin_avatars/\nLABEL ai-person.avatar-material-version="instructors-v2-20260914"\n'
        with client.open_sftp() as sftp:
            with sftp.open(RELEASE + '/Dockerfile', 'w') as stream:
                stream.write(dockerfile)
        run(client, f'docker tag {base_image} digital-human-service:before-instructors-v2-20260914')
        output = run(client, f'docker build --network=none -t digital-human-service:instructors-v2-20260914 {RELEASE}')
        (REPORT / 'image-build.log').write_text(output, encoding='utf-8')
        run(client, 'docker tag digital-human-service:instructors-v2-20260914 digital-human-service:cpu')
        image_id = run(client, "docker image inspect digital-human-service:cpu --format '{{.Id}}'")
        save('promotion.json', {**result, 'oldImage': base_image, 'newImage': image_id, 'release': RELEASE})
        print(json.dumps({'updated': IDS, 'newImage': image_id, 'backup': result['backup']}), flush=True)
    finally:
        client.close()


if __name__ == '__main__':
    main()