feat(v7-f06): live token streaming via /ws/gen/{job_id}

Honest-closure: stream_generate already had on_token callback support
in cppmega_v4/runtime/generate_stream.py, but no WS endpoint forwarded
the per-token events to the UI. The audit listed it as 'F06 streaming
generation tokens в UI: нет WS'.

- runtime/gen_event_bus.py: mirror of train_event_bus keyed by job_id.
  Thread-safe publish/subscribe with None sentinel for completion.
- jsonrpc/gen_run_method.py: GenRunParams.job_id; when set, gen.run
  routes via stream_generate(on_token=...) so each token publishes
  onto the bus. None sentinel fires on return.
- jsonrpc/server.py: /ws/gen/{job_id} endpoint subscribes to the
  bus and forwards each event as {job_id, event} frames + final
  {finish:'ok'}. Same 200ms-poll pattern as /ws/train/{run_id}.
- tests/v4/test_gen_event_bus.py: no-publish without job_id; full
  per-token + sentinel delivery; cross-job isolation.

F07 latency bench already swapped train(1) proxy for gen_run in
8f7ec3a; the streaming endpoint completes the F-block trio
(F01 → F02 → F06 + F03/F04 already wired in).

Author: apstenku123

cppmega_v4/jsonrpc/gen_run_method.py

@@ -22,7 +22,8 @@
 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 import gen_event_bus
+from cppmega_v4.runtime.generate_stream import stream_generate
 from cppmega_v4.runtime.samplers import (
     greedy, temperature_sample, top_k_sample, top_p_sample,
 )
@@ -53,6 +54,9 @@ class GenRunParams(BaseModel):
     # GenRunResult.kv_cache. 0 disables KV-cache reporting.
     kv_cache_layers: int = Field(0, ge=0, le=128)
     kv_cache_head_dim: int = Field(16, ge=1, le=4096)
+    # V7-F06: optional job_id; when set, gen_run publishes each token
+    # onto gen_event_bus so /ws/gen/{job_id} subscribers see streaming.
+    job_id: str | None = None
 
 
 class GenRunEvent(BaseModel):
@@ -141,12 +145,25 @@ def _step(last: int) -> int:
             kv_cache_state["growth_events"] += 1
         return tok
 
