Spaces:
Running
Running
File size: 6,914 Bytes
89e57b7 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | """Disposable maintenance qualification; rebuild externally between create/restore.
Uses SYSTEM_PAT in memory. State contains only resource IDs, markers and SHA.
Never point cleanup at an unowned/preexisting namespace.
"""
import argparse
import json
import os
import time
import subprocess
from pathlib import Path
from uuid import uuid4
import requests
def qualify(action, path, backup_checkout=None):
base=os.getenv('QUALIFY_API_URL','http://127.0.0.1:7860').rstrip('/')
session=requests.Session()
if base.startswith(('http://127.0.0.1:', 'http://localhost:')):
session.trust_env=False
session.headers['BrainPAT']=os.environ['SYSTEM_PAT']
def call(method,route,body=None,expected=200):
r=session.request(method,base+route,json=body,timeout=180)
if r.status_code!=expected:
raise RuntimeError(f'Qualification HTTP {r.status_code} at {route}; response suppressed')
return r.json()
def save(state):
path.write_text(json.dumps(state,indent=2)+'\n')
def retrieve(state):
for route in ('search','context'):
result=call('POST',f"/brains/{state['slug']}/agent/{route}",{'query':state['marker']})
assert state['marker'] in json.dumps(result),'Qualification fact absent from '+route
def policy(health):
storage=health['storage'];assert storage['zero_loss'] is False
assert isinstance(storage['rpo_seconds'],int) and storage['rpo_seconds']>0
if not storage['checkpointed_production_accepted'] and storage['local']=='ephemeral':
assert storage['production_ready'] is False
if action=='create':
assert not path.exists(),'State already exists; refusing a second namespace'
suffix=uuid4().hex[:12];slug='maintenance-qa-'+suffix
before=call('GET','/system/workspaces')['workspaces']
assert not any(w['slug']==slug for w in before)
workspace=call('POST','/system/workspaces',{'slug':slug,'display_name':'Disposable maintenance qualification'},201)
state={'slug':slug,'workspace_id':workspace['id'],'brain_id':workspace['brain_id'],
'marker':'MAINTENANCE_FACT_'+uuid4().hex,'baseline_slugs':[w['slug'] for w in before]}
brain=next(b for b in call('GET','/system/brains-list') if b['name_key']==state['brain_id'])
state['canonical_brain_id']=brain['id'];save(state)
accepted=call('POST',f'/brains/{slug}/agent/memory',{'text':'The maintenance qualification compass is copper. '+state['marker']},202)
state['task_id']=accepted['task_id'];assert isinstance(state['task_id'],str) and state['task_id']
save(state)
# Request barrier immediately, before polling: accepted work must drain.
result=call('POST','/system/backup/checkpoint?timeout_seconds=120')
assert result['state']=='durable' and result['remote_verified'] and result['dirty'] is False
state['checkpoint_a']=result['checkpoint'];save(state)
task=call('GET',f"/brains/{slug}/agent/tasks/{state['task_id']}")
assert str(task['status']).lower() in {'completed','success'}
retrieve(state)
rejected=call('POST',f'/brains/{slug}/agent/memory',{'text':'Must be rejected during maintenance'},503)
assert not rejected.get('task_id')
control=call('GET','/system/backup/status');assert control['quiesced'] and control['pending_tasks']==0
health=call('GET','/health');policy(health)
state['pre_rebuild_startup_id']=health['backup']['startup_id'];save(state)
return {'checkpoint':state['checkpoint_a'],'accepted_work_drained':True,'retrieval':'passed','new_writes':'blocked'}
state=json.loads(path.read_text());slug=state['slug']
workspace=call('GET','/system/workspaces/'+slug);assert workspace['id']==state['workspace_id']
if action=='restored':
health=call('GET','/health');policy(health)
assert health['backup']['startup_id']!=state['pre_rebuild_startup_id']
restored=health['restored_checkpoint']
if restored!=state['checkpoint_a']:
assert backup_checkout is not None,'Descendant requires an operator-fetched checkout for ancestry proof'
head=subprocess.check_output(['git','-C',str(backup_checkout),'rev-parse','HEAD'],text=True).strip()
assert head==restored
subprocess.run(['git','-C',str(backup_checkout),'merge-base','--is-ancestor',state['checkpoint_a'],restored],check=True,capture_output=True)
state['verified_descendant']=restored;save(state)
assert health['remote_checkpoint']==restored and health['checkpoint_lag']==0
assert health['backup']['restored'] is True
assert call('GET','/system/backup/status')['quiesced']
brain=next(b for b in call('GET','/system/brains-list') if b['name_key']==state['brain_id'])
assert brain['id']==state['canonical_brain_id']
retrieve(state)
assert call('POST',f'/brains/{slug}/agent/memory',{'text':'Rejected until explicit resume'},503)
state['restoration_passed']=True;save(state)
resumed=call('POST','/system/backup/resume');assert resumed['quiesced'] is False
call('PATCH','/system/workspaces/'+slug,{'display_name':'Disposable qualification resumed'})
state['resume_checkpoint']=resumed['checkpoint'];save(state)
return {'restored_checkpoint':health['restored_checkpoint'],'remote_checkpoint':health['remote_checkpoint'],
'first_replacement_retrieval':'passed','identities':'preserved','explicit_resume':'passed','storage':health['storage'],'checks':health['checks']}
if action=='cleanup':
assert state.get('restoration_passed'),'Cleanup only after restore success'
call('DELETE','/system/brains/'+state['brain_id'])
call('PATCH','/system/workspaces/'+slug,{'archived':True})
result=call('POST','/system/backup/checkpoint');assert result['remote_verified']
state['deletion_checkpoint']=result['checkpoint'];save(state)
assert not any(b['name_key']==state['brain_id'] for b in call('GET','/system/brains-list'))
released=call('POST','/system/backup/resume');assert released['quiesced'] is False
state['cleanup_release_checkpoint']=released['checkpoint'];save(state)
return {'deleted_only':slug,'deletion_checkpoint':result['checkpoint'],'final_checkpoint':released['checkpoint'],'writes_resumed':True}
raise RuntimeError('Unknown qualification action')
if __name__=='__main__':
parser=argparse.ArgumentParser();parser.add_argument('action',choices=['create','restored','cleanup']);parser.add_argument('--state',type=Path,required=True)
parser.add_argument('--backup-checkout',type=Path,help='Fetched main checkout for verifying a descendant of checkpoint A')
args=parser.parse_args();print(json.dumps(qualify(args.action,args.state,args.backup_checkout)))
|