File size: 2,528 Bytes
3bc7cb3
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""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()