Compress investigation evidence losslessly and handle source EOF safely

scripts/measure_investigation_context.py

@@ -0,0 +1,60 @@
+"""Compare archived project investigation prompts offline; never invoke an LLM."""
+import argparse
+import copy
+import json
+
+from viano_bug_analysis.analysis.investigation_prompt import build_prompt
+from viano_bug_analysis.analysis.investigation_view import working_view, thin_view
+from viano_bug_analysis.case.contracts import read_json
+from viano_bug_analysis.package import PackageReader
+
+
+def restore_tables(view):
+    for entry in view['evidence'].values():
+        for capture in entry['content'].get('items', []):
+            result = capture.get('result', {})
+            table = result.get('items')
+            if not isinstance(table, dict):
+                continue
+            if table.get('encoding') == 'source_line_table_v1':
+                result['items'] = [{**table['common'], **dict(zip(table['columns'], row))} for row in table['rows']]
+            elif table.get('encoding') == 'database_row_table_v1':
+                result['items'] = [{**table['common'], 'row': dict(zip(table['columns'], row))} for row in table['rows']]
+    return view
+
+
+def measure(path):
+    records = []
+    with PackageReader(path) as reader:
+        for ref in sorted(reader.files):
+            if not ref.startswith('analysis/investigation/') or '/contexts/' not in ref or not ref.endswith('.json'):
+                continue
+            context = read_json(reader, ref, 'investigation-context', 4 * 1024**2)
+            if context.get('prompt_version') != 'investigation_project_v4':
+                continue
+            old = build_prompt(context)
+            projected = copy.deepcopy(context)
+            projected['prompt_version'] = 'investigation_project_v5'
+            new = build_prompt(projected)
+            old_view = working_view(projected)
+            new_view = thin_view(projected)
+            if json.dumps(restore_tables(copy.deepcopy(new_view)), sort_keys=True) != json.dumps(old_view, sort_keys=True):
+                raise ValueError('Model view roundtrip failed')
+            records.append({'context_ref': ref,
+                            'old_characters': len(old['messages'][1]['content']),
+                            'new_characters': len(new['messages'][1]['content']),
+                            'old_components': {k: len(json.dumps(v, ensure_ascii=False, separators=(',', ':'))) for k, v in old_view.items()},
+                            'new_components': {k: len(json.dumps(v, ensure_ascii=False, separators=(',', ':'))) for k, v in new_view.items()}})
+    old = sum(r['old_characters'] for r in records)
+    new = sum(r['new_characters'] for r in records)
+    return {'stored_contexts': len(records), 'old_user_characters': old, 'new_user_characters': new,
+            'reduction_percent': round(100 * (1 - new / old), 2) if old else None,
+            'lossless_roundtrip': True, 'unit': 'characters_not_provider_tokens',
+            'scope': 'stored_contexts_not_billed_calls', 'contexts': records}
+
+
+if __name__ == '__main__':
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('package')
+    args = parser.parse_args()
+    print(json.dumps(measure(args.package), indent=2))