"""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: # Non-profile assignee is deliberately excluded from the native gateway. 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: # Reload explicit owner configuration. Never infer/renew a mission. 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()