-    tokens, reason, raw_events = collect_stream(
+    # V7-F06: when a job_id was supplied, route through stream_generate
+    # with an on_token callback that publishes onto the bus, then mirror
+    # the same collected events for the synchronous return path.
+    raw_events: list[dict] = []
+
+    def _on_token(ev: dict) -> None:
+        raw_events.append(ev)
+        if params.job_id:
+            gen_event_bus.publish(params.job_id, ev)
+
+    tokens, reason = stream_generate(
         initial_tokens=list(params.prompt_tokens),
         step_fn=_step,
         eos_token_id=params.eos_token_id,
         max_new_tokens=params.max_new_tokens,
+        on_token=_on_token,
     )
+    if params.job_id:
+        gen_event_bus.publish(params.job_id, None)
     if kv_obj is not None:
         kv_cache_state["total_bytes"] = int(kv_obj.total_bytes())
         kv_cache_state["lengths_per_layer"] = [

cppmega_v4/jsonrpc/server.py

@@ -79,6 +79,41 @@ async def rpc(payload: dict) -> dict:
         response = await _dispatch(payload, cache)
         return response.model_dump(mode="json", exclude_none=True)
 
+    @app.websocket("/ws/gen/{job_id}")
+    async def ws_gen(socket: WebSocket, job_id: str) -> None:
+        """V7-F06: live token stream from gen.run.
+
+        Client opens this WS *before* calling gen.run with the same
+        job_id, then receives {job_id, event:{step, token_id,
+        finish_reason}} frames until {finish:'ok'} fires."""
+        from cppmega_v4.runtime import gen_event_bus
+        import queue as _queue
+        await socket.accept()
+        q = gen_event_bus.subscribe(job_id)
+
+        def _try_get(timeout: float = 0.2):
+            try:
+                return q.get(timeout=timeout)
+            except _queue.Empty:
+                return _SENTINEL_EMPTY
+
+        try:
+            while True:
+                ev = await asyncio.to_thread(_try_get, 0.2)
+                if ev is _SENTINEL_EMPTY:
+                    await asyncio.sleep(0)
+                    continue
+                if ev is None:
+                    await socket.send_json({"finish": "ok",
+                                             "job_id": job_id})
+                    return
+                await socket.send_json({"job_id": job_id,
+                                         "event": ev})
+        except WebSocketDisconnect:
+            pass
+        finally:
+            gen_event_bus.unsubscribe(job_id, q)
+
     @app.websocket("/ws/train/{run_id}")
     async def ws_train(socket: WebSocket, run_id: str) -> None:
         """V7-H05: live per-step training metric stream.

cppmega_v4/runtime/gen_event_bus.py

@@ -0,0 +1,59 @@
+"""V7-F06: cross-thread pub/sub for live token-by-token generation events.
+
+Mirrors train_event_bus but keyed by a generation job_id supplied
+to gen.run via params.run_id. stream_generate's on_token callback
+publishes each {step, token_id, finish_reason} frame onto the bus
+and the WS endpoint /ws/gen/{job_id} forwards them to the UI.
+"""
+
+from __future__ import annotations
+
+import queue
+import threading
+from typing import Any
+
+
+_LOCK = threading.Lock()
+_QUEUES: dict[str, list[queue.Queue]] = {}
+
+
+def publish(job_id: str | None, event: dict[str, Any] | None) -> None:
+    if not job_id:
+        return
+    with _LOCK:
+        subs = list(_QUEUES.get(job_id, []))
+    for q in subs:
+        try:
+            q.put_nowait(event)
+        except Exception:
+            pass
+
+
+def subscribe(job_id: str) -> queue.Queue:
+    q: queue.Queue = queue.Queue(maxsize=4096)
+    with _LOCK:
+        _QUEUES.setdefault(job_id, []).append(q)
+    return q
+
+
+def unsubscribe(job_id: str, q: queue.Queue) -> None:
+    with _LOCK:
+        subs = _QUEUES.get(job_id, [])
+        if q in subs:
+            subs.remove(q)
+        if not subs:
+            _QUEUES.pop(job_id, None)
+
+
+def reset() -> None:
+    with _LOCK:
+        _QUEUES.clear()
+
+
+def subscriber_count(job_id: str) -> int:
+    with _LOCK:
+        return len(_QUEUES.get(job_id, []))
+
+
+__all__ = ["publish", "subscribe", "unsubscribe", "reset",
+            "subscriber_count"]

tests/v4/test_gen_event_bus.py

@@ -0,0 +1,66 @@
+"""V7-F06: gen_event_bus pub/sub + gen.run job_id wiring."""
+
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from cppmega_v4.jsonrpc import create_app
+from cppmega_v4.jsonrpc.gen_run_method import GenRunParams, gen_run
+from cppmega_v4.runtime import gen_event_bus as bus
+
+
+@pytest.fixture(autouse=True)
+def _clean():
+    bus.reset()
+    yield
+    bus.reset()
+
+
+@pytest.fixture
+def client():
+    return TestClient(create_app(cache_capacity=2))
+
+
+def test_gen_run_without_job_id_does_not_publish():
+    res = gen_run(GenRunParams(prompt_tokens=[0], eos_token_id=-1,
+                                 max_new_tokens=3))
+    # No subscribers, no job_id — bus stays empty.
+    assert bus.subscriber_count("anything") == 0
+    assert len(res.events) == 3
+
+
+def test_gen_run_with_job_id_publishes_per_token_and_finish():
+    q = bus.subscribe("job-7")
+    res = gen_run(GenRunParams(prompt_tokens=[0], eos_token_id=-1,
+                                 max_new_tokens=4,
+                                 job_id="job-7"))
+    received: list = []
+    while True:
+        ev = q.get(timeout=1.0)
+        if ev is None:
+            break
+        received.append(ev)
+    assert len(received) == 4
+    for ev in received:
+        assert "token_id" in ev
+        assert "step" in ev
+    assert res.finish_reason == "length"
+
+
+def test_gen_run_unsubscribe_isolates_jobs():
+    q1 = bus.subscribe("job-A")
+    q2 = bus.subscribe("job-B")
+    gen_run(GenRunParams(prompt_tokens=[0], eos_token_id=-1,
+                          max_new_tokens=2, job_id="job-A"))
+    # job-A queue receives events + sentinel; job-B stays untouched.
+    a_count = 0
+    while not q1.empty():
+        ev = q1.get_nowait()
+        if ev is None:
+            break
+        a_count += 1
+    assert a_count == 2
+    assert q2.empty()
+    bus.unsubscribe("job-A", q1)
+    bus.unsubscribe("job-B", q2)