131 lines
7.3 KiB
Python
131 lines
7.3 KiB
Python
"""Allowlisted failure details: never response text, headers or provider errors."""
|
|
REASONS = {'stop','length','content_filter','tool_calls','function_call','MAX_TOKENS','SAFETY','STOP','OTHER','unknown','missing'}
|
|
COUNTS = {'input_tokens','output_tokens','reasoning_tokens','cached_input_tokens',
|
|
'cache_created_tokens','cache_ttl_seconds','response_bytes','output_characters','max_output_tokens',
|
|
'parse_line','parse_column'}
|
|
KEYWORDS = {'type','required','additionalProperties','enum','const','pattern','minItems','maxItems','minimum','maximum','anyOf','oneOf'}
|
|
VALIDATION_FIELDS = frozenset({'assessment','semantic_score','reason','update','confidence',
|
|
'working_brief','format_version','findings','retired','revision_reason','direction',
|
|
'text','scope','evidence_refs','id','uncertainties','next_question','resolve_questions',
|
|
'add_questions','next','kind','operation','arguments','executor','question','steps',
|
|
'prerequisites','requested_observations','environment','countercheck_refs','path',
|
|
'start_line','end_line','limit','offset','table','columns','filters','column','operator',
|
|
'value','text','case_sensitive','path_prefix','source_ref','source_refs','note','causal_review'})
|
|
|
|
|
|
def schema_feedback(error, _depth=0, *, include_keyword=False):
|
|
"""Schema location/required field names only; never echo response values."""
|
|
if _depth < 8 and error.validator in {'oneOf','anyOf'} and isinstance(error.instance,dict):
|
|
branches=error.schema.get(error.validator,[])
|
|
matching=[]
|
|
for index,branch in enumerate(branches):
|
|
if not isinstance(branch,dict):continue
|
|
properties=branch.get('properties',{})
|
|
compared=False;matches=True
|
|
for discriminator in ('kind','operation'):
|
|
rule=properties.get(discriminator,{})
|
|
if not isinstance(rule,dict):continue
|
|
if discriminator not in error.instance:continue
|
|
if 'const' in rule:
|
|
compared=True;matches &= error.instance[discriminator]==rule['const']
|
|
elif 'enum' in rule:
|
|
compared=True;matches &= error.instance[discriminator] in rule['enum']
|
|
if compared and matches:matching.append(index)
|
|
if len(matching)==1:
|
|
# Reuse errors resolved against the full schema; revalidating an
|
|
# isolated branch loses its root $defs and original field path.
|
|
nested=next((child for child in error.context if child.schema_path
|
|
and child.schema_path[0]==matching[0]),None)
|
|
if nested is not None:
|
|
return schema_feedback(nested,_depth+1,include_keyword=include_keyword)
|
|
path = '/'.join(str(p) if type(p) is int and 0 <= p <= 1000000 else
|
|
p if isinstance(p,str) and p in VALIDATION_FIELDS else '?'
|
|
for p in list(error.absolute_path)[:8])
|
|
fields = []
|
|
if error.validator == 'required' and isinstance(error.instance,dict):
|
|
fields = [f for f in error.validator_value if f not in error.instance and f in VALIDATION_FIELDS][:8]
|
|
elif error.validator == 'additionalProperties' and isinstance(error.instance,dict):
|
|
fields = sorted((set(error.instance)-set(error.schema.get('properties',{}))) & VALIDATION_FIELDS)[:8]
|
|
result={'validation_path':'/'+path, 'validation_fields':fields}
|
|
if include_keyword and error.validator in KEYWORDS:
|
|
result['validation_keyword']=error.validator
|
|
return result
|
|
|
|
|
|
def count(value):
|
|
return value if type(value) is int and 0 <= value <= 10**12 else None
|
|
|
|
|
|
def sanitize(data):
|
|
if not isinstance(data,dict):
|
|
return {}
|
|
result = {k:count(v) for k,v in data.items() if k in COUNTS}
|
|
if isinstance(data.get('cache_status'), str) and data['cache_status'] in {'disabled','below_minimum','created','reused','fallback'}:
|
|
result['cache_status'] = data['cache_status']
|
|
if 'finish_reason' in data:
|
|
value = data['finish_reason']
|
|
result['finish_reason'] = value if isinstance(value,str) and value in REASONS else ('missing' if value is None else 'unknown')
|
|
for key in ('has_refusal','has_tool_calls'):
|
|
if type(data.get(key)) is bool:
|
|
result[key] = data[key]
|
|
if isinstance(data.get('validation_keyword'),str) and data['validation_keyword'] in KEYWORDS:
|
|
result['validation_keyword'] = data['validation_keyword']
|
|
for key, allowed in {
|
|
'response_stage':{'http_envelope','candidate','output_json'},
|
|
'response_error':{'invalid_json','json_policy','invalid_structure','root_not_object'},
|
|
}.items():
|
|
if isinstance(data.get(key),str) and data[key] in allowed:
|
|
result[key] = data[key]
|
|
path=data.get('validation_path')
|
|
if isinstance(path,str) and len(path)<=240 and path.startswith('/'):
|
|
tokens=path[1:].split('/')
|
|
if len(tokens)<=8 and all(t in VALIDATION_FIELDS or t in {'','?'} or t.isascii() and t.isdigit() and len(t)<=7 for t in tokens):
|
|
result['validation_path']=path
|
|
fields=data.get('validation_fields')
|
|
if isinstance(fields,list):
|
|
result['validation_fields']=[f for f in fields if isinstance(f,str) and f in VALIDATION_FIELDS][:8]
|
|
return result
|
|
|
|
|
|
def parse_response_object(raw, *, stage):
|
|
"""Strict JSON parsing with bounded diagnostics, never exception text."""
|
|
import json
|
|
from viano_bug_analysis.package.manifest import parse_manifest
|
|
from .errors import LLMError
|
|
try:
|
|
result = parse_manifest(raw)
|
|
except json.JSONDecodeError as error:
|
|
raise LLMError('LLM_RESPONSE', diagnostics={'response_stage':stage,
|
|
'response_error':'invalid_json','parse_line':error.lineno,
|
|
'parse_column':error.colno}) from None
|
|
except (ValueError, UnicodeError, RecursionError):
|
|
raise LLMError('LLM_RESPONSE', diagnostics={'response_stage':stage,
|
|
'response_error':'json_policy'}) from None
|
|
if not isinstance(result,dict):
|
|
raise LLMError('LLM_RESPONSE', diagnostics={'response_stage':stage,
|
|
'response_error':'root_not_object'})
|
|
return result
|
|
|
|
|
|
def envelope_details(envelope, max_output_tokens, response_bytes):
|
|
result = {'max_output_tokens':max_output_tokens,'response_bytes':response_bytes}
|
|
if isinstance(envelope,dict):
|
|
usage = envelope.get('usage')
|
|
if isinstance(usage,dict):
|
|
result.update(input_tokens=usage.get('prompt_tokens'),output_tokens=usage.get('completion_tokens'))
|
|
prompt_details = usage.get('prompt_tokens_details')
|
|
if isinstance(prompt_details,dict):
|
|
result['cached_input_tokens'] = prompt_details.get('cached_tokens')
|
|
details = usage.get('completion_tokens_details')
|
|
if isinstance(details,dict):
|
|
result['reasoning_tokens'] = details.get('reasoning_tokens')
|
|
choices = envelope.get('choices')
|
|
if isinstance(choices,list) and choices and isinstance(choices[0],dict):
|
|
result['finish_reason'] = choices[0].get('finish_reason')
|
|
message = choices[0].get('message')
|
|
if isinstance(message,dict):
|
|
result.update(has_refusal=bool(message.get('refusal')),has_tool_calls=bool(message.get('tool_calls')))
|
|
if isinstance(message.get('content'),str):
|
|
result['output_characters'] = len(message['content'])
|
|
return sanitize(result)
|