Bind idle resident handoffs to fresh session and conditioning identity

roadscore/prototype/resident_session.py

@@ -0,0 +1,80 @@
+"""Idle-only resident handoff bound to an exact fresh session and conditioning bank."""
+from contextlib import contextmanager
+import fcntl
+import json
+import math
+from pathlib import Path
+import re
+import time
+import uuid
+
+
+@contextmanager
+def session_lease(generated):
+    with (Path(generated) / 'session.lock').open('a') as handle:
+        fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        try:
+            yield
+        finally:
+            fcntl.flock(handle, fcntl.LOCK_UN)
+
+
+def session_active(generated):
+    try:
+        with session_lease(generated):
+            return False
+    except BlockingIOError:
+        return True
+
+
+def validate_session(request):
+    profile = request.get('profile')
+    seed = request.get('generation_seed')
+    policy = request.get('composition_policy')
+    bank = request.get('bank_sha256')
+    if profile not in ('prism', 'aurora'):
+        raise ValueError('Unsupported resident profile')
+    if type(seed) is not int or not 0 <= seed < 2**32:
+        raise ValueError('Fresh session seed must be an unsigned 32-bit integer')
+    if policy != 'hook-cache-v1':
+        raise ValueError('Resident handoff requires hook-cache-v1')
+    if not isinstance(bank, str) or re.fullmatch('[0-9a-f]{64}', bank) is None:
+        raise ValueError('Resident handoff requires the conditioning bank SHA-256')
+    return profile, seed, policy, bank
+
+
+def request_preparation(generated, profile, seed, composition_policy, bank_sha256, timeout=1560):
+    generated = Path(generated)
+    selection = {'profile': profile, 'generation_seed': seed,
+                 'composition_policy': composition_policy, 'bank_sha256': bank_sha256}
+    validate_session(selection)
+    if not math.isfinite(timeout) or timeout <= 0:
+        raise ValueError('Preparation timeout must be finite and positive')
+    with (generated / 'session_command.lock').open('a') as handle:
+        fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        request = generated / 'session_request.json'
+        if request.exists():
+            raise RuntimeError('An unacknowledged preparation exists; inspect it before retrying')
+        if session_active(generated):
+            raise RuntimeError('Playback or preparation owns the resident session')
+        token = uuid.uuid4().hex
+        payload = {'id': token, **selection}
+        temporary = request.with_suffix('.tmp')
+        temporary.write_text(json.dumps(payload))
+        temporary.replace(request)
+        deadline = time.monotonic() + timeout
+        while time.monotonic() < deadline:
+            try:
+                result = json.loads((generated / 'session_result.json').read_text())
+            except (OSError, ValueError):
+                result = {}
+            if isinstance(result, dict) and result.get('id') == token:
+                if result.get('error'):
+                    raise RuntimeError(str(result['error']))
+                if validate_session(result) != validate_session(selection):
+                    raise RuntimeError('Resident acknowledgment belongs to a different session')
+                if result.get('phase') != 'READY' or not result.get('preparation_id'):
+                    raise RuntimeError('Resident preparation did not report READY with provenance')
+                return result
+            time.sleep(min(.2, max(0, deadline - time.monotonic())))
+        raise TimeoutError('Preparation still pending; do not silently retry or restart')

roadscore/prototype/test_resident_session.py

@@ -0,0 +1,62 @@
+import json
+import fcntl
+from pathlib import Path
+import tempfile
+import threading
+import time
+import unittest
+from resident_session import request_preparation, session_active, session_lease, validate_session
+
+SELECTION = dict(profile='prism', generation_seed=123, composition_policy='hook-cache-v1', bank_sha256='a'*64)
+
+class ResidentTests(unittest.TestCase):
+ def setUp(self):
+  self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name)
+ def tearDown(self):self.tmp.cleanup()
+ def request(self, timeout=.5):
+  return request_preparation(self.root,'prism',123,'hook-cache-v1','a'*64,timeout)
+ def responder(self, change=None):
+  def run():
+   deadline=time.monotonic()+1
+   path=self.root/'session_request.json'
+   while not path.exists() and time.monotonic()<deadline:time.sleep(.005)
+   if not path.exists():return
+   payload=json.loads(path.read_text());payload.update(phase='READY',preparation_id='fresh-123')
+   if change:payload.update(change)
+   (self.root/'session_result.json').write_text(json.dumps(payload))
+  t=threading.Thread(target=run);t.start();self.addCleanup(t.join)
+ def test_identity_validation(self):
+  self.assertEqual(validate_session(SELECTION),('prism',123,'hook-cache-v1','a'*64))
+  for field,value in [('generation_seed',None),('generation_seed',True),('generation_seed',2**32),('profile','unknown'),('composition_policy','prepared-v1'),('bank_sha256','abc')]:
+   with self.subTest(field=field,value=value),self.assertRaises(ValueError):validate_session({**SELECTION,field:value})
+ def test_active_playback_blocks_requests(self):
+  with session_lease(self.root):
+   self.assertTrue(session_active(self.root))
+   with self.assertRaisesRegex(RuntimeError,'owns'):self.request()
+   self.assertFalse((self.root/'session_request.json').exists())
+  self.assertFalse(session_active(self.root))
+ def test_exact_ack_ignores_stale_ready(self):
+  (self.root/'session_result.json').write_text(json.dumps({'id':'old','phase':'READY'}))
+  self.responder();result=self.request()
+  self.assertEqual(result['generation_seed'],123);self.assertNotEqual(result['id'],'old')
+ def test_wrong_identity_ack_fails(self):
+  self.responder({'bank_sha256':'b'*64})
+  with self.assertRaisesRegex(RuntimeError,'different session'):self.request()
+ def test_serialized_commands(self):
+  with (self.root/'session_command.lock').open('a') as handle:
+   fcntl.flock(handle,fcntl.LOCK_EX|fcntl.LOCK_NB)
+   with self.assertRaises(BlockingIOError):self.request()
+   self.assertFalse((self.root/'session_request.json').exists())
+ def test_ready_without_preparation_provenance_fails(self):
+  self.responder({'preparation_id':None})
+  with self.assertRaisesRegex(RuntimeError,'provenance'):self.request()
+ def test_worker_failure_propagates(self):
+  self.responder({'error':'quality rejected'})
+  with self.assertRaisesRegex(RuntimeError,'quality rejected'):self.request()
+ def test_timeout_preserves_request_and_refuses_retry(self):
+  with self.assertRaises(TimeoutError):self.request(.02)
+  original=(self.root/'session_request.json').read_bytes()
+  with self.assertRaisesRegex(RuntimeError,'unacknowledged'):self.request()
+  self.assertEqual(original,(self.root/'session_request.json').read_bytes())
+
+if __name__=='__main__':unittest.main()