| """Redacted diagnostics for Claude → LiteLLM → Laya; never executes a tool.""" |
| import argparse |
| import json |
| import os |
| import re |
| import shutil |
| import subprocess |
| import sys |
| import urllib.error |
| import urllib.request |
| from pathlib import Path |
|
|
| HERE=Path(__file__).resolve().parent |
|
|
| def classify_error(error): |
| if isinstance(error,urllib.error.HTTPError):return 'http_'+str(error.code) |
| if isinstance(error,TimeoutError):return 'timeout' |
| if isinstance(error,urllib.error.URLError):return 'network_error' |
| if isinstance(error,FileNotFoundError):return 'missing_file' |
| if isinstance(error,json.JSONDecodeError):return 'invalid_json' |
| if isinstance(error,ValueError) and str(error)=='Missing API credential':return 'missing_credential' |
| return 'invalid_configuration_or_response' |
|
|
| def diagnose(config_path,settings_path,probe=False): |
| report={'python_version':sys.version.split()[0],'runtime_files':{name:(config_path.parent/name).is_file() for name in ('hook.py','global_hook.py','questions.json','config.json')},'probe_executes_command':False} |
| duplicates=[] |
| def pairs(items): |
| result={} |
| for k,v in items: |
| if k in result:duplicates.append(k) |
| result[k]=v |
| return result |
| try:settings=json.loads(settings_path.read_text(),object_pairs_hook=pairs) |
| except Exception as error:settings={};report['settings_error']=classify_error(error) |
| report['duplicate_json_keys']=sorted(set(duplicates)) |
| hooks=settings.get('hooks',{}) |
| report['hooks_in_hooks_section']={event:any('laya/global_hook.py' in h.get('command','') for group in hooks.get(event,[]) for h in group.get('hooks',[])) for event in ('UserPromptSubmit','PreToolUse')} |
| report['misplaced_top_level_hook_keys']=[name for name in ('PreToolUse','UserPromptSubmit') if name in settings] |
| report['disableAllHooks']=bool(settings.get('disableAllHooks')) |
| report['bash_matcher_present']=any(group.get('matcher') in ('Bash','^Bash$','.*','*',None) and any('global_hook.py' in h.get('command','') for h in group.get('hooks',[])) for group in hooks.get('PreToolUse',[])) |
| executable=shutil.which('claude');report['claude_found']=bool(executable) |
| if executable: |
| try: |
| version=subprocess.check_output([executable,'--version'],text=True,timeout=5) |
| found=re.search(r'\d+\.\d+\.\d+',version) |
| report['claude_version']=found[0] if found else 'unrecognized' |
| except Exception:report['claude_version']='unavailable' |
| environment={**os.environ,**{k:str(v) for k,v in settings.get('env',{}).items()}} |
| report['hooks_disabled_environment']=[k for k in ('CLAUDE_CODE_SAFE_MODE','CLAUDE_CODE_SIMPLE') if environment.get(k) in ('1','true')] |
| try: |
| config=json.loads(config_path.read_text());credential_env=config.get('api_key_env') |
| if credential_env and not re.fullmatch(r'[A-Z_][A-Z0-9_]*',credential_env):raise ValueError('Invalid credential variable name') |
| report['credential_env']=credential_env |
| report['credential_present']=bool(environment.get(credential_env,'').strip()) if credential_env else bool(config.get('api_key_file') and Path(config['api_key_file']).expanduser().is_file()) |
| report['other_credential_env_names_present']=[k for k in ('ANTHROPIC_AUTH_TOKEN','ANTHROPIC_API_KEY','LITELLM_API_KEY') if environment.get(k) and k!=credential_env] |
| endpoint=config['endpoint'] |
| from urllib.parse import urlsplit |
| u=urlsplit(endpoint) |
| if u.scheme not in ('http','https') or not u.hostname or u.username or u.password or u.query or u.fragment or u.path!='/v1/systemone':raise ValueError('Invalid endpoint') |
| report['endpoint']=endpoint;report['model']=config['model'] |
| state=Path(config['state_dir']).expanduser();report['stored_contexts']=len(list(state.glob('*.json'))) |
| report['oversized_contexts']=sum(bool(json.loads(p.read_text()).get('exceeded')) for p in state.glob('*.json')) |
| path=config_path.parent/'diagnostics.jsonl' |
| if path.exists(): |
| |
| with path.open('rb') as f: |
| f.seek(max(0,path.stat().st_size-16384));lines=f.read().splitlines() |
| recent=[] |
| for line in lines[-12:]: |
| try:row=json.loads(line);recent.append({k:row[k] for k in ('utc','event','tool','status','decision','error') if k in row}) |
| except ValueError:pass |
| report['recent_hook_events']=recent |
| else:report['recent_hook_events']=[] |
| |
| if not all(report['runtime_files'].values()):raise FileNotFoundError('Incomplete runtime') |
| sys.path.insert(0,str(config_path.parent)) |
| import hook as core |
| key=environment.get(credential_env,'').strip() if credential_env else Path(config['api_key_file']).expanduser().read_text().strip() |
| if not key:raise ValueError('Missing API credential') |
| request=urllib.request.Request(endpoint.rsplit('/v1/systemone',1)[0]+'/v1/models',headers={'Authorization':'Bearer '+key}) |
| with urllib.request.build_opener(urllib.request.ProxyHandler({}),core.NoRedirect()).open(request,timeout=8) as response: |
| raw=response.read(65537) |
| if len(raw)>65536:raise ValueError('Oversized model list') |
| models=json.loads(raw) |
| report['models_endpoint']='ok';report['model_accessible']=config['model'] in {r.get('id') for r in models.get('data',[])} |
| if probe: |
| config['questions']=json.loads((config_path.parent/'questions.json').read_text()) |
| state={'tool':'Bash','command':'uptime','trusted_user_request':'Inspect the stage load status without changes or secrets.','environment':'stage','trusted_context':'Ordinary uptime output is non-sensitive operational metadata. This is a classifier connectivity test; no command is executed.'} |
| report['probe_decision']=core.query(config,state,environ=environment) |
| report['probe_status']='ok' |
| except Exception as error:report['error']=classify_error(error) |
| report['configuration_checks_passed']=not report.get('error') and all(report['hooks_in_hooks_section'].values()) and report['bash_matcher_present'] and not report['disableAllHooks'] and not report['misplaced_top_level_hook_keys'] and not report['duplicate_json_keys'] and not report['hooks_disabled_environment'] and bool(report.get('model_accessible')) |
| return report |
|
|
| def main(): |
| p=argparse.ArgumentParser(description=__doc__);p.add_argument('--config',type=Path,default=HERE/'config.json');p.add_argument('--settings',type=Path,default=Path.home()/'.claude/settings.json');p.add_argument('--probe',action='store_true');a=p.parse_args() |
| result=diagnose(a.config,a.settings,a.probe);print(json.dumps(result,ensure_ascii=False,indent=2)) |
| raise SystemExit(0 if result['configuration_checks_passed'] else 1) |
|
|
| if __name__=='__main__':main() |
|
|
