172 lines
10 KiB
Python
172 lines
10 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,reconcile_completed
|
|
|
|
|
|
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]
|
|
# Single-cycle children exit naturally; never interrupt a native writer.
|
|
if any(p.poll() is None for p in child['processes'].values()):return False
|
|
for stream in child['logs'].values():stream.close()
|
|
del self.children[did]
|
|
return True
|
|
|
|
def tick(self,bindings):
|
|
now=time.time();problems=[];live={};threads=set();sessions=set()
|
|
journal_path=self.root/'schedule.json'
|
|
journal=json.loads(journal_path.read_text(encoding='utf-8')) if journal_path.exists() else {}
|
|
notifications={}
|
|
if self.cfg.get('notification_state_file'):
|
|
notification_path=Path(self.cfg['notification_state_file'])
|
|
if notification_path.exists():notifications=json.loads(notification_path.read_text(encoding='utf-8')).get('deliveries',{})
|
|
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)
|
|
child=self.children[did]
|
|
mirror=child['processes'].get('mirror')
|
|
if mirror and mirror.poll()==0:
|
|
journal.setdefault(did,{}).setdefault('fact_synced',{}).update(child.get('terminal_facts',{}))
|
|
if binding and child_config(self.cfg,binding,self.root)[0]!=self.children[did]['cfg']:
|
|
problems.append('binding_changed:'+did);continue
|
|
if not self.stop(did) and (not binding or binding.get('archive_status')=='archived'):
|
|
problems.append('archive_requires_reconciliation:'+did)
|
|
candidates=[]
|
|
for did,binding in live.items():
|
|
if binding.get('archive_status')=='archived':continue
|
|
candidate,runtime=child_config(self.cfg,binding,self.root)
|
|
demand=binding.get('demand')
|
|
state_path=Path(candidate['state_file'])
|
|
local_busy=busy(state_path)
|
|
worker=self.children.get(did,{}).get('processes',{}).get('worker')
|
|
worker_alive=worker is not None and worker.poll() is None
|
|
if local_busy and not worker_alive and demand is not None and not demand.get('uncertain',0) and not demand.get('inflight',0):
|
|
try:
|
|
identity=json.dumps({'control':self.cfg['control'],'session_id':binding['session_id']},sort_keys=True)
|
|
with session_lock(identity):
|
|
if reconcile_completed(state_path,Client(self.cfg['control']),binding['session_id'],binding['thread_id']):local_busy=False
|
|
except Exception:pass
|
|
if demand is None:
|
|
problems.append('queue_summary_unavailable:'+did);demand={}
|
|
uncertain=bool(demand.get('uncertain',0) or local_busy)
|
|
facts={key:r for key,r in notifications.items() if r.get('decision_id')==did and r.get('event_kind')=='fact_answered'}
|
|
terminal_facts={key:str(r.get('completed_at') or r.get('turn_id') or 'done') for key,r in facts.items() if r.get('processing_done')}
|
|
synced=journal.get(did,{}).get('fact_synced',{})
|
|
fact_demand=any(not r.get('processing_done') for r in facts.values()) or any(synced.get(key)!=value for key,value in terminal_facts.items())
|
|
needed=bool(demand.get('queued',0) or demand.get('inflight',0) or uncertain or fact_demand)
|
|
if not needed:
|
|
if did in journal:journal[did].pop('waiting_since',None)
|
|
continue
|
|
record=journal.setdefault(did,{})
|
|
record.setdefault('waiting_since',min(now,demand.get('oldest_queued_at') or now))
|
|
if did in self.children:
|
|
if uncertain and not worker_alive:problems.append('execution_requires_reconciliation:'+did)
|
|
continue
|
|
# Finite human boost: old ordinary work eventually precedes new human work.
|
|
score=now-record['waiting_since']+(60 if demand.get('human_queued',0) else 0)
|
|
candidates.append((-score,record['waiting_since'],record.get('last_served',0),did,binding,candidate,runtime,uncertain,terminal_facts))
|
|
candidates.sort(key=lambda row:row[:4]);waiting=0
|
|
for _,_,_,did,binding,candidate,runtime,uncertain,terminal_facts in candidates:
|
|
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(encoding='utf-8'))!=candidate:
|
|
problems.append('saved_binding_changed:'+did);continue
|
|
atomic_json(config_path,candidate)
|
|
child={'cfg':candidate,'runtime':runtime,'path':config_path,'processes':{},'logs':{},'terminal_facts':terminal_facts}
|
|
self.children[did]=child
|
|
journal[did].update(waiting_since=now,last_served=now)
|
|
atomic_json(journal_path,journal)
|
|
roles=('mirror',) if uncertain or binding.get('demand',{}).get('inflight',0) or not binding.get('demand',{}).get('queued',0) else ('worker','mirror')
|
|
if uncertain:problems.append('execution_requires_reconciliation:'+did)
|
|
for role in roles:
|
|
stream=open(runtime/(role+'.log'),'a',encoding='utf-8');child['logs'][role]=stream
|
|
command=[sys.executable,'-X','utf8',str(Path(__file__).with_name(role+'.py')),'--config',str(config_path),'--once']
|
|
child['processes'][role]=subprocess.Popen(command,stdout=stream,stderr=stream,
|
|
**({'creationflags':subprocess.CREATE_NO_WINDOW} if os.name=='nt' else {}))
|
|
atomic_json(journal_path,journal)
|
|
snapshot={'status':'failed' if problems else 'ok','updated':now,'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):
|
|
# Active cycles retain their durable state and host writer locks.
|
|
for did in list(self.children):
|
|
for stream in self.children[did]['logs'].values():stream.close()
|
|
self.stop(did)
|
|
|
|
|
|
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()
|