Reject hook continuation leading silence without flattening musical breaks

roadscore/experiments/ace_chestnut_20260916/ace_worker.py

@@ -16,7 +16,7 @@ sys.path.insert(0,str(R/'prototype'))
 from ace_profiles import selected
 from generation_seed import configured_seed,sample_seed
 base_seed=configured_seed(required=True)
-from quality_gate import QualifiedGenerator,POLICY
+from quality_gate import QualifiedGenerator,POLICY,HOOK_POLICY
 from link_health import LinkProbe
 from hook_service import Client
 from planned_composition import PlannedComposition
@@ -52,7 +52,7 @@ try:
    wave,latent,stats=sample(role,seed,previous);stats['power_limit_watts']=float(os.environ['AM_POWER_LIMIT']) if os.environ.get('AM_POWER_LIMIT') else None;stats['link_session']=probe.session;stats['host_peak_rss_kib']=resource.getrusage(resource.RUSAGE_SELF).ru_maxrss;stats['tracked_allocation_bytes']=probe.last.get('allocator_bytes');return wave,latent,stats
   except Exception as e:
    probe.sample('generation_exception',error=str(e),role=role,seed=seed);raise
- qualified=QualifiedGenerator(generate)
+ qualified=QualifiedGenerator(generate,policy=HOOK_POLICY if planned else POLICY)
  def record_for(job):
   folder=G/'quality'/str(job);folder.mkdir(parents=True,exist_ok=True)
   def record(attempt,wave,latent,row):

roadscore/experiments/ace_chestnut_20260916/quality_gate.py

@@ -29,6 +29,19 @@ class QualityPolicy:
  initial_buffer_seconds: float = 112.
 POLICY=QualityPolicy()
 
+@dataclass(frozen=True)
+class HookQualityPolicy(QualityPolicy):
+ version: str = 'prism-hook-precommit-v3'
+ max_leading_quiet_seconds: float = 1.
+HOOK_POLICY=HookQualityPolicy()
+
+def leading_seam_gap(result,policy):
+ limit=getattr(policy,'max_leading_quiet_seconds',None)
+ return (limit is not None and result['role']!='outro' and result['prefix_seconds']>0
+         and result['leading_quiet_seconds']>=limit-1e-6
+         and min(result['source_energy_rms'],result['new_energy_rms_p80'])>4*result['quiet_threshold'])
+
+
 def spans(mask,step):
  result=[];start=None
  for i,value in enumerate(list(mask)+[False]):
