"""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({'retire','finding_id','intent','hypothesis_id','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', 'next_step','goal','supports_if','weakens_if','investigate_if','profile','entity_refs'}) 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','json_error_at_end'): 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 { 'json_error_kind':{'missing_delimiter','missing_key','unterminated_string','invalid_escape','extra_data','unexpected_value','other'}, '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: kind=('missing_delimiter' if 'delimiter' in error.msg else 'missing_key' if error.msg.startswith('Expecting property name') else 'unterminated_string' if error.msg.startswith('Unterminated string') else 'invalid_escape' if error.msg.startswith('Invalid') else 'extra_data' if error.msg=='Extra data' else 'unexpected_value' if error.msg=='Expecting value' else 'other') raise LLMError('LLM_RESPONSE', diagnostics={'response_stage':stage, 'response_error':'invalid_json','parse_line':error.lineno, 'parse_column':error.colno,'json_error_kind':kind, 'json_error_at_end':error.pos>=len(error.doc.rstrip())}) 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)