191 lines
9.9 KiB
Python
191 lines
9.9 KiB
Python
"""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 measure_migration(path):
|
|
"""Actual v13 migration view; no made-up historical working summaries."""
|
|
rows = []
|
|
with PackageReader(path) as reader:
|
|
for ref in reader.files:
|
|
if not (ref.startswith('analysis/investigation/') and '/contexts/' in ref):
|
|
continue
|
|
context = json.loads(reader.read_bytes(ref))
|
|
if context.get('prompt_version') not in {'investigation_project_v11','investigation_project_v12'}:
|
|
continue
|
|
def sizes(request):
|
|
system, prefix, live = GeminiProvider._messages(request, stable_case_context=True)
|
|
return {'message_bytes':sum(len(m['content'].encode('utf8')) for m in request['messages']),
|
|
'prefix_bytes':size([system,prefix]), 'live_bytes':size(live)}
|
|
rows.append({'context':ref, 'old':sizes(build_prompt(context)),
|
|
'v13_migration':sizes(build_prompt(dict(context,prompt_version='investigation_project_v13')))})
|
|
old = sum(r['old']['live_bytes'] for r in rows)
|
|
new = sum(r['v13_migration']['live_bytes'] for r in rows)
|
|
return {'measurement':'production_v13_migration_offline', 'samples':rows,
|
|
'old_live_bytes':old, 'new_live_bytes':new,
|
|
'live_reduction_percent':round(100*(1-new/old),2) if old else None,
|
|
'limitations':['No old note history removed: historical runs have no persisted working brief.',
|
|
'Includes actual v13 schema/instructions; no invented working summaries.',
|
|
'No paid requests or token/quality claim.']}
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('package', type=Path)
|
|
parser.add_argument('--output', required=True, type=Path)
|
|
parser.add_argument('--production-v13', action='store_true', help='Measure actual v13 migration instead of speculative views')
|
|
args = parser.parse_args()
|
|
if args.production_v13:
|
|
report = measure_migration(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({k:v for k,v in report.items() if k!='samples'}))
|
|
return
|
|
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()
|