92 lines
4.6 KiB
Python
92 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Keep the local queue worker and visible-history mirror available without exporting auth."""
|
|
import argparse
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from worker import atomic_json,session_lock,reconcile_completed
|
|
from control import Client
|
|
from faults import report_process
|
|
|
|
|
|
def recover_worker(cfg,process=None):
|
|
"""Reconcile a stopped worker under its own lock; never execute its input."""
|
|
if process is not None and process.poll() is None:return False
|
|
identity=json.dumps({'control':cfg['control'],'session_id':cfg['session_id']},sort_keys=True)
|
|
try:
|
|
with session_lock(identity):
|
|
path=Path(cfg['state_file']).expanduser()
|
|
if not path.exists():return False
|
|
state=json.loads(path.read_text(encoding='utf-8'))
|
|
if not (state.get('inflight') and state.get('needs_reconciliation')):return False
|
|
if not cfg.get('thread_id'):return False
|
|
return reconcile_completed(path,Client(cfg['control']),cfg['session_id'],cfg['thread_id'])
|
|
except Exception:
|
|
# Offline, writer still locked, or evidence incomplete: preserve uncertainty.
|
|
return False
|
|
|
|
|
|
def main():
|
|
parser=argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('--config',required=True)
|
|
args=parser.parse_args()
|
|
config_path=Path(args.config).resolve()
|
|
cfg=json.loads(config_path.read_text(encoding='utf-8-sig'))
|
|
root=Path(cfg['state_file']).expanduser().resolve().parent
|
|
root.mkdir(parents=True,exist_ok=True,mode=0o700)
|
|
if os.name=='posix':os.umask(0o077)
|
|
identity='supervisor:'+json.dumps({'control':cfg['control'],'session_id':cfg['session_id']},sort_keys=True)
|
|
fault_path=root/'bridge-faults.json'
|
|
faults=json.loads(fault_path.read_text(encoding='utf-8')).get('roles',{}) if fault_path.exists() else {}
|
|
processes={}
|
|
started_at={}
|
|
logs={}
|
|
next_start={}
|
|
roles=['worker','mirror']
|
|
if cfg.get('notification_targets'):roles.append('notifier')
|
|
if cfg.get('task_report_config'):roles.append('report_relay')
|
|
if cfg.get('decision_fleet',{}).get('enabled'):roles.append('decision_fleet')
|
|
with session_lock(identity):
|
|
try:
|
|
while True:
|
|
state_path=Path(cfg['state_file']).expanduser()
|
|
state=json.loads(state_path.read_text(encoding='utf-8')) if state_path.exists() else {}
|
|
if recover_worker(cfg,processes.get('worker')):
|
|
state=json.loads(state_path.read_text(encoding='utf-8'))
|
|
for role in roles:
|
|
process=processes.get(role)
|
|
if process and cfg.get('notification_targets'):
|
|
report_process(cfg,root,role,process,faults,started_at.get(role))
|
|
if process and process.poll() is None:continue
|
|
if role=='worker' and (state.get('needs_reconciliation') or (state.get('inflight') and not state.get('pending_completion'))):
|
|
continue
|
|
if time.monotonic()<next_start.get(role,0):continue
|
|
if role not in logs:logs[role]=open(root/(role+'.log'),'a',encoding='utf-8')
|
|
command=[sys.executable,'-X','utf8',str(Path(__file__).with_name(role+'.py')),'--config',str(config_path)]
|
|
process=subprocess.Popen(command,stdout=logs[role],stderr=logs[role],
|
|
**({'creationflags':subprocess.CREATE_NO_WINDOW} if os.name=='nt' else {}))
|
|
processes[role]=process
|
|
started_at[role]=time.time()
|
|
next_start[role]=time.monotonic()+15
|
|
atomic_json(root/'bridge-service.json',{'pid':os.getpid(),'updated':time.time(),
|
|
'children':{name:{'pid':p.pid,'running':p.poll() is None} for name,p in processes.items()},
|
|
'worker_paused_for_reconciliation':bool((state.get('needs_reconciliation') or (state.get('inflight') and not state.get('pending_completion')))
|
|
and ('worker' not in processes or processes['worker'].poll() is not None))})
|
|
time.sleep(5)
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
for role,process in processes.items():
|
|
if role!='worker' and process.poll() is None:process.terminate()
|
|
for role,process in processes.items():
|
|
if role=='worker' and process.poll() is None:continue
|
|
try:process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:process.kill()
|
|
for stream in logs.values():stream.close()
|
|
|
|
|
|
if __name__=='__main__':main()
|