diff --git a/docs/QUERY-CONTEXT.md b/docs/QUERY-CONTEXT.md index d7821bc..cbf44dc 100644 --- a/docs/QUERY-CONTEXT.md +++ b/docs/QUERY-CONTEXT.md @@ -122,3 +122,16 @@ The query-role memory observation after first inference was 239,878,144 bytes private commitment and 274,141,184 bytes resident working set for repeat zero; these are different metrics and must not be summed. Native desktop first-input and complete search latency remain separate open checks. + +Benchmark contract hardening then replaced Python assertions with explicit +errors in both shared benchmark and query-context tooling. Model hashes, +CPU/device state, role/context, normalized vectors, source-window preservation, +MCP errors/filter behavior and batch parity remain checked under `python -O`. +Both tools refuse existing output reports. Four targeted invalid-vector/device +tests passed in normal and optimized modes. A genuine one-pair Google CPU +query/indexing run also passed under optimization, retaining twenty warmed +requests per role; this is a contract smoke rather than an additional latency +acceptance sample. Ignored report SHA-256: +`e611fd3c705a94a6c06ab6ae6a934e6f604665aacb17d0e86d6d5b162ac59e46` +(`.searchi/qa/query-context-optimized-contract.json`). Earlier recorded timing +reports were run without optimization and are unaffected by this correction. diff --git a/tools/embeddinggemma2/benchmark.py b/tools/embeddinggemma2/benchmark.py index a127b0f..7f4903e 100644 --- a/tools/embeddinggemma2/benchmark.py +++ b/tools/embeddinggemma2/benchmark.py @@ -38,6 +38,11 @@ def distribution(values): 'p50_ms': statistics.median(values), 'p95_ms': ordered[max(0, math.ceil(len(values) * .95) - 1)]} +def require(condition, message): + if not condition: + raise ValueError(message) + + class Session: def __init__(self, command): environment = os.environ.copy() @@ -81,7 +86,7 @@ class Session: response = self.next() if not response.get('ok'): raise RuntimeError(response.get('error', 'Worker request failed')) - assert response['device'] == 'cpu' + require(response['device'] == 'cpu', 'Benchmark contract failed at benchmark.py:84') return response, elapsed(start) def mcp(self, identifier, method, params): @@ -90,7 +95,7 @@ class Session: while True: response = self.next() if response.get('id') == identifier: - assert 'error' not in response, response + require('error' not in response, response) return response['result'], elapsed(start) def close(self): @@ -103,19 +108,19 @@ class Session: def vectors(response, count): result = response['vectors'] - assert len(result) == count + require(len(result) == count, 'Benchmark contract failed at benchmark.py:106') for vector in result: - assert len(vector) == 768 and all(math.isfinite(n) for n in vector) - assert abs(sum(n*n for n in vector) - 1) < .0001 + require(len(vector) == 768 and all(math.isfinite(n) for n in vector), 'Benchmark contract failed at benchmark.py:108') + require(abs(sum(n*n for n in vector) - 1) < .0001, 'Benchmark contract failed at benchmark.py:109') return result def native(args): manifest = json.loads((ROOT / 'tools/embeddinggemma2/model-manifest.json').read_text()) model = args.models / manifest['files'][0]['path'] start = time.perf_counter() - assert model.stat().st_size == manifest['files'][0]['bytes'] + require(model.stat().st_size == manifest['files'][0]['bytes'], 'Benchmark contract failed at benchmark.py:116') with model.open('rb') as stream: - assert hashlib.file_digest(stream, 'sha256').hexdigest() == manifest['files'][0]['sha256'] + require(hashlib.file_digest(stream, 'sha256').hexdigest() == manifest['files'][0]['sha256'], 'Benchmark contract failed at benchmark.py:118') verification_ms = elapsed(start) reports = [] for threads in args.threads: @@ -124,7 +129,7 @@ def native(args): try: hello = session.next() load_ms = elapsed(session.started) - assert hello['device'] == 'cpu' and hello['model'] == manifest['model_id'] + require(hello['device'] == 'cpu' and hello['model'] == manifest['model_id'], 'Benchmark contract failed at benchmark.py:127') first, first_ms = session.worker({'op':'query', 'text':'first unseen foreground query', 'debug_performance':True}) vectors(first, 1) @@ -143,8 +148,8 @@ def native(args): response, duration = session.worker({'op':'windows','text':passage,'title':'Agreement'}) encoded = passage.encode() for window in response['windows']: - assert encoded[window['start_byte']:window['end_byte']].decode() == window['text'] - assert response['windows'][-1]['end_byte'] == len(encoded) + require(encoded[window['start_byte']:window['end_byte']].decode() == window['text'], 'Benchmark contract failed at benchmark.py:146') + require(response['windows'][-1]['end_byte'] == len(encoded), 'Benchmark contract failed at benchmark.py:147') window_samples.append({'bytes':len(encoded),'windows':len(response['windows']), 'round_trip_ms':duration}) documents = [(f'Document {n}. Supplier cancellation requires sixty days written notice. ' @@ -158,7 +163,7 @@ def native(args): response, batch_ms = session.worker({'op':'documents','texts':documents,'title':'Agreement'}) batched = vectors(response, 8) cosines = [sum(a*b for a,b in zip(left,right)) for left,right in zip(singles,batched)] - assert min(cosines) > .99999 + require(min(cosines) > .99999, 'Benchmark contract failed at benchmark.py:161') reports.append({'threads':threads,'worker_info':hello,'process_load_ms':load_ms, 'first_query_ms':first_ms,'first_query_worker':first['performance'], 'warm_queries':distribution([sample['round_trip_ms'] for sample in samples]), @@ -174,8 +179,8 @@ def native(args): 'First query precedes warm samples. Documents measure IPC batch, not assumed parallelism.'} def mcp(args): - assert args.db and args.cli - assert args.db.is_file() + require(args.db and args.cli, 'Benchmark contract failed at benchmark.py:177') + require(args.db.is_file(), 'Benchmark contract failed at benchmark.py:178') session = Session([str(args.cli.resolve()),'mcp','--db',str(args.db.resolve()), '--model-dir',str(args.models.resolve()),'--embedding-worker',str(args.worker.resolve()), '--cpu-threads','1']) @@ -185,8 +190,8 @@ def mcp(args): startup_ms = elapsed(session.started) session.send({'jsonrpc':'2.0','method':'notifications/initialized'}) stats,_ = session.mcp(2,'tools/call',{'name':'stats','arguments':{}}) - assert stats['structuredContent']['mode'] == 'hybrid' - assert stats['structuredContent']['worker']['device'] == 'cpu' + require(stats['structuredContent']['mode'] == 'hybrid', 'Benchmark contract failed at benchmark.py:188') + require(stats['structuredContent']['worker']['device'] == 'cpu', 'Benchmark contract failed at benchmark.py:189') samples=[] identifier=3 cases = [('first_unseen',QUERIES[0],None)] @@ -199,11 +204,11 @@ def mcp(args): arguments['extension']=extension result,duration=session.mcp(identifier,'tools/call',{'name':'search','arguments':arguments}) identifier+=1 - assert not result.get('isError'), result + require(not result.get('isError'), result) response=result['structuredContent'] - assert response['mode']=='hybrid' + require(response['mode']=='hybrid', 'Benchmark contract failed at benchmark.py:204') if extension: - assert all(item['filename'].lower().endswith('.'+extension) for item in response['results']) + require(all(item['filename'].lower().endswith('.'+extension) for item in response['results']), 'Benchmark contract failed at benchmark.py:206') samples.append({'case':case,'query':query,'extension':extension,'request_ms':duration, 'engine_ms':response['metrics']['total_ms'], 'matches':[{'filename':item['filename'],'source':item['source'], @@ -230,7 +235,8 @@ def main(): parser.add_argument('--rust-profile',choices=['debug','release','not-applicable'],default='not-applicable') parser.add_argument('--label',default='') args=parser.parse_args() - assert 1<=args.repeats<=20 and all(1<=n<=8 for n in args.threads) + require(not args.output.exists(), 'Output must be new; preserve previous evidence') + require(1<=args.repeats<=20 and all(1<=n<=8 for n in args.threads), 'Benchmark contract failed at benchmark.py:233') def digest(path): with path.open('rb') as stream: return hashlib.file_digest(stream,'sha256').hexdigest() @@ -241,7 +247,8 @@ def main(): 'cli_sha256':digest(args.cli) if args.cli else None, 'results':native(args) if args.mode=='native' else mcp(args)} args.output.parent.mkdir(parents=True,exist_ok=True) - args.output.write_text(json.dumps(report,indent=2,ensure_ascii=False)+'\n',encoding='utf-8') + with args.output.open('x', encoding='utf-8') as output: + output.write(json.dumps(report,indent=2,ensure_ascii=False)+'\n') print(f'Saved {args.output}',flush=True) if __name__=='__main__': diff --git a/tools/embeddinggemma2/query_context_benchmark.py b/tools/embeddinggemma2/query_context_benchmark.py index b9eb746..42b3116 100644 --- a/tools/embeddinggemma2/query_context_benchmark.py +++ b/tools/embeddinggemma2/query_context_benchmark.py @@ -14,7 +14,7 @@ from pathlib import Path import platform import statistics import time -from benchmark import ROOT, Session, vectors +from benchmark import ROOT, Session, vectors, require def memory(pid): @@ -62,9 +62,9 @@ def main(): entry = manifest["files"][0] model = args.models / entry["path"] started = time.perf_counter() - assert model.stat().st_size == entry["bytes"] + require(model.stat().st_size == entry["bytes"], 'Benchmark contract failed at query_context_benchmark.py:65') with model.open("rb") as stream: - assert hashlib.file_digest(stream, "sha256").hexdigest() == entry["sha256"] + require(hashlib.file_digest(stream, "sha256").hexdigest() == entry["sha256"], 'Benchmark contract failed at query_context_benchmark.py:67') verification_ms = (time.perf_counter() - started) * 1000 samples = [] for repeat in range(args.repeats): @@ -74,8 +74,8 @@ def main(): try: hello = session.next() startup_ms = (time.perf_counter() - session.started) * 1000 - assert hello["device"] == "cpu" and not hello["vulkan_module_loaded"] - assert hello["role"] == role and hello["context_tokens"] == (512 if role == "query" else 2048) + require(hello["device"] == "cpu" and not hello["vulkan_module_loaded"], 'Benchmark contract failed at query_context_benchmark.py:77') + require(hello["role"] == role and hello["context_tokens"] == (512 if role == "query" else 2048), 'Benchmark contract failed at query_context_benchmark.py:78') sample = {"repeat": repeat, "role": role, "hello": hello, "startup_ms": startup_ms, "ready_memory": memory(session.process.pid)} response, first_ms = session.worker({"op": "query", "text": "cancel the supplier partnership", "debug_performance": True}) vectors(response, 1) @@ -92,8 +92,8 @@ def main(): if role == "query": session.send({"op": "documents", "texts": ["test"], "title": "none"}) rejected = session.next() - assert not rejected["ok"] and not rejected["fatal"] - assert "only supports" in rejected["error"] + require(not rejected["ok"] and not rejected["fatal"], 'Benchmark contract failed at query_context_benchmark.py:95') + require("only supports" in rejected["error"], 'Benchmark contract failed at query_context_benchmark.py:96') vectors(session.worker({"op": "code_query", "text": "retry failed requests"})[0], 1) samples.append(sample) finally: diff --git a/tools/embeddinggemma2/test_benchmark_contract.py b/tools/embeddinggemma2/test_benchmark_contract.py new file mode 100644 index 0000000..f76d5bc --- /dev/null +++ b/tools/embeddinggemma2/test_benchmark_contract.py @@ -0,0 +1,32 @@ +"""Benchmark validity checks must survive python -O.""" +import math +import unittest +from benchmark import Session, require, vectors + + +class Contract(unittest.TestCase): + def test_bad_vectors_rejected(self): + for result in ([], [[1.0]], [[math.nan] * 768], [[0.0] * 768]): + with self.subTest(result_length=len(result)), self.assertRaises(ValueError): + vectors({"vectors": result}, 1) + + def test_real_shape_normalized_vector_accepted(self): + vector = [1.0] + [0.0] * 767 + self.assertEqual(vectors({"vectors": [vector]}, 1), [vector]) + + def test_cpu_contract_cannot_be_bypassed(self): + class FakeSession: + def send(self, value): + pass + def next(self): + return {"ok": True, "device": "gpu"} + with self.assertRaises(ValueError): + Session.worker(FakeSession(), {"op": "query"}) + + def test_explicit_guard_survives_optimization(self): + with self.assertRaises(ValueError): + require(False, "invalid asset") + + +if __name__ == "__main__": + unittest.main()