"""Read-only offline audit; reports aggregate sizes/hashes, never source contents. All byte counts use UTF-8 or ZIP directory sizes. Potential savings overlap and must not be added together. This is a measurement, not a package migration. """ import argparse from collections import Counter, defaultdict import hashlib import io import json from pathlib import Path import tarfile import zipfile import zlib from viano_bug_analysis.package import PackageReader from viano_bug_analysis.llm.gemini import GeminiProvider def encoded(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':')).encode('utf-8') def digest(data): return hashlib.sha256(data).hexdigest() def category(name): if name.startswith('original/'): return 'original_archives' if name.endswith('/archive.tar.gz') else 'original_members' if name.startswith('llm/'): return 'llm_requests' if name.endswith('/request.json') else 'llm_other' if '/contexts/' in name: return 'analysis_contexts' return name.split('/')[0] def repeated(groups): result = [] for entries in groups.values(): if len(entries) > 1: result.append({'copies': len(entries), 'raw_duplicate_bytes': sum(e[1] for e in entries[1:]), 'compressed_duplicate_bytes': sum(e[2] for e in entries[1:]), 'paths': [e[0] for e in entries]}) return sorted(result, key=lambda v: v['raw_duplicate_bytes'], reverse=True) def audit(path): # Validate through the normal bounded reader before inspecting ZIP metadata. with PackageReader(path) as reader: manifest = reader.manifest files = set(reader.files) groups = defaultdict(list) canonical = defaultdict(list) categories = defaultdict(Counter) requests = [] archive_members = Counter() metadata_fields = Counter() deflate_estimate = Counter() compression_methods = Counter() evidence_items = {} source_lines = {} evidence_sizes = Counter() context_components = defaultdict(lambda: [0, 0, 0]) context_values = set() with zipfile.ZipFile(path) as z: infos = {v.filename: v for v in z.infolist() if not v.is_dir()} for name, info in infos.items(): data = z.read(name) compressor = zlib.compressobj(level=6, wbits=-15) compressed = compressor.compress(data) + compressor.flush() assert zlib.decompress(compressed, wbits=-15) == data deflate_size = len(compressed) deflate_estimate[category(name)] += deflate_size compression_methods[str(info.compress_type)] += 1 cat = categories[category(name)] cat.update(files=1, raw_bytes=len(data), compressed_bytes=info.compress_size) groups[digest(data)].append((name, len(data), info.compress_size)) if name.endswith('/archive.tar.gz'): prefix = name.removesuffix('archive.tar.gz') + 'members/' with tarfile.open(fileobj=io.BytesIO(data), mode='r|gz') as tar: for member in tar: if not member.isfile(): continue target = prefix + member.name if target in files and member.size == infos[target].file_size: with tar.extractfile(member) as stream: same = digest(stream.read(member.size + 1)) == digest(z.read(target)) if same: archive_members.update(files=1, raw_bytes=member.size, outer_zip_member_compressed_bytes=infos[target].compress_size) if not name.endswith('.json'): continue try: doc = json.loads(data) except (ValueError, UnicodeError): continue canonical[digest(encoded(doc))].append((name, len(data), info.compress_size)) if category(name) == 'analysis_contexts' and isinstance(doc, dict): for field, value in doc.items(): b = encoded(value) key = (field, digest(b)) context_components[field][0] += len(b) context_components[field][2] += 1 if key not in context_values: context_components[field][1] += len(b) context_values.add(key) if name.startswith('llm/operations/') and name.endswith('/request.json'): requests.append((name, doc)) by_prompt = defaultdict(list) system_groups = Counter() block_groups = defaultdict(lambda: [0, 0, 0]) per_call = [] def leaves(value, at='$'): if isinstance(value, dict): for k, v in value.items(): if isinstance(v, (dict, list)): yield from leaves(v, at + '.' + k) elif isinstance(v, str) and len(v.encode('utf-8')) >= 80: yield at + '.' + k, v elif isinstance(value, list): for v in value: yield from leaves(v, at + '[]') for name, req in requests: messages = req.get('messages', []) msg_bytes = sum(len(m['content'].encode('utf-8')) for m in messages) systems = [m['content'] for m in messages if m['role'] == 'system'] for value in systems: system_groups[(digest(value.encode()), len(value.encode()))] += 1 user_docs = [] strings = defaultdict(list) keys = Counter() def count_keys(value): if isinstance(value, dict): for k, v in value.items(): keys[k] += len(encoded(k)) + 1 count_keys(v) elif isinstance(value, list): for v in value: count_keys(v) for m in messages: if m['role'] != 'user': continue try: doc = json.loads(m['content']) except ValueError: continue user_docs.append(doc) view = doc.get('input', {}) if isinstance(doc, dict) else {} if isinstance(view, dict): for item in view.get('evidence', {}).values() if isinstance(view.get('evidence'), dict) else []: evidence_items[digest(encoded(item))] = (evidence_items.get(digest(encoded(item)), (0, len(encoded(item))))[0] + 1, len(encoded(item))) if isinstance(item, dict): for field, value in item.items(): evidence_sizes[field] += len(encoded(value)) shared = view.get('shared_sources', {}) for source in shared.values() if isinstance(shared, dict) else []: if not isinstance(source, dict): continue meta = source.get('metadata', {}) identity = digest(encoded(meta)) for block in source.get('blocks', []): for offset, line in enumerate(block.get('lines', [])): key = (identity, block['start_line'] + offset, digest(encoded(line))) count, size = source_lines.get(key, (0, len(encoded(line)))) source_lines[key] = (count + 1, size) count_keys(doc) for at, value in leaves(doc): strings[digest(value.encode())].append((at, len(encoded(value)))) if isinstance(doc, dict): for k, v in doc.items(): items = v.items() if k == 'input' and isinstance(v, dict) else [(k, v)] for field, item in items: label = 'input.' + field if k == 'input' else field b = encoded(item) key = (req.get('prompt_version'), label, digest(b)) block_groups[key][0] += 1 block_groups[key][1] += len(b) block_groups[key][2] = len(b) dup_strings = sum(sum(n for _, n in entries[1:]) for entries in strings.values()) metadata_fields.update(keys) partition = None if req.get('prompt_version', '').startswith('investigation_project_'): system, prefix, live = GeminiProvider._messages(req, stable_case_context=True) partition = {'prefix_bytes': len(encoded([system, prefix])), 'live_bytes': len(encoded(live)), 'prefix_hash': digest(encoded([system, prefix]))} row = {'ref': name, 'prompt': req.get('prompt_version'), 'message_utf8_bytes': msg_bytes, 'user_repeated_strings_ge80_bytes': dup_strings, 'json_key_bytes': sum(keys.values()), 'top_json_keys_by_bytes': keys.most_common(8), 'stable_partition': partition} per_call.append(row) by_prompt[req.get('prompt_version')].append(row) components = defaultdict(lambda: [0, 0, 0]) for (prompt, label, _), (count, total, size) in block_groups.items(): val = components[(prompt, label)] val[0] += total val[1] += size val[2] += count exact = repeated(groups) canon = repeated(canonical) summary = { 'path': path.as_posix(), 'identity': {k: manifest[k] for k in ('case_id', 'package_id', 'package_generation', 'state')}, 'zip_bytes': path.stat().st_size, 'raw_member_bytes': sum(v.file_size for v in infos.values()), 'compressed_member_bytes': sum(v.compress_size for v in infos.values()), 'members': len(infos), 'zip_compression_methods': dict(compression_methods), 'deflate6_estimated_payload_bytes': sum(deflate_estimate.values()), 'deflate6_by_category': dict(deflate_estimate), 'categories': dict(categories), 'stored_context_components': {k:{'total_bytes':v[0], 'unique_value_bytes':v[1], 'repeated_value_bytes':v[0]-v[1], 'occurrences':v[2]} for k,v in sorted(context_components.items(), key=lambda kv:kv[1][0], reverse=True)}, 'largest_members': [{'path': v.filename, 'raw_bytes': v.file_size, 'compressed_bytes': v.compress_size} for v in sorted(infos.values(), key=lambda v:v.compress_size, reverse=True)[:12]], 'exact_duplicate_groups': len(exact), 'exact_duplicate_raw_bytes': sum(v['raw_duplicate_bytes'] for v in exact), 'exact_duplicate_compressed_bytes': sum(v['compressed_duplicate_bytes'] for v in exact), 'largest_exact_duplicate_groups': exact[:8], 'canonical_json_duplicate_raw_bytes': sum(v['raw_duplicate_bytes'] for v in canon), 'original_tar_members_also_stored_separately': dict(archive_members), 'request_count': len(requests), 'research_repetition': { 'evidence_item_total_bytes': sum(n*s for n,s in evidence_items.values()), 'evidence_item_unique_bytes': sum(s for n,s in evidence_items.values()), 'evidence_item_fields_bytes':dict(evidence_sizes), 'source_line_total_bytes':sum(n*s for n,s in source_lines.values()), 'source_line_unique_bytes':sum(s for n,s in source_lines.values()), 'source_line_occurrences':sum(n for n,s in source_lines.values()), 'source_line_unique_count':len(source_lines)}, 'prompts': {p: {'calls':len(rows), 'message_utf8_bytes':sum(r['message_utf8_bytes'] for r in rows), 'within_call_repeated_strings_ge80_bytes':sum(r['user_repeated_strings_ge80_bytes'] for r in rows)} for p, rows in by_prompt.items()}, 'system_message_total_bytes':sum(size*count for (_,size),count in system_groups.items()), 'system_message_repeated_bytes':sum(size*(count-1) for (_,size),count in system_groups.items()), 'top_level_components': [{'prompt':p,'field':label,'total_canonical_bytes':v[0], 'unique_value_bytes':v[1], 'repeated_value_bytes':v[0]-v[1], 'occurrences':v[2]} for (p,label),v in sorted(components.items(), key=lambda kv:kv[1][0], reverse=True)], 'json_keys_by_bytes':metadata_fields.most_common(15), 'calls': per_call, } assert sum(v['raw_bytes'] for v in categories.values()) == summary['raw_member_bytes'] assert sum(v['compressed_bytes'] for v in categories.values()) == summary['compressed_member_bytes'] return summary def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('packages', nargs='+', type=Path) parser.add_argument('--output', required=True, type=Path) args = parser.parse_args() report = {'measurement':'redundancy_audit_v1', 'units':'bytes, UTF-8, decimal MB/GB', 'limitations': ['No paid API calls; no quality or token savings inferred.', 'Categories are disjoint; duplicate, archive-copy and repeated-string measures overlap.', 'Compressed duplicate sizes are an upper bound, not a compatible deletion proposal.', 'Repeated strings are counted within user messages, minimum 80 UTF-8 bytes; references and JSON framing not deducted.', 'Component repetition compares whole canonical top-level values within each prompt version, ignoring order of calls.', 'Saved requests are canonical gateway input, not literal cached HTTP payload.'], 'packages':[audit(p) for p in args.packages]} args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(report, ensure_ascii=False, indent=2)+'\n', encoding='utf-8') print(json.dumps({'output':str(args.output),'packages':len(report['packages'])})) if __name__ == '__main__': main()