fix: keep maintenance recovery moving across policy and release mismatches

scripts/native-maintenance/README.md

@@ -39,6 +39,12 @@ records release before checksum-checked removal of the exclusive temporary
 request/result files. Original completion/native receipts and every archive row
 remain intact. Incomplete proof and changed files stay retained.
 
+Requests with an older consumer-policy binding are ineligible for the current
+zero-scope proof and remain retained without blocking other eligible retirements.
+Every completion release must return the expected released status and request
+digest before the controller discards pending ownership; an invalid response
+keeps its durable release intent and local proof for restart.
+
 Existing receipts lacking the new header have an explicit, single-envelope
 recovery path. Under the existing orchestrator lock, invoke the installed script
 with its usual config/hash plus:

scripts/native-maintenance/maintain.py

@@ -117,7 +117,9 @@ def finish(self,pending):
     pending['request_header']={k:v for k,v in req.items() if k!='tasks'};pending['stage']='release_intent';self.checkpoint()
    # A killed process here resumes from durable local import proof, bypassing
    # broker result. Release is idempotent, including an already released request.
-   released=json.loads(self.call(self.c['broker']+['release',pending['sha256']],b''));pending['release']=released;pending['finished_at']=now();pending['stage']='released';self.checkpoint()
+   released=json.loads(self.call(self.c['broker']+['release',pending['sha256']],b''))
+   if released.get('status')!='released' or released.get('request_sha256')!=pending['sha256']:raise RuntimeError('completion release response binding invalid')
+   pending['release']=released;pending['finished_at']=now();pending['stage']='released';self.checkpoint()
   save(self.base/(pending['sha256']+'-receipt.json'),pending)
   # Keep released completion ownership until retirement intent is durable.
   # Restart re-enters finish without fetching/resubmitting the broker result.

scripts/native-maintenance/retention.py

@@ -328,6 +328,10 @@ def retire_zero_scope(cycle, before, native, zero):
         if not left or left != right:
             continue
         header = right['header']
+        # Old policy requests remain owned, but cannot obstruct compatible
+        # entries or strand replay of the completed scope-retirement operation.
+        if header['source_binding'].get('consumer_exclusions_sha256') != before['policy_sha256']:
+            continue
         if header['source_binding'].get('source_commit') != '6c9c1bea7d094397dcdfecfb50e693460592a5aa':
             continue
         if (header.get('repository'), header.get('model'), header.get('dimensions'), header.get('basis'), header.get('native_hash_version')) != ('openclaw/openclaw', 'text-embedding-3-small', 1536, 'title_original', 'embedding:v5'):

scripts/native-maintenance/test_maintain.py

@@ -39,8 +39,19 @@ def test_crash_exactly_after_release_before_checkpoint(self):
   x.update({'result':'result.json','result_sha256':m.sha(result),'native_receipt':str(native),'native_receipt_sha256':m.sha(native.read_bytes()),'matched':1});c.checkpoint();released=[]
   def die(argv,data=None,timeout=90):released.append(argv);raise KeyboardInterrupt('crash after remote release')
   c.call=die;self.assertRaises(KeyboardInterrupt,c.finish,x);disk=json.loads(c.statepath.read_bytes());self.assertEqual(disk['pending']['stage'],'release_intent')
-  def again(argv,data=None,timeout=90):self.assertEqual(argv[-2],'release');released.append(argv);return b'{"status":"released","removed":0}'
+  def again(argv,data=None,timeout=90):self.assertEqual(argv[-2],'release');released.append(argv);return json.dumps({'status':'released','request_sha256':argv[-1],'removed':0}).encode()
   resumed=m.Cycle(c.c,again);self.assertEqual(resumed.run()['status'],'recovered_release');self.assertEqual(len(released),2);self.assertNotIn('pending',resumed.state)
