| """Destructive-to-cache HTTP smoke test for an EXCLUSIVE, idle Strata test server. |
| |
| Sends synthetic conversations, deliberately evicts cached conversations, and |
| disconnects a streaming response. Never run while other clients are using it. |
| Dry-run by default; --run explicitly enables network traffic. Does not restart |
| or reconfigure the server. Use the private-engine parity gate for state hashes. |
| """ |
| import argparse |
| import json |
| from pathlib import Path |
| import time |
| import urllib.request |
| import uuid |
|
|
|
|
| def require(condition, message): |
| if not condition: |
| raise RuntimeError(message) |
|
|
|
|
| class Client: |
| def __init__(self, url): |
| self.url = url.rstrip('/') |
|
|
| def metrics(self): |
| with urllib.request.urlopen(self.url + '/metrics', timeout=10) as response: |
| return json.load(response) |
|
|
| def request(self, model, content, stream=False, count=8): |
| body = {'model': model, 'messages': [{'role': 'user', 'content': content}], |
| 'max_tokens': count, 'temperature': 0, 'stream': stream, |
| 'chat_template_kwargs': {'enable_thinking': False}} |
| request = urllib.request.Request(self.url + '/v1/chat/completions', |
| data=json.dumps(body).encode(), |
| headers={'Content-Type': 'application/json'}) |
| return urllib.request.urlopen(request, timeout=120) |
|
|
|
|
| def run(client): |
| initial = client.metrics() |
| require(initial['live']['state'] == 'idle' and initial['live']['queued'] == 0, 'server is busy') |
| info = initial['engine'] |
| slots = info.get('conversation_cache_slots', 0) |
| require(info.get('conversation_cache_mib', 0) > 0 and 1 <= slots <= 16, |
| 'requires an enabled cache with 1–16 slots') |
| require(not info.get('api_key'), 'use an unauthenticated private test endpoint') |
| marker = 'cache-smoke-' + uuid.uuid4().hex |
| records = [] |
|
|
| def before(): |
| metrics = client.metrics() |
| require(metrics['live']['state'] == 'idle' and metrics['live']['queued'] == 0, |
| 'concurrent client detected') |
| return max((r['time'] for r in metrics['requests']), default=0) |
|
|
| def after(stamp, label): |
| for _ in range(50): |
| metrics = client.metrics() |
| if metrics['live']['state'] == 'idle': |
| break |
| time.sleep(.2) |
| require(metrics['live']['state'] == 'idle', 'request did not drain') |
| fresh = [r for r in metrics['requests'] if r['time'] > stamp] |
| require(len(fresh) == 1 and metrics['live']['queued'] == 0, |
| 'missing request evidence or concurrent traffic') |
| record = {'name': label, **fresh[0]} |
| records.append(record) |
| return record |
|
|
| def prompt(label): |
| return marker + ' ' + label + '\n' + '\n'.join( |
| f'Record {i}: blue square, green triangle, red circle.' for i in range(128) |
| ) + '\nName one color. Answer with one word.' |
|
|
| def ask(label, content): |
| stamp = before() |
| with client.request(info['model'], content) as response: |
| result = json.load(response) |
| require(bool(result.get('choices')), 'missing completion') |
| record = after(stamp, label) |
| require(record['finish'] in ('length', 'stop'), 'request did not finish normally') |
| return result['choices'][0]['message'], record |
|
|
| a, _ = ask('A', prompt('A')) |
| ask('B', prompt('B')) |
| restored, record = ask('A-restored', prompt('A')) |
| require(record['reused'] > 0 and a == restored, 'A/B/A reuse or output mismatch') |
| for i in range(slots + 1): |
| ask(f'evict-{i}', prompt(f'evict-{i}')) |
| _, record = ask('A-evicted', prompt('A')) |
| require(record['reused'] == 0, 'old A survived slot eviction') |
|
|
| text = marker + ' cancel: ' + 'blue green red. ' * 256 |
| text += '\nCount from 1 to 1000, one number per line. Do not summarize.' |
| stamp = before() |
| with client.request(info['model'], text, stream=True, count=4096) as response: |
| for line in response: |
| if not line.startswith(b'data: ') or line.strip() == b'data: [DONE]': |
| continue |
| choices = json.loads(line[6:]).get('choices', []) |
| if choices and choices[0].get('delta', {}).get('content'): |
| break |
| else: |
| raise RuntimeError('no streamed content to cancel') |
| record = after(stamp, 'cancelled') |
| require(record['finish'] in ('disconnect', 'cancel') and 0 < record['output_tokens'] < 4096, |
| 'stream cancellation not observed') |
| ask('after-cancel-B', marker + ' Unrelated B: say hello.') |
| _, record = ask('cancelled-A-restored', text) |
| require(record['reused'] > 0, 'cancelled conversation was not restored') |
| return {'engine_info': info, 'records': records} |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument('--url', default='http://127.0.0.1:11434') |
| parser.add_argument('--output', type=Path, required=True, help='new results file; existing paths refused') |
| parser.add_argument('--run', action='store_true') |
| args = parser.parse_args() |
| if not args.run: |
| print('Dry run: HTTP A/B/A, slot eviction, cancellation and recovery. No network traffic.') |
| return |
| |
| with args.output.open('x', encoding='utf-8') as output: |
| try: |
| result = run(Client(args.url)) |
| except Exception as error: |
| json.dump({'passed': False, 'error': str(error)}, output, indent=2) |
| output.write('\n') |
| raise |
| json.dump({'passed': True, **result}, output, indent=2) |
| output.write('\n') |
| print(f'PASS: HTTP reuse, slot eviction, cancellation/recovery. Results: {args.output}') |
|
|
|
|
| if __name__ == '__main__': |
| main() |
|
|
