#!/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()