(4/7) feat(session): multi-process session-server data plane (routing + IPC + worker + thin router)

[guapisolo]

Summary

Opt-in multi-process session-server data plane, stacked on the SessionCore extraction (#1510) and the R3 client-strip (#1563). This PR adds the mechanism (routing + IPC + worker + thin router) but does not wire it into launch — the supervisor, the --session-server-workers flag, and the router_manager/rollout_manager wiring land in the follow-up control-plane PR. So this PR changes no runtime behavior: the default single-process path is untouched and nothing yet activates the multi-process path.

Stacked on #1569 → #1510 → #1563. Base = refactor/session-r3-client-strip. Review/merge after #1563.

What's here

• sharding.py — process-stable blake2b(session_id) % N sharding (router and every worker agree without coordination).
• ipc.py — minimal framed, multiplexed RPC over a UNIX socket: one frame per message, request_id multiplexing, cancel-safe request(), u64 frame lengths with a single 64 GiB sanity frame cap enforced at send time (an oversized frame is a per-request IpcError / ERROR-frame failure, never a channel teardown). No chunking / round-robin / send-budget / backpressure.
• worker.py — headless worker: one SessionCore + an httpx ProxyBackend, driven over IPC. No per-worker backpressure / parse-gate.
• router.py — thin client-facing router app: hash-routes to the owning worker and relays the opaque response body (never parses it); imports neither core nor worker, so the router process never loads the tokenizer/transformers stack.

Single-sourced across single- and multi-process (no drift)

• SessionCore.create_session(session_id=None) → SessionRegistry.create_session(session_id=None) — the router mints the id (uuid4 hex, so it can route by it) and the owning worker creates under exactly that id, with a loud collision guard; the single-process path passes None and the registry mints.
• core.error_response / build_session_core — used by both the FastAPI adapter and the worker, so the error/response/tokenizer contracts can't diverge between modes.

Cut vs. the original prototype (kept minimal)

Cut: IPC chunking/round-robin/send-budget, configurable size limits, router 503 backpressure, asyncio.shield, the router task-tracking set, worker backpressure + parse-gate, the 409/500 additions. Kept (load-bearing): request_id multiplexing, the sanity frame cap, cancel-safe request(). Rationale in docs/developer/multi-process-session-server.md.

Why u64 frame lengths (measured)

Session records are never pruned (the training data path consumes R3 from them via GET /sessions/{id}), so a full-records GET reply grows with turns × response size. At the production shape from the design doc (50 turns × ~64 MiB avg accumulated-R3 response) one session's GET reply measures ~3.15 GiB — already past a u32/2 GiB frame limit. With u32 the failure mode was the worst kind: the reply frame reached the router's reader, was indistinguishable from a corrupt length, and tore down the whole IPC channel — silently poisoning every other session on that shard (worker still alive, so the supervisor's is_alive() fail-fast can't see it; observed as 9/32 sessions failing with 503s and 0/32 GETs succeeding). With u64 + the 64 GiB send-side cap the same workload passes: 32/32 sessions, 1600/1600 turns, 0 chat errors, 30/32 GETs OK (2 client-side ReadTimeouts in the bench harness, not server failures) — each ~3.15 GiB reply crosses as one frame and an oversized frame can only fail its own request, never the channel.

Tests (tests/fast/router/)

• test_session_ipc — framing, out-of-order multiplexing, handler-error → IpcError, EOF teardown, cancel-safety, oversized-frame rejection.
• test_session_routing — determinism, distribution, and cross-process hash stability (independent of PYTHONHASHSEED).
• test_session_worker — op dispatch, create/get/delete lifecycle, missing→404, proxy passthrough, duplicate-id guard.
• test_session_dataplane — router ↔ N=3 workers over real socketpair IPC: routing determinism (≥2 shards; mis-route would 404), workers=N == workers=1 equivalence (status + body + content-type + content-length + 204 + 502 + TITO accumulated_token_ids across the spawn boundary).

python -m pytest tests/fast/router/ → 95 passed. black / isort / ruff clean.

Docs

docs/developer/multi-process-session-server.md — goal, functional requirements, module decomposition with data-path flowcharts (per-turn chat vs end-of-rollout full-records fetch), behavior parity, design targets, open decisions (incl. the measured records-GET head-of-line finding — accepted for this stack, fix decided and deferred to a follow-up records-path PR — and the kept-vs-cut rationale), and the not-covered list.

[gemini-code-assist]

Code Review

This pull request introduces a CI metric-history collection backend and regression gate storage contract, integrates it into the CI test runner, and implements a multi-process session server data plane (router, worker, IPC, and routing) to parallelize GIL-bound JSON parsing. It also updates DeepSeek-V4 training scripts to keep DSA indexer weights in BF16 and updates FP8 conversion tools. Feedback highlights several critical robustness improvements: resolving a potential TypeError when sorting None values in ci_utils.py, ensuring reliable IPC channel teardown via try...finally blocks in session_ipc.py, catching general exceptions in the router's health check to prevent 500 errors, and catching httpx.RequestError in the worker to properly handle timeouts.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[gemini-code-assist]

In Python 3, comparing None < None raises a TypeError. If there are multiple points in points where the step (p[0]) is None, sorting them with key=lambda p: (p[0] is None, p[0]) will fall back to comparing the second elements of the tuples (None vs None) when the first elements are both True. This will raise a TypeError and crash the CI run.

To prevent this, use a comparable fallback value (like 0) for None in the second element of the sorting key.

Suggested change

	points.sort(key=lambda p: (p[0] is None, p[0]))
	points.sort(key=lambda p: (p[0] is None, p[0] if p[0] is not None else 0))

[guapisolo]

Obsolete — tests/ci/ci_utils.py is not part of this PR anymore. It came from a lower commit in the original stack that dropped out when the PR was rebased onto main (the 1/7–3/7 commits are now merged). Current diff for this PR touches no CI utils; nothing to change here.

[gemini-code-assist]

To prevent resource leaks and ensure that cleanup logic is always executed, use a try...finally block instead of relying on catching specific exceptions. This guarantees that _teardown is called even in the case of unexpected exceptions or early returns.

Suggested change

	except (ConnectionError, OSError) as exc:
	self._teardown(IpcChannelClosed(f"write failed: {exc!r}"))
	try:
	# writer loop logic
	finally:
	self._teardown(IpcChannelClosed("writer loop exited"))

References

1. To prevent resource leaks (e.g., counters that are not decremented), use constructs like try...finally or a with statement to ensure cleanup logic is always executed, even in the case of exceptions or early returns.

[Shi-Dong]

Please fix this.

[guapisolo]

Fixed in 71ffd54. Both the writer and reader loops now wrap the loop body in try/finally so _teardown runs however the loop exits. The named socket-error / peer-closed messages are preserved, and a previously-silent unexpected exception now tears the channel down and surfaces as writer/reader loop crashed: <repr> on every pending request (_teardown is idempotent, so the finally is a no-op when the sibling loop already closed the channel).

[gemini-code-assist]

Similarly to the writer loop, to prevent resource leaks and ensure that cleanup logic is always executed, use a try...finally block to guarantee that _teardown is called even in the case of unexpected exceptions or early returns during packet parsing/dispatching.

Suggested change

	except (IpcError, ConnectionError, OSError) as exc:
	self._teardown(IpcChannelClosed(f"read failed: {exc!r}"))
	try:
	# reader loop logic
	finally:
	self._teardown(IpcChannelClosed("reader loop exited"))

References

1. To prevent resource leaks (e.g., counters that are not decremented), use constructs like try...finally or a with statement to ensure cleanup logic is always executed, even in the case of exceptions or early returns.

[Shi-Dong]

Same here.

[guapisolo]

Fixed in 71ffd54 together with the writer loop — same try/finally guaranteeing _teardown; IncompleteReadError ("peer closed connection") and the IpcError/ConnectionError/OSError reason are preserved, with a catch-all for the previously-silent case.

[gemini-code-assist]

If decode_envelope raises an unexpected exception (such as json.JSONDecodeError or struct.error due to corrupt/malformed data received from a worker), it will not be caught by the current exception handler. This would cause the /health endpoint to crash with a 500 Internal Server Error instead of returning a 503 Service Unavailable (or False). Catching Exception makes the health check much more robust.

Suggested change

	except (asyncio.TimeoutError, IpcChannelClosed, IpcError):
	return False
	except Exception:

[guapisolo]

Leaving this as a narrow catch on purpose. channel.request() returns bytes produced by our own encode_*; if decode_envelope raised struct.error/JSONDecodeError or meta["status"] KeyError, that is an internal protocol bug, not a "backend unhealthy" signal — broadening to except Exception would silently report it as unhealthy and mask the bug. The expected unhealthy modes (TimeoutError, IpcChannelClosed, IpcError) are already caught → False; a genuine protocol violation should surface loudly rather than be swallowed by the health probe.

[gemini-code-assist]

In httpx, TimeoutException (which includes ReadTimeout, ConnectTimeout, etc.) does not inherit from TransportError; they are siblings under RequestError. As a result, any timeout during the proxy request will propagate unhandled, causing the worker to raise an unhandled exception instead of gracefully returning a 502 status code. Catching httpx.RequestError handles both transport errors and timeouts.

        except httpx.RequestError as exc:

[guapisolo]

Not changing this — the premise is incorrect for our httpx (0.28.1). httpx.TimeoutException (and ReadTimeout/ConnectTimeout/…) are subclasses of TransportError: MRO is ReadTimeout → TimeoutException → TransportError → RequestError. So except httpx.TransportError already catches timeouts and returns 502 — verified: issubclass(httpx.ReadTimeout, httpx.TransportError) == True. TransportError is also the more precise base here (the extra RequestError leaves — DecodingError, TooManyRedirects, UnsupportedProtocol — do not arise for a raw non-redirecting proxy).

[Shi-Dong]

LGTM. Two of the Gemini bot comments are worth addressing.

[Shi-Dong]

Please fix this.

[Shi-Dong]

Same here.