Spaces:
Sleeping
Sleeping
| """CPU contract/integration tests against upstream-shaped duplex boundaries. | |
| These use real PCM/JPEG encoding, Pydantic parsing and FastAPI HTTP handling, | |
| plus native adapter source-ledger and lifecycle logic. They do not claim CUDA. | |
| """ | |
| import base64 | |
| import io | |
| import json | |
| import threading | |
| import time | |
| from concurrent.futures import ThreadPoolExecutor | |
| import numpy as np | |
| import pytest | |
| from fastapi.testclient import TestClient | |
| from PIL import Image | |
| from pydantic import ValidationError | |
| from gpu_service.app import create_app | |
| from gpu_service.contracts import AVChunk | |
| from gpu_service.native import MODEL_LOCK, NativeDuplex, NativeContractError, prepare_text_context, text_only_duplex | |
| TOKEN = "test-only-token-no-credentials-12345" | |
| SESSION = "contract-session-1234" | |
| class UpstreamDecoder: | |
| def __init__(self): | |
| self.units = 0 | |
| self.generated_tokens = [] | |
| def get_window_stats(self): | |
| return {"unit_count": min(self.units, 64), "cache_length": min(self.units, 64) * 100} | |
| def get_cache_length(self): | |
| return min(self.units, 64) * 100 | |
| def _unit_history(self): | |
| return [None] * min(self.units, 64) | |
| def reset(self): | |
| self.units = 0 | |
| class UpstreamShapedDuplex: | |
| """Faithful signatures and first-chunk/core consumption from pinned source.""" | |
| def __init__(self): | |
| self.decoder = UpstreamDecoder() | |
| self.audio_buffer = np.zeros(0, dtype=np.float32) | |
| self.calls = [] | |
| self.current_turn_ended = True | |
| self.total_hidden = [] | |
| self.total_ids = [] | |
| self.prefill_schema_tokens = [] | |
| self.res_ids = [] | |
| self.stopped = False | |
| self.last_audio = None | |
| self.last_frames = None | |
| self.first = True | |
| self.generated_count = 0 | |
| self.model = None | |
| self.prompts = [] | |
| self.generate_options = [] | |
| self.speak_token_id = 1 | |
| self.pending_logits = np.zeros((1, 4), dtype=np.float32) | |
| def prepare(self, *, prefix_system_prompt): | |
| self.calls.append("prepare") | |
| assert "English" in prefix_system_prompt | |
| assert "chain-of-thought" in prefix_system_prompt | |
| self.prompts.append(prefix_system_prompt) | |
| self.stopped = False | |
| self.decoder.reset() | |
| self._reset_streaming_state() | |
| def _reset_streaming_state(self): | |
| self.audio_buffer = np.zeros(0, dtype=np.float32) | |
| self.first = True | |
| def set_session_stop(self): | |
| self.stopped = True | |
| self.calls.append("stop") | |
| def streaming_prefill(self, *, audio_waveform, frame_list, max_slice_nums, batch_vision_feed): | |
| self.calls.append("prefill") | |
| assert audio_waveform.dtype == np.float32 and audio_waveform.shape == (16000,) | |
| assert max_slice_nums == 2 and batch_vision_feed is True | |
| self.last_audio, self.last_frames = audio_waveform.copy(), frame_list | |
| self.pending_logits = np.zeros((1, 4), dtype=np.float32) | |
| self.audio_buffer = np.concatenate([self.audio_buffer, audio_waveform]) | |
| if self.first: | |
| self.audio_buffer = np.concatenate([np.zeros(560, dtype=np.float32), self.audio_buffer]) | |
| self.audio_buffer = self.audio_buffer[16480:] | |
| self.first = False | |
| else: | |
| self.audio_buffer = self.audio_buffer[16000:] | |
| return {"success": True, "cost_all": 0.0001} | |
| def streaming_generate(self, *, max_new_speak_tokens_per_chunk, decode_mode, listen_top_k=None, listen_prob_scale=1.0): | |
| self.calls.append("generate") | |
| assert 8 <= max_new_speak_tokens_per_chunk <= 64 and decode_mode == "sampling" | |
| self.generate_options.append((max_new_speak_tokens_per_chunk, listen_top_k, listen_prob_scale)) | |
| self.generated_count += 1 | |
| self.decoder.units += 1 | |
| self.total_hidden.append([np.ones(10)]) | |
| return {"is_listen": True, "text": "", "audio_waveform": np.zeros(24000), "end_of_turn": False, "current_time": self.generated_count, "cost_all": 0.0001} | |
| def payload(sequence=0, start_ms=None, count=16000, epoch=0): | |
| start = sequence * 1000 if start_ms is None else start_ms | |
| # Dense continuously nonzero signal, encoded as actual little-endian float32. | |
| signal = (np.sin(np.arange(count) * 2 * np.pi * 313 / 16000) * 0.7).astype("<f4") | |
| image = Image.new("RGB", (64, 48), "green") | |
| stream = io.BytesIO() | |
| image.save(stream, "JPEG") | |
| return {"session_id": SESSION, "epoch": epoch, "sequence": sequence, | |
| "source_start_ms": start, "source_end_ms": start + count / 16, | |
| "sample_rate": 16000, "sample_count": count, | |
| "audio_b64": base64.b64encode(signal.tobytes()).decode(), | |
| "frames": [{"source_time_ms": start + count / 32, "jpeg_b64": base64.b64encode(stream.getvalue()).decode(), "timing_method": "requestVideoFrameCallback"}], | |
| "capture_time_ms": time.time() * 1000, "final_chunk": count < 16000, | |
| "evidence_ids": [f"e{epoch}c{sequence}"]} | |
| def client_and_native(): | |
| upstream = UpstreamShapedDuplex() | |
| adapter = NativeDuplex(upstream) | |
| return TestClient(create_app(adapter, TOKEN), headers={"Authorization": f"Bearer {TOKEN}"}), upstream | |
| def start(client, epoch=0): | |
| response = client.post("/sessions", json={"session_id": SESSION, "epoch": epoch, "objective": "Evaluate source evidence relevant to copper demand."}) | |
| assert response.status_code == 200, response.text | |
| def test_lock_identifies_exact_o45_and_same_remote_code(): | |
| assert MODEL_LOCK["model_id"] == "openbmb/MiniCPM-o-4_5" | |
| assert MODEL_LOCK["model_revision"] == MODEL_LOCK["inference_code_revision"] | |
| assert len(MODEL_LOCK["model_revision"]) == 40 | |
| def test_first_av_request_avoids_broken_optional_stats_and_retains_real_counts(monkeypatch): | |
| from unittest.mock import Mock | |
| client, upstream = client_and_native() | |
| broken_stats = Mock(side_effect=AttributeError("_system_prompt_template")) | |
| monkeypatch.setattr(upstream.decoder, "get_window_stats", broken_stats) | |
| with client: | |
| start(client) | |
| first = client.post(f"/sessions/{SESSION}/chunks", json=payload()) | |
| assert first.status_code == 200, first.text | |
| assert upstream.last_audio.shape == (16000,) and np.any(upstream.last_audio) | |
| assert upstream.last_frames[0].mode == "RGB" | |
| assert first.json()["retained_native_units"] == 1 | |
| assert first.json()["cache_tokens"] == 100 | |
| # A native window drop must still change retained context; removing the | |
| # broken helper must not replace real counts with constants/defaults. | |
| upstream.decoder.units = 0 | |
| second = client.post(f"/sessions/{SESSION}/chunks", json=payload(1)) | |
| assert second.status_code == 200, second.text | |
| assert second.json()["context_start_ms"] == 1000 | |
| assert second.json()["context_evidence_ids"] == ["e0c1"] | |
| broken_stats.assert_not_called() | |
| def test_context_prefill_preserves_prompt_tokens_and_protection_with_bounded_forward_calls(): | |
| from types import SimpleNamespace | |
| from unittest.mock import Mock | |
| # Token-shaped CPU decoder: compare the actual ordered context and protected | |
| # boundary, while proving long portfolios do not cause one forward per token. | |
| received = [] | |
| decoder = SimpleNamespace(_window_config=SimpleNamespace(sliding_window_mode="basic"), | |
| reset=Mock(side_effect=received.clear), embed_tokens=lambda ids: ids, | |
| feed=Mock(side_effect=received.extend)) | |
| protected = [] | |
| decoder.register_system_prompt = lambda: protected.append(len(received)) | |
| duplex = SimpleNamespace(generate_audio=False, decoder=decoder, | |
| tokenizer=SimpleNamespace(encode=lambda s, **kw: list(s.encode())), | |
| clear_break_event=Mock(), clear_session_stop=Mock(), | |
| _reset_streaming_state=Mock(), model=SimpleNamespace(init_streaming_processor=Mock())) | |
| prompt = "Observed source and portfolio context. " * 60 | |
| cancelled = threading.Event() | |
| prepare_text_context(duplex, prompt, cancelled) | |
| assert bytes(received).decode() == "<|im_start|>system\n" + prompt + "<|im_end|>" | |
| assert protected == [len(received)] and decoder.feed.call_count < 12 | |
| duplex.model.init_streaming_processor.assert_called_once() | |
| cancelled.set() | |
| forwards = decoder.feed.call_count | |
| with pytest.raises(NativeContractError, match="cancelled"): | |
| prepare_text_context(duplex, prompt, cancelled) | |
| assert decoder.feed.call_count == forwards | |
| def test_text_duplex_skips_unused_speech_renderer_and_restores_model_method(): | |
| from types import SimpleNamespace | |
| class Model: | |
| config = SimpleNamespace(init_vision=True, init_audio=True, init_tts=False) | |
| fail_after_init = False | |
| def init_tts(self, **kwargs): | |
| raise RuntimeError("Unused speech renderer must not be initialized") | |
| def as_duplex(self, *, generate_audio): | |
| self.init_tts(enable_float16=False, n_timesteps=10) | |
| if self.fail_after_init: | |
| raise ValueError("unrelated constructor failure") | |
| return SimpleNamespace(generate_audio=generate_audio, model=self) | |
| model = Model() | |
| with pytest.raises(RuntimeError, match="Unused speech renderer"): | |
| model.as_duplex(generate_audio=False) | |
| duplex = text_only_duplex(model) | |
| assert duplex.model is model and duplex.generate_audio is False | |
| assert model.config.init_vision and model.config.init_audio | |
| assert "init_tts" not in vars(model) | |
| model.fail_after_init = True | |
| with pytest.raises(ValueError, match="unrelated constructor failure"): | |
| text_only_duplex(model) | |
| assert "init_tts" not in vars(model) | |
| with pytest.raises(RuntimeError, match="Unused speech renderer"): | |
| model.init_tts() | |
| def test_eager_startup_error_identifies_stage_and_cause_without_secret_messages(monkeypatch): | |
| import importlib | |
| module = importlib.import_module("gpu_service.app") | |
| secret = "hf_" + "private-value-never-log" * 2 | |
| monkeypatch.setenv("HF_TOKEN", secret) | |
| monkeypatch.setenv("NATIVE_EAGER_LOAD", "1") | |
| class FailingLoad(NativeDuplex): | |
| def load(self): | |
| self.load_stage = "native_duplex_setup" | |
| try: | |
| raise ValueError("private download URL and credential: " + secret) | |
| except ValueError: | |
| raise ImportError("unhelpful upstream wrapper " + secret) from None | |
| monkeypatch.setattr(module, "NativeDuplex", FailingLoad) | |
| with pytest.raises(RuntimeError) as failure: | |
| with TestClient(module.create_app(token=TOKEN)): | |
| pass | |
| message = str(failure.value) | |
| assert "native_duplex_setup" in message | |
| assert "ImportError" in message and "ValueError" in message | |
| assert "test_gpu_contract.py:" in message | |
| assert secret not in message and "private download URL" not in message | |
| def test_portfolio_context_once_per_epoch_without_publication_cooldown(): | |
| from types import SimpleNamespace | |
| client, native = client_and_native() | |
| resets = [] | |
| native.model = SimpleNamespace(reset_session=lambda: resets.append(True)) | |
| holding = {"id": "copper-fund", "name": "User copper research holding", "ticker": "COPPER", | |
| "exchange": "User supplied", "currency": "CAD", "asset_type": "fund", "weight": 100} | |
| context = {"session_id": SESSION, "epoch": 0, "objective": "Evaluate source evidence relevant to copper demand.", | |
| "holdings": [holding], "output_max_tokens_per_unit": 32, "live_update_interval_seconds": 10} | |
| original = native.streaming_generate | |
| def first_speaks(**kwargs): | |
| result = original(**kwargs) | |
| if native.generated_count == 1: | |
| return {**result, "text": "Evidence: Demand is discussed. Holding: copper-fund", "is_listen": False, "end_of_turn": True} | |
| return result | |
| native.streaming_generate = first_speaks | |
| with client: | |
| assert client.post("/sessions", json=context).status_code == 200 | |
| assert client.post("/sessions", json=context).status_code == 200 | |
| assert client.post("/sessions", json={**context, "holdings": [{**holding, "name": "Changed name"}]}).status_code == 409 | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload()).status_code == 200 | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 200 | |
| assert native.generate_options == [(32, None, 1.0), (32, None, 1.0)] | |
| assert len(native.prompts) == 1 and '"id":"copper-fund"' in native.prompts[0] | |
| assert len(resets) == 1 | |
| assert "Holding: the exact user holding ID" in native.prompts[0] | |
| assert client.post(f"/sessions/{SESSION}/reset", json={"epoch": 1}).status_code == 200 | |
| assert len(native.prompts) == 2 and native.prompts[1] == native.prompts[0] | |
| assert len(resets) >= 2 | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload(0, start_ms=30000, epoch=1)).status_code == 200 | |
| assert native.generate_options[-1] == (32, None, 1.0) | |
| def test_native_silence_is_not_overridden_and_effective_settings_are_reported(): | |
| client, native = client_and_native() | |
| original = native.streaming_generate | |
| def decode(**kwargs): | |
| # Application must leave every pending logit unchanged, even after a | |
| # long silent stretch or the legacy interval setting. | |
| np.testing.assert_array_equal(native.pending_logits, np.zeros((1, 4))) | |
| return original(**kwargs) | |
| native.streaming_generate = decode | |
| with client: | |
| response = client.post('/sessions', json={'session_id': SESSION, 'epoch': 0, | |
| 'objective': 'Evaluate supplied media.', 'live_update_interval_seconds': 1}) | |
| assert response.status_code == 200 | |
| runtime = response.json()['runtime'] | |
| assert runtime['max_slice_nums'] == 2 and runtime['batch_vision_feed'] is True | |
| assert runtime['generate_audio'] is False and runtime['audio_understanding'] is True | |
| assert runtime['window_high_tokens'] == 8000 and runtime['window_low_tokens'] == 6000 | |
| for i in range(5): | |
| result = client.post(f'/sessions/{SESSION}/chunks', json=payload(i)).json() | |
| assert result['native_silence'] and result['text'] == '' | |
| timing = result['timing'] | |
| assert timing['new_source_ms'] == (995 if i == 0 else 1000) | |
| assert timing['processing_per_source_second'] == pytest.approx( | |
| (timing['preprocess_ms'] + timing['native_unit_ms']) / timing['new_source_ms']) | |
| assert native.calls.count('prepare') == 1 and native.calls.count('generate') == 5 | |
| assert client.post(f'/sessions/{SESSION}/finish', json={'epoch': 0}).status_code == 200 | |
| def test_real_pcm_jpeg_decode_and_native_request(monkeypatch): | |
| from gpu_service.contracts import Frame | |
| audio_decode, frame_decode = AVChunk.decode_audio, Frame.decode | |
| decoded = [] | |
| def audio_once(chunk): | |
| decoded.append(("audio", threading.current_thread().name)) | |
| return audio_decode(chunk) | |
| def frame_once(frame): | |
| decoded.append(("frame", threading.current_thread().name)) | |
| return frame_decode(frame) | |
| monkeypatch.setattr(AVChunk, "decode_audio", audio_once) | |
| monkeypatch.setattr(Frame, "decode", frame_once) | |
| client, native = client_and_native() | |
| with client: | |
| start(client) | |
| data = payload() | |
| second = io.BytesIO() | |
| Image.new("RGB", (32, 24), "blue").save(second, "JPEG") | |
| data["frames"].append({"source_time_ms": 750.125, | |
| "jpeg_b64": base64.b64encode(second.getvalue()).decode()}) | |
| response = client.post(f"/sessions/{SESSION}/chunks", json=data) | |
| assert response.status_code == 200, response.text | |
| result = response.json() | |
| assert native.last_frames[0].mode == "RGB" | |
| np.testing.assert_array_equal(native.last_audio, np.frombuffer(base64.b64decode(data["audio_b64"]), dtype="<f4")) | |
| assert [kind for kind, _ in decoded] == ["audio", "frame", "frame"] | |
| assert [frame.size for frame in native.last_frames] == [(64, 48), (32, 24)] | |
| assert result["timing"]["frame_source_times_ms"] == [500, 750.125] | |
| assert all(name.startswith("asyncio_") for _, name in decoded) | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=data).json()["idempotent_replay"] | |
| assert len(decoded) == 3 and native.calls.count("generate") == 1 | |
| assert result["consumed_source_end_ms"] == 995 | |
| assert result["synthetic_padding_samples"] == 560 | |
| assert result["context_evidence_ids"] == ["e0c0"] | |
| assert result["native_silence"] is True | |
| assert result["audio_kind"] == "audio evidence" | |
| assert len(native.total_hidden) == 0 | |
| def test_stop_during_cpu_preparation_discards_decoded_input_before_native_work(monkeypatch): | |
| from concurrent.futures import ThreadPoolExecutor | |
| entered, release = threading.Event(), threading.Event() | |
| decode = NativeDuplex.decode | |
| def blocked(chunk): | |
| entered.set() | |
| assert release.wait(3) | |
| return decode(chunk) | |
| monkeypatch.setattr(NativeDuplex, "decode", staticmethod(blocked)) | |
| client, native = client_and_native() | |
| with client, ThreadPoolExecutor(max_workers=1) as pool: | |
| start(client) | |
| request = pool.submit(client.post, f"/sessions/{SESSION}/chunks", json=payload()) | |
| try: | |
| assert entered.wait(2) | |
| before = time.perf_counter() | |
| assert client.post(f"/sessions/{SESSION}/stop", json={"epoch": 0}).status_code == 200 | |
| assert time.perf_counter() - before < 1 | |
| finally: | |
| release.set() | |
| assert request.result(timeout=3).status_code == 409 | |
| assert "prefill" not in native.calls and "generate" not in native.calls | |
| def test_ten_minute_dense_audio_actual_encoder_http_and_final_five_ms(): | |
| client, native = client_and_native() | |
| with client: | |
| start(client) | |
| for sequence in range(600): | |
| response = client.post(f"/sessions/{SESSION}/chunks", json=payload(sequence)) | |
| assert response.status_code == 200, (sequence, response.text) | |
| result = response.json() | |
| assert result["consumed_source_end_ms"] == 599995 | |
| assert len(native.total_hidden) == 0 | |
| finish = client.post(f"/sessions/{SESSION}/finish", json={"epoch": 0}) | |
| assert finish.status_code == 200, finish.text | |
| report = finish.json() | |
| assert report["model_consumed_sample_count"] == 600 * 16000 | |
| assert report["complete"] and report["model_consumed_end_ms"] == 600000 | |
| assert report["outputs"][0]["consumed_source_start_ms"] == 599995 | |
| assert report["outputs"][0]["consumed_source_end_ms"] == 600000 | |
| assert report["outputs"][0]["synthetic_tail"] | |
| assert report["outputs"][0]["context_end_ms"] == 600000 | |
| assert native.calls.count("prepare") == 1 | |
| assert native.calls.count("generate") == 601 | |
| def test_final_partial_preserves_every_sample(count): | |
| client, native = client_and_native() | |
| with client: | |
| start(client) | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload(0)).status_code == 200 | |
| response = client.post(f"/sessions/{SESSION}/chunks", json=payload(1, count=count)) | |
| assert response.status_code == 200, response.text | |
| completion = client.post(f"/sessions/{SESSION}/finish", json={"epoch": 0}).json() | |
| assert completion["complete"] | |
| assert completion["model_consumed_sample_count"] == 16000 + count | |
| assert completion["model_consumed_end_ms"] == 1000 + count / 16 | |
| def test_bad_encoding_intervals_rejected_before_model(mutation): | |
| client, native = client_and_native() | |
| with client: | |
| start(client) | |
| data = payload() | |
| data.update(mutation) | |
| if mutation.get("source_end_ms") == float("inf"): | |
| with pytest.raises(ValidationError): | |
| AVChunk(**data) | |
| else: | |
| response = client.post(f"/sessions/{SESSION}/chunks", json=data) | |
| assert response.status_code == 422 | |
| assert "prefill" not in native.calls | |
| def test_idempotent_retry_no_second_inference_and_conflict_rejected(): | |
| client, native = client_and_native() | |
| with client: | |
| start(client) | |
| data = payload() | |
| first = client.post(f"/sessions/{SESSION}/chunks", json=data) | |
| repeat = client.post(f"/sessions/{SESSION}/chunks", json=data) | |
| assert first.status_code == repeat.status_code == 200 | |
| assert repeat.json()["idempotent_replay"] | |
| assert native.calls.count("generate") == 1 | |
| data["audio_b64"] = base64.b64encode(np.zeros(16000, dtype="<f4").tobytes()).decode() | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=data).status_code == 409 | |
| def test_out_of_order_source_gap_seek_and_stale_epoch(): | |
| client, native = client_and_native() | |
| with client: | |
| start(client) | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 409 | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload(0)).status_code == 200 | |
| assert client.post(f"/sessions/{SESSION}/reset", json={"epoch": 1}).status_code == 200 | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 409 | |
| new = client.post(f"/sessions/{SESSION}/chunks", json=payload(0, start_ms=50000, epoch=1)) | |
| assert new.status_code == 200, new.text | |
| assert new.json()["context_start_ms"] == 50000 | |
| assert new.json()["context_evidence_ids"] == ["e1c0"] | |
| gap = client.post(f"/sessions/{SESSION}/chunks", json=payload(1, start_ms=53000, epoch=1)) | |
| assert gap.status_code == 502 | |
| def test_stop_no_additional_generation_finish_or_prepare(): | |
| client, native = client_and_native() | |
| with client: | |
| start(client) | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload()).status_code == 200 | |
| calls = native.calls.count("generate") | |
| response = client.post(f"/sessions/{SESSION}/stop", json={"epoch": 0}) | |
| assert response.status_code == 200 | |
| assert response.json()["gpu_uptime_billing_paused"] is False | |
| assert client.post(f"/sessions/{SESSION}/finish", json={"epoch": 0}).status_code == 409 | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 409 | |
| assert native.calls.count("generate") == calls | |
| assert native.calls.count("prepare") == 1 | |
| def test_stop_interrupts_control_immediately_and_discards_late_result(): | |
| client, native = client_and_native() | |
| entered, release = threading.Event(), threading.Event() | |
| original_generate = native.streaming_generate | |
| def slow(**kwargs): | |
| entered.set() | |
| release.wait(timeout=5) | |
| return original_generate(**kwargs) | |
| native.streaming_generate = slow | |
| with client, ThreadPoolExecutor(max_workers=1) as pool: | |
| start(client) | |
| future = pool.submit(client.post, f"/sessions/{SESSION}/chunks", json=payload()) | |
| assert entered.wait(2) | |
| before = time.monotonic() | |
| response = client.post(f"/sessions/{SESSION}/stop", json={"epoch": 0}) | |
| assert response.status_code == 200 | |
| assert time.monotonic() - before < 1 | |
| release.set() | |
| assert future.result(timeout=5).status_code == 409 | |
| def test_authentication_and_validation_never_echo_payload(): | |
| client, native = client_and_native() | |
| with client: | |
| assert client.get("/health", headers={"Authorization": "Bearer wrong"}).status_code == 401 | |
| assert client.post("/sessions", json={"session_id": "PRIVATE_SENTINEL", "epoch": -1, "objective": "PRIVATE_SENTINEL"}).status_code == 422 | |
| response = client.post("/sessions", json={"session_id": "PRIVATE_SENTINEL", "epoch": -1, "objective": "PRIVATE_SENTINEL"}) | |
| assert "PRIVATE_SENTINEL" not in response.text | |
| def test_malformed_native_result_fails_closed_and_requires_new_epoch(): | |
| client, native = client_and_native() | |
| native.streaming_generate = lambda **_: {"text": "unchecked", "is_listen": "false"} | |
| with client: | |
| start(client) | |
| response = client.post(f"/sessions/{SESSION}/chunks", json=payload()) | |
| assert response.status_code == 502 | |
| assert "unchecked" not in response.text | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload(1)).status_code == 409 | |
| def test_separate_inference_secret_supports_private_hf_ingress(): | |
| client, native = client_and_native() | |
| with client: | |
| response = client.get("/health", headers={"Authorization": "Bearer hf_ingress_placeholder", "X-Inference-Token": TOKEN}) | |
| assert response.status_code == 200 | |
| rejected = client.get("/health", headers={"Authorization": "Bearer " + TOKEN, "X-Inference-Token": "wrong"}) | |
| assert rejected.status_code == 401 | |
| def test_real_application_http_client_to_native_service_contract(): | |
| import asyncio | |
| from types import SimpleNamespace | |
| import httpx | |
| from pydantic import SecretStr | |
| from backend.perception import PerceptionClient | |
| from backend.schemas import AVChunk as AppChunk | |
| async def exercise(): | |
| upstream = UpstreamShapedDuplex() | |
| app = create_app(NativeDuplex(upstream), TOKEN) | |
| settings = SimpleNamespace(app_mode="real", inference_service_token=SecretStr(TOKEN), hf_token=None, | |
| minicpm_service_url="http://127.0.0.1:8090", inference_timeout_seconds=5) | |
| async with app.router.lifespan_context(app): | |
| async with httpx.AsyncClient(transport=httpx.ASGITransport(app=app), base_url=settings.minicpm_service_url, | |
| headers={"X-Inference-Token": TOKEN}) as http: | |
| client = PerceptionClient(settings, SESSION, client=http) | |
| await client.start("Report English audiovisual evidence") | |
| for sequence in range(3): | |
| encoded = payload(sequence) | |
| encoded.pop("session_id") | |
| encoded.pop("evidence_ids") | |
| chunk = AppChunk(**encoded) | |
| response = await client.process(chunk) | |
| assert response["sequence"] == sequence | |
| assert response["context_evidence_ids"][-1] == f"{SESSION}:e0:c{sequence}" | |
| report = await client.finish(0) | |
| assert report["complete"] and report["model_consumed_sample_count"] == 48000 | |
| assert await client.stop() | |
| assert upstream.calls.count("prepare") == 1 | |
| asyncio.run(exercise()) | |
| def test_native_timeout_does_not_retry_an_uncertain_request(monkeypatch): | |
| monkeypatch.setenv("NATIVE_CALL_TIMEOUT_SECONDS", "0.05") | |
| client, native = client_and_native() | |
| release = threading.Event() | |
| original_generate = native.streaming_generate | |
| def timeout(**kwargs): | |
| release.wait(2) | |
| return original_generate(**kwargs) | |
| native.streaming_generate = timeout | |
| with client: | |
| start(client) | |
| try: | |
| first = client.post(f"/sessions/{SESSION}/chunks", json=payload()) | |
| assert first.status_code == 504 | |
| assert client.post(f"/sessions/{SESSION}/chunks", json=payload()).status_code == 409 | |
| assert client.post(f"/sessions/{SESSION}/finish", json={"epoch": 0}).status_code == 409 | |
| finally: | |
| release.set() | |
| def test_stop_during_model_loading_never_prefills_prompt_after_stop(): | |
| upstream = UpstreamShapedDuplex() | |
| adapter = NativeDuplex(upstream) | |
| entered, release = threading.Event(), threading.Event() | |
| def slow_load(): | |
| entered.set() | |
| release.wait(3) | |
| adapter.load = slow_load | |
| client = TestClient(create_app(adapter, TOKEN), headers={"X-Inference-Token": TOKEN}) | |
| with client, ThreadPoolExecutor(max_workers=1) as pool: | |
| future = pool.submit(client.post, "/sessions", json={"session_id": SESSION, "epoch": 0, "objective": "Read evidence"}) | |
| assert entered.wait(2) | |
| stopped = client.post(f"/sessions/{SESSION}/stop", json={"epoch": 0}) | |
| assert stopped.status_code == 200 | |
| release.set() | |
| assert future.result(timeout=5).status_code == 409 | |
| assert "prepare" not in upstream.calls | |
| assert "prefill" not in upstream.calls | |
| assert "generate" not in upstream.calls | |
| def test_bounded_native_debug_history_preserves_selected_repetition_window(): | |
| upstream = UpstreamShapedDuplex() | |
| adapter = NativeDuplex(upstream) | |
| upstream.decoder.generated_tokens = list(range(12000)) | |
| upstream.decoder.generated_special_tokens = list(range(12000)) | |
| upstream.decoder.context = "x" * 20000 | |
| upstream.total_hidden = [object()] * 100 | |
| upstream.res_ids = list(range(3000)) | |
| recent = upstream.decoder.generated_tokens[-512:] | |
| adapter._bound_debug_state() | |
| assert upstream.decoder.generated_tokens[-512:] == recent | |
| assert len(upstream.decoder.generated_tokens) == 4096 | |
| assert len(upstream.decoder.generated_special_tokens) == 4096 | |
| assert len(upstream.decoder.context) == 8192 | |
| assert len(upstream.res_ids) == 512 and upstream.total_hidden == [] | |