`verification/fleet-classification/prepare.py`

72 lines
import collections, difflib, hashlib, json, pathlib, subprocess

ROOT = pathlib.Path('/tmp/tk11438-fleet-classification')
PROJECTS = pathlib.Path('/Users/macstudio3/Projects')
EVIDENCE = PROJECTS / 'tk11438-postgres-migration'
def read(name): return json.loads((ROOT / name).read_text())
def save(name, value): (ROOT / name).write_text(json.dumps(value, indent=2) + '\n')
def sha(data): return hashlib.sha256(data).hexdigest()

rows = read('root-classified.json')['rows'] + read('dw-classified.json') + read('norma-classified.json')
expected = {str(PROJECTS / p.removeprefix('./')) for p in (EVIDENCE / 'verification/original-218-paths.txt').read_text().splitlines()}
assert len(rows) == len(expected) == 218
assert len({r['path'] for r in rows}) == 218
assert {r['path'] for r in rows} == expected
checked = 0
for row in rows:
    p = pathlib.Path(row['path'])
    if not p.is_file():
        assert p == PROJECTS and row['disposition'] == 'invalid-directory-entry'
        continue
    data = p.read_bytes()
    assert sha(data) == (row.get('sha256_current') or row.get('sha256')), str(p)
    n = len(data.splitlines())
    for line in row.get('evidence_lines', []) + row.get('matched_line_numbers', []):
        assert isinstance(line, int) and 1 <= line <= n
    checked += 1
rows.sort(key=lambda r: r['path'])
counts = dict(sorted(collections.Counter(r['classification'] for r in rows).items()))
save('original-ledger.json', {'status': 'classified; migration incomplete', 'original_entries': 218,
 'files': 217, 'invalid_directories': 1, 'classification_counts': counts,
 'semantics': 'local-cli is a source candidate, not proof of execution or local deployment; unresolved paths remain open; linked worktrees are not independent production consumers',
 'rows': rows})

modules = read('shared-modules.json')
patchdir = ROOT / 'patches'; patchdir.mkdir(exist_ok=True)
manifest = []
for m in modules['modules']:
    path = pathlib.Path(m['path']); before = path.read_text()
    assert sha(path.read_bytes()) == m['sha256'], str(path)
    assert before.count(modules['proposed_old']) == 1
    after = before.replace(modules['proposed_old'], modules['proposed_new'])
    assert after.replace(modules['proposed_new'], modules['proposed_old']) == before
    check = subprocess.run(['node', '--check'], input=after, text=True, capture_output=True)
    assert check.returncode == 0, str(path)
    patch = ''.join(difflib.unified_diff(before.splitlines(True), after.splitlines(True),
      fromfile='a/lib/vendor-requests.js', tofile='b/lib/vendor-requests.js', n=0))
    name = path.parent.parent.name + '.patch'; (patchdir / name).write_text(patch)
    manifest.append({'project':path.parent.parent.name, 'path':str(path),
      'before_sha256':sha(before.encode()), 'after_sha256':sha(after.encode()),
      'patch':'patches/' + name, 'patch_sha256':sha(patch.encode()),
      'source_changes_applied':False, 'syntax':'PASS', 'replacement_roundtrip':'PASS',
      'current_pm2_names':[r['name'] for r in m['pm2']],
      'requires_before_rollout':['effective/durable PGHOST and DB identity', 'service startup side effects and email exclusion', 'auth and read-only operational baseline', 'file/PM2 backups and scoped rollback']})
save('patch-manifest.json', {'status':'prepared only', 'modules':manifest})

schedulers=read('schedulers.json'); checked_sources={}
for row in schedulers['rows']:
    for source in row.get('referenced_sources',[]):
        if source.get('sha256'):
            p=pathlib.Path(source['source'])
            checked_sources[str(p)] = sha(p.read_bytes()) == source['sha256']
drifted=[p for p,ok in checked_sources.items() if not ok]
save('preparation-proof.json', {'verdict':'PASS original ledger and patch preparation; scheduler snapshot drift unresolved; runtime migration incomplete',
 'original_coverage':218, 'unique_original_paths':218, 'source_hashes_verified':checked,
 'classification_counts':counts, 'candidate_patches':len(manifest),
 'syntax_and_replacement_roundtrips':len(manifest),
 'scheduler_source_hashes_checked':len(checked_sources),
 'scheduler_source_hashes_matching':sum(checked_sources.values()),
 'scheduler_sources_changed_since_scan':drifted,
 'application_mutations':0, 'email_sends':0, 'critical_path_status':'not exercised: no rollout authorized for this prepared batch'})
print(json.dumps({'files_verified':checked,'counts':counts,'patches':len(manifest),'scheduler_sources_checked':len(checked_sources),'scheduler_sources_drifted':len(drifted)}))