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