144 lines
7.7 KiB
Python
144 lines
7.7 KiB
Python
#!/usr/bin/env python3
|
|
"""Bridge already bound decision threads; host creation and archive stay explicit."""
|
|
import argparse
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from control import Client
|
|
from worker import atomic_json,session_lock
|
|
|
|
|
|
def busy(path):
|
|
if not path.exists():return False
|
|
state=json.loads(path.read_text(encoding='utf-8'))
|
|
return bool(state.get('inflight') or state.get('pending_completion') or state.get('needs_reconciliation'))
|
|
|
|
|
|
def child_config(base,binding,root):
|
|
did=binding['decision_id'];sid=binding['session_id'];thread=binding['thread_id']
|
|
if any(not isinstance(v,str) or not re.fullmatch(r'[A-Za-z0-9_.:-]{1,128}',v) for v in (did,sid)):
|
|
raise ValueError('Invalid decision binding')
|
|
if not re.fullmatch(r'[0-9a-fA-F-]{36}',thread):raise ValueError('Invalid real thread identity')
|
|
cwd=Path(binding['worktree']).expanduser().resolve()
|
|
if not cwd.is_dir():raise ValueError('Bound thread directory unavailable')
|
|
runtime=root/hashlib.sha256(did.encode()).hexdigest()[:24]
|
|
runtime.mkdir(parents=True,exist_ok=True,mode=0o700)
|
|
cfg={key:base[key] for key in ('control','codex') if key in base}
|
|
cfg.update(mode='desktop_queue',session_id=sid,thread_id=thread,cwd=str(cwd),
|
|
worker_id='decision-'+hashlib.sha256(sid.encode()).hexdigest()[:20],
|
|
state_file=str(runtime/'worker.json'),mirror_state_file=str(runtime/'mirror.json'),
|
|
include_progress_snapshot=False,poll_seconds=5,mirror_seconds=5,
|
|
max_turn_seconds=base.get('max_turn_seconds',600))
|
|
cfg['instructions']=('你是独立决策咨询会话，仅维护决策 '+did+'。只可通过初始化时提供的本决策有限凭据读取历史、解释或修订该决策、向主控提出事实核实请求。'
|
|
'不得修改项目代码、系统、其他任务或决策，不执行方案、接受用户提交、伪造检查或归档。需要主控事实时使用本决策facts接口并保留关联ID，不重复催问。'
|
|
'用户提交或主控接受不表示新增授权。一般项目问题与待授权事项定向交项目agent对话入口，不另起一般人类沟通。当前网页消息只用于本决策咨询。')
|
|
return cfg,runtime
|
|
|
|
|
|
class Fleet:
|
|
def __init__(self,cfg):
|
|
self.cfg=cfg
|
|
settings=cfg['decision_fleet']
|
|
self.root=Path(settings['runtime_root']).expanduser().resolve()
|
|
self.root.mkdir(parents=True,exist_ok=True,mode=0o700)
|
|
self.limit=settings.get('max_active',3)
|
|
if isinstance(self.limit,bool) or not isinstance(self.limit,int) or not 1<=self.limit<=3:
|
|
raise ValueError('Decision bridge bound must be between one and three')
|
|
self.children={}
|
|
self.health=Path(cfg['state_file']).parent/'decision_fleet-health.json'
|
|
|
|
def stop(self,did,force=False):
|
|
child=self.children[did]
|
|
if not force and busy(Path(child['cfg']['state_file'])):return False
|
|
for process in child['processes'].values():
|
|
if process.poll() is None:process.terminate()
|
|
for process in child['processes'].values():
|
|
try:process.wait(timeout=10)
|
|
except subprocess.TimeoutExpired:process.kill();process.wait(timeout=10)
|
|
for stream in child['logs'].values():stream.close()
|
|
del self.children[did]
|
|
return True
|
|
|
|
def tick(self,bindings):
|
|
live={}
|
|
waiting=0;problems=[]
|
|
threads=set();sessions=set()
|
|
for binding in bindings:
|
|
did=binding['decision_id']
|
|
if did in live or binding['thread_id'] in threads or binding['session_id'] in sessions:
|
|
raise ValueError('Duplicate decision, session or thread binding')
|
|
live[did]=binding;threads.add(binding['thread_id']);sessions.add(binding['session_id'])
|
|
for did in list(self.children):
|
|
binding=live.get(did)
|
|
if not binding or binding.get('archive_status')=='archived':
|
|
if not self.stop(did):problems.append('archive_requires_reconciliation:'+did)
|
|
for did,binding in sorted(live.items()):
|
|
if binding.get('archive_status')=='archived':continue
|
|
candidate,runtime=child_config(self.cfg,binding,self.root)
|
|
child=self.children.get(did)
|
|
if child and child['cfg']!=candidate:
|
|
problems.append('binding_changed:'+did);continue
|
|
if not child:
|
|
if len(self.children)>=self.limit:waiting+=1;continue
|
|
config_path=runtime/'bridge.json'
|
|
if config_path.exists() and json.loads(config_path.read_text())!=candidate:
|
|
problems.append('saved_binding_changed:'+did);continue
|
|
atomic_json(config_path,candidate)
|
|
child={'cfg':candidate,'runtime':runtime,'path':config_path,'processes':{},'logs':{},'retry_at':{},'started':{}}
|
|
self.children[did]=child
|
|
state_path=Path(candidate['state_file'])
|
|
for role in ('worker','mirror'):
|
|
process=child['processes'].get(role)
|
|
if process and process.poll() is None:
|
|
try:
|
|
health=json.loads((runtime/(role+'-health.json')).read_text())
|
|
healthy=health.get('status')=='ok' and time.time()-health.get('updated',0)<60
|
|
except (OSError,ValueError):healthy=False
|
|
if not healthy and time.time()-child['started'].get(role,0)>60:
|
|
problems.append('bridge_health_stale:'+did+':'+role)
|
|
continue
|
|
if role=='worker' and busy(state_path):
|
|
problems.append('execution_requires_reconciliation:'+did);continue
|
|
if time.monotonic()<child['retry_at'].get(role,0):continue
|
|
if role not in child['logs']:child['logs'][role]=open(runtime/(role+'.log'),'a',encoding='utf-8')
|
|
command=[sys.executable,'-X','utf8',str(Path(__file__).with_name(role+'.py')),'--config',str(child['path'])]
|
|
child['processes'][role]=subprocess.Popen(command,stdout=child['logs'][role],stderr=child['logs'][role],
|
|
**({'creationflags':subprocess.CREATE_NO_WINDOW} if os.name=='nt' else {}))
|
|
child['started'][role]=time.time()
|
|
child['retry_at'][role]=time.monotonic()+15
|
|
snapshot={'status':'failed' if problems else 'ok','updated':time.time(),'active':len(self.children),
|
|
'waiting':waiting,'problems':problems,'bindings':{did:{'session_id':c['cfg']['session_id'],
|
|
'thread_id':c['cfg']['thread_id'],'processes':{r:{'pid':p.pid,'running':p.poll() is None}
|
|
for r,p in c['processes'].items()}} for did,c in self.children.items()}}
|
|
atomic_json(self.health,snapshot)
|
|
return snapshot
|
|
|
|
def close(self):
|
|
for did in list(self.children):self.stop(did,force=True)
|
|
|
|
|
|
def main():
|
|
parser=argparse.ArgumentParser(description=__doc__);parser.add_argument('--config',required=True)
|
|
parser.add_argument('--once',action='store_true');args=parser.parse_args()
|
|
cfg=json.loads(Path(args.config).read_text(encoding='utf-8-sig'));fleet=Fleet(cfg)
|
|
if os.name=='posix':os.umask(0o077)
|
|
client=Client(cfg['control'])
|
|
with session_lock('decision-fleet:'+str(fleet.root)):
|
|
try:
|
|
while True:
|
|
try:fleet.tick(client.call('/api/decision-bridge-bindings')['bindings'])
|
|
except Exception:
|
|
atomic_json(fleet.health,{'status':'failed','updated':time.time(),'reason':'Inspect decision bindings and private bridge state'})
|
|
if args.once:raise
|
|
if args.once:return
|
|
time.sleep(5)
|
|
finally:fleet.close()
|
|
|
|
|
|
if __name__=='__main__':main()
|