← back to Chief Of Operations

chief_ops.py

217 lines

#!/usr/bin/env python3
"""Chief of Operations: read-first oversight for layout, dotbar, tickets, and Plannator."""
from __future__ import annotations
import argparse, json, os, subprocess, time, urllib.request, urllib.error
import socket
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

HOME = Path.home()
ROUTER = HOME / '.claude/skills/dot-screen-router'
DOTBAR = HOME / 'Projects/desktop-dotbar'
TICKET_LIB = HOME / 'Projects/ticket-system/lib.js'
TK = HOME / 'Projects/ticket-system/tk'
STATE = Path(__file__).with_name('state')
OPS_AGENT = 'chief-of-operations'
STATUS = HOME / '.local/state/abrams-terminal-status/ttys016.json'
TICKET = 'TK-11882-chief-of-operations-harness-for-terminal'
SCREENSHOTS = [HOME/'Desktop/arrange-click-screen1.png', HOME/'Desktop/arrange-click-screen2.png']
ARCHIVE = HOME/'.claude/skills/_disabled-layout-legacy-20260917'


def cmd(args, timeout=4):
    try:
        p = subprocess.run(args, text=True, capture_output=True, timeout=timeout)
        return p.returncode, (p.stdout or '').strip(), (p.stderr or '').strip()
    except Exception as e:
        return 1, '', str(e)


def launchd(label):
    rc, out, err = cmd(['launchctl','print',f'gui/{os.getuid()}/{label}'])
    return {'name': label, 'ok': rc == 0 and 'state = running' in out, 'detail': next((x.strip() for x in out.splitlines() if x.strip().startswith('state =')), err or 'not loaded')}


def http_health(port, path='/health'):
    try:
        with urllib.request.urlopen(f'http://127.0.0.1:{port}{path}', timeout=2) as r:
            return {'ok': r.status == 200, 'body': json.loads(r.read().decode() or '{}')}
    except urllib.error.HTTPError as e:
        if e.code in (401, 403):
            return {'ok': None, 'status': f'auth-required-{e.code}', 'error': str(e)}
        return {'ok': False, 'error': str(e)}
    except Exception as e:
        msg = str(e)
        # Workspace sandboxes deny localhost sockets with EPERM. Keep that
        # distinct from a real connection refusal so the report says
        # "unverified" instead of falsely declaring an outage.
        if 'Operation not permitted' in msg:
            return {'ok': None, 'status': 'unverified-in-sandbox', 'error': msg}
        return {'ok': False, 'error': msg}

def exo_ring():
    env=HOME/'ai-cluster/.env'
    hosts=[]
    if env.exists():
        for line in env.read_text(errors='replace').splitlines():
            if line.startswith('EXO_HOSTS='):
                hosts=[x.strip() for x in line.split('=',1)[1].split(',') if x.strip()]
    results=[]
    for item in hosts:
        host,_,port=item.rpartition(':'); port=int(port or 52415)
        last_error='timeout'
        for attempt in range(3):
            s=socket.socket(); s.settimeout(2); started=time.time()
            try:
                s.connect((host,port)); results.append({'host':host,'port':port,'online':True,'latency_ms':round((time.time()-started)*1000,1),'attempt':attempt+1}); break
            except Exception as e:
                last_error=str(e)
                if 'Operation not permitted' in last_error:
                    results.append({'host':host,'port':port,'online':None,'error':last_error}); break
            finally: s.close()
        else:
            results.append({'host':host,'port':port,'online':False,'error':last_error,'attempts':3})
    online=sum(x['online'] is True for x in results); unknown=sum(x['online'] is None for x in results)
    ok=None if unknown else (bool(results) and online==len(results))
    return {'ok':ok,'online':online,'unknown':unknown,'configured':len(results),'hosts':results,
            'status':'unverified-in-sandbox' if unknown else ('all-online' if online==len(results) else ('partial' if online else 'offline'))}

def watch_exo():
    """Persist the Exo ring verdict and log only state transitions."""
    STATE.mkdir(exist_ok=True)
    verdict=exo_ring(); path=STATE/'exo-watch.json'; previous={}
    if path.exists():
        try: previous=json.loads(path.read_text())
        except Exception: previous={}
    verdict['checked_at']=time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())
    path.write_text(json.dumps(verdict,indent=2)+'\n')
    old=previous.get('status'); new=verdict.get('status')
    if old != new and new in ('partial','offline'):
        cmd([str(TK),'log',TICKET,f'Chief Ops Exo watchdog: ring {new} ({verdict.get("online",0)}/{verdict.get("configured",0)} online); investigate host reachability or Mac keepalive.'], timeout=10)
    return verdict


