diff --git a/inference_engine/server/grpc_app.py b/inference_engine/server/grpc_app.py index 7446c0ed..c9aaf5b0 100644 --- a/inference_engine/server/grpc_app.py +++ b/inference_engine/server/grpc_app.py @@ -42,11 +42,34 @@ ) from inference_engine.session import ( AppendTokensCoordinator, + DoneEvent, + GenerationCoordinator, + HistoryTruncatedEvent, InvariantViolation, SessionNotFoundError, SessionStore, + STOP_REASON_CANCELLED, + STOP_REASON_EOS, + STOP_REASON_MAX_TOKENS, + STOP_REASON_TRUNCATED, + TokenEvent, ) + +# Mapping from GenerationCoordinator's string stop reasons to the +# protobuf enum. Defined at module level so reviewers can audit the +# 1:1 correspondence at a glance. +_STOP_REASON_TO_PROTO = { + STOP_REASON_MAX_TOKENS: + runtime_pb2.GenerateDone.STOP_REASON_MAX_TOKENS, + STOP_REASON_EOS: + runtime_pb2.GenerateDone.STOP_REASON_EOS, + STOP_REASON_CANCELLED: + runtime_pb2.GenerateDone.STOP_REASON_CANCELLED, + STOP_REASON_TRUNCATED: + runtime_pb2.GenerateDone.STOP_REASON_TRUNCATED, +} + _logger = logging.getLogger(__name__) DEFAULT_BIND_ADDRESS = "127.0.0.1:50051" @@ -113,18 +136,25 @@ def __init__( session_store: SessionStore, *, append_coordinator: Optional[AppendTokensCoordinator] = None, + generation_coordinator: Optional[GenerationCoordinator] = None, ) -> None: """Construct a Servicer. - ``append_coordinator`` is the wiring point added in PR-B2. When - ``None`` (the PR-B1 mode, preserved for tests that don't need a - verifier), ``AppendTokens`` returns ``UNIMPLEMENTED`` — the same - framework default used in PR-B1. When non-None, ``AppendTokens`` - runs the §2.3 byte-exact prefill-incremental contract through - the coordinator and surfaces the typed error mapping above. + ``append_coordinator`` is the PR-B2 wiring point: when None + (PR-B1 mode, preserved for tests that don't need a verifier), + ``AppendTokens`` returns ``UNIMPLEMENTED``; when non-None, + ``AppendTokens`` runs the §2.3 byte-exact prefill-incremental + contract. + + ``generation_coordinator`` is the PR-B3 wiring point: same + optional-default contract for ``Generate``. When None, the + Generate stream returns ``UNIMPLEMENTED``; when non-None, + Generate streams TokenEvents / HistoryTruncatedEvents / + DoneEvent through the gRPC server-streaming response. """ self._store = session_store self._append = append_coordinator + self._generate = generation_coordinator async def CreateSession( # noqa: N802 — gRPC-generated method casing self, @@ -184,6 +214,111 @@ async def AppendTokens( # noqa: N802 — gRPC-generated method casing history_length=new_history_length, ) + async def Generate( # noqa: N802 — gRPC-generated method casing + self, + request: runtime_pb2.GenerateRequest, + context: grpc.aio.ServicerContext, + ): + """Stream tokens generated against ``request.session_id``. + + Yields ``runtime_pb2.GenerateResponse`` frames carrying one of: + + * ``token_id``: a committed token, in generation order. + * ``truncated``: ``HistoryTruncated`` event, emitted at most + once per call before the first ``token_id`` (per the proto + contract). + * ``done``: ``GenerateDone`` terminal frame. + + When this Servicer was constructed without a + ``generation_coordinator``, returns ``UNIMPLEMENTED`` (PR-B2 + regression contract preserved). + + Cancellation: the loop polls ``context.cancelled()`` after + every event the coordinator yields. On cancellation we emit + a ``GenerateDone(STOP_REASON_CANCELLED)`` frame and return. + Cancellation latency is bounded by one generation step on + the worst case (the in-flight forward pass finishes before + the next poll). + """ + if self._generate is None: + await context.abort( + grpc.StatusCode.UNIMPLEMENTED, + "Generate not configured on this Servicer " + "(coordinator not provided)", + ) + + seed = request.seed if request.HasField("seed") else None + temperature = ( + request.temperature + if request.HasField("temperature") else None + ) + top_p = request.top_p if request.HasField("top_p") else None + top_k = request.top_k if request.HasField("top_k") else None + + # GenerationCoordinator.generate is a generator function; the + # call itself returns a generator object without executing + # any of the body, so no exception is raised here. All typed + # errors (SessionNotFoundError, ValueError, InvariantViolation) + # propagate from the inner `for event in event_stream:` loop + # below and are caught there. + event_stream = self._generate.generate( + session_id=request.session_id, + max_tokens=request.max_tokens, + seed=seed, + temperature=temperature, + top_p=top_p, + top_k=top_k, + ) + + token_count_so_far = 0 + + try: + for event in event_stream: + if context.cancelled(): + yield runtime_pb2.GenerateResponse( + done=runtime_pb2.GenerateDone( + stop_reason=_STOP_REASON_TO_PROTO[ + STOP_REASON_CANCELLED + ], + generated_token_count=token_count_so_far, + prefill_duration_seconds=0.0, + total_duration_seconds=0.0, + ), + ) + return + + if isinstance(event, TokenEvent): + token_count_so_far += 1 + yield runtime_pb2.GenerateResponse( + token_id=event.token_id, + ) + elif isinstance(event, HistoryTruncatedEvent): + yield runtime_pb2.GenerateResponse( + truncated=runtime_pb2.HistoryTruncated( + dropped_token_count=event.dropped_token_count, + ), + ) + else: + # DoneEvent — the only remaining event type per + # the GenerateEvent union. + assert isinstance(event, DoneEvent) + yield runtime_pb2.GenerateResponse( + done=runtime_pb2.GenerateDone( + stop_reason=_STOP_REASON_TO_PROTO[ + event.stop_reason + ], + generated_token_count=event.generated_token_count, + prefill_duration_seconds=event.prefill_seconds, + total_duration_seconds=event.total_seconds, + ), + ) + except SessionNotFoundError as exc: + await context.abort(grpc.StatusCode.NOT_FOUND, str(exc)) + except ValueError as exc: + await context.abort(grpc.StatusCode.INVALID_ARGUMENT, str(exc)) + except InvariantViolation as exc: + await context.abort(grpc.StatusCode.FAILED_PRECONDITION, str(exc)) + async def CloseSession( # noqa: N802 self, request: runtime_pb2.CloseSessionRequest, @@ -233,6 +368,7 @@ def create_grpc_server( *, session_store: SessionStore, append_coordinator: Optional[AppendTokensCoordinator] = None, + generation_coordinator: Optional[GenerationCoordinator] = None, config: Optional[GrpcServerConfig] = None, ) -> grpc.aio.Server: """Build, but do not start, a configured gRPC asyncio server. @@ -264,7 +400,9 @@ def create_grpc_server( ) runtime_pb2_grpc.add_RuntimeServiceServicer_to_server( RuntimeServiceServicer( - session_store, append_coordinator=append_coordinator, + session_store, + append_coordinator=append_coordinator, + generation_coordinator=generation_coordinator, ), server, ) diff --git a/inference_engine/session/__init__.py b/inference_engine/session/__init__.py index 5cae8c9f..2e5ebe3e 100644 --- a/inference_engine/session/__init__.py +++ b/inference_engine/session/__init__.py @@ -20,6 +20,17 @@ AppendTokensCoordinator, VerifierProtocol, ) +from inference_engine.session.generator import ( + DoneEvent, + GenerateEvent, + GenerationCoordinator, + HistoryTruncatedEvent, + STOP_REASON_CANCELLED, + STOP_REASON_EOS, + STOP_REASON_MAX_TOKENS, + STOP_REASON_TRUNCATED, + TokenEvent, +) from inference_engine.session.store import ( CacheInspector, InvariantViolation, @@ -32,10 +43,19 @@ __all__ = [ "AppendTokensCoordinator", "CacheInspector", + "DoneEvent", + "GenerateEvent", + "GenerationCoordinator", + "HistoryTruncatedEvent", "InvariantViolation", + "STOP_REASON_CANCELLED", + "STOP_REASON_EOS", + "STOP_REASON_MAX_TOKENS", + "STOP_REASON_TRUNCATED", "Session", "SessionNotFoundError", "SessionStore", "SessionStoreError", + "TokenEvent", "VerifierProtocol", ] diff --git a/inference_engine/session/generator.py b/inference_engine/session/generator.py new file mode 100644 index 00000000..a34efd26 --- /dev/null +++ b/inference_engine/session/generator.py @@ -0,0 +1,242 @@ +"""GenerationCoordinator — ADR 0008 PR-B3 (Phase B). + +Session-aware token generation against a verifier. v0.3 ships +**greedy decoding only**; speculative-decoding integration (the +DLM proposer + AR verifier rejection sampling that is Kakeya's +distinguishing feature) is reserved for a later PR. The wire +contract — :class:`runtime_pb2.GenerateResponse` with its +``token_id`` / ``done`` / ``truncated`` ``oneof`` payload — is +algorithm-agnostic, so the upgrade path lands without breaking +clients. + +The coordinator yields a stream of typed events: + + * :class:`TokenEvent` — one per committed token, in order + * :class:`HistoryTruncatedEvent` — emitted at most once at the + start of a Generate call when the session is already operating + in sink+window-truncated mode (per `runtime.proto` contract: + "Emitted at most once per Generate call, before any token_id + event in that call.") + * :class:`DoneEvent` — terminal; emitted exactly once at the end + +Layering note: this coordinator depends on the same +:class:`VerifierProtocol` PR-B2 introduced. It does not call +``verifier.prefill`` — that is the AppendTokens path's +responsibility (PR-B2). Generate operates on whatever cache state +:meth:`AppendTokensCoordinator.append_tokens` left behind, which +is precisely the byte-exact prefill-incremental contract from +ADR 0008 §2.3 in action. + +Anomaly invariants: + + * INV-1 (parallel-sequence consistency): enforced after every + generated token via :meth:`SessionStore.append_tokens`'s + INV-1 check (the same check PR-B2's coordinator triggers on + user-submitted tokens). + * INV-2 (position monotonicity): enforced after every token + via :meth:`SessionStore.record_position_advance`. + * INV-3 (continuation-path determinism): for the same + ``(session_id, history_token_ids)`` pair under greedy + decoding, repeated Generate calls produce bit-identical token + sequences. Tested with a deterministic ``FakeVerifier`` in + the unit suite and against the real Qwen3 verifier under + ``tests/core/`` on Mac M4. +""" + +from __future__ import annotations + +import time +from dataclasses import dataclass +from typing import Iterator, Optional, Union + +import torch + +from inference_engine.session.coordinator import VerifierProtocol +from inference_engine.session.store import SessionStore + + +# Stop-reason string constants. Mirror the protobuf enum but defined +# here so this module has no protobuf dependency (the gRPC servicer +# does the string -> enum translation). +STOP_REASON_MAX_TOKENS = "max_tokens" +STOP_REASON_EOS = "eos" +STOP_REASON_CANCELLED = "cancelled" +STOP_REASON_TRUNCATED = "truncated" + + +@dataclass(frozen=True) +class TokenEvent: + """One committed token, yielded in generation order.""" + + token_id: int + + +@dataclass(frozen=True) +class HistoryTruncatedEvent: + """Cache no longer holds the full session history. + + ``dropped_token_count`` is the difference between the session's + full history length and what the verifier's sink+window cache + currently holds. Per the runtime contract this event is + emitted at most once per Generate call, before any TokenEvent. + """ + + dropped_token_count: int + + +@dataclass(frozen=True) +class DoneEvent: + """Terminal event for a Generate call. + + ``prefill_seconds`` is 0.0 in PR-B3 because Generate has no + separate prefill phase — the prefill ran inside the preceding + AppendTokens call. The field is preserved on the wire for + forward-compatibility with future PRs that re-introduce a + prefill step (e.g., for speculative-decoding warmup). + """ + + stop_reason: str + generated_token_count: int + prefill_seconds: float + total_seconds: float + + +GenerateEvent = Union[TokenEvent, HistoryTruncatedEvent, DoneEvent] + + +class GenerationCoordinator: + """Greedy session-aware token generation against a verifier.""" + + def __init__( + self, + store: SessionStore, + verifier: VerifierProtocol, + ) -> None: + self._store = store + self._verifier = verifier + + def generate( + self, + session_id: str, + *, + max_tokens: int, + seed: Optional[int] = None, + temperature: Optional[float] = None, + top_p: Optional[float] = None, + top_k: Optional[int] = None, + ) -> Iterator[GenerateEvent]: + """Yield a stream of GenerateEvents for ``session_id``. + + Raises: + * :class:`SessionNotFoundError` — unknown / closed / evicted + session id. + * :class:`ValueError` — invalid argument (e.g., + ``max_tokens < 1``, sampling param set in v0.3 greedy + mode, no AppendTokens preceded this call). + * :class:`InvariantViolation` — INV-1 / INV-2 violation + during a generation step. + + v0.3 greedy contract: + * ``temperature`` / ``top_p`` / ``top_k`` MUST be unset + (or in their no-op default of 0 / unset / 1 + respectively). Setting any of them raises ValueError — + the runtime refuses to silently downgrade a non-greedy + request to greedy (per ADR 0008 §2.10 "no graceful + degradation"). + * ``seed`` is accepted (per OQ-4 default) but ignored: + greedy decoding has no RNG to seed, and the + byte-exact contract over a fixed seed reduces to the + byte-exact contract under any seed because there is no + seed-dependent randomness. + """ + if max_tokens < 1: + raise ValueError( + f"max_tokens must be >= 1, got {max_tokens}" + ) + if temperature is not None and float(temperature) != 0.0: + raise ValueError( + f"v0.3 supports only greedy decoding; temperature " + f"must be 0 or unset, got {temperature}" + ) + if top_p is not None: + raise ValueError( + "v0.3 supports only greedy decoding; top_p must be " + "unset (greedy ignores it)" + ) + if top_k is not None and int(top_k) != 1: + raise ValueError( + f"v0.3 supports only greedy decoding; top_k must be " + f"1 or unset, got {top_k}" + ) + # seed is accepted but not used in greedy; explicitly ignore. + del seed + + session = self._store.get_session(session_id) + if session.next_global_position == 0: + raise ValueError( + "session has no history yet; AppendTokens must " + "precede Generate (the first token's logits are " + "the prefill's last position)" + ) + + # Emit HistoryTruncated at start if the cache is already in + # truncated mode. Per the proto contract, this event is + # emitted at most once per Generate call and BEFORE any + # token_id event — we honor both by checking once at the + # start and never emitting again during this call. + history_len = len(session.history_token_ids) + cached_len = len(session.cached_token_sequence) + if history_len > cached_len: + yield HistoryTruncatedEvent( + dropped_token_count=history_len - cached_len, + ) + + eos_set = set(session.eos_token_ids) + t0 = time.perf_counter() + # Generate has no separate prefill phase in PR-B3; report 0. + prefill_seconds = 0.0 + generated_count = 0 + + for _step in range(max_tokens): + # Greedy: argmax of the verifier's last next_token_logits. + next_token = int( + torch.argmax(self._verifier.next_token_logits).item() + ) + + # Forward + commit (forwarded == accepted for prompt-mode + # appends; same contract used by AppendTokens, just one + # token at a time). + block_logits = self._verifier.forward_block([next_token]) + self._verifier.commit_or_truncate(forwarded=1, accepted=1) + self._verifier.next_token_logits = block_logits[-1].clone() + + # Mirror state from verifier onto session BEFORE the + # store's INV-1 check runs (it compares + # session.cached_token_sequence length against + # verifier.k_seq_length). + session.cached_token_sequence = list( + self._verifier.cached_token_sequence, + ) + self._store.append_tokens(session_id, [next_token]) + self._store.record_position_advance( + session_id, self._verifier.next_global_position, + ) + generated_count += 1 + + yield TokenEvent(token_id=next_token) + + if next_token in eos_set: + yield DoneEvent( + stop_reason=STOP_REASON_EOS, + generated_token_count=generated_count, + prefill_seconds=prefill_seconds, + total_seconds=time.perf_counter() - t0, + ) + return + + yield DoneEvent( + stop_reason=STOP_REASON_MAX_TOKENS, + generated_token_count=generated_count, + prefill_seconds=prefill_seconds, + total_seconds=time.perf_counter() - t0, + ) diff --git a/results/platform-tests/pr-b3-mac-generator-tests-1780323650.coverage.xml b/results/platform-tests/pr-b3-mac-generator-tests-1780323650.coverage.xml new file mode 100644 index 00000000..96790ba5 --- /dev/null +++ b/results/platform-tests/pr-b3-mac-generator-tests-1780323650.coverage.xml @@ -0,0 +1,82 @@ + + + + + + /Users/fluffy314/Documents/Kakeya-LLM-Inference-engine + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/results/platform-tests/pr-b3-mac-generator-tests-1780323650.json b/results/platform-tests/pr-b3-mac-generator-tests-1780323650.json new file mode 100644 index 00000000..d2933a19 --- /dev/null +++ b/results/platform-tests/pr-b3-mac-generator-tests-1780323650.json @@ -0,0 +1,209 @@ +{ + "schema_version": 1, + "kind": "pr_b3_mac_generator_tests", + "host": { + "platform": "macOS-26.5-arm64-arm-64bit-Mach-O", + "machine": "arm64", + "python": "3.13.12" + }, + "junit": { + "tests": 31, + "failures": 0, + "errors": 0, + "skipped": 0, + "cases": [ + { + "classname": "tests.inference_engine.session.test_generator.TestGreedyHappyPath", + "name": "test_yields_token_then_done", + "time": 0.001, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestGreedyHappyPath", + "name": "test_max_tokens_caps_token_emission", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestGreedyHappyPath", + "name": "test_done_is_terminal_and_unique", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestGreedyHappyPath", + "name": "test_done_includes_total_seconds", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestGreedyAdvancesVerifier", + "name": "test_each_token_advances_position_by_one", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestGreedyAdvancesVerifier", + "name": "test_session_history_grows", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestGreedyAdvancesVerifier", + "name": "test_session_cached_token_sequence_mirrors_verifier", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestGreedyAdvancesVerifier", + "name": "test_each_token_calls_forward_then_commit", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestEos", + "name": "test_eos_token_terminates_with_eos_stop_reason", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestEos", + "name": "test_no_eos_runs_to_max_tokens", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestEos", + "name": "test_empty_eos_set_runs_to_max_tokens", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestHistoryTruncated", + "name": "test_emitted_at_start_when_already_truncated", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestHistoryTruncated", + "name": "test_not_emitted_when_under_capacity", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestHistoryTruncated", + "name": "test_at_most_one_per_call", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_max_tokens_zero_rejected", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_max_tokens_negative_rejected", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_temperature_nonzero_rejected", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_temperature_zero_accepted", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_top_p_set_rejected", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_top_k_other_than_one_rejected", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_top_k_one_accepted", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_seed_accepted_and_ignored_in_greedy", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_no_appendtokens_first_rejected", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestValidation", + "name": "test_unknown_session_raises_session_not_found", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestInvariants", + "name": "test_inv1_violation_propagates", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestInvariants", + "name": "test_inv2_violation_propagates", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestDeterminism", + "name": "test_two_runs_with_same_history_produce_same_tokens", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestConstructorAndEventDataclasses", + "name": "test_constructor_stores_references", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestConstructorAndEventDataclasses", + "name": "test_token_event_is_frozen", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestConstructorAndEventDataclasses", + "name": "test_history_truncated_event_is_frozen", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.session.test_generator.TestConstructorAndEventDataclasses", + "name": "test_done_event_is_frozen", + "time": 0.0, + "outcome": "passed" + } + ] + }, + "coverage": { + "line_rate": 1.0, + "branch_rate": 0.0, + "lines_covered": 63, + "lines_valid": 63 + } +} \ No newline at end of file diff --git a/results/platform-tests/pr-b3-mac-generator-tests-1780323650.junit.xml b/results/platform-tests/pr-b3-mac-generator-tests-1780323650.junit.xml new file mode 100644 index 00000000..a05a9094 --- /dev/null +++ b/results/platform-tests/pr-b3-mac-generator-tests-1780323650.junit.xml @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/results/platform-tests/pr-b3-mac-grpc-appender-smoke-1780323650.json b/results/platform-tests/pr-b3-mac-grpc-appender-smoke-1780323650.json new file mode 100644 index 00000000..d841e6cb --- /dev/null +++ b/results/platform-tests/pr-b3-mac-grpc-appender-smoke-1780323650.json @@ -0,0 +1,132 @@ +{ + "schema_version": 1, + "kind": "grpc_appender_smoke", + "host": { + "platform": "macOS-26.5-arm64-arm-64bit-Mach-O", + "machine": "arm64", + "python": "3.13.12", + "grpc": "1.81.0" + }, + "steps_total": 10, + "steps_passed": 10, + "steps_failed": 0, + "all_passed": true, + "steps": [ + { + "step": "CreateSession", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "session_id": "sess-bde5b331375e4a76a82af3d4f91306af" + }, + "elapsed_ms": 0.99 + }, + { + "step": "AppendTokens (cold prefill)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "history_length": 3, + "verifier_calls": [ + "prefill" + ] + }, + "elapsed_ms": 0.48 + }, + { + "step": "GetSessionInfo (after cold)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "history_length": 3, + "inv1_violations": 0, + "inv2_violations": 0 + }, + "elapsed_ms": 0.32 + }, + { + "step": "AppendTokens (incremental)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "history_length": 5, + "verifier_calls_total": [ + "prefill", + "forward_block", + "commit_or_truncate" + ] + }, + "elapsed_ms": 0.34 + }, + { + "step": "GetSessionInfo (after incremental)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "history_length": 5 + }, + "elapsed_ms": 0.31 + }, + { + "step": "AppendTokens (empty list, no-op)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "history_length": 5, + "verifier_calls_after_empty": [ + "prefill", + "forward_block", + "commit_or_truncate" + ] + }, + "elapsed_ms": 0.29 + }, + { + "step": "AppendTokens (unknown session)", + "expected": "NOT_FOUND", + "observed": "NOT_FOUND", + "passed": true, + "detail": { + "details": "session_id 'sess-nonexistent' not found" + }, + "elapsed_ms": 0.32 + }, + { + "step": "AppendTokens (after CloseSession)", + "expected": "NOT_FOUND", + "observed": "NOT_FOUND", + "passed": true, + "detail": { + "details": "session_id 'sess-bde5b331375e4a76a82af3d4f91306af' not found" + }, + "elapsed_ms": 0.3 + }, + { + "step": "Generate (PR-B3, still UNIMPLEMENTED)", + "expected": "UNIMPLEMENTED", + "observed": "UNIMPLEMENTED", + "passed": true, + "detail": { + "phase": "PR-B3" + }, + "elapsed_ms": 0.35 + }, + { + "step": "AppendTokens (INV-1 violation)", + "expected": "FAILED_PRECONDITION", + "observed": "FAILED_PRECONDITION", + "passed": true, + "detail": { + "details": "INV-1 violation in session 'sess-89aefa3e276144388ce057c6fbf27c80': cached_token_sequence length (3) != K/V tensor seque", + "kind": "INV-1" + }, + "elapsed_ms": 0.3 + } + ] +} \ No newline at end of file diff --git a/results/platform-tests/pr-b3-mac-grpc-generator-smoke-1780323650.json b/results/platform-tests/pr-b3-mac-grpc-generator-smoke-1780323650.json new file mode 100644 index 00000000..13c6f0a8 --- /dev/null +++ b/results/platform-tests/pr-b3-mac-grpc-generator-smoke-1780323650.json @@ -0,0 +1,132 @@ +{ + "schema_version": 1, + "kind": "grpc_generator_smoke", + "host": { + "platform": "macOS-26.5-arm64-arm-64bit-Mach-O", + "machine": "arm64", + "python": "3.13.12", + "grpc": "1.81.0" + }, + "steps_total": 10, + "steps_passed": 10, + "steps_failed": 0, + "all_passed": true, + "steps": [ + { + "step": "CreateSession", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "session_id": "sess-c9d9f77c02d4498a975518cb6a61319a" + }, + "elapsed_ms": 0.98 + }, + { + "step": "AppendTokens (cold prefill)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "history_length": 3 + }, + "elapsed_ms": 0.48 + }, + { + "step": "Generate (max_tokens=3, no EOS)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "frames": [ + "token_id", + "token_id", + "token_id", + "done" + ], + "tokens_emitted": 3, + "done_frames": 1 + }, + "elapsed_ms": 0.63 + }, + { + "step": "Generate (EOS triggers, max_tokens=10)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "tokens": [ + 6 + ], + "stop_reason": "STOP_REASON_EOS" + }, + "elapsed_ms": 1.01 + }, + { + "step": "Generate (no AppendTokens prior)", + "expected": "INVALID_ARGUMENT", + "observed": "INVALID_ARGUMENT", + "passed": true, + "detail": { + "details": "session has no history yet; AppendTokens must precede Generate (the first token'" + }, + "elapsed_ms": 0.57 + }, + { + "step": "Generate (max_tokens=0)", + "expected": "INVALID_ARGUMENT", + "observed": "INVALID_ARGUMENT", + "passed": true, + "detail": { + "details": "max_tokens must be >= 1, got 0" + }, + "elapsed_ms": 0.33 + }, + { + "step": "Generate (temperature=0.7, non-greedy rejected)", + "expected": "INVALID_ARGUMENT", + "observed": "INVALID_ARGUMENT", + "passed": true, + "detail": { + "details": "v0.3 supports only greedy decoding; temperature must be 0 or" + }, + "elapsed_ms": 0.31 + }, + { + "step": "Generate (unknown session)", + "expected": "NOT_FOUND", + "observed": "NOT_FOUND", + "passed": true, + "detail": { + "details": "session_id 'sess-nonexistent' not found" + }, + "elapsed_ms": 0.34 + }, + { + "step": "Generate (truncated state -> truncated frame)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "frames": [ + "truncated", + "token_id", + "token_id", + "done" + ], + "dropped_token_count": 2 + }, + "elapsed_ms": 1.5 + }, + { + "step": "Generate (no coordinator wired)", + "expected": "UNIMPLEMENTED", + "observed": "UNIMPLEMENTED", + "passed": true, + "detail": { + "details": "Generate not configured on this Servicer (coordinator not provided)" + }, + "elapsed_ms": 0.72 + } + ] +} \ No newline at end of file diff --git a/results/platform-tests/pr-b3-mac-grpc-runtime-smoke-1780323650.json b/results/platform-tests/pr-b3-mac-grpc-runtime-smoke-1780323650.json new file mode 100644 index 00000000..ece46352 --- /dev/null +++ b/results/platform-tests/pr-b3-mac-grpc-runtime-smoke-1780323650.json @@ -0,0 +1,129 @@ +{ + "schema_version": 1, + "kind": "grpc_runtime_smoke", + "host": { + "platform": "macOS-26.5-arm64-arm-64bit-Mach-O", + "machine": "arm64", + "python": "3.13.12", + "grpc": "1.81.0" + }, + "steps_total": 10, + "steps_passed": 10, + "steps_failed": 0, + "all_passed": true, + "steps": [ + { + "step": "CreateSession", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "session_id": "sess-7304326ad3a7452a8aef5c4f877042df", + "port": 54787 + }, + "elapsed_ms": 1.03 + }, + { + "step": "GetSessionInfo (initial)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "history_length": 0, + "kv_live_bytes": 0, + "inv1_violations": 0, + "inv2_violations": 0, + "idle_seconds": 0.0004 + }, + "elapsed_ms": 0.33 + }, + { + "step": "CloseSession", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "final_history_length": 0 + }, + "elapsed_ms": 0.29 + }, + { + "step": "CloseSession (double-close)", + "expected": "NOT_FOUND", + "observed": "NOT_FOUND", + "passed": true, + "detail": { + "details": "session_id 'sess-7304326ad3a7452a8aef5c4f877042df' not found" + }, + "elapsed_ms": 0.36 + }, + { + "step": "GetSessionInfo (after close)", + "expected": "NOT_FOUND", + "observed": "NOT_FOUND", + "passed": true, + "detail": { + "details": "session_id 'sess-7304326ad3a7452a8aef5c4f877042df' not found" + }, + "elapsed_ms": 0.3 + }, + { + "step": "AppendTokens (PR-B2)", + "expected": "UNIMPLEMENTED", + "observed": "UNIMPLEMENTED", + "passed": true, + "detail": { + "phase": "PR-B2" + }, + "elapsed_ms": 0.29 + }, + { + "step": "Generate (PR-B3)", + "expected": "UNIMPLEMENTED", + "observed": "UNIMPLEMENTED", + "passed": true, + "detail": { + "phase": "PR-B3" + }, + "elapsed_ms": 0.34 + }, + { + "step": "CreateSession (eos + client_label)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "session_id": "sess-6458d33e57bd446e9e720c43a29cc191", + "eos_token_ids_recorded": [ + 7, + 11, + 13 + ], + "client_label_recorded": "smoke-demo" + }, + "elapsed_ms": 0.34 + }, + { + "step": "CreateSession (pool slab #1 / 1)", + "expected": "ok", + "observed": "ok", + "passed": true, + "detail": { + "session_id": "sess-7c0f52bf475e480193b78ca8c9740878" + }, + "elapsed_ms": 0.77 + }, + { + "step": "CreateSession (pool exhausted)", + "expected": "RESOURCE_EXHAUSTED", + "observed": "RESOURCE_EXHAUSTED", + "passed": true, + "detail": { + "details": "slab pool exhausted: all 1 slabs in use; admission control must reject or queue this session", + "pool_in_use": 1, + "pool_available": 0 + }, + "elapsed_ms": 0.49 + } + ] +} \ No newline at end of file diff --git a/results/platform-tests/pr-b3-mac-grpc-tests-1780323650.coverage.xml b/results/platform-tests/pr-b3-mac-grpc-tests-1780323650.coverage.xml new file mode 100644 index 00000000..759a0162 --- /dev/null +++ b/results/platform-tests/pr-b3-mac-grpc-tests-1780323650.coverage.xml @@ -0,0 +1,107 @@ + + + + + + /Users/fluffy314/Documents/Kakeya-LLM-Inference-engine + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + diff --git a/results/platform-tests/pr-b3-mac-grpc-tests-1780323650.json b/results/platform-tests/pr-b3-mac-grpc-tests-1780323650.json new file mode 100644 index 00000000..22da2706 --- /dev/null +++ b/results/platform-tests/pr-b3-mac-grpc-tests-1780323650.json @@ -0,0 +1,257 @@ +{ + "schema_version": 1, + "kind": "pr_b3_mac_grpc_tests", + "host": { + "platform": "macOS-26.5-arm64-arm-64bit-Mach-O", + "machine": "arm64", + "python": "3.13.12" + }, + "junit": { + "tests": 39, + "failures": 0, + "errors": 0, + "skipped": 0, + "cases": [ + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_session_returns_server_issued_id", + "time": 0.021, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_session_records_eos_token_ids", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_session_records_client_label", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_session_default_eos_is_empty_tuple", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_session_pool_exhausted_returns_resource_exhausted", + "time": 0.003, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_close_session_returns_final_history_length", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_close_session_returns_zero_for_empty_session", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_close_session_unknown_id_returns_not_found", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_close_session_double_close_returns_not_found", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_get_session_info_initial_state", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_get_session_info_reflects_history_growth", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_get_session_info_reflects_kv_live_bytes_when_pool_present", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_get_session_info_unknown_id_returns_not_found", + "time": 0.001, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_get_session_info_after_close_returns_not_found", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_append_tokens_returns_unimplemented_when_no_coordinator", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_append_tokens_first_call_triggers_prefill", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_append_tokens_subsequent_call_triggers_incremental", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_append_tokens_unknown_session_returns_not_found", + "time": 0.001, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_append_tokens_invariant_violation_returns_failed_precondition", + "time": 0.003, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_append_tokens_value_error_returns_invalid_argument", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_returns_unimplemented_when_no_coordinator", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_streams_tokens_then_done", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_eos_stops_with_eos_stop_reason", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_history_truncated_emitted", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_unknown_session_returns_not_found", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_no_history_returns_invalid_argument", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_max_tokens_zero_returns_invalid_argument", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_temperature_nonzero_returns_invalid_argument", + "time": 0.003, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_seed_is_accepted", + "time": 0.002, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_invariant_violation_returns_failed_precondition", + "time": 0.005, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_grpc_server_accepts_generation_coordinator", + "time": 0.001, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_generate_cancellation_emits_cancelled_done", + "time": 0.001, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_default_bind_address_is_loopback", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_grpc_server_config_defaults", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_grpc_server_config_is_frozen", + "time": 0.0, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_grpc_server_default_config_binds_default_address", + "time": 0.001, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_grpc_server_with_no_config_uses_defaults", + "time": 0.001, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_grpc_server_with_max_concurrent_rpcs", + "time": 0.001, + "outcome": "passed" + }, + { + "classname": "tests.inference_engine.server.test_grpc_app", + "name": "test_create_grpc_server_accepts_append_coordinator", + "time": 0.001, + "outcome": "passed" + } + ] + }, + "coverage": { + "line_rate": 1.0, + "branch_rate": 0.0, + "lines_covered": 88, + "lines_valid": 88 + } +} \ No newline at end of file diff --git a/results/platform-tests/pr-b3-mac-grpc-tests-1780323650.junit.xml b/results/platform-tests/pr-b3-mac-grpc-tests-1780323650.junit.xml new file mode 100644 index 00000000..1882be1d --- /dev/null +++ b/results/platform-tests/pr-b3-mac-grpc-tests-1780323650.junit.xml @@ -0,0 +1 @@ + \ No newline at end of file diff --git a/scripts/review_pr_b3_on_mac.sh b/scripts/review_pr_b3_on_mac.sh new file mode 100755 index 00000000..7d9e4d2c --- /dev/null +++ b/scripts/review_pr_b3_on_mac.sh @@ -0,0 +1,183 @@ +#!/usr/bin/env bash +# Mac M4 review aid for PR-B3 (ADR 0008 Phase B, Generate RPC). +# +# Per ADR 0008 §9, PR-B3 is the FIRST Phase-B PR with a mandatory +# Mac M4 integration test report — Linux CI is necessary but not +# sufficient because Generate exercises the verifier-sampler path +# that has MLX-specific behavior (argmax on bf16 tensors, post-trim +# K/V tensor shapes, etc.). This script produces 5 artifacts under +# results/platform-tests/: +# +# 1. pr-b3-mac-generator-tests-.json +# pytest tests/inference_engine/session/test_generator.py +# (31 tests; 100% line coverage on +# inference_engine/session/generator.py). +# +# 2. pr-b3-mac-grpc-tests-.json +# pytest tests/inference_engine/server/test_grpc_app.py +# (39 tests after PR-B3 additions; 100% line coverage on +# inference_engine/server/grpc_app.py). +# +# 3. pr-b3-mac-grpc-runtime-smoke-.json +# Regression smoke: smoke_grpc_runtime.py (PR-B1 contract, +# AppendTokens + Generate stay UNIMPLEMENTED with a bare +# Servicer). +# +# 4. pr-b3-mac-grpc-appender-smoke-.json +# Regression smoke: smoke_grpc_appender.py (PR-B2 contract, +# AppendTokens reachable + Generate still UNIMPLEMENTED). +# +# 5. pr-b3-mac-grpc-generator-smoke-.json +# New PR-B3 smoke: smoke_grpc_generator.py (10 RPC scenarios +# with Generate fully wired, including HistoryTruncated and +# STOP_REASON_EOS frames). +# +# Usage (from repo root, on Mac M4 / arm64): +# +# bash scripts/review_pr_b3_on_mac.sh +# +# Then commit the artifacts: +# +# git add results/platform-tests/pr-b3-mac-* +# git commit -m "Mac M4 review evidence for PR-B3" +# git push +# +# Same `coverage run -m pytest` + `--include` filter pattern as +# review_pr_b2_on_mac.sh — sidesteps the Python 3.13 / coverage / +# torch race documented in commit 9cb1c56 + 9d1a250 on PR #45. + +set -euo pipefail + +ROOT="$(cd "$(dirname "$0")/.." && pwd)" +cd "$ROOT" + +stamp="$(date +%s)" +out_dir="results/platform-tests" +mkdir -p "$out_dir" + +gen_junit="$out_dir/pr-b3-mac-generator-tests-${stamp}.junit.xml" +gen_cov="$out_dir/pr-b3-mac-generator-tests-${stamp}.coverage.xml" +gen_report="$out_dir/pr-b3-mac-generator-tests-${stamp}.json" + +grpc_junit="$out_dir/pr-b3-mac-grpc-tests-${stamp}.junit.xml" +grpc_cov="$out_dir/pr-b3-mac-grpc-tests-${stamp}.coverage.xml" +grpc_report="$out_dir/pr-b3-mac-grpc-tests-${stamp}.json" + +runtime_smoke="$out_dir/pr-b3-mac-grpc-runtime-smoke-${stamp}.json" +appender_smoke="$out_dir/pr-b3-mac-grpc-appender-smoke-${stamp}.json" +generator_smoke="$out_dir/pr-b3-mac-grpc-generator-smoke-${stamp}.json" + + +_summarize_pytest() { + # $1 junit $2 coverage $3 out $4 kind label + PYTHONPATH=. python3 - "$1" "$2" "$3" "$4" <<'PY' +import json, platform, sys, xml.etree.ElementTree as ET +junit_path, cov_path, out_path, kind = sys.argv[1:5] +jr = ET.parse(junit_path).getroot() + +# Aggregate counts from elements (pytest's --junitxml +# emits the counts on the inner , not on the +# wrapper). Same fix as commit 9d1a250. +testsuites = list(jr.iter("testsuite")) +total_tests = sum(int(ts.get("tests", "0")) for ts in testsuites) +total_failures = sum(int(ts.get("failures", "0")) for ts in testsuites) +total_errors = sum(int(ts.get("errors", "0")) for ts in testsuites) +total_skipped = sum(int(ts.get("skipped", "0")) for ts in testsuites) + +cases = [] +for tc in jr.iter("testcase"): + cases.append({ + "classname": tc.get("classname"), + "name": tc.get("name"), + "time": float(tc.get("time", 0.0)), + "outcome": ( + "failed" if tc.find("failure") is not None + else "errored" if tc.find("error") is not None + else "skipped" if tc.find("skipped") is not None + else "passed" + ), + }) + +cov_root = ET.parse(cov_path).getroot() +report = { + "schema_version": 1, + "kind": kind, + "host": { + "platform": platform.platform(), + "machine": platform.machine(), + "python": platform.python_version(), + }, + "junit": { + "tests": total_tests, + "failures": total_failures, + "errors": total_errors, + "skipped": total_skipped, + "cases": cases, + }, + "coverage": { + "line_rate": float(cov_root.get("line-rate", "0.0")), + "branch_rate": float(cov_root.get("branch-rate", "0.0")), + "lines_covered": int(cov_root.get("lines-covered", "0")), + "lines_valid": int(cov_root.get("lines-valid", "0")), + }, +} +with open(out_path, "w", encoding="utf-8") as fh: + json.dump(report, fh, indent=2) +print(f" -> {out_path}") +PY +} + + +echo "==> [1/5] generator unit tests" +PYTHONPATH=. python3 -m coverage erase +PYTHONPATH=. python3 -m coverage run \ + -m pytest tests/inference_engine/session/test_generator.py \ + --junitxml="$gen_junit" -v +python3 -m coverage report \ + --include='inference_engine/session/generator.py' \ + --fail-under=100 -m +python3 -m coverage xml \ + --include='inference_engine/session/generator.py' \ + -o "$gen_cov" +_summarize_pytest "$gen_junit" "$gen_cov" "$gen_report" \ + "pr_b3_mac_generator_tests" + +echo +echo "==> [2/5] gRPC tests (PR-B1 + PR-B2 + PR-B3 surface)" +PYTHONPATH=. python3 -m coverage erase +PYTHONPATH=. python3 -m coverage run \ + -m pytest tests/inference_engine/server/test_grpc_app.py \ + --junitxml="$grpc_junit" -v +python3 -m coverage report \ + --include='inference_engine/server/grpc_app.py' \ + --fail-under=100 -m +python3 -m coverage xml \ + --include='inference_engine/server/grpc_app.py' \ + -o "$grpc_cov" +_summarize_pytest "$grpc_junit" "$grpc_cov" "$grpc_report" \ + "pr_b3_mac_grpc_tests" + +echo +echo "==> [3/5] runtime smoke (PR-B1 contract regression)" +PYTHONPATH=. python3 scripts/smoke_grpc_runtime.py --report "$runtime_smoke" + +echo +echo "==> [4/5] appender smoke (PR-B2 contract regression)" +PYTHONPATH=. python3 scripts/smoke_grpc_appender.py --report "$appender_smoke" + +echo +echo "==> [5/5] generator smoke (PR-B3 new)" +PYTHONPATH=. python3 scripts/smoke_grpc_generator.py --report "$generator_smoke" + +echo +echo "==> Done." +echo " Generator tests : $gen_report" +echo " gRPC tests : $grpc_report" +echo " Runtime smoke : $runtime_smoke" +echo " Appender smoke : $appender_smoke" +echo " Generator smoke : $generator_smoke" +echo +echo "Next:" +echo " git add $out_dir/pr-b3-mac-*" +echo " git commit -m 'Mac M4 review evidence for PR-B3'" +echo " git push" diff --git a/scripts/smoke_grpc_generator.py b/scripts/smoke_grpc_generator.py new file mode 100755 index 00000000..831e4dd1 --- /dev/null +++ b/scripts/smoke_grpc_generator.py @@ -0,0 +1,363 @@ +"""End-to-end smoke for the PR-B3 Generate server-streaming RPC. + +Spins up a real ``grpc.aio.Server`` with both the AppendTokens +coordinator (PR-B2) and the Generation coordinator (PR-B3) wired +in, and walks the Generate scenarios this PR ships: + + 1. CreateSession -> success + 2. AppendTokens (cold prefill) -> success + 3. Generate (max_tokens=3, no EOS) -> 3 token_id frames + done(MAX_TOKENS) + 4. Generate (max_tokens=10, EOS in token 6) -> 1 token + done(EOS) (deterministic by FakeVerifier) + 5. Generate (no AppendTokens prior) -> INVALID_ARGUMENT + 6. Generate (max_tokens=0) -> INVALID_ARGUMENT + 7. Generate (temperature=0.7) -> INVALID_ARGUMENT + 8. Generate (unknown session) -> NOT_FOUND + 9. Generate (large prefill -> truncated state) -> 1 truncated frame + tokens + done + 10. Generate (no coordinator wired) -> UNIMPLEMENTED + +Each step prints one JSON-Lines record with expected vs observed +outcome. Exit code 0 iff all 10 scenarios match. + +Same review-affordance pattern as PR-B1 (smoke_grpc_runtime.py) and +PR-B2 (smoke_grpc_appender.py). + +Usage:: + + PYTHONPATH=. python3 scripts/smoke_grpc_generator.py \\ + --report results/platform-tests/grpc-generator-smoke-$(date +%s).json +""" + +from __future__ import annotations + +import argparse +import asyncio +import json +import platform +import sys +import time +import traceback +from dataclasses import asdict, dataclass, field +from typing import Any, AsyncIterator, Optional + +import grpc + +from inference_engine.server.grpc_app import RuntimeServiceServicer +from inference_engine.server.proto_gen.kakeya.v1 import ( + runtime_pb2, + runtime_pb2_grpc, +) +from inference_engine.session import ( + AppendTokensCoordinator, + GenerationCoordinator, + SessionStore, +) + +from tests.inference_engine.session.test_coordinator import FakeVerifier + + +@dataclass +class StepResult: + step: str + expected: str + observed: str + passed: bool + detail: dict = field(default_factory=dict) + elapsed_ms: float = 0.0 + + def asline(self) -> str: + return json.dumps(asdict(self), separators=(",", ":")) + + +async def _serve( + store: SessionStore, + append_coord: Optional[AppendTokensCoordinator] = None, + gen_coord: Optional[GenerationCoordinator] = None, +) -> AsyncIterator[tuple[runtime_pb2_grpc.RuntimeServiceStub, grpc.aio.Server, int]]: + server = grpc.aio.server() + runtime_pb2_grpc.add_RuntimeServiceServicer_to_server( + RuntimeServiceServicer( + store, + append_coordinator=append_coord, + generation_coordinator=gen_coord, + ), + server, + ) + port = server.add_insecure_port("127.0.0.1:0") + await server.start() + channel = grpc.aio.insecure_channel(f"127.0.0.1:{port}") + stub = runtime_pb2_grpc.RuntimeServiceStub(channel) + try: + yield stub, server, port + finally: + await channel.close() + await server.stop(grace=0.1) + + +async def _step(name: str, expected: str, body) -> StepResult: + t0 = time.perf_counter() + try: + observed, detail = await body() + passed = observed == expected + except Exception as exc: # noqa: BLE001 + observed = f"unexpected exception: {type(exc).__name__}" + detail = {"traceback": traceback.format_exc()} + passed = False + elapsed_ms = (time.perf_counter() - t0) * 1000 + return StepResult( + step=name, expected=expected, observed=observed, + passed=passed, detail=detail, elapsed_ms=round(elapsed_ms, 2), + ) + + +async def run_smoke(verbose: bool = True) -> list[StepResult]: + results: list[StepResult] = [] + + def emit(r: StepResult) -> None: + results.append(r) + if verbose: + print(r.asline(), flush=True) + + # ----- Server #1: full coordinator wiring (Append + Generate) ----- + fv = FakeVerifier() + store = SessionStore(capacity=4, cache_inspector=fv) + append_coord = AppendTokensCoordinator(store, fv) + gen_coord = GenerationCoordinator(store, fv) + s_ctx = _serve(store, append_coord, gen_coord).__aiter__() + stub, _, _ = await s_ctx.__anext__() + try: + async def _step1(): + r = await stub.CreateSession(runtime_pb2.CreateSessionRequest()) + return ("ok", {"session_id": r.session_id}) + emit(await _step("CreateSession", "ok", _step1)) + sid = results[-1].detail["session_id"] + + async def _step2(): + r = await stub.AppendTokens( + runtime_pb2.AppendTokensRequest( + session_id=sid, token_ids=[1, 2, 3], + ), + ) + return ("ok", {"history_length": r.history_length}) + emit(await _step("AppendTokens (cold prefill)", "ok", _step2)) + + async def _step3(): + frames = [] + async for f in stub.Generate( + runtime_pb2.GenerateRequest(session_id=sid, max_tokens=3), + ): + frames.append(f.WhichOneof("payload")) + done = frames.count("done") + tokens = frames.count("token_id") + return ("ok", { + "frames": frames, + "tokens_emitted": tokens, + "done_frames": done, + }) + emit(await _step( + "Generate (max_tokens=3, no EOS)", "ok", _step3, + )) + + # Step 4: EOS scenario. FakeVerifier's _logits_for produces argmax + # = sum(history[-3:]) % 16; with history [1,2,3] -> argmax=6. + # Create fresh session with eos=[6]; first generated token is 6. + async def _step4(): + r_create = await stub.CreateSession( + runtime_pb2.CreateSessionRequest(eos_token_ids=[6]), + ) + sid2 = r_create.session_id + await stub.AppendTokens( + runtime_pb2.AppendTokensRequest( + session_id=sid2, token_ids=[1, 2, 3], + ), + ) + tokens = [] + stop_reason_name = None + async for f in stub.Generate( + runtime_pb2.GenerateRequest(session_id=sid2, max_tokens=10), + ): + kind = f.WhichOneof("payload") + if kind == "token_id": + tokens.append(f.token_id) + elif kind == "done": + stop_reason_name = ( + runtime_pb2.GenerateDone.StopReason.Name( + f.done.stop_reason, + ) + ) + return ("ok", { + "tokens": tokens, + "stop_reason": stop_reason_name, + }) + emit(await _step( + "Generate (EOS triggers, max_tokens=10)", "ok", _step4, + )) + + # Step 5: Generate without prior AppendTokens. + async def _step5(): + r_create = await stub.CreateSession(runtime_pb2.CreateSessionRequest()) + try: + async for _f in stub.Generate( + runtime_pb2.GenerateRequest( + session_id=r_create.session_id, max_tokens=1, + ), + ): + return ("ok", {}) # would be a bug + return ("ok", {}) + except grpc.aio.AioRpcError as e: + return (e.code().name, {"details": e.details()[:80]}) + emit(await _step( + "Generate (no AppendTokens prior)", "INVALID_ARGUMENT", _step5, + )) + + # Step 6: max_tokens=0 + async def _step6(): + try: + async for _f in stub.Generate( + runtime_pb2.GenerateRequest(session_id=sid, max_tokens=0), + ): + return ("ok", {}) + return ("ok", {}) + except grpc.aio.AioRpcError as e: + return (e.code().name, {"details": e.details()[:60]}) + emit(await _step( + "Generate (max_tokens=0)", "INVALID_ARGUMENT", _step6, + )) + + # Step 7: temperature=0.7 (non-greedy rejected in v0.3) + async def _step7(): + try: + async for _f in stub.Generate( + runtime_pb2.GenerateRequest( + session_id=sid, max_tokens=1, temperature=0.7, + ), + ): + return ("ok", {}) + return ("ok", {}) + except grpc.aio.AioRpcError as e: + return (e.code().name, {"details": e.details()[:60]}) + emit(await _step( + "Generate (temperature=0.7, non-greedy rejected)", + "INVALID_ARGUMENT", _step7, + )) + + # Step 8: unknown session + async def _step8(): + try: + async for _f in stub.Generate( + runtime_pb2.GenerateRequest( + session_id="sess-nonexistent", max_tokens=1, + ), + ): + return ("ok", {}) + return ("ok", {}) + except grpc.aio.AioRpcError as e: + return (e.code().name, {"details": e.details()[:60]}) + emit(await _step( + "Generate (unknown session)", "NOT_FOUND", _step8, + )) + + # Step 9: large prefill -> truncated state + async def _step9(): + r_create = await stub.CreateSession(runtime_pb2.CreateSessionRequest()) + await stub.AppendTokens( + runtime_pb2.AppendTokensRequest( + session_id=r_create.session_id, + token_ids=[10, 20, 30, 40, 50, 60, 70, 80], + ), + ) + frame_kinds = [] + truncated_dropped = None + async for f in stub.Generate( + runtime_pb2.GenerateRequest( + session_id=r_create.session_id, max_tokens=2, + ), + ): + kind = f.WhichOneof("payload") + frame_kinds.append(kind) + if kind == "truncated": + truncated_dropped = f.truncated.dropped_token_count + return ("ok", { + "frames": frame_kinds, + "dropped_token_count": truncated_dropped, + }) + emit(await _step( + "Generate (truncated state -> truncated frame)", "ok", _step9, + )) + finally: + try: + await s_ctx.__anext__() + except StopAsyncIteration: + pass + + # ----- Server #2: no generation_coordinator wired ----- + fv2 = FakeVerifier() + store2 = SessionStore(capacity=2) + s2_ctx = _serve(store2, None, None).__aiter__() + stub2, _, _ = await s2_ctx.__anext__() + try: + async def _step10(): + try: + async for _f in stub2.Generate( + runtime_pb2.GenerateRequest( + session_id="any", max_tokens=1, + ), + ): + return ("ok", {}) + return ("ok", {}) + except grpc.aio.AioRpcError as e: + return (e.code().name, {"details": e.details()[:80]}) + emit(await _step( + "Generate (no coordinator wired)", "UNIMPLEMENTED", _step10, + )) + finally: + try: + await s2_ctx.__anext__() + except StopAsyncIteration: + pass + + return results + + +def _summary(results: list[StepResult]) -> dict[str, Any]: + return { + "schema_version": 1, + "kind": "grpc_generator_smoke", + "host": { + "platform": platform.platform(), + "machine": platform.machine(), + "python": platform.python_version(), + "grpc": grpc.__version__, + }, + "steps_total": len(results), + "steps_passed": sum(r.passed for r in results), + "steps_failed": sum(not r.passed for r in results), + "all_passed": all(r.passed for r in results), + "steps": [asdict(r) for r in results], + } + + +def _parse_args(argv: Optional[list[str]] = None) -> argparse.Namespace: + p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + p.add_argument("--report", type=str, default=None) + p.add_argument("--quiet", action="store_true") + return p.parse_args(argv) + + +def main(argv: Optional[list[str]] = None) -> int: + args = _parse_args(argv) + results = asyncio.run(run_smoke(verbose=not args.quiet)) + summary = _summary(results) + print(json.dumps( + {"summary": {k: v for k, v in summary.items() if k not in ("steps", "host")}}, + separators=(",", ":"), + )) + print(json.dumps({"host": summary["host"]}, separators=(",", ":"))) + if args.report: + with open(args.report, "w", encoding="utf-8") as fh: + json.dump(summary, fh, indent=2) + print(f"report written: {args.report}") + return 0 if summary["all_passed"] else 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/inference_engine/server/test_grpc_app.py b/tests/inference_engine/server/test_grpc_app.py index ae0cf759..fa59b74c 100644 --- a/tests/inference_engine/server/test_grpc_app.py +++ b/tests/inference_engine/server/test_grpc_app.py @@ -485,9 +485,15 @@ def append_tokens(self, session_id, token_ids): # --------------------------------------------------------------------------- -async def test_generate_returns_unimplemented(grpc_pair): - """Generate lands in PR-B3. The framework default UNIMPLEMENTED - must hold until that PR explicitly overrides the method.""" +# --------------------------------------------------------------------------- +# Generate (PR-B3) — wired via GenerationCoordinator + FakeVerifier +# --------------------------------------------------------------------------- + + +async def test_generate_returns_unimplemented_when_no_coordinator(grpc_pair): + """Servicer constructed without a GenerationCoordinator (the + PR-B1 / PR-B2 default) keeps the framework UNIMPLEMENTED for + Generate. Regression contract.""" stub, _, _ = grpc_pair with pytest.raises(grpc.aio.AioRpcError) as exc_info: async for _event in stub.Generate( @@ -497,6 +503,350 @@ async def test_generate_returns_unimplemented(grpc_pair): assert exc_info.value.code() == grpc.StatusCode.UNIMPLEMENTED +@pytest_asyncio.fixture +async def grpc_pair_with_generator() -> AsyncIterator[ + tuple[ + runtime_pb2_grpc.RuntimeServiceStub, + SessionStore, + FakeVerifier, + grpc.aio.Server, + ] +]: + """gRPC pair with both AppendTokens and Generate coordinators + wired (so we can prep a session via AppendTokens, then call + Generate against it).""" + from inference_engine.session import ( + AppendTokensCoordinator, + GenerationCoordinator, + ) + + fv = FakeVerifier() + store = SessionStore(capacity=4, cache_inspector=fv) + append_coord = AppendTokensCoordinator(store, fv) + gen_coord = GenerationCoordinator(store, fv) + server = grpc.aio.server() + runtime_pb2_grpc.add_RuntimeServiceServicer_to_server( + RuntimeServiceServicer( + store, + append_coordinator=append_coord, + generation_coordinator=gen_coord, + ), + server, + ) + port = server.add_insecure_port("127.0.0.1:0") + await server.start() + channel = grpc.aio.insecure_channel(f"127.0.0.1:{port}") + stub = runtime_pb2_grpc.RuntimeServiceStub(channel) + try: + yield stub, store, fv, server + finally: + await channel.close() + await server.stop(grace=0.1) + + +async def _prep_session(stub, token_ids=(1, 2, 3)): + """Create + prefill a session, return its session_id.""" + create = await stub.CreateSession(runtime_pb2.CreateSessionRequest()) + await stub.AppendTokens( + runtime_pb2.AppendTokensRequest( + session_id=create.session_id, token_ids=list(token_ids), + ), + ) + return create.session_id + + +async def test_generate_streams_tokens_then_done(grpc_pair_with_generator): + stub, _, _, _ = grpc_pair_with_generator + sid = await _prep_session(stub) + events = [] + async for resp in stub.Generate( + runtime_pb2.GenerateRequest(session_id=sid, max_tokens=3), + ): + events.append(resp) + # Three token frames followed by one done frame. + payload_kinds = [r.WhichOneof("payload") for r in events] + assert payload_kinds == ["token_id", "token_id", "token_id", "done"] + done = events[-1].done + assert done.stop_reason == runtime_pb2.GenerateDone.STOP_REASON_MAX_TOKENS + assert done.generated_token_count == 3 + assert done.prefill_duration_seconds == 0.0 + + +async def test_generate_eos_stops_with_eos_stop_reason( + grpc_pair_with_generator, +): + stub, store, _, _ = grpc_pair_with_generator + # Pre-load history that makes the first argmax = 6 (FakeVerifier's + # _logits_for hashes recent 3 tokens to argmax = sum % 16). + create = await stub.CreateSession( + runtime_pb2.CreateSessionRequest(eos_token_ids=[6]), + ) + await stub.AppendTokens( + runtime_pb2.AppendTokensRequest( + session_id=create.session_id, token_ids=[1, 2, 3], + ), + ) + events = [] + async for resp in stub.Generate( + runtime_pb2.GenerateRequest( + session_id=create.session_id, max_tokens=10, + ), + ): + events.append(resp) + payload_kinds = [r.WhichOneof("payload") for r in events] + assert payload_kinds == ["token_id", "done"] + assert events[0].token_id == 6 + assert events[-1].done.stop_reason == \ + runtime_pb2.GenerateDone.STOP_REASON_EOS + + +async def test_generate_history_truncated_emitted(grpc_pair_with_generator): + stub, store, _, _ = grpc_pair_with_generator + # FakeVerifier's default budget is sink+window = 2+4 = 6. + # Prefill 8 tokens so we're in truncated state at start of Generate. + create = await stub.CreateSession(runtime_pb2.CreateSessionRequest()) + await stub.AppendTokens( + runtime_pb2.AppendTokensRequest( + session_id=create.session_id, + token_ids=[10, 20, 30, 40, 50, 60, 70, 80], + ), + ) + events = [] + async for resp in stub.Generate( + runtime_pb2.GenerateRequest( + session_id=create.session_id, max_tokens=2, + ), + ): + events.append(resp) + payload_kinds = [r.WhichOneof("payload") for r in events] + # First frame is truncated, then tokens, then done. + assert payload_kinds[0] == "truncated" + assert events[0].truncated.dropped_token_count == 2 # 8 - 6 + # Tokens follow. + assert payload_kinds[1:3] == ["token_id", "token_id"] + assert payload_kinds[3] == "done" + + +async def test_generate_unknown_session_returns_not_found( + grpc_pair_with_generator, +): + stub, _, _, _ = grpc_pair_with_generator + with pytest.raises(grpc.aio.AioRpcError) as exc_info: + async for _resp in stub.Generate( + runtime_pb2.GenerateRequest( + session_id="sess-nonexistent", max_tokens=1, + ), + ): + pass # pragma: no cover - stream raises before yielding + assert exc_info.value.code() == grpc.StatusCode.NOT_FOUND + + +async def test_generate_no_history_returns_invalid_argument( + grpc_pair_with_generator, +): + """Session created but no AppendTokens preceded — Generate has + no prefill state to start from. Must surface INVALID_ARGUMENT, + not crash on argmax of uninitialized logits.""" + stub, _, _, _ = grpc_pair_with_generator + create = await stub.CreateSession(runtime_pb2.CreateSessionRequest()) + with pytest.raises(grpc.aio.AioRpcError) as exc_info: + async for _resp in stub.Generate( + runtime_pb2.GenerateRequest( + session_id=create.session_id, max_tokens=1, + ), + ): + pass # pragma: no cover + assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT + assert "AppendTokens must precede" in exc_info.value.details() + + +async def test_generate_max_tokens_zero_returns_invalid_argument( + grpc_pair_with_generator, +): + stub, _, _, _ = grpc_pair_with_generator + sid = await _prep_session(stub) + with pytest.raises(grpc.aio.AioRpcError) as exc_info: + async for _resp in stub.Generate( + runtime_pb2.GenerateRequest(session_id=sid, max_tokens=0), + ): + pass # pragma: no cover + assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT + + +async def test_generate_temperature_nonzero_returns_invalid_argument( + grpc_pair_with_generator, +): + stub, _, _, _ = grpc_pair_with_generator + sid = await _prep_session(stub) + with pytest.raises(grpc.aio.AioRpcError) as exc_info: + async for _resp in stub.Generate( + runtime_pb2.GenerateRequest( + session_id=sid, max_tokens=1, temperature=0.7, + ), + ): + pass # pragma: no cover + assert exc_info.value.code() == grpc.StatusCode.INVALID_ARGUMENT + + +async def test_generate_seed_is_accepted(grpc_pair_with_generator): + """Seed must be accepted on the wire (proto3 optional uint64). + In greedy mode it's ignored; the run must complete normally.""" + stub, _, _, _ = grpc_pair_with_generator + sid = await _prep_session(stub) + events = [] + async for resp in stub.Generate( + runtime_pb2.GenerateRequest( + session_id=sid, max_tokens=2, seed=42, + ), + ): + events.append(resp) + assert any(r.WhichOneof("payload") == "done" for r in events) + + +async def test_generate_invariant_violation_returns_failed_precondition(): + """An INV-1 violation during a generation step must surface as + FAILED_PRECONDITION, not INTERNAL.""" + from inference_engine.session import ( + AppendTokensCoordinator, + GenerationCoordinator, + ) + + fv = FakeVerifier() + store = SessionStore(capacity=2, cache_inspector=fv) + append_coord = AppendTokensCoordinator(store, fv) + gen_coord = GenerationCoordinator(store, fv) + server = grpc.aio.server() + runtime_pb2_grpc.add_RuntimeServiceServicer_to_server( + RuntimeServiceServicer( + store, + append_coordinator=append_coord, + generation_coordinator=gen_coord, + ), + server, + ) + port = server.add_insecure_port("127.0.0.1:0") + await server.start() + channel = grpc.aio.insecure_channel(f"127.0.0.1:{port}") + stub = runtime_pb2_grpc.RuntimeServiceStub(channel) + try: + # Set up a valid session via honest AppendTokens. + create = await stub.CreateSession(runtime_pb2.CreateSessionRequest()) + await stub.AppendTokens( + runtime_pb2.AppendTokensRequest( + session_id=create.session_id, token_ids=[1, 2, 3], + ), + ) + # Now make k_seq_length lie so the FIRST generation step's + # INV-1 check fires. + fv.k_seq_length = lambda session: 999 + with pytest.raises(grpc.aio.AioRpcError) as exc_info: + async for _resp in stub.Generate( + runtime_pb2.GenerateRequest( + session_id=create.session_id, max_tokens=1, + ), + ): + pass # pragma: no cover - stream aborts mid-way + assert exc_info.value.code() == grpc.StatusCode.FAILED_PRECONDITION + assert "INV-1" in exc_info.value.details() + finally: + await channel.close() + await server.stop(grace=0.1) + + +async def test_create_grpc_server_accepts_generation_coordinator(): + """The factory must accept the new keyword and plumb it through.""" + from inference_engine.session import GenerationCoordinator + + fv = FakeVerifier() + store = SessionStore(capacity=2) + coord = GenerationCoordinator(store, fv) + server = create_grpc_server( + session_store=store, + generation_coordinator=coord, + config=GrpcServerConfig(bind_address="127.0.0.1:0"), + ) + assert server is not None + + +async def test_generate_cancellation_emits_cancelled_done(): + """Drive the Servicer's Generate directly with a fake gRPC context + that flips ``cancelled()`` to True after the first event. The + servicer must: + + 1. Yield the first TokenEvent normally. + 2. On the next loop turn, observe context.cancelled() == True. + 3. Emit a final GenerateDone(STOP_REASON_CANCELLED) frame and + return — without continuing to generate. + + Direct-invocation test rather than through-the-channel: the + real-channel cancellation closes the connection from the client + side, so the server-emitted CANCELLED frame is observable only + in-process. This test exercises the server-side branch. + """ + from inference_engine.session import ( + AppendTokensCoordinator, + GenerationCoordinator, + ) + + fv = FakeVerifier() + store = SessionStore(capacity=1, cache_inspector=fv) + append_coord = AppendTokensCoordinator(store, fv) + gen_coord = GenerationCoordinator(store, fv) + sess = store.create_session() + append_coord.append_tokens(sess.session_id, [1, 2, 3]) + + servicer = RuntimeServiceServicer( + store, + append_coordinator=append_coord, + generation_coordinator=gen_coord, + ) + + class _FakeContext: + """Minimal stand-in for grpc.aio.ServicerContext. + + Tracks how many times ``cancelled()`` has been polled; flips + to True after the first poll so the very first iteration of + the servicer's loop yields a TokenEvent normally and the + second iteration observes the cancellation. ``abort`` is not + used in this happy-path-of-cancellation test. + """ + def __init__(self) -> None: + self._polls = 0 + self.poll_history: list[bool] = [] + + def cancelled(self) -> bool: + self._polls += 1 + verdict = self._polls > 1 + self.poll_history.append(verdict) + return verdict + + async def abort(self, code, details): # pragma: no cover + raise AssertionError( + f"abort should not be called: {code} {details!r}", + ) + + ctx = _FakeContext() + request = runtime_pb2.GenerateRequest( + session_id=sess.session_id, max_tokens=10, + ) + + events = [] + async for resp in servicer.Generate(request, ctx): + events.append(resp) + + # First frame is a token, then CANCELLED done — no more. + assert len(events) == 2 + assert events[0].WhichOneof("payload") == "token_id" + assert events[1].WhichOneof("payload") == "done" + assert events[1].done.stop_reason == \ + runtime_pb2.GenerateDone.STOP_REASON_CANCELLED + assert events[1].done.generated_token_count == 1 + # cancelled() polled twice: once on first iteration (returned + # False, allowed token to flow), once on second (returned True, + # tripped CANCELLED branch). + assert ctx.poll_history == [False, True] + + # --------------------------------------------------------------------------- # create_grpc_server factory + GrpcServerConfig # --------------------------------------------------------------------------- diff --git a/tests/inference_engine/session/test_generator.py b/tests/inference_engine/session/test_generator.py new file mode 100644 index 00000000..fce16ffe --- /dev/null +++ b/tests/inference_engine/session/test_generator.py @@ -0,0 +1,433 @@ +"""Unit tests for :mod:`inference_engine.session.generator` (PR-B3). + +Coverage target: 100% on ``inference_engine/session/generator.py``. + +Test strategy mirrors :mod:`tests.inference_engine.session.test_coordinator`: +the dispatch + state-mirroring + error-mapping logic is tested with +the deterministic :class:`FakeVerifier` (Linux-runnable, no model +weights). Real Qwen3 verifier integration lives under +:mod:`tests.core` (Mac-only) and runs on the §9 Mac M4 gate. +""" + +from __future__ import annotations + +import pytest +import torch + +from inference_engine.session import ( + AppendTokensCoordinator, + DoneEvent, + GenerationCoordinator, + HistoryTruncatedEvent, + InvariantViolation, + SessionNotFoundError, + SessionStore, + STOP_REASON_EOS, + STOP_REASON_MAX_TOKENS, + TokenEvent, +) + +# Reuse the FakeVerifier from PR-B2's test module rather than +# re-defining it. It already mirrors the real verifier's mutation +# contract (sink+window trim in commit_or_truncate, parallel-sequence +# growth in forward_block, deterministic logits). +from tests.inference_engine.session.test_coordinator import FakeVerifier + + +def _build( + *, + sink_size: int = 2, + window_size: int = 4, + eos_token_ids=(), + initial_tokens=(1, 2, 3), +): + """Construct (store, fv, gen_coord, session) ready for Generate. + + Runs an AppendTokens via the PR-B2 coordinator first so the + session has prefilled state — Generate against an empty session + is a documented ValueError, tested separately. + """ + fv = FakeVerifier( + sink_size=sink_size, window_size=window_size, vocab_size=16, + ) + store = SessionStore(capacity=2, cache_inspector=fv) + append_coord = AppendTokensCoordinator(store, fv) + gen_coord = GenerationCoordinator(store, fv) + sess = store.create_session(eos_token_ids=eos_token_ids) + if initial_tokens: + append_coord.append_tokens(sess.session_id, list(initial_tokens)) + return store, fv, gen_coord, sess + + +# --------------------------------------------------------------------------- +# Greedy dispatch + happy path +# --------------------------------------------------------------------------- + + +class TestGreedyHappyPath: + def test_yields_token_then_done(self): + store, fv, coord, sess = _build() + events = list(coord.generate(sess.session_id, max_tokens=1)) + assert len(events) == 2 + assert isinstance(events[0], TokenEvent) + assert isinstance(events[1], DoneEvent) + assert events[1].stop_reason == STOP_REASON_MAX_TOKENS + assert events[1].generated_token_count == 1 + + def test_max_tokens_caps_token_emission(self): + store, fv, coord, sess = _build() + events = list(coord.generate(sess.session_id, max_tokens=3)) + token_events = [e for e in events if isinstance(e, TokenEvent)] + done_events = [e for e in events if isinstance(e, DoneEvent)] + assert len(token_events) == 3 + assert len(done_events) == 1 + assert done_events[0].stop_reason == STOP_REASON_MAX_TOKENS + assert done_events[0].generated_token_count == 3 + + def test_done_is_terminal_and_unique(self): + store, fv, coord, sess = _build() + events = list(coord.generate(sess.session_id, max_tokens=2)) + # Done event is exactly one and is the last. + done_indices = [ + i for i, e in enumerate(events) if isinstance(e, DoneEvent) + ] + assert len(done_indices) == 1 + assert done_indices[0] == len(events) - 1 + + def test_done_includes_total_seconds(self): + store, fv, coord, sess = _build() + events = list(coord.generate(sess.session_id, max_tokens=1)) + done = events[-1] + assert isinstance(done, DoneEvent) + assert done.total_seconds >= 0.0 + # PR-B3 has no separate prefill phase. + assert done.prefill_seconds == 0.0 + + +class TestGreedyAdvancesVerifier: + def test_each_token_advances_position_by_one(self): + store, fv, coord, sess = _build() + pos_before = fv.next_global_position + list(coord.generate(sess.session_id, max_tokens=4)) + assert fv.next_global_position == pos_before + 4 + + def test_session_history_grows(self): + store, fv, coord, sess = _build() + list(coord.generate(sess.session_id, max_tokens=3)) + assert len(sess.history_token_ids) == 3 + 3 # initial + generated + + def test_session_cached_token_sequence_mirrors_verifier(self): + store, fv, coord, sess = _build(sink_size=2, window_size=4) + list(coord.generate(sess.session_id, max_tokens=10)) + assert sess.cached_token_sequence == fv.cached_token_sequence + + def test_each_token_calls_forward_then_commit(self): + store, fv, coord, sess = _build() + fv.call_log.clear() + list(coord.generate(sess.session_id, max_tokens=2)) + # 2 (forward_block, commit_or_truncate) pairs. + kinds = [c[0] for c in fv.call_log] + assert kinds == [ + "forward_block", "commit_or_truncate", + "forward_block", "commit_or_truncate", + ] + + +# --------------------------------------------------------------------------- +# EOS handling +# --------------------------------------------------------------------------- + + +class TestEos: + def test_eos_token_terminates_with_eos_stop_reason(self): + # FakeVerifier._logits_for hashes recent tokens to an argmax. + # We pre-load history that makes the next argmax a known + # token, then put that token in eos_token_ids. + # The FakeVerifier formula: argmax = sum(history[-3:]) % 16. + # With initial=[1, 2, 3], next argmax = 6. + store, fv, coord, sess = _build( + initial_tokens=(1, 2, 3), eos_token_ids=(6,), + ) + events = list(coord.generate(sess.session_id, max_tokens=10)) + token_events = [e for e in events if isinstance(e, TokenEvent)] + done_events = [e for e in events if isinstance(e, DoneEvent)] + # Exactly one TokenEvent, then Done with EOS. + assert len(token_events) == 1 + assert token_events[0].token_id == 6 + assert done_events[0].stop_reason == STOP_REASON_EOS + assert done_events[0].generated_token_count == 1 + + def test_no_eos_runs_to_max_tokens(self): + # Use a token id that cannot be produced (vocab size 16; eos + # set to 99 cannot match any argmax). + store, fv, coord, sess = _build(eos_token_ids=(99,)) + events = list(coord.generate(sess.session_id, max_tokens=4)) + done_events = [e for e in events if isinstance(e, DoneEvent)] + assert done_events[0].stop_reason == STOP_REASON_MAX_TOKENS + + def test_empty_eos_set_runs_to_max_tokens(self): + store, fv, coord, sess = _build(eos_token_ids=()) + events = list(coord.generate(sess.session_id, max_tokens=2)) + done = next(e for e in events if isinstance(e, DoneEvent)) + assert done.stop_reason == STOP_REASON_MAX_TOKENS + + +# --------------------------------------------------------------------------- +# HistoryTruncated event +# --------------------------------------------------------------------------- + + +class TestHistoryTruncated: + def test_emitted_at_start_when_already_truncated(self): + # sink+window = 2+4 = 6 capacity. Append 8 tokens → cache + # holds 6, history holds 8 → already truncated state. + store, fv, coord, sess = _build( + sink_size=2, window_size=4, + initial_tokens=(10, 20, 30, 40, 50, 60, 70, 80), + ) + events = list(coord.generate(sess.session_id, max_tokens=1)) + # First non-token event should be HistoryTruncated, before + # any TokenEvent. + assert isinstance(events[0], HistoryTruncatedEvent) + assert events[0].dropped_token_count == 8 - 6 # 2 dropped + # A TokenEvent must follow before Done. + assert isinstance(events[1], TokenEvent) + + def test_not_emitted_when_under_capacity(self): + # sink+window = 6; initial = 3 tokens; cache == history. + store, fv, coord, sess = _build( + sink_size=2, window_size=4, initial_tokens=(1, 2, 3), + ) + events = list(coord.generate(sess.session_id, max_tokens=2)) + # No HistoryTruncated event present. + assert not any( + isinstance(e, HistoryTruncatedEvent) for e in events + ) + + def test_at_most_one_per_call(self): + # Even after generation pushes well past the boundary, only + # one HistoryTruncated per Generate call (per the proto + # contract: "Emitted at most once per Generate call"). + store, fv, coord, sess = _build( + sink_size=2, window_size=4, + initial_tokens=(10, 20, 30, 40, 50, 60, 70, 80), + ) + events = list(coord.generate(sess.session_id, max_tokens=10)) + truncated_events = [ + e for e in events if isinstance(e, HistoryTruncatedEvent) + ] + assert len(truncated_events) == 1 + + +# --------------------------------------------------------------------------- +# Validation: max_tokens, sampling params, no AppendTokens prior +# --------------------------------------------------------------------------- + + +class TestValidation: + def test_max_tokens_zero_rejected(self): + store, fv, coord, sess = _build() + with pytest.raises(ValueError, match="max_tokens must be >= 1"): + list(coord.generate(sess.session_id, max_tokens=0)) + + def test_max_tokens_negative_rejected(self): + store, fv, coord, sess = _build() + with pytest.raises(ValueError, match="max_tokens must be >= 1"): + list(coord.generate(sess.session_id, max_tokens=-3)) + + def test_temperature_nonzero_rejected(self): + store, fv, coord, sess = _build() + with pytest.raises(ValueError, match="greedy"): + list(coord.generate( + sess.session_id, max_tokens=1, temperature=0.5, + )) + + def test_temperature_zero_accepted(self): + store, fv, coord, sess = _build() + # Temperature=0 is greedy's no-op default; accept silently. + events = list(coord.generate( + sess.session_id, max_tokens=1, temperature=0.0, + )) + assert any(isinstance(e, TokenEvent) for e in events) + + def test_top_p_set_rejected(self): + store, fv, coord, sess = _build() + with pytest.raises(ValueError, match="top_p"): + list(coord.generate( + sess.session_id, max_tokens=1, top_p=0.9, + )) + + def test_top_k_other_than_one_rejected(self): + store, fv, coord, sess = _build() + with pytest.raises(ValueError, match="top_k"): + list(coord.generate( + sess.session_id, max_tokens=1, top_k=50, + )) + + def test_top_k_one_accepted(self): + store, fv, coord, sess = _build() + events = list(coord.generate( + sess.session_id, max_tokens=1, top_k=1, + )) + assert any(isinstance(e, TokenEvent) for e in events) + + def test_seed_accepted_and_ignored_in_greedy(self): + store, fv, coord, sess = _build() + # Seed shouldn't affect greedy output. Two runs with + # different seeds must produce identical token streams. + store_a, fv_a, coord_a, sess_a = _build() + store_b, fv_b, coord_b, sess_b = _build() + events_a = [ + e for e in coord_a.generate( + sess_a.session_id, max_tokens=4, seed=1, + ) + if isinstance(e, TokenEvent) + ] + events_b = [ + e for e in coord_b.generate( + sess_b.session_id, max_tokens=4, seed=999, + ) + if isinstance(e, TokenEvent) + ] + assert [e.token_id for e in events_a] == [ + e.token_id for e in events_b + ] + + def test_no_appendtokens_first_rejected(self): + # Session created but no AppendTokens called — no prefill, + # so next_token_logits is meaningless. Reject loudly. + fv = FakeVerifier() + store = SessionStore(capacity=1, cache_inspector=fv) + coord = GenerationCoordinator(store, fv) + sess = store.create_session() + with pytest.raises(ValueError, match="AppendTokens must precede"): + list(coord.generate(sess.session_id, max_tokens=1)) + + def test_unknown_session_raises_session_not_found(self): + store, fv, coord, _ = _build() + with pytest.raises(SessionNotFoundError): + list(coord.generate("sess-unknown", max_tokens=1)) + + +# --------------------------------------------------------------------------- +# INV-1 / INV-2 propagation through Generate +# --------------------------------------------------------------------------- + + +class TestInvariants: + def test_inv1_violation_propagates(self): + # Drive AppendTokens with an honest inspector, then patch the + # inspector to lie just before Generate. The lying inspector + # makes the first generation step's INV-1 check fail because + # session.cached_token_sequence (mirrored from verifier) won't + # match the lie's reported k_seq_length. + fv = FakeVerifier() + store = SessionStore(capacity=1, cache_inspector=fv) + append_coord = AppendTokensCoordinator(store, fv) + gen_coord = GenerationCoordinator(store, fv) + sess = store.create_session() + # AppendTokens with the honest FakeVerifier — works. + append_coord.append_tokens(sess.session_id, [1, 2, 3]) + # Now monkey-patch the inspector to lie. Note: SessionStore's + # _assert_inv1 calls self._cache_inspector.k_seq_length(session), + # which dispatches to the patched bound method. + fv.k_seq_length = lambda session: 999 # type: ignore[assignment] + with pytest.raises(InvariantViolation) as exc: + list(gen_coord.generate(sess.session_id, max_tokens=1)) + assert exc.value.kind == "1" + with pytest.raises(SessionNotFoundError): + store.get_session(sess.session_id) + + def test_inv2_violation_propagates(self): + # AppendTokens uses verifier.prefill, NOT commit_or_truncate, + # so the FIRST commit_or_truncate the Verifier sees is from + # the first generation step. Trip the regress on call #1. + class _RegressingVerifier(FakeVerifier): + def __init__(self): + super().__init__() + self._calls = 0 + + def commit_or_truncate(self, *, forwarded, accepted): + super().commit_or_truncate( + forwarded=forwarded, accepted=accepted, + ) + self._calls += 1 + if self._calls == 1: # first generation step's commit + self.next_global_position = 0 # regress + + fv = _RegressingVerifier() + store = SessionStore(capacity=1, cache_inspector=fv) + append_coord = AppendTokensCoordinator(store, fv) + gen_coord = GenerationCoordinator(store, fv) + sess = store.create_session() + append_coord.append_tokens(sess.session_id, [1, 2, 3]) + with pytest.raises(InvariantViolation) as exc: + list(gen_coord.generate(sess.session_id, max_tokens=1)) + assert exc.value.kind == "2" + + +# --------------------------------------------------------------------------- +# Determinism (INV-3 byte-exact under greedy) +# --------------------------------------------------------------------------- + + +class TestDeterminism: + def test_two_runs_with_same_history_produce_same_tokens(self): + # INV-3 byte-exact at the GenerationCoordinator level: two + # parallel sessions with identical history produce identical + # token sequences under greedy decoding. + store_a, fv_a, coord_a, sess_a = _build( + initial_tokens=(7, 11, 13, 17, 19), + ) + store_b, fv_b, coord_b, sess_b = _build( + initial_tokens=(7, 11, 13, 17, 19), + ) + tokens_a = [ + e.token_id + for e in coord_a.generate(sess_a.session_id, max_tokens=8) + if isinstance(e, TokenEvent) + ] + tokens_b = [ + e.token_id + for e in coord_b.generate(sess_b.session_id, max_tokens=8) + if isinstance(e, TokenEvent) + ] + assert tokens_a == tokens_b + + +# --------------------------------------------------------------------------- +# Constructor / event types +# --------------------------------------------------------------------------- + + +class TestConstructorAndEventDataclasses: + def test_constructor_stores_references(self): + fv = FakeVerifier() + store = SessionStore(capacity=1, cache_inspector=fv) + coord = GenerationCoordinator(store, fv) + # Coordinator accepts the references; we verify by exercising. + sess = store.create_session() + AppendTokensCoordinator(store, fv).append_tokens( + sess.session_id, [1], + ) + events = list(coord.generate(sess.session_id, max_tokens=1)) + assert any(isinstance(e, TokenEvent) for e in events) + + def test_token_event_is_frozen(self): + e = TokenEvent(token_id=5) + with pytest.raises(Exception): # FrozenInstanceError + e.token_id = 6 # type: ignore[misc] + + def test_history_truncated_event_is_frozen(self): + e = HistoryTruncatedEvent(dropped_token_count=3) + with pytest.raises(Exception): + e.dropped_token_count = 4 # type: ignore[misc] + + def test_done_event_is_frozen(self): + e = DoneEvent( + stop_reason=STOP_REASON_MAX_TOKENS, + generated_token_count=1, + prefill_seconds=0.0, total_seconds=0.0, + ) + with pytest.raises(Exception): + e.generated_token_count = 2 # type: ignore[misc]