Study focused investigation working context offline

scripts/study_investigation_context.py

@@ -0,0 +1,157 @@
+"""Offline design study, never a live prompt protocol or an LLM call.
+
+Keep evidence intact in the conservative variant. The fresh-only variant is an
+explicitly unsafe size experiment, not a deployment recommendation.
+"""
+import argparse
+import copy
+import json
+from pathlib import Path
+
+from viano_bug_analysis.analysis.investigation_prompt import build_prompt
+from viano_bug_analysis.analysis.model_context import model_json
+from viano_bug_analysis.llm.gemini import GeminiProvider
+from viano_bug_analysis.package import PackageReader
+
+
+def size(value):
+    return len(model_json(value).encode('utf-8'))
+
+
+def brief(hypothesis):
+    notes = hypothesis.get('notes', [])
+    reviews = [n for n in notes if n.get('causal_review')]
+    review = reviews[-1].get('causal_review') if reviews else None
+    assessments = [n for n in notes if n.get('kind') == 'assessment']
+    return {
+        'claim': hypothesis['proposal']['mechanism'],
+        'confidence': hypothesis['confidence'],
+        'causal_balance': copy.deepcopy(review) if review else {
+            'status': 'not_recorded_in_this_legacy_run',
+            'meaning': 'No support/against summary can be inferred from confidence or novelty.'},
+        'last_working_note': assessments[-1]['text'] if assessments else None,
+        'note_status': 'prior_model_claim_not_independent_evidence',
+        'open_questions': [q for q in hypothesis.get('questions', []) if not q.get('resolved')],
+    }
+
+
+def administrative_trim(document):
+    """Drop controller bookkeeping/history, preserve every research payload.
+
+    Unknown fields are retained. Deliberate note-history omissions are explicit;
+    this is not advertised as lossless. Stable case fields remain identical.
+    """
+    result = copy.deepcopy(document)
+    view = result['input']
+    selected = view['selected_hypothesis']
+    for hid, h in view['hypotheses'].items():
+        if hid != selected:
+            view['hypotheses'][hid] = {
+                'proposal': h['proposal'], 'confidence': h['confidence'],
+                'disposition': h['disposition']}
+        else:
+            h['task_brief'] = brief(h)
+            # The last note is in the brief; resolved question IDs/text stay.
+            h['omitted_note_count'] = len(h.pop('notes', []))
+        view['hypotheses'][hid].pop('last_record_ref', None)
+        view['hypotheses'][hid].pop('invalid_responses', None)
+        view['hypotheses'][hid].pop('iterations', None)
+    for entry in view['evidence'].values():
+        entry.pop('record_ref', None)
+        record = entry['record']
+        for key in ('format', 'format_version', 'id', 'case_id', 'package_id',
+                    'analysis_revision', 'content_ref', 'content_sha256',
+                    'novelty_score', 'duplicate_of', 'action_id', 'origin'):
+            record.pop(key, None)
+        # Scope, queries, snapshot/deployment provenance and result types remain
+        # in the unmodified content and shared source/result tables.
+    budget = view['budget']
+    account = budget['accounts'][selected]
+    view['budget'] = {'selected_account': account}
+    return result
+
+
+def fresh_only(document):
+    """Deliberate omission experiment: exposes why a working memory is needed."""
+    result = administrative_trim(document)
+    view = result['input']
+    previous = view.get('previous_result')
+    keep = {previous['evidence_id']} if previous else set()
+    omitted = set(view['evidence']) - keep
+    view['evidence'] = {eid: entry for eid, entry in view['evidence'].items() if eid in keep}
+    # Tables remain intact: removing referenced material would distort the size
+    # comparison and risk losing fresh-result scope/metadata. This is conservative.
+    view['omitted_evidence_ids'] = sorted(omitted)
+    view['omission_warning'] = 'Offline experiment: old evidence payloads unavailable; not safe to deploy.'
+    return result, sorted(omitted)
+
+
+def transport(req, document):
+    changed = copy.deepcopy(req)
+    changed['messages'][1]['content'] = model_json(document)
+    system, prefix, live = GeminiProvider._messages(changed, stable_case_context=True)
+    return {'message_bytes': sum(len(m['content'].encode()) for m in changed['messages']),
+            'live_transport_bytes': size(live), 'prefix_transport_bytes': size([system, prefix])}
+
+
+def measure(path):
+    rows = []
+    example = None
+    with PackageReader(path) as reader:
+        identity = {k: reader.manifest[k] for k in ('case_id', 'package_id', 'package_generation')}
+        for name in reader.files:
+            if not (name.startswith('analysis/investigation/') and '/contexts/' in name):
+                continue
+            context = json.loads(reader.read_bytes(name))
+            if context.get('prompt_version') not in {'investigation_project_v11', 'investigation_project_v12'}:
+                continue
+            req = build_prompt(context)
+            doc = json.loads(req['messages'][1]['content'])
+            trimmed = administrative_trim(doc)
+            unsafe, omitted = fresh_only(doc)
+            assert doc['input']['evidence'].keys() == trimmed['input']['evidence'].keys()
+            for eid, original in doc['input']['evidence'].items():
+                assert original['content'] == trimmed['input']['evidence'][eid]['content']
+            for key in ('shared_sources', 'shared_results', 'shared_metadata', 'sources', 'interpretation'):
+                assert doc['input'].get(key) == trimmed['input'].get(key)
+            row = {'context':name, 'hypothesis':context['selected_hypothesis'],
+                'evidence_count':len(context['evidence']),
+                'baseline':transport(req, doc), 'administrative_trim':transport(req, trimmed),
+                'fresh_only_unsafe':transport(req, unsafe), 'unsafe_omitted_evidence':omitted,
+                'selected_notes':len(context['hypotheses'][context['selected_hypothesis']]['notes'])}
+            rows.append(row)
+            if example is None or row['evidence_count'] > example['evidence_count']:
+                h = trimmed['input']['hypotheses'][context['selected_hypothesis']]
+                example = {'context':name, 'evidence_count':row['evidence_count'],
+                           'selected_hypothesis':context['selected_hypothesis'],
+                           'task_brief':h['task_brief']}
+    totals = {variant:{unit:sum(r[variant][unit] for r in rows)
+        for unit in ('message_bytes', 'live_transport_bytes', 'prefix_transport_bytes')}
+        for variant in ('baseline', 'administrative_trim', 'fresh_only_unsafe')}
+    for variant in ('administrative_trim', 'fresh_only_unsafe'):
+        base = totals['baseline']['live_transport_bytes']
+        totals[variant]['live_reduction_percent'] = round(100*(1-totals[variant]['live_transport_bytes']/base),2) if base else None
+    return {'identity':identity, 'contexts':len(rows), 'totals':totals,
+            'largest_evidence_example':example, 'samples':rows}
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('package', type=Path)
+    parser.add_argument('--output', required=True, type=Path)
+    args = parser.parse_args()
+    report = {'study':'offline_investigation_context_v1',
+        'limitations':['Not a runnable prompt protocol; production unchanged.',
+            'No quality test or token/cost claim; bytes are UTF-8.',
+            'Old note history removed in both variants; no semantic summary fabricated.',
+            'Fresh-only deliberately lacks evidence; not recommended for deployment.',
+            'Historical v11 has no structured causal review; a reliable pro/con brief needs a new persisted contract.',
+            'Same original system/schema for fair data-size comparison; a live protocol needs consistent revised instructions.'],
+        **measure(args.package)}
+    args.output.parent.mkdir(parents=True, exist_ok=True)
+    args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2)+'\n', encoding='utf8')
+    print(json.dumps(report['totals']))
+
+
+if __name__ == '__main__':
+    main()