def router_evidence():
    log = ROUTER/'data/router.jsonl'
    if not log.exists(): return {'ok': False, 'error': 'router evidence missing'}
    rows=[]
    for line in log.read_text(errors='replace').splitlines()[-40:]:
        try: rows.append(json.loads(line))
        except Exception: pass
    rows=[r for r in rows if 'green_misplaced' in r]
    last=rows[-1] if rows else {}
    green_bad = last.get('green_misplaced', 99)
    attn_bad = last.get('attn_misplaced', 99)
    front_exempt = bool(last.get('skipped_front'))
    return {'ok': bool(rows) and green_bad == 0 and (attn_bad == 0 or (attn_bad == 1 and front_exempt)),
            'frontmost_exempt': front_exempt, 'last': last, 'samples': len(rows),
            'cellfails': sum(1 for x in log.read_text(errors='replace').splitlines()[-40:] if '"event":"cellfail"' in x)}

def plannator_state():
    root = HOME/'.claude/skills/plannator/state'
    files = sorted(root.glob('*.json')) if root.exists() else []
    active=[]; blocked=[]
    for p in files:
        try:
            d=json.loads(p.read_text())
            states=[str(s.get('status','')).lower() for s in d.get('steps',[]) if isinstance(s,dict)]
            if any(s in ('running','pending') for s in states): active.append(p.stem)
            if 'blocked' in states: blocked.append(p.stem)
        except Exception: continue
    return {'ok': root.exists(), 'manifest_count': len(files), 'active': active[-20:], 'blocked': blocked[-20:]}

def ticket_queue(write=False):
    js = "const l=require(process.argv[1]); console.log(JSON.stringify([...l.tickets().values()].map(t=>({id:t.id,title:t.title,project:t.project,status:t.status,updated_at:t.updated_at,assignee:t.assignee||''}))));"
    rc,out,err=cmd(['node','-e',js,str(TICKET_LIB)], timeout=12)
    if rc or not out: return {'ok':False,'error':err or 'ticket ledger unavailable','items':[]}
    try: rows=json.loads(out)
    except Exception as e: return {'ok':False,'error':str(e),'items':[]}
    dw_terms=('designer wallcoverings','designerwallcoverings','dw-commerce','dw-catalog','dw_unified','shopify','wallcovering','wallpaper','vendor','microsite','catalog')
    client_terms=('client','customer','live','production','revenue','merchant','shopify','storefront')
    items=[]
    for t in rows:
        if t.get('status') not in ('open','blocked','doing'): continue
        text=(str(t.get('title',''))+' '+str(t.get('project',''))).lower()
        dw=any(x in text for x in dw_terms); client=any(x in text for x in client_terms)
        status_score={'blocked':70,'open':60,'doing':35}.get(t.get('status'),0)
        score=status_score + (100 if dw else 0) + (35 if client else 0)
        if t.get('status')=='blocked': next_step='review blocker and route the next reversible step'
        elif t.get('status')=='open': next_step='claim or dispatch through Plannator'
        else: next_step='verify active work and unblock dependencies'
        items.append({**t,'score':score,'dw_priority':dw,'client_impact':client,'next_step':next_step})
    items.sort(key=lambda x:(-x['score'], x.get('updated_at') or ''))
    result={'ok':True,'generated_at':time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime()),'total':len(items),'dw_count':sum(x['dw_priority'] for x in items),'blocked_count':sum(x['status']=='blocked' for x in items),'items':items[:40]}
    if write:
        STATE.mkdir(exist_ok=True); (STATE/'queue.json').write_text(json.dumps(result,indent=2)+'\n')
    return result

def auto_claim():
    # TK-12235: the ticket front desk (tk claim / POST /api/claim, lease-based) is the ONE claimer.
    # Chief-of-ops only labelled tickets without working them (held TK-11889 for 8 days, 10k "hold"
    # log lines), so the tk-take path is removed entirely -- no re-enable switch that could bypass leases.
    return {'ok':True,'action':'disabled','reason':'claiming moved to the ticket front desk (TK-12235)'}

