Download Uebergabe/verify_operator.py from smlflg/MetaMetaMeta: direct link, hf CLI and curl.
- Browser
- Download file 2.53 kB
-
https://huggingface.co/smlflg/MetaMetaMeta/resolve/main/Uebergabe/verify_operator.py
- Command line
-
hf download hf://smlflg/MetaMetaMeta/Uebergabe/verify_operator.py
-
curl -L -o verify_operator.py https://huggingface.co/smlflg/MetaMetaMeta/resolve/main/Uebergabe/verify_operator.py
2.53 kB
| """Read-only board verification of the observed Operator trial.""" | |
| import hashlib | |
| import json | |
| from pathlib import Path | |
| import sqlite3 | |
| def main(): | |
| here=Path(__file__).resolve().parent | |
| path='/home/smlflg/.hermes/kanban/boards/agenten-uebergaben/kanban.db' | |
| checks={};tasks={} | |
| with sqlite3.connect('file:'+path+'?mode=ro',uri=True) as c: | |
| c.row_factory=sqlite3.Row | |
| for tid,profile,state in [('t_599ec1c2','wggesucht','blocked'),('t_e52c3df1','verifier','review')]: | |
| t=dict(c.execute('SELECT id,status,assignee FROM tasks WHERE id=?',(tid,)).fetchone());tasks[tid]=t | |
| event=json.loads(c.execute("SELECT payload FROM task_events WHERE task_id=? AND kind='operator_routed' ORDER BY id DESC LIMIT 1",(tid,)).fetchone()[0]) | |
| checks[tid+'_assigned']=t['assignee']=='hai:'+profile | |
| checks[tid+'_state']=t['status']==state | |
| checks[tid+'_operator_selected']=event['profile']==profile and bool(event['operator_pid']) | |
| checks[tid+'_operator_comment']=bool(c.execute("SELECT 1 FROM task_comments WHERE task_id=? AND author='operator'",(tid,)).fetchone()) | |
| checks['result_followed']=bool(c.execute("SELECT 1 FROM task_comments WHERE task_id='t_e52c3df1' AND author='operator' AND body LIKE '%meldet review%'").fetchone()) | |
| checks['no_notifications']=c.execute('SELECT count(*) FROM kanban_notify_subs').fetchone()[0]==0 | |
| job=json.loads((here/'evidence/t_e52c3df1/11/job.json').read_text()) | |
| result=json.loads((here/'evidence/t_e52c3df1/11/result.json').read_text()) | |
| checks['existing_verifier_profile']=job['grant']['worker_profile']=='verifier' | |
| checks['source_read']=any(a['operation']=='read' and a['ok'] for a in result['actions']) | |
| checks['actual_result']=result['ok'] and '139' in result['answer'] | |
| report={'passed':all(checks.values()),'checks':checks,'tasks':tasks,'worker_lease':job['lease'], | |
| 'limits':['Bounded current mission; no permanent service or global profile changes.', | |
| 'WG-Gesucht is assigned but not solved: its execution grant is absent.', | |
| 'External reviewer acceptance is separate from builder checks.'], | |
| 'code_sha256':{p.name:hashlib.sha256(p.read_bytes()).hexdigest() for p in here.glob('*.py')}} | |
| (here/'evidence/OPERATOR_VERIFICATION.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)) | |
| print(json.dumps({'passed':report['passed'],'checks':len(checks),'tasks':tasks})) | |
| assert report['passed'] | |
| if __name__=='__main__':main() | |