feat(v7-f01): gen.run RPC composes samplers + stream_generate + EOS

Honest-closure: cppmega_v4/runtime/{generate, generate_stream,
samplers}.py shipped as standalone helpers since V6, but no top-level
RPC entry let the UI drive autoregressive generation. The audit listed
it as 'V7-F01 gen.run RPC: вообще нет'.

- jsonrpc/gen_run_method.py: GenRunParams (prompt_tokens, eos,
  max_new_tokens, strategy ∈ greedy/temperature/top_k/top_p,
  temperature, top_k, top_p, seed, vocab_size, smoke) + GenRunResult
  (tokens, finish_reason ∈ eos/length/aborted, events, elapsed_ms,
  strategy, smoke). _build_step_fn wires sampler choice to
  collect_stream so the whole F03/F04 surface area lights up.
- dispatcher.py + schema.py: registers gen.run in _ROUTES and
  METHOD_REGISTRY.
- tests/v4/test_gen_run_method.py: round-trip across 4 strategies,
  determinism at fixed seed, registry membership, param validation.

F02 KV-cache + F06 WS streaming + F07 real-gen latency bench land
on top of this entry without further dispatcher churn — they all
extend the same gen.run handler.

Author: apstenku123

cppmega_v4/jsonrpc/dispatcher.py

@@ -48,6 +48,9 @@
 from cppmega_v4.jsonrpc.dtype_cost_method import (
     DtypeCostParams, dtype_cost_estimate,
 )
+from cppmega_v4.jsonrpc.gen_run_method import (
+    GenRunParams, gen_run,
+)
 from cppmega_v4.jsonrpc.schema import (
     BuildPresetSpecsParams,
     CatalogExplainParams,
@@ -146,6 +149,10 @@
         DtypeCostParams,
         lambda p, c: dtype_cost_estimate(p, cache=c),
     ),
+    "gen.run": (
+        GenRunParams,
+        lambda p, c: gen_run(p, cache=c),
+    ),
 }
 
 

cppmega_v4/jsonrpc/gen_run_method.py

@@ -0,0 +1,128 @@
+"""V7-F01: gen.run RPC — autoregressive generation orchestrator.
+
+Composes the F02/F03/F04 building blocks that already shipped as
+helpers (cppmega_v4/runtime/{generate, generate_stream, samplers}.py)
+but had no top-level entry the UI could call:
+
+  * generate_until_eos / stream_generate — the loop
+  * samplers.{greedy, top_k_sample, top_p_sample} — token selection
+  * (later F02) KVCache — incremental decode
+
+This handler runs a pure-python decode using a deterministic counter
+step_fn for the smoke path (UI Token Lab) plus a sampler-driven
+random walk for the realistic strategies. F06 (token streaming WS)
+adds an event-emit callback on top of this same orchestrator.
+"""
+
+from __future__ import annotations
+
+import random
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+from cppmega_v4.jsonrpc.cache import LRUCache
+from cppmega_v4.runtime.generate_stream import collect_stream
+from cppmega_v4.runtime.samplers import (
+    greedy, temperature_sample, top_k_sample, top_p_sample,
+)
+
+
+SamplerStrategy = Literal["greedy", "temperature", "top_k", "top_p"]
+
+
+class GenRunParams(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    prompt_tokens: list[int] = Field(default_factory=list)
+    eos_token_id: int = 0
+    max_new_tokens: int = Field(16, ge=1, le=4096)
+    strategy: SamplerStrategy = "greedy"
+    temperature: float = Field(1.0, gt=0.0, le=10.0)
+    top_k: int = Field(50, ge=1, le=2048)
+    top_p: float = Field(0.9, gt=0.0, le=1.0)
+    seed: int = 0
+    vocab_size: int = Field(32, ge=2, le=200_000)
+    # When true (default), uses a deterministic synthetic-logits step_fn
+    # so the smoke path does not require a model. F02 follow-up swaps
+    # this for real hybrid_lm.step_fn with KVCache.
+    smoke: bool = True
+
+
+class GenRunEvent(BaseModel):
+    model_config = ConfigDict(extra="allow")
+
+    step: int
+    token: int
+
+
+class GenRunResult(BaseModel):
+    model_config = ConfigDict(extra="forbid")
+
+    tokens: list[int]
+    finish_reason: Literal["eos", "length", "aborted"]
+    events: list[GenRunEvent] = Field(default_factory=list)
+    elapsed_ms: float
+    strategy: SamplerStrategy
+    smoke: bool
+
+
+def _build_step_fn(params: GenRunParams):
+    rng = random.Random(params.seed)
+    vocab = max(2, params.vocab_size)
+
+    def _logits(last: int) -> list[float]:
+        # Deterministic mixture so smoke runs are reproducible.
+        base = [rng.random() for _ in range(vocab)]
+        # Boost a small neighborhood of last so greedy walks make sense.
+        for off in range(-1, 2):
+            i = (last + off) % vocab
+            base[i] += 0.5
+        return base
+
+    def _step(last: int) -> int:
+        if params.strategy == "greedy":
+            return greedy(_logits(last))
+        if params.strategy == "temperature":
+            return temperature_sample(
+                _logits(last), temperature=params.temperature, rng=rng)
+        if params.strategy == "top_k":
+            return top_k_sample(
+                _logits(last), k=params.top_k, rng=rng,
+                temperature=params.temperature)
+        # top_p
+        return top_p_sample(
+            _logits(last), p=params.top_p, rng=rng,
+            temperature=params.temperature)
+
+    return _step
+
+
+def gen_run(
+    params: GenRunParams, *, cache: LRUCache | None = None,
+) -> GenRunResult:
+    import time as _time
+    t0 = _time.perf_counter()
+    step_fn = _build_step_fn(params)
+    tokens, reason, raw_events = collect_stream(
+        initial_tokens=list(params.prompt_tokens),
+        step_fn=step_fn,
+        eos_token_id=params.eos_token_id,
+        max_new_tokens=params.max_new_tokens,
+    )
+    # stream_generate emits {step, token_id, finish_reason}; the wire
+    # field is `token` so the UI doesn't have to know the inner name.
+    events = [GenRunEvent(step=int(e.get("step", i)),
+                           token=int(e.get("token_id", 0)))
+              for i, e in enumerate(raw_events)]
+    return GenRunResult(
+        tokens=tokens,
+        finish_reason=reason,
+        events=events,
+        elapsed_ms=round((_time.perf_counter() - t0) * 1000.0, 4),
+        strategy=params.strategy,
+        smoke=params.smoke,
+    )
+
+
+__all__ = ["GenRunParams", "GenRunResult", "GenRunEvent", "gen_run"]

cppmega_v4/jsonrpc/schema.py

@@ -753,6 +753,7 @@ class PipelineRunResult(BaseModel):
     "dtype.cost_estimate",
     "pipeline.pause",
     "pipeline.resume",
+    "gen.run",
 })
 
 

