#!/usr/bin/env python3 """Bounded read-only adapter; reuses Hermes readers without importing its server.""" import hashlib import json import os import sys import time from datetime import datetime, timezone from itertools import islice from pathlib import Path # The fleet runtime root (.pi/agent-fleet), not the project root: the Hermes # readers this reuses are siblings of .pi/agent-fleet/scripts/, and both move together. FLEET_ROOT = Path(__file__).resolve().parent.parent sys.path.insert(0, str(FLEET_ROOT / 'hermes/plugins/agent-fleet-herdr/dashboard')) sys.path.insert(0, str(FLEET_ROOT / 'hermes/skills/hub-watchdog/scripts')) import activity import herdr_source from watchdog_transport import discover, MonitorTransport def read_pane(project, peer): try: matches = [a for a in herdr_source.list_agents() if herdr_source.peer_key(a) == (project, peer) and a.get('agent') == 'pi'] if len(matches) != 1: return {'paneId':None, 'state':'unknown', 'reason':'ambiguous' if matches else 'unavailable'} a = matches[0] return {'paneId':a.get('pane_id'), 'state':herdr_source.normalize_state(a.get('agent_status')), 'reason':''} except herdr_source.HerdrUnavailable: return {'paneId':None, 'state':'unknown', 'reason':'herdr unavailable'} def read_activity(entry, cursor, root=None): transcript, reason = activity.transcript_for_entry(entry, root=root) if transcript is None: return {**activity.unavailable(reason), 'cursor':cursor, 'gap':True} info = transcript.stat() identity = hashlib.sha256(f'{transcript}:{info.st_dev}:{info.st_ino}'.encode()).hexdigest() after = cursor.get('seq', 0) reset = bool(cursor) and (cursor.get('identity') != identity or after > info.st_size) if reset: after = 0 steps = activity.steps_for_path(transcript) fresh = [s for s in steps if s['seq'] > after] result = activity.activity_for_entry(entry, after=after, limit=50, root=root) # The canonical reader trims to newest whole lines. Surface that loss explicitly. gap = reset or (info.st_size > activity.TAIL_BYTES and after < max(0, info.st_size-activity.TAIL_BYTES)) or len(fresh) > len(result['steps']) return {**result, 'gap':gap, 'cursor':{'identity':identity, 'seq':result['seq']}, 'source':'pi_transcript'} TASK_FIELDS = ('id','generation','kind','state','specialist','parentId','parentGeneration','updatedAt','outputSequence') def read_tasks(transport, tasks, owner, hub, cursors): selected = [t for t in tasks if t.get('ownerSessionId') == owner] # Fair bounded window: oldest unread outputs first, then current task metadata. selected.sort(key=lambda t: (cursors.get(f"{owner}:{hub}:{t['id']}:{t['generation']}", 0) >= t.get('outputSequence', 0), t.get('updatedAt',''))) results, next_cursors, gap = [], dict(cursors), len(selected) > 100 reads, pending = 0, 0 for task in selected[:100]: item = {k:task[k] for k in TASK_FIELDS if k in task} item.update(ownerSessionId=owner, hubInstanceId=hub) key = f"{owner}:{hub}:{task['id']}:{task['generation']}" after = cursors.get(key, 0) if reads < 8 and (key not in cursors or task.get('outputSequence', 0) > after): reads += 1 try: output = transport._call({'type':'output','taskId':task['id'],'generation':task['generation'],'afterSequence':after})['output'] text = output.get('text', '') item['output'] = text[-32768:] item['outputGap'] = bool(output.get('truncated') or len(text) > 32768 or output.get('firstSequence', 0) > after + 1) gap = gap or item['outputGap'] next_cursors[key] = output.get('sequence', after) except Exception: item['outputUnavailable'] = True; gap = True if reads >= 8 and (key not in next_cursors or task.get('outputSequence',0) > next_cursors.get(key,0)): item['outputDeferred'] = True; pending += 1 results.append(item) active_keys = {f"{owner}:{hub}:{t['id']}:{t['generation']}" for t in selected} return {'tasks':results, 'pendingOutputs':pending, 'cursor':{k:v for k,v in next_cursors.items() if k in active_keys}, 'gap':gap} def read_monitor(entry, cursor, wait_ms=0): runtime = Path(os.environ.get('AGENT_FLEET_MONITOR_RUNTIME_DIR', f'/tmp/agent-fleet-monitor-{os.getuid()}')) profile = os.environ.get('AGENT_FLEET_PROFILE_ID') pattern = (hashlib.sha256(profile.encode()).hexdigest() if profile else '*') + '/*/discovery-*.json' candidates = [] try: if runtime.is_symlink() or runtime.stat().st_uid != os.getuid(): raise ValueError() paths = list(islice(runtime.glob(pattern), 33)) if len(paths) > 32: raise ValueError() for p in paths: try: d = discover(str(runtime), p.parent.parent.name, p.parent.name, datetime.now(timezone.utc)) if d['lease'].get('pid') == entry['pid']: candidates.append(d) except Exception: continue except Exception: return {'available':False, 'reason':'monitor discovery unavailable', 'tasks':[], 'cursor':cursor, 'gap':True} if len(candidates) != 1: return {'available':False, 'reason':'monitor missing or ambiguous for selected Pi process', 'tasks':[], 'cursor':cursor, 'gap':True} d = candidates[0]; transport = MonitorTransport(d, timeout=2) owner = d['owner']; hub = d['lease']['hub'] reset = bool(cursor) and cursor.get('owner') != owner previous = {} if reset else cursor gap = reset try: if wait_ms: time.sleep(min(wait_ms, 2000)/1000) snap = transport.snapshot()['snapshot'] tasks = read_tasks(transport, snap.get('tasks',[]), owner, hub, previous.get('outputs',{})) return {'available':True, 'source':'hub_monitor', 'tasks':tasks['tasks'], 'pendingOutputs':tasks['pendingOutputs'], 'gap':gap or tasks['gap'], 'cursor':{'owner':owner,'outputs':tasks['cursor']}} except Exception: return {'available':False, 'reason':'monitor unavailable', 'tasks':[], 'cursor':cursor, 'gap':True} def project_replies(rows, pending): """Match Pi's own respond() audit to the assistant text it used, never a guessed next turn.""" replies, latest = [], None wanted = set(pending) for _, line in rows: try: record = json.loads(line) except ValueError: continue if record.get('type') == 'message' and record.get('message',{}).get('role') == 'assistant': content = record['message'].get('content') if isinstance(content,str): latest = content elif isinstance(content,list): latest = '\n'.join(c['text'] for c in content if c.get('type') == 'text' and isinstance(c.get('text'),str)) data = record.get('data',{}) if record.get('customType') == 'coms-log' and data.get('event') in ('outbound_response','outbound_response_failed') and data.get('msg_id') in wanted and latest is not None: replies.append({'wireId':data['msg_id'], 'state':'failed' if data.get('error') else 'result', 'result':latest[:16000], 'truncated':len(latest)>16000}) return replies def read(request): entry = request['entry']; cursor = request.get('cursor') or {} pane = read_pane(request['project'], entry['name']) transcript = read_activity(entry, cursor.get('activity',{})) monitor = read_monitor(entry, cursor.get('monitor',{}), request.get('waitMs',0)) path, _ = activity.transcript_for_entry(entry) replies = project_replies(activity.read_tail(path), request.get('pending', [])) if path else [] return {'replies':replies, 'observedAt':datetime.now(timezone.utc).isoformat(), 'pane':pane, 'activity':transcript, 'monitor':monitor, 'partial':not transcript['available'] or not monitor['available'] or transcript['gap'] or monitor['gap'] or bool(monitor.get('pendingOutputs')), 'cursor':{'activity':transcript['cursor'],'monitor':monitor['cursor']}} if __name__ == '__main__': try: request = json.loads(sys.stdin.buffer.read(65537)) print(json.dumps(read(request), ensure_ascii=False)) except Exception: print(json.dumps({'error':'read sources unavailable'})); sys.exit(1)