diff --git a/scripts/native-maintenance/README.md b/scripts/native-maintenance/README.md index deaa6d5..651f992 100644 --- a/scripts/native-maintenance/README.md +++ b/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: diff --git a/scripts/native-maintenance/maintain.py b/scripts/native-maintenance/maintain.py index 23dd4f3..08fce81 100755 --- a/scripts/native-maintenance/maintain.py +++ b/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. diff --git a/scripts/native-maintenance/retention.py b/scripts/native-maintenance/retention.py index 3193d17..a73d884 100644 --- a/scripts/native-maintenance/retention.py +++ b/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'): diff --git a/scripts/native-maintenance/test_maintain.py b/scripts/native-maintenance/test_maintain.py index 5317368..6bcf9af 100644 --- a/scripts/native-maintenance/test_maintain.py +++ b/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(): diff --git a/scripts/native-maintenance/test_retention.py b/scripts/native-maintenance/test_retention.py index c6453c7..344c70b 100644 --- a/scripts/native-maintenance/test_retention.py +++ b/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())