"""Experiment 001: real loopback provider + real isolated Redis; Kafka is a boundary stub. Repository is mounted read-only. Only HTTP timeout duration is reduced (120s -> .15s). """ import asyncio import json import re import threading import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path import httpx import pytest import redis from fastapi.testclient import TestClient from budget import Budget from reservation import Reservation import gateway.proxy as proxy import gateway.ingest as ingest from gateway.outbox import flush_once from ingestion import CustodyConfig, CustodyContext, TokenEventCustody EVIDENCE = Path('/evidence') REAL_CLIENT = httpx.AsyncClient def record(name, data): with (EVIDENCE / 'observations.jsonl').open('a') as f: f.write(json.dumps({'case': name, **data}) + '\n') class Producer: def __init__(self): self.messages = []; self.error = None def produce(self, topic, *, key, value, on_delivery=None): self.messages.append(json.loads(value)) if on_delivery: on_delivery(self.error, None) def poll(self, timeout): return 0 @pytest.fixture def env(monkeypatch): r = redis.Redis(host='127.0.0.1', decode_responses=True) r.flushdb() # ONLY the disposable Redis container, no published host ports. Budget(r, cache={}).configure('experiment_customer', 10) p = Producer() calls = [] class Provider(BaseHTTPRequestHandler): def do_POST(self): body = json.loads(self.rfile.read(int(self.headers['Content-Length']))) item = {'provider_request_id': f'mock-{len(calls)+1}', 'processed': True, 'usage': {'prompt_tokens': 100, 'completion_tokens': 50}, 'body': body, 'correlation_headers': {k:v for k,v in self.headers.items() if k.lower() != 'authorization'}} calls.append(item) record('provider_processed', item) # Processing is already committed; receipt remains private to provider. if body.get('stream'): self.send_response(200); self.send_header('Content-Type', 'text/event-stream') self.end_headers(); self.wfile.flush() time.sleep(.6) # No receipt reaches Gateway before its .15s read timeout. def log_message(self, *args): pass server = ThreadingHTTPServer(('127.0.0.1', 0), Provider) thread = threading.Thread(target=server.serve_forever, daemon=True); thread.start() monkeypatch.setattr(proxy, 'UPSTREAM_BASE', f'http://127.0.0.1:{server.server_port}/v1') monkeypatch.setattr(proxy.httpx, 'AsyncClient', lambda **kw: REAL_CLIENT(timeout=httpx.Timeout(.15, connect=.15), trust_env=False)) monkeypatch.setattr(ingest, 'get_kafka_producer', lambda: p) import auth, gateway.routes monkeypatch.setattr(auth, '_redis', lambda: r) monkeypatch.setattr(gateway.routes, 'get_redis', lambda: r) yield r, p, calls server.shutdown(); server.server_close(); thread.join() def completion(r, stream=False): return proxy.ProxiedCompletion(r, customer_id='experiment_customer', body={'model':'gpt-4o-mini', 'max_tokens':100, 'messages':[{'role':'user','content':'synthetic'}], 'stream':stream}, provider_auth='Bearer mock-only', parent_span_id='experiment_span', session_id='experiment_session', key_id=None, tenant_id=None) def timeout_request(r): c = completion(r) with pytest.raises(httpx.ReadTimeout): asyncio.run(c.run()) return c def test_nonstream_unknown_and_expiry(env): r,p,calls = env c = timeout_request(r) rid = c.reservation_id state = r.hgetall('reservation:'+rid) assert len(calls) == 1 and calls[0]['processed'] assert state['customer_id'] == 'experiment_customer' assert state['parent_span_id'] == 'experiment_span' assert float(r.get('budget:experiment_customer:held_usd')) > 0 assert p.messages == [] and r.zcard('gateway:outbox:pending') == 0 record('nonstream_timeout', {'reservation_id':rid, 'reservation':state, 'keys':sorted(r.keys('*')), 'upstream_calls':len(calls), 'usage_events':len(p.messages), 'automatic_retry':False, 'explicit_UNKNOWN_state':False}) deadline = r.zscore('gateway:reservations:pending',rid) assert Reservation(r).expire_due(now=deadline+1) == 1 assert Reservation(r).expire_due(now=deadline+2) == 0 assert float(r.get('budget:experiment_customer:held_usd')) == 0 assert float(r.get('budget:experiment_customer:balance_usd')) == 10 alert = json.loads(r.lindex('gateway:reservation:alerts',0)) assert alert['reservationId'] == rid and alert['customerId'] == 'experiment_customer' assert not r.exists('reservation:'+rid) record('expiry', {'alert':alert, 'balance':r.get('budget:experiment_customer:balance_usd'), 'reservation_deleted':True}) def test_http_retry_same_client_id_creates_second_execution(env): r,p,calls = env from gateway_app import app client = TestClient(app, raise_server_exceptions=False) headers = {'X-FluxMeter-Customer-Id':'experiment_customer','Authorization':'Bearer mock-only', 'Idempotency-Key':'experiment-request-001','X-Request-ID':'experiment-request-001'} body = {'model':'gpt-4o-mini','max_tokens':100,'messages':[{'role':'user','content':'synthetic'}]} responses = [client.post('/v1/chat/completions',headers=headers,json=body) for _ in range(2)] assert [x.status_code for x in responses] == [500,500] assert len(calls) == 2 ids = r.zrange('gateway:reservations:pending',0,-1) assert len(ids) == 2 and ids[0] != ids[1] assert all('X-FluxMeter-Reservation-Id' not in x.headers for x in responses) record('client_retry', {'status_codes':[x.status_code for x in responses], 'upstream_calls':len(calls),'reservation_ids':ids,'response_bodies':[x.text for x in responses], 'client_correlation_forwarded':any('X-Request-ID' in x['correlation_headers'] for x in calls)}) def test_stream_without_receipt_emits_zero_usage(env): r,p,calls = env c = completion(r,stream=True) async def run(): response = await c.run() async for _ in response.body_iterator: pass with pytest.raises(httpx.ReadTimeout): asyncio.run(run()) assert len(calls) == 1 and len(p.messages) == 1 event = p.messages[0] assert event['payload']['inputTokens'] == event['payload']['outputTokens'] == 0 assert event['reservation']['reservationId'] == c.reservation_id record('stream_timeout_zero_event', {'envelope':event,'provider_truth':calls[0]['usage']}) custody = TokenEventCustody(r,p,CustodyConfig(topic='token-events',quarantine_topic='quarantine',timeout_seconds=1)) corrected = dict(event['payload'], inputTokens=100, outputTokens=50) result = custody.accept(corrected,CustodyContext(source='gateway', reservation_id=c.reservation_id,reserved_usd=c.reserved_usd)) assert result['status'] == 'conflict' record('stream_later_receipt', {'result':result,'reason':'same reservation-derived event ID already accepted with zero usage'}) def test_manual_later_receipt_custody_idempotency(env): r,p,calls = env c = timeout_request(r) # Operator supplies receipt and correlation from test instrumentation; no automatic recovery exists. event = {'customerId':'experiment_customer','modelId':'gpt-4o-mini', 'inputTokens':100,'outputTokens':50,'timestamp':int(time.time()*1000)} custody = TokenEventCustody(r,p,CustodyConfig(topic='token-events',quarantine_topic='quarantine',timeout_seconds=1)) ctx = CustodyContext(source='gateway',reservation_id=c.reservation_id,reserved_usd=c.reserved_usd) first = custody.accept(event,ctx); again = custody.accept(dict(event),ctx) assert first['status'] == again['status'] == 'accepted' and again['idempotent'] assert len(p.messages) == 1 record('manual_later_receipt', {'first':first,'replay':again,'envelope':p.messages[0], 'automatic_recovery':False,'billing_verified_here':False}) def test_gateway_ingest_regenerated_timestamp_conflicts(env): r,p,calls = env c = timeout_request(r) args = dict(customer_id=c.customer_id, model_id=c.model, input_tokens=100,output_tokens=50, reservation_id=c.reservation_id,reserved_usd=c.reserved_usd) first = ingest.ingest_usage(r,**args) time.sleep(.02) second = ingest.ingest_usage(r,**args) assert first['status'] == 'accepted' and second['status'] == 'conflict' assert len(p.messages) == 1 record('gateway_ingest_repeated_receipt', {'first':first,'second':second, 'reason':'ingest_usage generates a new timestamp included in payload fingerprint'}) def test_received_receipt_outbox_replay(env): r,p,calls = env c = timeout_request(r) p.error = RuntimeError('synthetic Kafka unavailable') result = ingest.ingest_usage(r,customer_id=c.customer_id,model_id=c.model, input_tokens=100,output_tokens=50,reservation_id=c.reservation_id,reserved_usd=c.reserved_usd) assert result['status'] == 'buffered' and r.zcard('gateway:outbox:pending') == 1 original = p.messages[0] p.error = None assert flush_once(r,p,'token-events',1) assert p.messages[1] == original and r.zcard('gateway:outbox:pending') == 0 record('outbox_replay', {'result':result,'identical_envelope':True, 'event_id':original['payload']['eventId'],'provider_calls':len(calls), 'scope':'real Redis, Kafka ACK boundary stub'}) @pytest.mark.parametrize('expire_first',[False,True]) def test_actual_billing_lua_late_receipt_and_replay(env,expire_first): r,p,calls = env c = timeout_request(r) if expire_first: Reservation(r).expire_due(now=r.zscore('gateway:reservations:pending',c.reservation_id)+1) source = Path('/repo/src/main/java/io/fluxmeter/sink/BudgetEnforcerSink.java').read_text() section = source.split('private static final String SINK_LUA_SCRIPT =',1)[1].split(';\n',1)[0] lua = ''.join(json.loads(s) for s in re.findall(r'"(?:[^"\\]|\\.)*"',section)) keys = [f'exp:lua:{i}' for i in range(1,31)] keys[15] = 'budget:experiment_customer:balance_usd' keys[26] = 'exp:window:reservations'; keys[27] = 'budget:experiment_customer:held_usd' keys[28] = 'gateway:reservations:pending' r.sadd(keys[26],c.reservation_id) cost = proxy.get_catalog().quote_usd('gpt-4o-mini',input_tokens=100,output_tokens=50,monthly_tokens_before=0) args = ['100','50','150','1',str(cost),'0','0','.1',str(int(time.time()*1000)),'experiment_customer'] first = r.eval(lua,30,*keys,*args); replay = r.eval(lua,30,*keys,*args) assert first[0] == 'OK' and replay[0] == 'SKIP' assert float(r.get(keys[15])) == pytest.approx(10-cost) assert float(r.get(keys[27])) == 0 assert not r.exists('reservation:'+c.reservation_id) assert r.get(keys[4]) == '1' record('billing_lua', {'expire_first':expire_first,'first':first,'replay':replay, 'cost':cost,'balance':r.get(keys[15]),'held':r.get(keys[27]), 'scope':'exact production Lua on real Redis, synthetic window keys/args; no Flink execution'})