import json from importlib.resources import files from viano_bug_analysis.lifecycle.storage import json_bytes PROMPT_VERSION="investigation_v1" def output_schema(context=None, *, strict=False): if context and context.get('lean_protocol')==2: from .investigation_lean_v20 import schema return schema(context,strict=strict) if context and "lean_session" in context: from .investigation_lean import schema return schema(context,strict=strict) name = "investigation-project-output-v1.schema.json" if context and "x-project-bindings" in context["tool_schema"] else "investigation-output-v1.schema.json" schema=json.loads(files("viano_bug_analysis.schemas").joinpath(name).read_text()) if context and context['tool_schema'].get('x-project-bindings', {}).get('relation_profiles'): from viano_bug_analysis.tools.database_relations import extend_tool_schema schema['properties']['next'] = extend_tool_schema(schema['properties']['next']) if context and context['tool_schema'].get('x-project-bindings', {}).get('source_analysis'): from viano_bug_analysis.tools.source_context import extend_tool_schema protocol=context['tool_schema']['x-project-bindings']['source_analysis']['version'] schema['properties']['next'] = extend_tool_schema(schema['properties']['next'],protocol) if context and context.get('prompt_version') in {'investigation_project_v12','investigation_project_v13','investigation_project_v14','investigation_project_v15','investigation_project_v16','investigation_project_v17','investigation_project_v18'}: schema['properties']['update']['properties']['causal_review'] = json.loads( files('viano_bug_analysis.schemas').joinpath('investigation-causal-review-v1.schema.json').read_text()) schema['properties']['update']['required'].append('causal_review') if context and context.get('prompt_version') == 'investigation_project_v13': schema['properties']['update']['properties']['working_brief'] = json.loads( files('viano_bug_analysis.schemas').joinpath('investigation-working-brief-v1.schema.json').read_text()) schema['properties']['update']['properties']['working_brief']['properties'].pop('next_question_details',None) schema['properties']['update']['required'].append('working_brief') schema['properties']['update']['properties']['note']['maxLength'] = 400 if context and context.get("prompt_version") in {"investigation_v2","investigation_project_v3","investigation_v3","investigation_project_v4","investigation_project_v5","investigation_project_v6","investigation_project_v7","investigation_project_v8","investigation_project_v9","investigation_project_v10","investigation_project_v11","investigation_project_v12","investigation_project_v13","investigation_project_v14","investigation_project_v15","investigation_project_v16","investigation_project_v17","investigation_project_v18"}: schema["properties"]["next"]["oneOf"].extend(json.loads(files("viano_bug_analysis.schemas").joinpath("investigation-early-actions-v1.schema.json").read_text())["oneOf"]) if context and context.get('prompt_version') in {'investigation_project_v8','investigation_project_v9'}: schema['properties']['next']['oneOf'].append({ 'type':'object','additionalProperties':False,'required':['kind','evidence_refs'], 'properties':{'kind':{'const':'recall'},'evidence_refs':{ 'type':'array','minItems':1,'maxItems':4,'uniqueItems':True, 'items':{'type':'string','pattern':'^E[0-9]{6,}$'}}}}) if context and context.get('prompt_version') in {'investigation_project_v14','investigation_project_v15','investigation_project_v16','investigation_project_v17','investigation_project_v18'}: schema['properties']['update'] = json.loads(files('viano_bug_analysis.schemas').joinpath('investigation-update-v14.schema.json').read_text()) for branch in schema['properties']['assessment']['anyOf']: if branch.get('type') == 'object': branch['properties'].pop('evidence_refs') branch['required'].remove('evidence_refs') if context and context.get('prompt_version') in {'investigation_project_v15','investigation_project_v16','investigation_project_v17','investigation_project_v18'}: branches=schema['properties']['next']['oneOf'] operations=[b['properties']['operation']['const'] for b in branches if b['properties']['kind'].get('const')=='tool'] if not strict: branches=[b for b in branches if b['properties']['kind'].get('const')!='tool'] branches.append({'type':'object','additionalProperties':False,'required':['kind','operation','arguments'], 'properties':{'kind':{'const':'tool'},'operation':{'enum':operations},'arguments':{'type':'object'}}}) branches.append({'type':'object','additionalProperties':False,'required':['kind','evidence_refs'], 'properties':{'kind':{'const':'recall'},'evidence_refs':{'type':'array','minItems':1,'maxItems':4, 'uniqueItems':True,'items':{'type':'string','pattern':'^E[0-9]{6,}$'}}}}) schema['properties']['next']['oneOf']=branches if not strict: schema['properties']['next']={'type':'object','required':['kind'], 'properties':{'kind':{'enum':['tool','test','finish','reject','request_diagnosis','defer','recall']}}} if not strict: def compact(value): if isinstance(value,list):return [compact(v) for v in value] if not isinstance(value,dict):return value return {k:compact(v) for k,v in value.items() if k not in {'$schema','pattern','minLength','maxLength','uniqueItems'}} schema=compact(schema) if context and context.get('prompt_version') in {'investigation_project_v16','investigation_project_v17','investigation_project_v18'}: from .investigation_compact import preparing_review if preparing_review(context): schema['properties']['next'] = {'type':'object','additionalProperties':False, 'required':['kind','reason'], 'properties':{ 'kind':{'enum':['defer','reject']}, 'reason':{'type':'string','minLength':1,'maxLength':2000}}} schema['properties']['update']['properties']['add_questions']['maxItems'] = 0 if context and context.get('prompt_version') == 'investigation_project_v18': from .investigation_compact import preparing_review if not preparing_review(context): schema['properties']['update']['properties']['next_step'] = json.loads( files('viano_bug_analysis.schemas').joinpath('investigation-next-step-v1.schema.json').read_text()) schema['properties']['update']['required'].append('next_step') return schema def build_prompt(context): if context.get('response_repair'): from viano_bug_analysis.llm.response_recovery import repair_request return repair_request(context) request = _build_prompt(context) from .tool_guidance import instructions, HEADER guide = instructions(context) if guide and HEADER not in request['messages'][0]['content']: request['messages'][0]['content'] += guide from .tool_guidance import finding_repair_instructions request['messages'][0]['content'] += finding_repair_instructions(context) if context.get('policy',{}).get('response_recovery_version')==1: from viano_bug_analysis.llm.response_recovery import MARKER request['output_schema'][MARKER]=1 body=json.loads(request['messages'][1]['content']) if 'output_schema' in body:body['output_schema']=request['output_schema'] request['messages'][1]['content']=json_bytes(body).decode('utf-8') if context.get('format_notice'): body=json.loads(request['messages'][1]['content']) body['format_notice']={'normalized_previous_reply':context['format_notice'], 'instruction':'Use ordinary JSON objects next time; no extra research or reply needed.'} request['messages'][1]['content']=json_bytes(body).decode('utf-8') return request def _build_prompt(context): if context.get('lean_protocol')==2: from .investigation_lean_v20 import build return build(context) if "lean_session" in context: from .investigation_lean import build return build(context) schema=output_schema(context) version = "investigation_project_v1" if "x-project-bindings" in context["tool_schema"] else PROMPT_VERSION if context.get('prompt_version') in {'investigation_project_v2','investigation_project_v3','investigation_project_v4','investigation_project_v5','investigation_project_v6','investigation_project_v7','investigation_project_v8','investigation_project_v9','investigation_project_v10','investigation_project_v11','investigation_project_v12','investigation_project_v13','investigation_project_v14','investigation_project_v15','investigation_project_v16','investigation_project_v17','investigation_project_v18'}: from .investigation_view import working_view from .model_context import model_json version = context['prompt_version'] if version in {'investigation_project_v15','investigation_project_v16','investigation_project_v17','investigation_project_v18'}: from .investigation_compact import model_view visible = model_view(context) elif version in {'investigation_project_v13','investigation_project_v14'}: from .working_brief import model_view visible = model_view(context) elif version in {'investigation_project_v10','investigation_project_v11','investigation_project_v12'}: from .investigation_focus import full_history_view visible = full_history_view(context) elif version in {'investigation_project_v8','investigation_project_v9'}: from .investigation_focus import focused_view visible = focused_view(context) elif version == 'investigation_project_v7': from .investigation_blocks import block_view visible = block_view(context) elif version == 'investigation_project_v6': from .investigation_consolidation import consolidated_view visible = consolidated_view(context) elif version == 'investigation_project_v5': from .investigation_view import thin_view visible = thin_view(context) else: visible = working_view(context) encode = model_json else: visible = compact_context(context) if version != PROMPT_VERSION else context encode = lambda value: json_bytes(value).decode() if context.get("prompt_version") in {"investigation_v2","investigation_v3"}:version=context["prompt_version"] prompt_file = version task = 'investigate_hypothesis' if version in {'investigation_project_v16','investigation_project_v17','investigation_project_v18'}: from .investigation_compact import preparing_review, review_view if preparing_review(context): visible = review_view(context) prompt_file = 'investigation_review_v16' task = 'assess_pending_result_for_early_review' elif version == 'investigation_project_v17': from .investigation_step import step_view visible['investigation_step'] = step_view(context) task = 'advance_selected_question' elif version == 'investigation_project_v18': from .investigation_plan import framed_view visible = framed_view(context,visible) task = 'advance_investigation_goal' bootstrap_version='investigation_bootstrap_v22' if context.get('bootstrap_framing_version')==2 else 'investigation_bootstrap_v21' return {"prompt_version":bootstrap_version if context.get("bootstrap_json_framing") else version,"privacy":"configured","output_schema":schema, "messages":[{"role":"system","content":files("viano_bug_analysis.llm.prompts").joinpath(prompt_file+".txt").read_text(encoding="utf-8")}, {"role":"user","content":encode({"task":task,"input":visible,"output_schema":schema})}]} def compact_context(context): """A reproducible prompt view; the full immutable context stays in the case. Retain every evidence ID and provenance. Only the selected hypothesis's evidence carries payload; other hypotheses are available as bounded summaries. The previous result is already in evidence and need not be sent twice. """ import copy view=copy.deepcopy(context) hid=context["selected_hypothesis"] view["case"].pop("inputs",None) quality=view["case"].get("data_quality",{}) quality.pop("unclaimed_files",None) for block in ("source_context","database_context"): refs=view["case"][block]["refs"] view["case"][block]={"reference_count":len(refs), "refs":refs[:12], "references_truncated":len(refs)>12} for other, hypothesis in view["hypotheses"].items(): if other != hid: hypothesis["notes"] = hypothesis["notes"][-2:] for evidence_id,entry in view["evidence"].items(): if entry["record"]["hypothesis_id"] != hid: entry["content"]={"kind":entry["content"]["kind"],"scope":entry["content"]["scope"], "coverage":entry["content"]["coverage"],"items_omitted":True} previous=context["previous_result"] if previous: view["previous_result"]={"evidence_id":previous["record"]["id"],"see":"evidence"} view["tool_schema"]={"project":context["tool_schema"]["x-project-bindings"], "request_arguments":"Use the schema of next/tool in output_schema"} return view