File size: 7,017 Bytes
aa73ff9 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | """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():
# Read a bounded tail; report only known code fields.
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']=[]
# Import the actual installed client only after reporting missing files.
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()
|