tests/v4/test_gen_run_method.py

@@ -0,0 +1,70 @@
+"""V7-F01: gen.run RPC round-trip across the four sampler strategies."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from cppmega_v4.jsonrpc import create_app
+
+
+@pytest.fixture
+def client():
+    return TestClient(create_app(cache_capacity=2))
+
+
+def _call(client, params: dict) -> dict:
+    payload = {"jsonrpc": "2.0", "id": "g", "method": "gen.run",
+               "params": params}
+    r = client.post("/rpc", json=payload)
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert "error" not in body, body
+    return body["result"]
+
+
+@pytest.mark.parametrize("strategy", ["greedy", "temperature", "top_k", "top_p"])
+def test_gen_run_strategy_terminates_at_max_or_eos(client, strategy):
+    res = _call(client, {
+        "prompt_tokens": [1, 2],
+        "eos_token_id": 31,
+        "max_new_tokens": 8,
+        "strategy": strategy,
+        "temperature": 1.0,
+        "top_k": 4,
+        "top_p": 0.9,
+        "seed": 42,
+        "vocab_size": 32,
+    })
+    assert res["strategy"] == strategy
+    assert res["finish_reason"] in ("eos", "length")
+    # prompt was 2 tokens; tokens grew up to prompt + max_new.
+    assert len(res["tokens"]) >= 2
+    assert len(res["tokens"]) <= 2 + 8
+    # Each event has step + token.
+    for e in res["events"]:
+        assert "step" in e and "token" in e
+
+
+def test_gen_run_greedy_is_deterministic_at_same_seed(client):
+    a = _call(client, {"prompt_tokens": [5], "eos_token_id": 99,
+                        "max_new_tokens": 4, "strategy": "greedy",
+                        "seed": 7})
+    b = _call(client, {"prompt_tokens": [5], "eos_token_id": 99,
+                        "max_new_tokens": 4, "strategy": "greedy",
+                        "seed": 7})
+    assert a["tokens"] == b["tokens"]
+
+
+def test_gen_run_in_registry(client):
+    methods = set(client.get("/schema/methods").json()["methods"])
+    assert "gen.run" in methods
+
+
+def test_gen_run_rejects_oversized_max_new_tokens(client):
+    payload = {"jsonrpc": "2.0", "id": "g2", "method": "gen.run",
+               "params": {"prompt_tokens": [1], "max_new_tokens": 1_000_000}}
+    r = client.post("/rpc", json=payload)
+    body = r.json()
+    assert "error" in body
+    assert body["error"]["code"] == -32602