Uebergabe/operator_loop.py

"""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()<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()