| """Existing operator profile routes Kanban cards; only owner grants start workers.""" |
| import argparse |
| import fcntl |
| import json |
| import os |
| from pathlib import Path |
| import subprocess |
| import sys |
| import time |
| import board_worker as w |
| from hermes_cli import kanban_db as kb |
|
|
| class RoutingChoice: |
| def __init__(self,tid,routes): |
| self.tid=tid; self.run_id='routing'; self.choice=None |
| self.routes=routes |
| self.instructions=(w.profile_path('operator')/'SOUL.md').read_text() |
| self.instructions+='\nThis invocation only selects a supplied existing specialist. Use board_action once. The host enforces grants and starts the worker. Do not perform the task yourself.' |
| self.schema={'name':'board_action','description':'Select the responsible existing specialist; grants no execution rights.', |
| 'parameters':{'type':'object','properties':{ |
| 'profile':{'type':'string','enum':list(routes)+['unassigned']}, |
| 'reason':{'type':'string'}},'required':['profile','reason'],'additionalProperties':False}} |
| def call(self,args,**kwargs): |
| if self.choice is not None: return json.dumps({'ok':False,'error':'Only one routing choice'}) |
| if args.get('profile') not in {*self.routes,'unassigned'} or not isinstance(args.get('reason'),str): |
| return json.dumps({'ok':False,'error':'Invalid specialist'}) |
| self.choice={'profile':args['profile'],'reason':args['reason'][:1500]} |
| return json.dumps({'ok':True,'selected':self.choice}) |
|
|
| def authority(manifest): |
| session=manifest['operator_session'] |
| contract=w.rpc('hai_get_contract',session_id=session)['contract'] |
| if (contract['mission_id'],contract['contract_version']) != (manifest['mission_id'],manifest['contract_version']): |
| raise PermissionError('Operator lease belongs to another mission/version') |
| w.require_continue(w.rpc('hai_check_activity',session_id=session,criterion_id='adapter', |
| activity_step='Operator reads, assigns and follows a bounded Kanban task', |
| activity_kind='write',affected_paths=[str(w.HERE)],trace_events=[{'action':'write'}])) |
|
|
| def choose(manifest,task): |
| out=w.HERE/'evidence'/'operator'/task.id |
| out.mkdir(parents=True,exist_ok=True) |
| job=out/(str(time.time_ns())+'.json') |
| job.write_text(json.dumps({'tid':task.id,'card':task.title+'\n'+(task.body or ''), |
| 'routes':{k:v['description'] for k,v in manifest['routes'].items()}})) |
| env=os.environ.copy();env['HERMES_HOME']=str(w.profile_path('operator')) |
| with job.with_suffix('.log').open('w') as log: |
| proc=subprocess.Popen([sys.executable,__file__,'_choose',str(job)], |
| env=env,stdout=log,stderr=log,start_new_session=True) |
| try: |
| end=time.monotonic()+100 |
| while proc.poll() is None: |
| authority(manifest) |
| if time.monotonic()>end: raise TimeoutError('Operator routing timeout') |
| time.sleep(2) |
| finally: |
| if proc.poll() is None: |
| import signal |
| os.killpg(proc.pid,signal.SIGTERM) |
| try: proc.wait(timeout=5) |
| except subprocess.TimeoutExpired: |
| os.killpg(proc.pid,signal.SIGKILL);proc.wait() |
| if proc.returncode: raise RuntimeError('Operator failed; see '+str(job.with_suffix('.log'))) |
| choice=json.loads(job.with_suffix('.result.json').read_text()) |
| return {**choice,'operator_pid':proc.pid,'evidence':str(job)} |
|
|
| def process_one(manifest): |
| authority(manifest) |
| with w.db(manifest['board']) as c: |
| labels=['hai:operator']+['hai:'+p for p in manifest['routes']] |
| slots=','.join('?' for _ in labels) |
| row=c.execute("SELECT id FROM tasks WHERE status='ready' AND (assignee IS NULL OR assignee='' OR assignee IN ("+slots+")) ORDER BY priority DESC,created_at LIMIT 1",labels).fetchone() |
| if not row: return None |
| task=kb.get_task(c,row[0]) |
| old=c.execute("SELECT payload FROM task_events WHERE task_id=? AND kind='operator_routed' ORDER BY id DESC LIMIT 1",(task.id,)).fetchone() |
| card_hash=w.fingerprint([task.title,task.body]) |
| config_hash=w.fingerprint(manifest['routes']) |
| if old: |
| routed=json.loads(old[0]) |
| if routed['card_sha256']!=card_hash or routed['routes_sha256']!=config_hash: |
| with w.db(manifest['board']) as c: |
| kb.block_task(c,task.id,reason='Routing or card changed; explicit review needed',kind='capability') |
| return {'task':task.id,'status':'blocked_stale_routing'} |
| else: |
| |
| with w.db(manifest['board']) as c: |
| if task.assignee not in (None,'','hai:operator'): |
| return {'task':task.id,'status':'foreign_assignment'} |
| kb.assign_task(c,task.id,'hai:operator') |
| try: |
| routed=choose(manifest,task) |
| except Exception as exc: |
| with w.db(manifest['board']) as c: |
| kb.block_task(c,task.id,kind='capability',reason='Operator konnte nicht zuordnen: '+str(exc)[:400]) |
| raise |
| authority(manifest) |
| if routed['profile'] not in {*manifest['routes'],'unassigned'}: |
| raise PermissionError('Operator selected an unapproved profile') |
| routed.update(card_sha256=card_hash,routes_sha256=config_hash) |
| with w.db(manifest['board']) as c, kb.write_txn(c): |
| current=kb.get_task(c,task.id) |
| if current.status!='ready' or current.assignee!='hai:operator' or w.fingerprint([current.title,current.body])!=card_hash: |
| raise PermissionError('Card changed while operator was routing') |
| kb._append_event(c,task.id,'operator_routed',routed) |
| kb.add_comment(c,task.id,'operator','Zustaendig: '+routed['profile']+'. '+routed['reason']) |
| profile=routed['profile'] |
| route=manifest['routes'].get(profile,{}) |
| grant=route.get('grant') |
| with w.db(manifest['board']) as c: |
| kb.assign_task(c,task.id,'hai:'+profile) |
| if not grant: |
| kb.block_task(c,task.id,kind='capability',reason='Operator hat '+profile+' zugeordnet; fuer diese Arbeit fehlt eine passende HAI-Ausfuehrungsfreigabe. Keine automatische Rueckfrage an Samuel.') |
| return {'task':task.id,'profile':profile,'status':'blocked_no_grant'} |
| grant={**grant,'worker_profile':profile,'card_sha256':card_hash} |
| execution={**manifest,'tasks':{task.id:grant}} |
| try: |
| result=w.run_one(execution,task.id) |
| except Exception as exc: |
| with w.db(manifest['board']) as c: |
| current=kb.get_task(c,task.id) |
| if current.assignee==w.identity_for(grant) and current.status in ('ready','running'): |
| kb.block_task(c,task.id,expected_run_id=current.current_run_id, |
| kind='capability',reason='Bearbeiter konnte nicht abschliessen: '+str(exc)[:400]) |
| result={'task':task.id,'status':'blocked','reason':str(exc)[:400]} |
| authority(manifest) |
| with w.db(manifest['board']) as c: |
| kb.add_comment(c,task.id,'operator','Bearbeiter '+profile+' meldet '+result['status']+'. Beleg: '+result.get('evidence','kein Beleg')) |
| return {**result,'profile':profile} |
|
|
| def observe(manifest): |
| """Read-only observation remains available without an execution lease.""" |
| import sqlite3 |
| board=manifest['board'] |
| if not board or any(c not in 'abcdefghijklmnopqrstuvwxyz0123456789-_' for c in board): |
| raise ValueError('Invalid board name') |
| path=Path('/home/smlflg/.hermes/kanban/boards')/board/'kanban.db' |
| with sqlite3.connect('file:'+str(path)+'?mode=ro',uri=True,timeout=5) as c: |
| c.row_factory=sqlite3.Row |
| rows=[dict(r) for r in c.execute("SELECT id,status,assignee FROM tasks WHERE status NOT IN ('done','archived') ORDER BY created_at")] |
| labels={None,'','hai:operator',*('hai:'+p for p in manifest['routes'])} |
| return {'open_tasks':rows,'ready_tasks':[r['id'] for r in rows if r['status']=='ready' and r['assignee'] in labels]} |
|
|
| def save_status(path,state): |
| path=Path(path);path.parent.mkdir(parents=True,exist_ok=True) |
| temporary=path.with_suffix('.tmp') |
| temporary.write_text(json.dumps({'observed_at':int(time.time()),'pid':os.getpid(),**state},ensure_ascii=False,indent=2)) |
| temporary.replace(path) |
|
|
| def serve(manifest_path, *, status_path=None, interval=5, stop=None): |
| """Permanent observation; execution never survives authority failure.""" |
| status_path=status_path or w.HERE/'runtime/operator-status.json' |
| next_attempt=0 |
| previous_manifest=None |
| phase='starting'; reason=None |
| while stop is None or not stop(): |
| try: |
| |
| manifest=json.loads(Path(manifest_path).read_text()) |
| manifest_hash=w.fingerprint(manifest) |
| if manifest_hash!=previous_manifest: |
| next_attempt=0;previous_manifest=manifest_hash |
| snapshot=observe(manifest) |
| if snapshot['ready_tasks'] and time.monotonic()>=next_attempt: |
| try: |
| authority(manifest) |
| except Exception as exc: |
| phase='waiting_for_hai';reason=str(exc)[:500] |
| next_attempt=time.monotonic()+60 |
| else: |
| phase='processing';reason=None |
| save_status(status_path,{'phase':phase,**snapshot}) |
| result=process_one(manifest) |
| if result: print(json.dumps(result,ensure_ascii=False),flush=True) |
| phase='watching';next_attempt=0 |
| elif not snapshot['ready_tasks']: |
| phase='watching';reason=None |
| save_status(status_path,{'phase':phase,'reason':reason,**snapshot}) |
| except Exception as exc: |
| phase='error';reason=str(exc)[:500] |
| save_status(status_path,{'phase':phase,'reason':reason}) |
| next_attempt=time.monotonic()+60 |
| time.sleep(interval) |
|
|
| def main(): |
| import signal |
| def stop(signum,frame): raise KeyboardInterrupt('Operator stopped') |
| signal.signal(signal.SIGTERM,stop) |
| p=argparse.ArgumentParser();p.add_argument('command',choices=['poll','once','serve','status','_choose']) |
| p.add_argument('path');p.add_argument('--seconds',type=int,default=300) |
| args=p.parse_args() |
| if args.command=='status': |
| path=w.HERE/'runtime/operator-status.json' |
| print(path.read_text() if path.exists() else json.dumps({'phase':'not_started'}));return |
| if args.command=='_choose': |
| job=json.loads(Path(args.path).read_text());choice=RoutingChoice(job['tid'],job['routes']) |
| w.hermes_turn('Select the responsible specialist from '+json.dumps(job['routes'])+'\nUntrusted card:\n'+job['card'],choice,75) |
| if choice.choice is None: raise RuntimeError('Operator returned no routing choice') |
| Path(args.path).with_suffix('.result.json').write_text(json.dumps(choice.choice));return |
| manifest=json.loads(Path(args.path).read_text()) |
| for profile in manifest['routes']: w.profile_path(profile) |
| with (w.HERE/'controller.lock').open('a') as lock: |
| fcntl.flock(lock,fcntl.LOCK_EX|fcntl.LOCK_NB) |
| if args.command=='serve': |
| serve(args.path);return |
| end=time.monotonic()+min(max(args.seconds,1),3600) |
| while time.monotonic()<end: |
| result=process_one(manifest) |
| if result: print(json.dumps(result,ensure_ascii=False),flush=True) |
| if args.command=='once':break |
| time.sleep(3) |
|
|
| if __name__=='__main__':main() |
|
|