@@ -59,6 +72,7 @@ def inspect(wave,rate,prefix_seconds,*,role='verse',policy=POLICY,endpoint_error
  # A quiet interval can straddle a measurement window; reserve one window
  # so a nominal two-second gap does not evade the gate through block alignment.
  result['quiet_span_uncertainty_seconds']=policy.window_seconds
+ if leading_seam_gap(result,policy):reasons.append('leading_continuation_gap')
  if role=='outro':
   if leading>=min(policy.outro_leading_limit_seconds,len(new)/rate*.5):reasons.append('outro_silent_before_final_phrase')
  else:

roadscore/experiments/ace_chestnut_20260916/test_hook_quality.py

@@ -0,0 +1,34 @@
+import unittest
+import numpy as np
+from quality_gate import HOOK_POLICY,POLICY,inspect,leading_seam_gap,QualifiedGenerator,reroll_seed
+
+class HookSeamTests(unittest.TestCase):
+ def tone(self):
+  t=np.arange(36*48000)/48000
+  return np.repeat((.1*np.sin(2*np.pi*220*t))[:,None],2,axis=1)
+ def test_leading_gap_rejected_only_by_hook_gate(self):
+  wave=self.tone();wave[8*48000:round(9.5*48000)]=0
+  self.assertTrue(inspect(wave,48000,8,policy=POLICY)['accepted'])
+  result=inspect(wave,48000,8,policy=HOOK_POLICY)
+  self.assertFalse(result['accepted']);self.assertIn('leading_continuation_gap',result['reasons'])
+  self.assertTrue(inspect(wave,48000,8,role='outro',policy=HOOK_POLICY)['accepted'])
+ def test_intentional_interior_break_and_sparse_opening_preserved(self):
+  wave=self.tone();wave[15*48000:round(16.5*48000)]=0
+  self.assertTrue(inspect(wave,48000,8,policy=HOOK_POLICY)['accepted'])
+  wave=self.tone();wave[8*48000:15*48000]*=.1
+  self.assertTrue(inspect(wave,48000,8,policy=HOOK_POLICY)['accepted'])
+ def test_recorded_user_review_cases(self):
+  first=dict(role='verse',prefix_seconds=8,leading_quiet_seconds=1.5,source_energy_rms=.0480834596,new_energy_rms_p80=.0811132708,quiet_threshold=.0015026081)
+  third=dict(role='verse',prefix_seconds=8,leading_quiet_seconds=0,source_energy_rms=.1008403855,new_energy_rms_p80=.1471485812,quiet_threshold=.003)
+  self.assertTrue(leading_seam_gap(first,HOOK_POLICY));self.assertFalse(leading_seam_gap(first,POLICY));self.assertFalse(leading_seam_gap(third,HOOK_POLICY))
+  for changes in ({'prefix_seconds':0},{'role':'outro'},{'source_energy_rms':.0001},{'leading_quiet_seconds':.9}):
+   self.assertFalse(leading_seam_gap({**first,**changes},HOOK_POLICY))
+ def test_existing_deterministic_bounded_retry_keeps_parent(self):
+  calls=[];prior=np.zeros((1,700,64));wave=self.tone();bad=wave.copy();bad[8*48000:round(9.5*48000)]=0
+  def generate(role,seed,previous):
+   self.assertIs(previous,prior);calls.append((role,seed))
+   return (bad if len(calls)==1 else wave),np.zeros((1,900,64)),dict(prefix_seconds=8,cold=False)
+  _,_,result=QualifiedGenerator(generate,policy=HOOK_POLICY).run('verse',123,prior)
+  self.assertEqual(calls,[('verse',123),('verse',reroll_seed(123,1))]);self.assertTrue(result['quality_accepted']);self.assertEqual(result['rerolls'],1)
+
+if __name__=='__main__':unittest.main()

roadscore/experiments/ace_chestnut_20260916/validate_hook_sessions.py

@@ -13,7 +13,7 @@ sys.path.insert(0, str(Path(__file__).resolve().parents[2] / 'prototype'))
 from generation_seed import sample_seed
 from hook_planning import PlanCache, digest, next_section, request_plan
 from host_hook_adapter import HostHookAdapter
-from quality_gate import inspect
+from quality_gate import inspect, HOOK_POLICY
 from session_seed import select_session
 from window_policy import retained_end
 
@@ -38,8 +38,10 @@ def main():
     for session in sessions:
         select_session(session['generation_seed'])
     report = {'sessions': sessions, 'backend': 'Mac MLX offline; not native runtime equivalence',
+              'quality_policy': HOOK_POLICY.version,
               'seed_selection': 'normal fresh-session selector; no rerolls or seed shopping',
               'musical_acceptance': 'pending listening; energy checks cannot establish hook quality',
+              'quality_artifacts': 'quality/*_committed.wav is raw before gain, includes retained prefix; paired latent/quality records',
               'retry_of': str(args.session_manifest) if args.session_manifest else None,
               'runs': [], 'status': 'starting'}
     if args.resume:
@@ -116,7 +118,12 @@ def main():
             decode_seconds = time.monotonic() - tick
             frames, endpoint = retained_end(wave, 48000, request.prefix_seconds, 28 if index == 0 else 36)
             committed = wave[:frames * 1920]
-            quality = inspect(committed, 48000, request.prefix_seconds, role=role)
+            quality = inspect(committed, 48000, request.prefix_seconds, role=role, policy=HOOK_POLICY)
+            diagnostics = run / 'quality'
+            diagnostics.mkdir(exist_ok=True)
+            sf.write(diagnostics / f'{index:02d}_committed.wav', committed, 48000, subtype='FLOAT')
+            np.save(diagnostics / f'{index:02d}_committed.npy', latent.detach().float().cpu().numpy()[:, :frames])
+            (diagnostics / f'{index:02d}_quality.json').write_text(json.dumps(quality, indent=2))
             new = committed[request.prefix_seconds * 48000:]
             sf.write(run / f'{index:02d}_{role}.wav', new * np.float32(.65), 48000, subtype='FLOAT')
             np.save(run / 'prefix.npy', latent.detach().float().cpu().numpy()[:, frames-200:frames])