Measure package and prompt redundancy in retained trials

scripts/redundancy_audit.py

@@ -0,0 +1,271 @@
+"""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()