def check():
    status={}
    if STATUS.exists():
        try:
            d=json.loads(STATUS.read_text()); status={'ok': d.get('state') == 'green', 'state': d.get('state'), 'title': d.get('title'), 'ticket': d.get('ticket')}
        except Exception as e: status={'ok':False,'error':str(e)}
    archived=[p.name for p in (ARCHIVE.iterdir() if ARCHIVE.exists() else [])]
    screenshots=[]
    for p in SCREENSHOTS:
        age_s=(time.time()-p.stat().st_mtime) if p.exists() else None
        screenshots.append({'path':str(p),'ok':bool(p.exists() and age_s is not None and age_s < 86400),'age_s':round(age_s) if age_s is not None else None})
    checks={
      'terminal_status': status,
      'router_launchd': launchd('com.steve.dot-screen-router'),
      'dotbar_launchd': launchd('com.steve.desktop-dotbar'),
      'supervisor_launchd': launchd('com.steve.chief-of-operations'),
      'dotbar_health': http_health(9787),
      'plannator_health': http_health(9772),
      'plannator_state': plannator_state(),
      'exo_ring': exo_ring(),
      'ticket_queue': ticket_queue(write=True),
      'router_evidence': router_evidence(),
      'legacy_conflicts_archived': {'ok': all(x in archived for x in ('arrange','tile-blue-right','tile-green-left')), 'items': archived},
      'screenshots': {'ok': all(x['ok'] for x in screenshots), 'items': screenshots},
    }
    hard=[v.get('ok') for v in checks.values()]
    checks['overall']={'ok': all(x is not False for x in hard), 'unverified': sum(x is None for x in hard), 'checked_at':time.strftime('%Y-%m-%dT%H:%M:%SZ',time.gmtime())}
    return checks


def arrange():
    try:
        req=urllib.request.Request('http://127.0.0.1:9787/api/arrange', method='POST')
        with urllib.request.urlopen(req, timeout=50) as r: return json.loads(r.read().decode())
    except Exception as e: return {'ok':False,'error':str(e)}

HTML='''<!doctype html><meta charset="utf-8"><title>Chief of Operations</title><style>body{font:15px -apple-system;background:#101318;color:#e9edf3;margin:32px}h1{margin:0 0 8px}.sub{color:#9aa4b2;margin-bottom:24px}.grid{display:grid;grid-template-columns:repeat(auto-fit,minmax(280px,1fr));gap:12px}.card{background:#1b2029;border:1px solid #303947;border-radius:12px;padding:16px}.ok{color:#54d17a}.bad{color:#ff7676}.muted{color:#9aa4b2;font-size:12px;word-break:break-word}button{background:#315dff;color:white;border:0;border-radius:8px;padding:10px 14px;font-weight:700;cursor:pointer}pre{white-space:pre-wrap;font-size:12px;color:#c9d2df}</style><h1>Chief of Operations</h1><div class="sub">Read-first oversight for the master terminal router, desktop bar, Plannator, and evidence.</div><button onclick="arrange()">Arrange master now</button> <button onclick="load()">Refresh checks</button><div id="out" class="grid" style="margin-top:20px"></div><script>async function load(){let d=await fetch('/api/check').then(r=>r.json());out.innerHTML=Object.entries(d).map(([k,v])=>`<section class="card"><b class="${v.ok?'ok':'bad'}">${v.ok?'●':'●'} ${k}</b><pre>${JSON.stringify(v,null,2)}</pre></section>`).join('')}async function arrange(){let r=await fetch('/api/arrange',{method:'POST'}).then(r=>r.json());alert(JSON.stringify(r));load()}load();setInterval(load,10000)</script>'''

class Handler(BaseHTTPRequestHandler):
    def log_message(self,*args): pass
    def do_GET(self):
        if self.path == '/api/check': self.send_json(check()); return
        if self.path == '/api/queue': self.send_json(ticket_queue(write=True)); return
        self.send_response(200); self.send_header('Content-Type','text/html'); self.end_headers(); self.wfile.write(HTML.encode())
    def do_POST(self):
        if self.path == '/api/arrange': self.send_json(arrange()); return
        self.send_error(404)
    def send_json(self, data):
        b=json.dumps(data, indent=2).encode(); self.send_response(200); self.send_header('Content-Type','application/json'); self.send_header('Content-Length',str(len(b))); self.end_headers(); self.wfile.write(b)

def main():
    ap=argparse.ArgumentParser(); ap.add_argument('action', choices=['check','arrange','queue','claim','watch-exo','serve'], nargs='?', default='check'); ap.add_argument('--port',type=int,default=9896); ap.add_argument('--write',action='store_true'); a=ap.parse_args()
    if a.action=='check': print(json.dumps(check(), indent=2)); return
    if a.action=='arrange': print(json.dumps(arrange(), indent=2)); return
    if a.action=='queue':
        q=ticket_queue(write=True)
        # --write (the 60s supervisor loop) prints a one-line summary; the full queue lives in state/queue.json.
        # Dumping the whole JSON every minute grew state/supervisor.log to 225 MB (TK-12235).
        print(json.dumps({k:q.get(k) for k in ('ok','generated_at','total','dw_count','blocked_count','error') if k in q}) if a.write else json.dumps(q, indent=2)); return
    if a.action=='claim': print(json.dumps(auto_claim(), indent=2)); return
    if a.action=='watch-exo': print(json.dumps(watch_exo(), indent=2)); return
    print(f'Chief of Operations on http://127.0.0.1:{a.port}', flush=True); ThreadingHTTPServer(('127.0.0.1',a.port), Handler).serve_forever()
if __name__=='__main__': main()