+ def test_unbound_release_keeps_intent_and_local_proof_for_restart(self):
+  for response in [{'status':'pending'}, {'status':'released','request_sha256':'wrong'}, {}]:
+   with self.subTest(response=response):
+    c,p,calls=self.make();self.pending(c,p);x=c.state['pending'];result=b'{"results":[]}';(p/'result.json').write_bytes(result);native=p/'native.json';m.save(native,{'request_sha256':x['sha256'],'response_sha256':m.sha(result),'imported':1,'unchanged':0,'committed_at':'t'})
+    x.update(result='result.json',result_sha256=m.sha(result),native_receipt=str(native),native_receipt_sha256=m.sha(native.read_bytes()),matched=1);c.checkpoint()
+    c.call=lambda argv,data=None,timeout=90:json.dumps(response).encode()
+    with self.assertRaisesRegex(RuntimeError,'release response binding invalid'):c.finish(x)
+    disk=json.loads(c.statepath.read_bytes());self.assertEqual(disk['pending']['stage'],'release_intent');self.assertTrue((p/'request.json').exists());self.assertTrue((p/'result.json').exists());self.assertFalse((p/(x['sha256']+'-receipt.json')).exists())
+    def released(argv,data=None,timeout=90):
+     self.assertEqual(argv[-2:],[ 'release',x['sha256']]);calls.append(argv);return json.dumps({'status':'released','request_sha256':x['sha256'],'removed':0}).encode()
+    resumed=m.Cycle(c.c,released);self.assertEqual(resumed.run()['status'],'recovered_release');self.assertNotIn('pending',resumed.state);self.assertEqual(len(calls),1)
  def test_new_exclusion_drops_old_bytes_and_exports_clean(self):
   c,p,calls=self.make();self.pending(c,p);old=c.state['pending']['sha256'];policy={'format':'gitcrawl.embedding.exclusions.v1','dataset':'gitcrawl/history/threads','excluded_thread_ids':['1']};m.save(c.policy,policy);c.policy_refresh=lambda:'new-policy';exports=[]
   def new():

scripts/native-maintenance/test_retention.py

@@ -265,6 +265,33 @@ def test_new_due_task_or_policy_change_preserves_envelope(self):
         self.assertEqual(retention.retire_zero_scope(c, before, native, zero)['retired'], [])
         self.assertEqual(calls, [])
 
+    def test_old_policy_entry_does_not_strand_scope_retirement(self):
+        c, entry, native, zero, calls = self.zero_fixture()
+        old = copy.deepcopy(entry)
+        old.update(request='old-policy-request.json', invalidated='consumer_policy_changed')
+        old.pop('result'); old.pop('result_sha256')
+        request = json.loads((c.base / entry['request']).read_bytes())
+        request['source_binding']['consumer_exclusions_sha256'] = 'earlier-same-set-binding'
+        raw = retention.encoded(request)
+        (c.base / old['request']).write_bytes(raw)
+        old['sha256'] = retention.digest(raw)
+        c.state['retained'].insert(0, old)
+        before = retention.capture_scope(c)
+        self.assertIn(old['sha256'], before['entries'])
+        c.state['scope_retirement_pending'] = {'before': before, 'publication': native}
+        c.checkpoint()
+        resumed = maintain.Cycle(c.c, c.call)
+        resumed.zero_export = lambda: zero
+        resumed.finish_scope_retirement()
+        self.assertNotIn('scope_retirement_pending', resumed.state)
+        self.assertEqual(resumed.state['retained'], [old])
+        self.assertEqual((c.base / old['request']).read_bytes(), raw)
+        self.assertEqual([call[-1] for call in calls], [entry['sha256']])
+        again = maintain.Cycle(c.c, c.call)
+        again.finish_scope_retirement()
+        self.assertEqual(again.state['retained'], [old])
+        self.assertEqual(len(calls), 1)
+
     def test_changed_input_cycles_do_not_grow_retained_cap(self):
         c, entry, native, zero, calls = self.zero_fixture()
         original_request = json.loads((c.base / entry['request']).read_bytes())