Spaces:
Sleeping
Sleeping
| """Real HTTP encoder/client/native service boundary, with a fake CUDA decoder only.""" | |
| import asyncio | |
| import httpx | |
| import pytest | |
| from pydantic import SecretStr | |
| from backend.perception import PerceptionClient | |
| from backend.schemas import AVChunk, StartRequest | |
| from backend.session import Session | |
| from backend.settings import Settings | |
| from gpu_service.app import create_app | |
| from gpu_service.native import NativeDuplex | |
| from tests.test_api import media_chunk, start_body | |
| class Decoder: | |
| def __init__(self): | |
| self.units = 0 | |
| def get_window_stats(self): | |
| return {"unit_count": min(self.units, 30), "cache_length": min(self.units * 200, 8000)} | |
| def get_cache_length(self): | |
| return min(self.units * 200, 8000) | |
| def _unit_history(self): | |
| return [None] * min(self.units, 30) | |
| def reset(self): | |
| self.units = 0 | |
| class DuplexFixture: | |
| """Mirrors the locked native first-chunk/sample-remainder contract, not media encoding.""" | |
| def __init__(self): | |
| import numpy as np | |
| self.decoder = Decoder() | |
| self.audio_buffer = np.zeros(0, dtype=np.float32) | |
| self.current_turn_ended = True | |
| self.first = True | |
| self.calls = 0 | |
| self.speak_token_id = 1 | |
| self.pending_logits = np.zeros((1, 4), dtype=np.float32) | |
| def prepare(self, **kwargs): | |
| self._reset_streaming_state() | |
| def _reset_streaming_state(self): | |
| import numpy as np | |
| self.audio_buffer = np.zeros(0, dtype=np.float32) | |
| self.first = True | |
| self.decoder.reset() | |
| def set_session_stop(self): | |
| pass | |
| def streaming_prefill(self, audio_waveform, frame_list, **kwargs): | |
| import numpy as np | |
| assert audio_waveform.dtype == np.float32 | |
| assert audio_waveform.ndim == 1 | |
| assert all(frame.mode == "RGB" for frame in frame_list) | |
| prefix = np.zeros(560 if self.first else 0, dtype=np.float32) | |
| self.audio_buffer = np.concatenate([self.audio_buffer, prefix, audio_waveform]) | |
| self.audio_buffer = self.audio_buffer[16480 if self.first else 16000:] | |
| self.first = False | |
| self.pending_logits = np.zeros((1, 4), dtype=np.float32) | |
| self.decoder.units += 1 | |
| return {"success": True} | |
| def streaming_generate(self, **kwargs): | |
| self.calls += 1 | |
| return {"text": "Evidence: The fixture reports an uncertain supply outlook.\nHolding: xom\nImplication: If supply disruptions affect the issuer operations, delivery reliability could weaken.\nUncertainty: The issuer exposure is unverified.\nVerify: Check affected facility locations in issuer disclosures." if self.calls % 20 == 1 else "", | |
| "is_listen": self.calls % 20 != 1, "end_of_turn": True, "current_time": self.calls} | |
| async def test_full_600_seconds_through_app_client_and_native_encoder_contract(): | |
| native = NativeDuplex(DuplexFixture()) | |
| gpu_app = create_app(native=native, token="fixture-service-token-long-enough") | |
| cfg = Settings(app_mode="real", inference_service_token=SecretStr("fixture-service-token-long-enough"), | |
| mock_latency_seconds=0) | |
| # Real perception HTTP adapter + actual GPU-service parser/decode/worker. | |
| # Only the expensive CUDA decoder is replaced by a deterministic fixture. | |
| async with gpu_app.router.lifespan_context(gpu_app): | |
| async with httpx.AsyncClient(transport=httpx.ASGITransport(app=gpu_app), base_url="http://gpu", | |
| headers={"X-Inference-Token": "fixture-service-token-long-enough"}) as http: | |
| session = Session(StartRequest.model_validate(start_body(600000)), cfg) | |
| session.perception = PerceptionClient(cfg, session.id, client=http) | |
| await session.start() | |
| for sequence in range(600): | |
| await session.accept(AVChunk.model_validate(media_chunk(sequence))) | |
| # Simulates finite-source pacing instead of discarding queued speech. | |
| await session.queue.join() | |
| await session.complete() | |
| await session.completion_task | |
| assert session.state == "completed" | |
| assert session.coverage["processed_chunks"] == 600 | |
| assert session.coverage["processed_audio_ms"] == 600000 | |
| assert session.coverage["model_consumed_end_ms"] == 600000 | |
| assert session.coverage["complete"], session.coverage | |
| assert native.sample_count_consumed == 9_600_000 | |
| assert native.duplex.calls == 601 # Native retained5ms is drained at EOF. | |
| assert session.brief is not None | |
| assert session.brief["implications"][0]["holding_id"] == "xom" | |
| assert session.coverage["final_summary_requests"] == 0 | |
| assert session.coverage["attempts"] == 600 | |
| assert len(session.evidence) <= 720 | |