baram2584/MagnetismHeroes
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Integrate immutable voice input snapshots and history validation

Browse Source
This commit is contained in:
NINI
2026-10-05 02:24:47 +09:00
parent b2eded808e 8afff106da
commit 7b3eb62925
15 changed files with 2616 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/audio/generate_voice.py
+47 -1
View File
@@ -92,6 +92,31 @@ def waveform_metadata(path):
(['clipping candidate'] if peak >= .999 else [])}
def load_run_inputs(manifest, root):
run_id = manifest.get('runId', '')
if not isinstance(run_id, str) or not re.fullmatch(r'[A-Za-z0-9_]+', run_id):
raise ValueError('invalid input snapshot runId')
references = manifest.get('inputSnapshots', {'catalog': 'catalog.json', 'profile': 'profile.json'})
if references != {'catalog': 'catalog.json', 'profile': 'profile.json'}:
raise ValueError('input snapshots must reside in their immutable run directory')
run_dir = Path(root) / 'generated' / run_id
catalog = load_json(run_dir / references['catalog'])
profile = load_json(run_dir / references['profile'])
if not isinstance(catalog, dict) or not isinstance(profile, dict):
raise ValueError('input snapshots must contain JSON objects')
if digest(catalog) != manifest.get('catalogHash') or digest(profile) != manifest.get('profileHash'):
raise ValueError('input snapshot hash differs from manifest')
validate_inputs(catalog, profile)
if profile['modelRevision'] != manifest.get('modelRevision'):
raise ValueError('input snapshot model revision differs from manifest')
clips = {clip['id']: clip for clip in catalog['clips']}
for entry in manifest.get('clips', []):
clip = clips.get(entry.get('id'))
if not clip or entry.get('textHash') != digest(clip['text']) or entry.get('settingsHash') != digest(profile):
raise ValueError(f"{entry.get('id')}: input snapshot differs from clip text/settings hashes")
return catalog, profile
def clip_errors(entry, clip, profile, root):
errors = []
if entry.get('result') not in ('generated', 'reused'):
@@ -108,8 +133,21 @@ def clip_errors(entry, clip, profile, root):
errors.append(f"{clip['id']}: {key} mismatch")
if actual['channels'] != 1:
errors.append(f"{clip['id']}: expected mono")
run_id = entry['originalRunId']
if not isinstance(run_id, str) or not re.fullmatch(r'[A-Za-z0-9_]+', run_id):
raise ValueError('invalid original generation runId')
origin = load_json(root / 'generated' / run_id / 'manifest.json')
if origin.get('runId') != run_id:
raise ValueError('original manifest runId mismatch')
load_run_inputs(origin, root)
original_entries = [c for c in origin['clips'] if c.get('id') == clip['id']]
if len(original_entries) != 1 or original_entries[0].get('result') != 'generated':
raise ValueError('missing successful original generation entry')
for key in ('file', 'sha256', 'textHash', 'settingsHash'):
if original_entries[0].get(key) != entry.get(key):
raise ValueError(f'original generation {key} differs')
except (KeyError, OSError, ValueError, sf.LibsndfileError) as error:
errors.append(f"{clip['id']}: invalid WAV: {error}")
errors.append(f"{clip['id']}: invalid source artifact: {error}")
return errors
@@ -117,6 +155,10 @@ def validate_manifest(manifest, catalog, profile, root, ids=None):
validate_inputs(catalog, profile)
requested = select_clips(catalog, ids, required_only=ids is None)
errors = []
try:
load_run_inputs(manifest, root)
except (KeyError, OSError, ValueError, TypeError) as error:
errors.append(f'input snapshots invalid: {error}')
if manifest.get('schemaVersion') != 1 or manifest.get('modelRevision') != profile['modelRevision']:
errors.append('manifest schema/revision differs')
if manifest.get('catalogHash') != digest(catalog) or manifest.get('profileHash') != digest(profile):
@@ -139,6 +181,7 @@ def reusable_entries(root, catalog, profile):
for path in sorted((root / 'generated').glob('*/manifest.json'), reverse=True):
try:
manifest = load_json(path)
load_run_inputs(manifest, root)
for entry in manifest.get('clips', []):
clip = by_id.get(entry.get('id'))
if clip and clip['id'] not in result and not clip_errors(entry, clip, profile, root):
@@ -177,10 +220,13 @@ def generate(catalog, profile, root, ids=None, synthesize=None, environment=None
run_id = datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S%fZ_') + uuid.uuid4().hex[:8]
run_dir = root / 'generated' / run_id
run_dir.mkdir(parents=True)
write_json(run_dir / 'catalog.json', catalog)
write_json(run_dir / 'profile.json', profile)
manifest = {'schemaVersion': 1, 'runId': run_id, 'createdUtc': datetime.now(timezone.utc).isoformat(),
'modelRevision': profile['modelRevision'], 'catalogHash': digest(catalog),
'profileHash': digest(profile), 'scope': 'selected' if ids else 'all',
'regenerateRequested': regenerate,
'inputSnapshots': {'catalog': 'catalog.json', 'profile': 'profile.json'},
'requestedIds': [c['id'] for c in selected], 'environment': environment or {}, 'clips': []}
try:
for clip in selected:
tools/audio/test_generate_voice.py
+71
View File
@@ -152,6 +152,77 @@ class VoiceHistoryTests(unittest.TestCase):
self.assertEqual(voice.file_hash(self.root / source['manifest']), source['manifestSha256'])
self.assertEqual({'test': True}, source['environment'])
def test_run_input_snapshots_preserve_text_and_settings_after_shared_files_change(self):
old_catalog, old_profile = copy.deepcopy(self.catalog), copy.deepcopy(self.profile)
first = self.run_generation()
run_dir = self.root / 'generated' / first['runId']
original_wav = (self.root / first['clips'][0]['file']).read_bytes()
self.catalog['clips'][0]['text'] = '하나예요.'
self.profile.update(seed=43, instruct='밝게 읽어 주세요.', dtype='float16', attention='eager')
voice.write_json(self.root / 'voice-catalog.json', self.catalog)
voice.write_json(self.root / 'qwen-profile.json', self.profile)
self.run_generation()
self.assertEqual(old_catalog, voice.load_json(run_dir / 'catalog.json'))
self.assertEqual(old_profile, voice.load_json(run_dir / 'profile.json'))
self.assertEqual(first['catalogHash'], voice.digest(voice.load_json(run_dir / 'catalog.json')))
self.assertEqual(first['profileHash'], voice.digest(voice.load_json(run_dir / 'profile.json')))
self.assertEqual(original_wav, (self.root / first['clips'][0]['file']).read_bytes())
self.assertTrue(voice.validate_manifest(first, self.catalog, self.profile, self.root))
def test_run_input_snapshots_missing_corrupt_or_hash_mismatch_fail_validation(self):
manifest = self.run_generation()
run_dir = self.root / 'generated' / manifest['runId']
for name, data in [('catalog.json', self.catalog), ('profile.json', self.profile)]:
path = run_dir / name
for contents in (None, 'not json', '{}'):
if contents is None:
path.unlink(missing_ok=True)
else:
path.write_text(contents, encoding='utf-8')
self.assertTrue(voice.validate_manifest(manifest, self.catalog, self.profile, self.root))
voice.write_json(path, data)
changed = copy.deepcopy(manifest)
changed['profileHash'] = 'f' * 64
self.assertTrue(voice.validate_manifest(changed, self.catalog, self.profile, self.root))
def test_reused_run_rejects_missing_original_input_snapshot(self):
first = self.run_generation()
second = self.run_generation()
self.assertEqual('reused', second['clips'][0]['result'])
(self.root / 'generated' / first['runId'] / 'profile.json').unlink(missing_ok=True)
self.assertTrue(voice.validate_manifest(second, self.catalog, self.profile, self.root))
def test_snapshot_validation_cli_does_not_load_model_or_use_network(self):
import sys
manifest = self.run_generation()
catalog = self.root / 'voice-catalog.json'
profile = self.root / 'qwen-profile.json'
voice.write_json(catalog, self.catalog)
voice.write_json(profile, self.profile)
args = ['generate_voice.py', '--catalog', str(catalog), '--profile', str(profile),
'--validate-only', '--manifest', str(self.root / 'generated' / manifest['runId'] / 'manifest.json')]
with mock.patch.object(sys, 'argv', args), \
mock.patch.object(voice, 'load_synthesizer', side_effect=AssertionError('model loaded')) as loader, \
mock.patch.object(voice.subprocess, 'run', side_effect=AssertionError('device command ran')), \
mock.patch.dict(sys.modules, {'torch': None, 'qwen_tts': None, 'huggingface_hub': None}):
self.assertEqual(0, voice.main())
snapshot = self.root / 'generated' / manifest['runId'] / 'catalog.json'
for contents in (None, 'not json', '{}'):
if contents is None:
snapshot.unlink(missing_ok=True)
else:
snapshot.write_text(contents, encoding='utf-8')
self.assertEqual(1, voice.main())
loader.assert_not_called()
def test_legacy_manifest_uses_verified_sidecar_without_manifest_rewrite(self):
manifest = self.run_generation()
legacy = copy.deepcopy(manifest)
legacy.pop('inputSnapshots')
saved_manifest = (self.root / 'generated' / manifest['runId'] / 'manifest.json').read_bytes()
self.assertEqual([], voice.validate_manifest(legacy, self.catalog, self.profile, self.root))
self.assertEqual(saved_manifest, (self.root / 'generated' / manifest['runId'] / 'manifest.json').read_bytes())
if __name__ == '__main__':
unittest.main()