← back to Dw Gemini Skip TK11321

scripts/deploy-owner-skip.py

70 lines

"""TK-11321 reviewed-file deployment with original-byte checks and owned lock."""
from pathlib import Path
import datetime
import hashlib
import json
import os
import subprocess
import tempfile

source = Path('/Users/macstudio3/Projects/dw-gemini-skip-TK11321')
target = Path('/Users/macstudio3/Projects/dw-rotation-activator')
evidence = source / 'verification/tk11321'
files = ['rotate-activate.js', 'lib/owner-review-policy.js', 'lib/hourly-activation-count.js',
         'config/gemini-review-policy.json', 'test/owner-review-policy.test.js', 'test/hourly-activation-count.test.js']
manifest_path = evidence / 'deployment-manifest.json'
lock = target / '.drain.lock'


def git_blob(repo, revision, name):
    r = subprocess.run(['git', '-C', str(repo), 'show', f'{revision}:{name}'], capture_output=True)
    return r.stdout if r.returncode == 0 else None


def main():
    assert not manifest_path.exists(), 'Inspect existing deployment before retry'
    prepared = []
    for name in files:
        before = git_blob(target, 'f9e5125', name)
        current = (target/name).read_bytes() if (target/name).exists() else None
        assert current == before, f'Production file drift: {name}'
        after = git_blob(source, '3d2ba05', name)
        assert after is not None and (source/name).read_bytes() == after, f'Reviewed file drift: {name}'
        prepared.append((name, before, after))
    # Rehearse exact source restoration in a disposable directory before deploy.
    with tempfile.TemporaryDirectory(prefix='tk11321-rollback-') as directory:
        scratch = Path(directory)
        for name, before, after in prepared:
            p = scratch/name; p.parent.mkdir(parents=True, exist_ok=True); p.write_bytes(after)
            if before is None: p.unlink()
            else: p.write_bytes(before)
            assert (p.read_bytes() if p.exists() else None) == before
            p.write_bytes(after); assert p.read_bytes() == after
    (evidence/'rollback-rehearsal.json').write_text(json.dumps({'verdict':'PASS', 'files':len(prepared),
        'method':'Disposable-copy restoration of exact preimages followed by exact reviewed-byte reapplication', 'production_mutations':0},indent=2)+'\n')
    lock.mkdir()
    (lock/'TK-11321-owner.json').write_text(json.dumps({'ticket':'TK-11321','purpose':'approved Gemini-skip rollout and canary',
        'created_at':datetime.datetime.now(datetime.timezone.utc).isoformat()}))
    manifest = {'ticket':'TK-11321','authorization':'Steve: skip gemeni and go', 'status':'locked', 'lock':str(lock),
        'started_at':datetime.datetime.now(datetime.timezone.utc).isoformat(), 'files':[]}
    for name, before, after in prepared:
        backup = evidence/'before'/name
        if before is not None:
            backup.parent.mkdir(parents=True,exist_ok=True); backup.write_bytes(before)
        manifest['files'].append({'relative':name,'target':str(target/name),'backup':str(backup) if before is not None else None,
            'before_sha256':hashlib.sha256(before).hexdigest() if before is not None else None,'after_sha256':hashlib.sha256(after).hexdigest()})
    manifest_path.write_text(json.dumps(manifest,indent=2)+'\n')
    for name, before, after in prepared:
        destination = target/name
        assert (destination.read_bytes() if destination.exists() else None) == before, f'Concurrent change: {name}'
        destination.parent.mkdir(parents=True,exist_ok=True)
        temporary = destination.with_name(destination.name+'.TK11321-stage')
        temporary.write_bytes(after); temporary.chmod((source/name).stat().st_mode & 0o777)
        os.replace(temporary,destination)
    manifest['status']='installed-awaiting-live-verification'
    manifest_path.write_text(json.dumps(manifest,indent=2)+'\n')
    print(json.dumps({'status':manifest['status'],'installed_files':len(prepared),'rollback':'PASS','lock_retained':str(lock)}))


if __name__ == '__main__': main()