Repository navigation
fix(server): route a session's requests to the worker that holds it - #3905
Conversation
|
Auto-sync is disabled for draft pull requests in this repository. Workflows must be run manually. Contributors can view more details about this message here. |
9d1062f to
94f8f25
Compare
be0e2b2 to
98063d9
Compare
23f6dd6 to
a18c2a7
Compare
…ith generation cuts (NVIDIA-NeMo#3883) ## What changed and why A policy model server with `checkpoint_policy: true` becomes a checkpoint participant. The checkpoint never waits for generations. - **Held responses, and new calls that wait.** Closing admission holds every response that has not been delivered yet. Only a response that is already streaming blocks readiness. - A new policy call that arrives while a checkpoint is open waits until resume before it is admitted, as a resources server's requests do. It used to get 409 `admission_closed`. A rollout that keeps running through the checkpoint, such as a restart, would have shown that refusal to the model or failed. - A waiting call is not admitted, so prepare never counts it, and its agent is still at the boundary it recorded before the call. If its attempt is retired while it waits, it is refused with `stale_attempt` at once instead of running after resume. - The gate tracks the calls waiting at closed admission by capture key. When an attempt is marked retired, each waiting call of that attempt or an earlier one is flagged as refused for good, and woken. The flag stays set after the controller forgets the rollout, and it is set by the mark, which comes before any work the retire's deadline limits, so a retire cut short by its deadline still refuses the call. Without it, a call that had not run again by the time the controller forgot the rollout found no mark and was served once admission reopened. - A retire returns only after those waiting calls have left the gate with `stale_attempt`. A waiting call leaves the tracking when it leaves the gate, so the tracking holds only calls still waiting. - **Undelivered calls are left out, because they have no ledger row.** An undelivered call is excluded from the checkpoint, so after a restore the agent simply makes the call again. - A call's capture-ledger row is published when its response starts, at its first answer chunk, not when the worker stages its tokens. Headers and keep-alive comments publish nothing. Behind the gate, the capture handler leaves the row on the call's capture context (`deferred_commit`), and the gate writes it after `deliver_response` and before the chunk is sent. - Rows used to be written at staging. Each policy server left out its own undelivered calls by `model_call_id`, but a second policy server sharing the ledger could not see them and exported their rows; and a call whose capture context attached after the commit's snapshot could land its row before the export read the ledger. Either way a checkpoint could hold the answer of a call the caller never received. - The write cannot be cut short by a cancellation, and the call stays tracked until the write or its failure handling is done, so a retire or a checkpoint waits for it. A failed write poisons the call and no answer is sent. A call that leaves before its response starts has no row and is not poisoned; its staged tokens are left for the training framework's cleanup, as before. - The export still leaves out undelivered calls' rows, now only as a guard. - With checkpointing on, every model server named in the config of an agent with token capture must be a checkpoint policy server, since the capture ledger is run-wide and a server outside the gate would write rows at staging. Otherwise the model servers refuse to start, with an error naming the server, the agent and the policy server. - **Generation cuts** (`checkpoint_generation_cuts: true`). Prepare asks each inference worker to cut its in-flight calls at a durable token prefix. The replacement call carries the cut in its capture admission, and the worker continues the prefix instead of regenerating it. The wire models (`GenerationCutInventory`, `GenerationCutReceipt`, `GenerationCutPrefixAck`) keep the field layout NeMo RL already uses. A call with constrained decoding (`tool_choice` other than `auto` or `none`, a non-text `response_format`, guided or structured outputs) is never cut and never continues a cut, because a token prefix cannot restore its structured decoder. A cut gets half of the prepare time remaining, so a worker that hangs on its cut still leaves time for the stages after the model. - **Token lineage.** Ledger rows are restored under the replacement attempt's capture key, so parent resolution continues the lineage by normal fingerprint matching, with no parent header relay. - **Several uvicorn workers.** A policy model server may run `num_workers > 1`. It is still one participant to the controller: a coordinator in the main process owns the phases, lease, storage, and restored cuts, and each worker runs a gate (admission, held responses, its own cut round) that reports to the coordinator over a Unix socket. A worker that has not reported, or that disconnects during a checkpoint, blocks the checkpoint instead of weakening it. A worker can also disconnect after the commit checked readiness but before the coordinator collects the workers' snapshots; the rows of that worker's undelivered calls would then be exported as delivered. The commit therefore fails with `export_failed` if a worker was lost or fewer workers answered than expected, and that failure is final for the checkpoint, because a later export could take a later cut. With one worker nothing changes. - **Generation cuts at a per-backend address** (from NVIDIA-NeMo#3971, by Teodor-Dumitru Ene). `generation_cut_control_url` lists one control URL per `base_url`, for backends that serve cuts somewhere other than their OpenAI URL. It is refused alongside `endpoint_file`, and a backend without an address fails its cut, which only costs regenerating the call. - **One runtime directory per server process.** The coordinator socket of a server with several workers lives in `/tmp/ng-<pid>-<random>/` (`nemo_gym/runtime_dir.py`), under `/tmp` because AF_UNIX paths are limited to about 100 bytes. It used to be a socket file in the temporary directory that nothing ever removed. - The directory is removed when the process exits. - Creating one first removes directories that killed servers left behind, but only a directory whose owning process no longer exists **and** none of whose sockets accepts a connection. The second check keeps a live server in another container that shares `/tmp` but not its process namespace, and a connection that times out counts as alive. Nothing is removed by age. - NVIDIA-NeMo#3905 adds the same file for session routing's sockets, so either PR can merge first. - **Scale fixes.** Commit and restore read and write ledgers on a worker thread. A restore installs ledgers without syncing each file (the checkpoint is the durable copy, and importing again is a no-op), with one directory listing and one directory sync: importing 15,686 episodes drops from 3.2 s to 1.5 s. Readiness is cheap to recompute. An episode keeps only its latest cut. The commit reply lists `staging_keys`, the staged token-store keys the checkpoint still refers to, so the training framework knows which rows to keep. Cut requests use the reserved control connection pool, and cut failures are counted in readiness and logged. - **Restored attempts the controller no longer continues.** The model tracks every attempt a restore continues. An attempt stops being pending at its first admitted call, on any worker. With several workers, only the first reopen after an install gives the workers its attempts, so a later reopen cannot make an attempt a worker has started pending again. - An unclaimed restored cut, one no re-issued call has claimed yet, is exported again by the next checkpoint while the controller's commit scope still continues its episode, so a second crash still continues the prefix. - Once the scope leaves out an attempt that never started, the commit drops its cut and deletes its ledger without a fence (`delete_restored`), because the controller may still start the rollout over as that attempt. A started replacement is never touched, so its training rows are kept. - Several policy servers can share one run-wide capture ledger, as `policy_model` and `policy_model_reasoning_off` do, but each tracks its restored attempts on its own. When the first call of a restored attempt reaches a server, before the call can write a row or be held, that server creates a shared "started" marker next to the attempt's ledger, `<capture key>.lineage.started` (`mark_started()`). The marker is created with `O_CREAT|O_EXCL` under the rollout's lock. It is not synced to the directory: other servers only need to see it, and after a crash the next step is a restore, which deletes started markers with the other stale capture files. The commit deletes an unscoped restored attempt through `delete_unstarted()`, which checks for the marker under the same lock and keeps the ledger if any server sharing it has started the attempt. Without the marker, a server that never saw the attempt start would delete the rows another server wrote for it, including the rows of a call that is held and has no row yet. - The marker ends with the attempt's ledger: a ledger retire or delete removes it, and so does a restore that clears what a dead execution left for the attempt. A retired attempt is never marked. - The commit deletes the ledger first and only then drops the attempt from its restored bookkeeping and its saved cut, in one process and with several workers. A delete that fails therefore leaves the attempt pending, and the retried commit deletes it again. - In one process, a restored cut is kept until the call that uses it delivers its response, so a call still undelivered at the next checkpoint continues from it again if its own cut fails. - After each cut round, the newest cut replaces the restored cut a retry would use. For each capture key, the cut of the latest undelivered call, by admission time, as export chooses it, takes the restored cut's place, and that call claims it. The checkpoint exports the newer cut and keeps only its staged rows, so a call that ends before its response starts leaves the newer cut for its retry, not a restored cut whose staged rows the training framework may already have deleted. Delivering the response frees it as before. - With several workers, a worker claims a restored cut from the coordinator before the call runs, and a call that used it never gives it back. A retry of that call therefore regenerates instead of continuing a stale cut. - **Restoring the same checkpoint again.** The capture directory survives a Gym restart. If the replacement attempt makes model calls and Gym crashes before the next checkpoint, restoring the same checkpoint again finds that attempt's files. - Install receives the restore scope and clears the target attempt of every episode in it, including an episode the checkpoint holds no model record for. The import first deletes what dead executions left for the target attempt and for every later attempt of the same rollout. It uses the `delete` operations of NVIDIA-NeMo#3938 and NVIDIA-NeMo#3939: ledgers, token records, intents, incomplete markers, capture state, and **retire fences**. - A fence left in place would silently discard the restored episode's rows once the episode reached that attempt again. Attempt numbers after a restore are deterministic, and the training framework now retires every finished rollout, so fences are common. - The restore then retires the ledgers of the attempts it continues, and of earlier attempts, in one batch. Attempt N continues as attempt N+1, so attempt N's ledger would otherwise stay on disk. - The cleanup lists the ledger directory once and indexes it by rollout. Rescanning it per rollout took about 19 minutes for 16k rollouts over 96k files; the index takes about 0.23 seconds. - A model server that has served any call refuses to restore, because a ledger it wrote may belong to a live episode. With several workers, each worker reports this to the coordinator, and once any worker has left, restore is refused until the server restarts, since a fresh replacement cannot report what the old worker served. This replaces tracking every ledger the process appended to, which grew without bound. - **Worker messages carry what the checkpoint writer carries.** Messages between the coordinator and its workers are framed with the standard `json` encoder, as the checkpoint writer is. `orjson` refused integers beyond 64 bits and silently wrote NaN and infinities, such as a `-inf` logprob, as `null`. A reply that still cannot be framed is answered with a typed error, so the caller does not wait until its deadline. A message's send and reply share one timeout, and a timeout or transport error is `CoordinatorUnavailableError`, so a cut claim against a stalled coordinator falls back to regenerating. A ticket's exit reports to the coordinator even if giving back its cut fails. - **Capture-ledger cleanup stays consistent with checkpoints.** The training framework retires and deletes ledgers through the model server's control routes (NVIDIA-NeMo#3938). - While a checkpoint is open on the model server, those routes refuse with 409, and the caller retries after resume. A commit reads the ledger of every episode it continues, so a retire landing between prepare and commit would leave the checkpoint holding an episode with no lineage, with no error. - A checkpoint `retire` of attempt N cancels the calls of attempt N and every earlier attempt, each only once, and waits for them, then retires those attempts' ledgers, so their files are freed and no late row recreates them. The policy gate refuses the retired attempt's late calls until the controller forgets the rollout. A retire skips calls without a rollout prefix, such as evaluation, judge, or health-probe calls, and a malformed rollout prefix is a 400 `invalid_rollout_id`. With several workers, the coordinator marks the attempt on every worker before any of them stops its calls, relays `forget`, and gives a worker that registers later the current marks. - **Calls a restore carried over are verified against their own capture key.** A restored episode continues as attempt N+1, but the calls it made before the checkpoint stay staged under the earlier attempt's capture key. Rebuilding the episode's rollout rejected them as `wrong_rollout`, so the trainer dropped the row. Restore now stamps each carried-over ledger row with the capture key it was staged under (`CallRecord.capture_key`), and rebuild accepts the same or an earlier attempt of the rollout; anything else fails as `foreign_capture_key`. A chain of restores keeps each row's original stamp. - **Staged keys per episode.** The commit reply also returns `staging_keys_by_episode`: each committed episode's staged keys under its capture key, built from the same records as the flat `staging_keys` list. The controller keeps only the rows of episodes it continues, and a key's name does not always say whose it is: generation-cut keys name no rollout. - **`PolicyGate.deliver_response`** lets a response start, holding it while a checkpoint is open; it was `release_response`. - **A response starts with its first answer data, not its headers.** `StreamingResponse` sends `http.response.start` before it reads its body, and the buffered chat-streaming path sends SSE keep-alive comments while the backend is still generating. - The admission middleware used to call `deliver_response()` on the headers, so prepare treated a call that had produced nothing as streaming and waited for it instead of cutting it. If the snapshot fence held the call's terminal token write, the call could not finish and prepare stalled. - It now calls `deliver_response()` on the first body chunk that carries data, skipping empty chunks and, for an event stream, `SSE_KEEPALIVE`. `SSE_KEEPALIVE` moved to `nemo_gym/chat_streaming.py`, so the middleware and the streaming path share one definition of a liveness frame. - So headers and keep-alives leave the call cuttable, its first answer chunk is held while admission is closed, and an answer that already started still drains before prepare completes. This is the change proposed in NVIDIA-NeMo#4555. - **Generation cut wire names match.** The cut inventory and acknowledgement name the attempt `attempt`, as `EpisodeId` does, and the continuation a re-issued call carries names its prefix `prefix_token_count` and `prefix_digest`, as the worker's acknowledgement does. Conflict note: `CaptureContext` on `main` gained `external_worker_response_seen` (NVIDIA-NeMo#3670); this PR adds `admission_hook` next to it, and both fields are kept. ## How it works ### Where this PR sits in the overall flow The highlighted part is what this PR adds. ```mermaid flowchart LR C["Controller<br/>NeMo RL, or rollout collection"] CO["Coordination<br/>prepare, commit, restore, resume,<br/>retire, forget"] K["Control plane on every server<br/>phases, lease, storage,<br/>retire: refuse, stop, free;<br/>forget"] subgraph G["One participant per Gym server"] E["Environment server<br/>episode steps"] M["Policy model<br/>held responses, generation cuts"] A["Agent<br/>sessions parked at boundaries"] R["Resources server<br/>session state"] end W["Inference worker<br/>stages cut prefixes"] D[("Checkpoint directory<br/>records, then manifest")] L[("Capture ledger<br/>retire and delete from NVIDIA-NeMo#3938, NVIDIA-NeMo#3939")] C --> CO --> K K --> E & M & A & R M --> W M --> L G --> D classDef this fill:#fde68a,stroke:#b45309,stroke-width:2px,color:#1f2937 class M this ``` One checkpoint, a crash, and the restore, end to end: ```mermaid sequenceDiagram participant C as Controller participant G as Gym participants participant D as Checkpoint directory C->>G: prepare, in order environment, model, agent, resources Note over G: admission closes, in-flight work parks at a boundary,<br/>undelivered model responses are held G-->>C: prepared, or blockers at the deadline C->>G: commit with the episodes the controller continues G->>D: each participant writes its records, then its manifest Note over G: restored state the commit's scope leaves out is deleted C->>C: publish the checkpoint with the controller's own state C->>G: resume, in order resources, agent, model, environment Note over C,G: crash - every Gym process dies C->>G: restore the checkpoint in fresh processes, all or nothing D-->>G: records installed under attempt N+1, attempt N's capture ledger retired C->>G: resume C->>G: /run as attempt N+1 continues each episode from its boundary Note over C,G: retiring an episode, only while no checkpoint is open C->>G: retire - environment, then agent, then model and resources Note over G: each server marks the attempt, stops its work, waits, frees its state, then replies C->>G: forget, once the rollout is done for good - same order Note over G: the mark refusing the retired attempt goes away ``` ### This PR A model call across a checkpoint and a crash. ```mermaid sequenceDiagram participant A as Agent participant P as Policy model server participant W as Inference worker A->>P: model call for rollout r, attempt 0 P->>W: generate Note over P: prepare closes admission - new calls park,<br/>responses not yet delivered are held P->>W: generation cut, if enabled - stage the prefix so far W-->>P: receipt with staging keys per call Note over P: commit exports ledger rows minus undelivered calls,<br/>plus the latest cut of each episode Note over A,W: crash, restore, resume A->>P: the same call again, as attempt 1 P->>W: continue from the staged prefix, or generate again ``` With several uvicorn workers, one coordinator in the main process is the participant; every worker keeps its own gate and forwards control calls to the coordinator. ```mermaid flowchart LR CO["Coordination"] -->|"control call on the shared port"| W1["uvicorn worker 1<br/>gate: tickets, held responses"] CO -->|"control call on the shared port"| W2["uvicorn worker 2<br/>gate: tickets, held responses"] W1 <-->|"Unix socket: forward, close, reports, cut claims"| C["Coordinator in the main process<br/>phases, lease, ledger, restored cuts"] W2 <-->|"Unix socket"| C ``` The capture ledger across checkpoints, restores and the training framework's cleanup. ```mermaid sequenceDiagram participant T as Training framework participant P as Policy model server participant L as Capture ledger Note over P: checkpoint open - ledger retire and delete routes answer 409 T->>P: retry retire and delete after resume Note over P,L: restore - delete what dead executions left for attempt N+1 and later, fences included P->>L: import attempt N's rows as attempt N+1, then retire attempts 0 to N Note over P,L: checkpoint retire of attempt N - stop its calls, then retire the ledgers of attempts 0 to N ``` ## Where this sits in the stack This is one PR in a stack of draft PRs that re-cut partial-rollout checkpointing onto environment servers. Each PR is one commit, and its base is the branch of the PR before it, so each diff shows only that PR's change. The stack builds on the token-capture cleanup PRs NVIDIA-NeMo#3938 (capture ledger `retire` and `delete`) and NVIDIA-NeMo#3939 (complete-record `retire` and `delete`), which have merged. NVIDIA-NeMo#3882 has merged too, so NVIDIA-NeMo#3883's base is `main`. The checkpoint stack uses those operations to free the ledgers of retired attempts and to clear a dead execution's capture files before a restore. The capture-ledger cache bound (NVIDIA-NeMo#4268) and the striped lock files (NVIDIA-NeMo#3937) are independent of the stack. 1. NVIDIA-NeMo#3882 (`ananthsub/partial-ckpt-core`): feat(checkpoint): add the participant control plane, episode steps, and coordination (merged) 2. NVIDIA-NeMo#3883 (`ananthsub/partial-ckpt-policy-model`): feat(checkpoint): make policy model servers checkpoint participants with generation cuts **(this PR)** 3. NVIDIA-NeMo#3884 (`ananthsub/partial-ckpt-model-worker-cuts`): feat(token-capture): add worker staging helpers for generation cuts 4. NVIDIA-NeMo#3885 (`ananthsub/partial-ckpt-environment`): feat(checkpoint): continue environment server episodes from their boundaries 5. NVIDIA-NeMo#3886 (`ananthsub/partial-ckpt-resources`): feat(checkpoint): add the resources server participant with asynchronous session hooks 6. NVIDIA-NeMo#3887 (`ananthsub/partial-ckpt-verifier-declarations`): feat(checkpoint): declare five training verifiers stateless with replayable verification 7. NVIDIA-NeMo#3888 (`ananthsub/partial-ckpt-agent`): feat(checkpoint): add the agent session participant and Simple Agent continuation 8. NVIDIA-NeMo#3889 (`ananthsub/partial-ckpt-e2e`): test(checkpoint): add a process-level end-to-end suite driven by coordination 9. NVIDIA-NeMo#3893 (`ananthsub/partial-ckpt-rollout-collection`): feat(checkpoint): checkpoint evaluation runs from rollout collection 10. NVIDIA-NeMo#3903 (`ananthsub/partial-ckpt-telemetry`): feat(checkpoint): spans and metrics for partial-rollout checkpoints 11. NVIDIA-NeMo#3909 (`ananthsub/partial-ckpt-multi-worker`): feat(checkpoint): resources, agent, and environment servers with several workers 12. NVIDIA-NeMo#4197 (`ananthsub/partial-ckpt-fault-tests`): test(checkpoint): check checkpoint invariants over random sequences of operations Partial-rollout checkpointing lets a training controller, such as NeMo RL, checkpoint Gym while rollouts are in flight and, after a crash, continue those rollouts from their last safe point instead of starting them over. Checkpointing is off by default; with the `checkpoint:` block unset, no server installs any checkpoint routes or behavior. ## Relationship to the old stack (NVIDIA-NeMo#2939 to NVIDIA-NeMo#2946) - Supersedes NVIDIA-NeMo#2941 (policy model participant). The checkpointable ledger wrapper and parent relay are replaced by `main`'s capture ledger plus rows re-imported under the next attempt; admission no longer waits for generations. - Takes the generation-cut models from NVIDIA-NeMo#2940, under `nemo_gym._checkpoint.generation_cut` instead of `model_control_contracts`. - Multi-worker support follows the shape of NVIDIA-NeMo#3563 (a companion in the main process, a Unix socket, and a restored cut with a single owner) with much less machinery: no proof files, evidence files, or worker journals. The scale fixes cover the remaining needs listed in NVIDIA-NeMo#3755 (reserved control pool, restore I/O off the event loop). Constrained-decoding restart comes from NVIDIA-NeMo#3412. ## Issue No tracking issue exists for this re-cut. The design and the mapping from the old stack are described in this PR series, and the old stack's PRs (NVIDIA-NeMo#2939 to NVIDIA-NeMo#2946) carry the original discussion. ## Validation Run on this branch, on the current `main`, which includes NVIDIA-NeMo#3882: - `RAY_TMPDIR=/tmp .venv/bin/python -m pytest -q -p no:cacheprovider tests/unit_tests/test_checkpoint_*.py`: passed. - `RAY_TMPDIR=/tmp .venv/bin/python -m pytest -q -p no:cacheprovider tests/unit_tests/test_token_capture_*.py tests/unit_tests/test_token_id_capture.py tests/unit_tests/test_base_responses_api_model.py`: passed. - `RAY_TMPDIR=/tmp .venv/bin/python -m pytest -q -p no:cacheprovider responses_api_models/vllm_model/tests`: passed. - Among the checkpoint tests, each failing without its change: - restoring the same checkpoint again after the replacement made a call (without the fix: `capture ledger for roll-1-a1 already holds rows from another execution`), with the files and fences of the target and later attempts deleted and look-alike rollouts untouched; - a restore retires the ledgers of the attempts it continues; - a model server that has served calls refuses to restore, with one process and with two workers; - ledger retire and delete return 409 while a checkpoint is open and succeed after resume; - a retire stops a running generation before it replies, and retires the ledgers of attempts 0 to N but not N+1; - a retire leaves no fence on any worker, including one that joins later; - worker messages carry NaN, infinities, and integers beyond 64 bits unchanged; - a commit that no longer names a started replacement leaves its ledger alone, and deletes a restored attempt nothing started without fencing it; with two workers, an attempt any worker started is no longer pending, and a reopen does not make it pending again; - restoring again clears a dead execution of an in-scope episode without a model record; - untagged calls neither break a retire nor surface as untyped errors; - a hung cut leaves the rest of the prepare deadline for the stages after it; - a message that cannot be sent or is never answered is a typed unavailable error, and a ticket leaving is reported even if giving back its cut fails; - restore is refused after a worker left, even if its replacement is fresh; - a call that arrives during a checkpoint waits for resume without being admitted, and is refused at once if its attempt is retired while it waits; - two policy servers sharing a ledger, one holding a completed response: neither snapshot, nor the restored ledger, holds its row (without the fix, the second server exported it); - a call whose capture context attaches after the commit's snapshot cannot land its row in the checkpoint (without the fix, it did); - a row is published at the first answer chunk, not at headers or keep-alives; a call that leaves first has no row and is not poisoned; a failed write poisons the call and sends no answer; a retire or a checkpoint during the write waits for it; each worker of a server with several workers follows the same rule; - a model server outside the checkpoint gate that an agent with token capture names is refused at startup; - a retried retire waits for a call an earlier retire cancelled without cancelling it again (without the fix, its cleanup was cancelled a second time); - marking a restored attempt started does not sync the ledger directory, and a restore after a crash removes the markers the dead execution left; - headers, empty bodies and keep-alives do not start a response, for JSON and event streams; a stream sending only keep-alives during a checkpoint is cut and its output is held until resume (without the fix, both time out); an event stream that already started finishes while admission is closed; - a worker that disconnects after the commit checked readiness and before the snapshot fails the commit, and a retried commit fails the same way; - a restored attempt started on one policy server keeps its ledger when a commit on another server sharing the ledger leaves it out of scope, both once the started call's row is recorded and while the call is admitted but not yet recorded; with several workers, a worker marks the attempt started in the shared ledger; - a restored attempt no server started is deleted, and no marker is left behind; - the started marker is removed by a ledger retire, a ledger delete, and a restore that clears the attempt; - a failed delete of a restored ledger keeps the attempt pending, and the next commit attempt deletes the remaining rows, in one process and with several workers; - a call waits at closed admission, its attempt is marked and retired, the controller forgets the rollout, and admission reopens; the call is refused with `stale_attempt` and nothing is served; - the same with a retire cut short by its deadline, in one process and on a worker of a server with several workers, where the waiting call also leaves the gate's tracking; - a resumed call that continues a restored cut is cut again by the next checkpoint, which exports the newer cut, and then ends before its response starts; its retry continues the newer cut, not the restored one; - with several workers, a retry of a call that used its restored cut gets no cut after a newer checkpoint, and the coordinator no longer holds the restored cut. - `RAY_TMPDIR=/tmp .venv/bin/python -m pytest -q -p no:cacheprovider tests/unit_tests/test_runtime_dir.py`: passed. A dead server's directory with closed sockets is removed; a live process's directory is kept; a directory whose socket accepts is kept even when its process looks dead; only runtime directory names are considered; and a process creates its directory once, with mode 0700, and removes it at exit. - The full core unit suite (`tests/unit_tests`, eight processes, without `test_opensandbox_compose.py`, which needs the optional `opensandbox` package): passed except a sandbox retry test, which fails the same way on `main` in this development environment. - `pre-commit run --files <files changed by this PR>`: all hooks passed, and no hook modified a file. - Every commit carries a `Signed-off-by` line. ## Rollout evidence - **Real model:** see the rollout evidence in NVIDIA-NeMo#3889, whose opt-in tests drove this PR's code with `Qwen/Qwen3-4B-Instruct-2507` on vLLM: saved rollouts continued after a kill and restore, restarts started over, and every rollout finished with reward 1.0. A real vLLM worker serving generation cuts is not covered: stock vLLM does not serve the cut route. - Scale, on one workstation with every Gym server and a fake inference backend (weather rollouts through the typed `single_agent_turn` request path, token capture and generation cuts on, each rollout's second model call held so most rollouts are in flight at the checkpoint; checkpoint, kill every Gym process group, restart, restore, run the replacement attempts). After the scale fixes in this PR, measured on the branch this stack was cut from, with the same commits: | Rollouts | Policy workers | In flight at checkpoint (exported) | Prepare (s) | Commit (s) | Restore (s) | Replacement failures | |---|---|---|---|---|---|---| | 8,000 | 1 | 4,201 | 0.67 | 0.60 | 0.60 | 0 | | 8,000 | 2 | 4,083 | 0.56 | 0.69 | 0.56 | 0 | | 16,000 | 2 | 14,320 | 7.57 (model 7.38) | 4.64 | 4.97 | 0 | | 16,000 | 2 (rerun) | 12,228 | 3.60 (model 3.53) | 1.83 | 3.23 | 0 | In the first 16,000-rollout run, the policy model logged 8 uvloop "File descriptor N is used by transport" errors and 16 requests failed with "ledger lineage resolution failed". The cause is a libuv accept race while clients disconnect, not checkpoint code; the ledger failed closed, clients retried, and no replacement failed. The rerun and every 8,000-rollout run had none. ## Compatibility - New model server config: `checkpoint_policy` (default `false`) makes the server a participant, and `checkpoint_generation_cuts` (default `false`) turns on generation cuts. Both only take effect with the global `checkpoint:` block enabled. - Users of the old stack's `token_id_capture.generation_prefix_cuts_enabled` key (never on `main`) switch to `checkpoint_generation_cuts`. - Generation-cut models are imported from `nemo_gym._checkpoint.generation_cut`. - The `CheckpointableLedger` protocol gains `mark_started()` and `delete_unstarted()`, so a custom ledger used with a checkpoint participant must implement both. `PolicyWorkerLink` takes a `ledger=` argument, which a worker needs to mark restored attempts started. - A restored attempt's first call writes a `<capture key>.lineage.started` file in the capture ledger directory. It is removed with the attempt's ledger. - With checkpointing off, model servers behave as before. Signed-off-by: Ananth Subramaniam <ansubramania@nvidia.com> Co-authored-by: Anish Mahishi <amahishi@nvidia.com> Co-authored-by: Teodor-Dumitru Ene <teodord.ene@gmail.com>
a18c2a7 to
ee788e6
Compare
8b6e4a6 to
bbe9e7a
Compare
|
A session whose seed never read the session was not claimed. The previous version claimed a new resources session only if its handler read or wrote This version applies one rule: a session ID a client can name is a claimed session.
Tests that fail on the previous version: a seed through the default This version also adds a comment on the router's per-claimant |
|
/ok to test bbe9e7a |
bbe9e7a to
939644b
Compare
uvicorn's workers share one listening socket, so each request goes to whichever worker accepts the connection, while session state lives in the memory of the worker that created the session. With the stateful counter resources server at 4 workers, 200 concurrent episodes through /run scored 1.0 on only 7; swe_rebench (8 workers) can lose a session's sandbox at /verify and score 0. With num_workers > 1, resources and agent servers now route every request by one rule: 1. find the request's session ID; 2. look up the session's claim, which names the worker that created it; 3. forward the request to that worker's private Unix socket, unchanged, and stream its reply back unchanged. If this worker holds the claim, or nobody does, the request is handled here. The session ID comes from the request: - on a resources server, the session cookie's session_id; for an MCP tool call, or a request with an MCP session token and no cookie, the token's sid; for a typed /seed_session or /close_session, the body's resources_session_id; - on an agent server, the cookie's agent_session_id, or the body's agent_session_id on the agent session seed and close. A typed resources seed now gives its cookie its resources_session_id as the session_id, so the seed, its close and every request in between name the session by one ID. If the cookie and the body name different sessions, the reply is 409, with one worker too. A resources request with no session ID gets a new ID from the router. A session start (/seed_session, or another route in checkpoint_session_starts) claims it before its handler runs, since its reply's cookie names the session; the default seed_session, which an agent's own /run reaches with its untyped body, never reads the session. Any other request claims its new ID only if the handler read or wrote the session, before the reply leaves the worker; a request that keeps nothing, such as a health check, /aggregate_metrics or a checkpoint control route, leaves no claim and gets no session cookie, so no cookie names a session nobody claims. A request whose cookie names a session whose claim ended is claimed the same way by the worker that serves it. If another worker claimed the same ID meanwhile, the request gets 409 rather than being served twice. A claim is a file in the server's runtime directory, created atomically by the worker that creates the session: on a seed with or without a cookie, an agent seed, another session start, a resources session whose handler used it, or a close of an unknown ID, so a delayed seed is refused as on one worker. A request over a private socket is always handled where it arrives, and a claimant whose socket is gone gives 410 instead of running against empty state. Cookies and MCP tokens no longer carry the creating worker's ID; one minted by an earlier version is read for its session ID only. Every claim ends. An agent releases its claim when it drops the session's last record: the close receipt or the failed seed, session_close_retry_window_seconds after it was written. A resources claim lasts while the session is open, from a successful seed (or another route in checkpoint_session_starts) until its close, its /verify or checkpoint_session_ended (a /verify that says a close follows, as single_agent_turn's does, leaves it open for that close), and otherwise ends session_close_retry_window_seconds (a new resources server setting, default 300) after the last request for the session. Each worker sweeps its own expired claims, so they end even when no more requests arrive. An agent seed whose state was set up but cannot be registered, for example because the checkpoint participant refuses it, now releases that state and its claim at once, or keeps it as a failed seed if the release fails. The claims of a worker that exited, and the whole directory, go when the server exits. With one worker nothing is claimed. ClosedResourcesSessionRecords keeps a closed resources session for the same window, counted from its first close, so the record never outlives the claim. NOOA keeps its closed sessions in it, and its tool routes refuse a cookie whose counter is gone, so a stale cookie cannot recreate a counter that no close would remove. An agent drops a failed seed that nobody closed once the window has passed, running its close hook in a background task with a fresh context, under the session's own rollout. If the hook raises, the record keeps its state for a later close and the release is tried again a window later. Seeds and closes of any size are routed: the router reads the whole body, as the app does anyway. Each worker's private socket is served by a second uvicorn server, which leaves uvicorn's process-wide loggers as the worker's main server configured them and writes no access lines of its own, so uvicorn_logging_show_200_ok logs each request once. Agent sessions no longer require num_workers=1. Codex and NOOA drop their own copies of the check, so the harnesses (Codex, Hermes, NOOA, OpenClaw, OpenCode and Pi) all seed sessions on several workers, and each one's tests check a seed with two workers. Their READMEs and the Hermes SWE-bench Pro guide no longer ask for one worker. The resources server guide gains a section on state with several workers. The counter resources server gains the module-level app that multi-worker uvicorn imports. An e2e test (NEMO_GYM_MULTIWORKER_E2E=1) runs it with 4 workers behind the /run relay (`legacy_agent`) and Simple Agent, and the unit-tests workflow runs it when the routing code, or the server code it wires into, changes. An opt-in stress test (NEMO_GYM_MULTIWORKER_STRESS=1) runs thousands of concurrent episodes through single_agent_turn over 8 resources and 4 agent workers, cookieless seeds and closes, untyped seeds that never read the session, MCP sequences, a lost worker and the claims' lifetime. Signed-off-by: Ananth Subramaniam <ansubramania@nvidia.com>
939644b to
c46fd97
Compare
What changed and why
uvicorn's workers share one listening socket, and each request goes to whichever worker accepts the connection. Session state lives in the memory of the worker that created the session, and nothing keeps a session's later requests on that worker.
Measured with the stateful counter resources server (
example_session_state_mgmt) and 200 concurrent episodes through/run:Servers on main with this shape:
swe_rebenchruns 8 workers. It stores each session's sandbox in an in-memory dict at/seed_sessionand pops it at/verify. A/verifyon another worker finds nothing, the patch is recorded as empty, and the reward is 0./mcp. They send the signed MCP session token minted at/seed_session, and no Gym session cookie. With 4 workers and 200 concurrent counter sequences over MCP, 65 of 200 ended with the right count.One rule
With
num_workers > 1, resources servers and agent servers route every request the same way:A request that arrives over a private socket is always handled where it arrives, so nothing is forwarded twice. If the claimant's socket is gone, because its worker exited and uvicorn replaced it, the request gets a 410 instead of running silently against empty state.
The cookie, the MCP session token and the body only supply the session ID:
session_id(SESSION_ID_KEY)/mcp, or a request with an MCP session token and no cookiesid/seed_sessionor/close_sessionresources_session_idagent_session_id/v1/agent_sessions,/v1/agent_sessions/close)agent_session_idSeed and close routes are matched bare or under a rollout prefix. Some details of the rule:
resources_session_idas thesession_id, so the seed, its close and every request in between name the session by one ID. Handlers that readrequest.session[SESSION_ID_KEY]keep working, and see the same ID the body carries. An agent session already had one ID./seed_session, or another route incheckpoint_session_starts) claims it before the handler runs, whether or not the handler reads the session, because its reply's cookie names the session. This covers an agent's own/runseeding with its untyped body, which the defaultseed_sessionanswers without reading the session. Any other request claims its new ID only if the handler read or wrote the session, before the reply leaves the worker, so the client's next request through any worker finds the claim; this covers a server that creates a session implicitly on its first tool call. A request whose handler never touches the session, such as a health check,/aggregate_metricsor a checkpoint control route, leaves no claim and gets no session cookie, so no cookie ever names a session nobody claims. If another worker claimed the same random ID meanwhile, the request gets 409 rather than being served twice.Claims
A claim is a file in the server's runtime directory,
/tmp/ng-<pid>-<random>/session-claims/, named by a hash of the session ID and holding the claimant's worker ID. The worker that creates a session creates its claim with a hard link from a complete temporary file, which fails if the claim exists, so two workers can never both claim an ID and a reader never sees a partial claim.Claims are created on every path that creates a session:
Every claim ends:
session_close_retry_window_secondsafter it was written/seed_session(or another route incheckpoint_session_starts, such as a Gymnasium/reset) until its/close_session, its/verify, orcheckpoint_session_ended. A/verifythat says a close follows (x-ng-checkpoint-close-follows, whichsingle_agent_turnsends) leaves it open for that closesession_close_retry_window_secondsafter the last request for it: after its close or verify, after a failed seed, or for a session that never had a seedsession_close_retry_window_secondsis a new resources server setting, default 300 seconds, as agents already had. It covers the caller's retried and cleanup closes, which reach the claimant within the window.ClosedResourcesSessionRecordsinnemo_gym.base_resources_serverkeeps a closed resources session for the same window, counted from its first close, so the record never outlives the claim. While the claim lasts, a late seed reaches the worker that remembers the close and is refused; after it ends, every worker treats the ID as new. NOOA keeps its closed sessions in it, and its tool routes refuse a cookie whose counter is gone.sequenceDiagram participant C as Caller participant B as Worker B participant A as "Worker A (claimant)" C->>B: POST /verify with session cookie B->>B: read session_id from the cookie B->>B: read the session's claim: worker A B->>A: same request over A's private socket A->>A: SessionMiddleware and handler with A's state A-->>B: reply with A's Set-Cookie B-->>C: reply streamed back unchangedsequenceDiagram participant E as "Environment server" participant B as Worker B participant A as Worker A E->>A: POST /seed_session, resources_session_id s, no cookie A->>A: claim s for A, seed it A-->>E: 500, no Set-Cookie E->>B: POST /close_session, resources_session_id s, no cookie B->>B: read s from the body, claim of s: worker A B->>A: same request over A's private socket A->>A: close releases the partial state A-->>E: 200, through BAlso in this change
main, an agent server refuses to seed an environment server session whennum_workers > 1. That guard is removed, and Codex and NOOA drop their own copies of it, so the harnesses (Codex, Hermes, NOOA, OpenClaw, OpenCode and Pi) all seed sessions on several workers, like Simple Agent. Each harness's tests check that a seed with two workers opens its session. The harness READMEs and the Hermes SWE-bench Pro guide no longer ask for one worker.uvicorn_logging_show_200_oklogs each forwarded request once, on the worker that received it.session-routing-e2ejob inunit-tests.ymlrunsNEMO_GYM_MULTIWORKER_E2E=1 pytest tests/e2e/session_routingwhen the routing code, or the server code it wires into, changes.example_session_state_mgmtgains the module-levelappthat multi-worker uvicorn imports.How it fits together
SessionMiddlewareand reads the cookie by decoding it with the same secret. A forwarded request never reaches the receiving worker'sSessionMiddleware, so the reply carries only the claimant'sSet-Cookie.SimpleResourcesServerandSimpleResponsesAPIAgentthroughroutes_sessions_to_owner, which a server that keeps no per-session state can set toFalse. Model servers keep no session state and are not routed.UnixConnectorper claimant socket; a claimant whose socket refuses connections has its client discarded.dateandserverheaders./runcalls that carry no session ID by the worker their header names, which the router handles before it falls back to handling the request locally.Related work
This fix targets main and does not depend on the rest of the partial-rollout checkpointing stack (#3886 onward; #3882 to #3885 have merged). #3909 builds on it to support environment, agent, and resources servers with several workers. The adapter PR #4790, stacked on this one, moves more resources servers onto
ClosedResourcesSessionRecords, whose name and interface are unchanged.Issue
None known. The bug was found while measuring multi-worker behavior for partial-rollout checkpointing. This is the fix for main, separate from that work.
Validation
Run on this branch, on the current
main:pytest tests/unit_tests/test_session_routing.py: passed. In-process workers share a real socket directory. The tests check, table-driven where it helps:/aggregate_metrics, a control-style POST) and an agent request without a session create none and get no session cookie, a new session is claimed before its reply arrives, a cookie whose claim ended is claimed again by the worker that uses it, a minted ID another worker claimed meanwhile gets 409, and sessions minted on two workers at once are distinct;checkpoint_session_ended, a session started without a seed a window after its last request, and a failed seed a window after the cleanup close, all without another request;Set-Cookie;pytest tests/unit_tests -n 16, in a venv with CI's extras: passed.codex_agent,opencode_agent,openclaw_agent,nooa_agentandhermes_agenttests, each in that agent's own environment: passed. Thepi_agenttests: passed except one test that runs the installed Pi CLI, which fails the same way onmainbecause of the local Node.js version.simple_agent,example_session_state_mgmt,example_single_tool_callandsingle_agent_turntests: passed.NEMO_GYM_MULTIWORKER_E2E=1 pytest tests/e2e/session_routing: passed.NEMO_GYM_MULTIWORKER_STRESS=1 pytest tests/e2e/session_routing/test_session_routing_stress.py: passed; see below.pre-commit run --files <changed files>: passed.Rollout evidence
Qwen/Qwen3-4B-Instruct-2507on vLLM 0.29.0 and two workers per server, every rollout that continued after a crash and restore, and every restarted rollout, finished with reward 1.0 in three runs in a row. See feat(checkpoint): resources, agent, and environment servers with several workers #3909's rollout evidence.The opt-in e2e test runs real server processes against a fake OpenAI-compatible backend, with the counter resources server at 4 workers behind the
/runrelay (legacy_agent) and Simple Agent: every/runepisode, every seed-then-verify session and every MCP sequence scores correctly, and killing a worker gives exactly its sessions a 410.The opt-in stress test (
tests/e2e/session_routing/test_session_routing_stress.py) runs the NOOA counter resources server at 8 workers and Simple Agent at 4 workers behindsingle_agent_turn, against the same fake backend, withsession_close_retry_window_secondsset to 5 seconds:single_agent_turn, episode requestssingle_agent_turn_legacy)/runsends), then five tool calls and an end that read it, each on a new connectionRouting's cost, over one kept-alive connection to the 8-worker server against the same requests to a 1-worker copy: a tool call took 1.9 ms at the median on one worker and 3.1 ms when the accepting worker forwarded it, so forwarding adds about 1.2 ms at the median and about 0.4 ms at p99 (8.8 ms against 9.2 ms). The claim lookup alone, opening and reading a small file, takes about 8 microseconds at the median.
With a real model: the checkpoint stack's opt-in tests ran
Qwen/Qwen3-4B-Instruct-2507on vLLM with two workers per agent, resources and environment server (#3909), so every session's requests were routed to the worker that holds it, and every rollout finished with reward 1.0. They ran against the previous version of this change and will run again on this one with #3909.Pending:
swe_rebenchrun at 8 workers with real sandboxes;Compatibility
resources_session_id, and a typed seed or close whose cookie names another session gets 409.request.session[SESSION_ID_KEY]and the MCP endpoint's token check are unaffected.agent_session_id. An agent that keeps state across requests by the cookie'ssession_idinstead, outside agent sessions, is not routed by it, and should keep one worker.num_workers > 1now accept agent session seeds instead of raisingAgent sessions require num_workers=1.session_close_retry_window_secondsrefuses a delayed seed through another worker only within the window;ClosedResourcesSessionRecordskeeps the two in step. State keyed by an ID that the request carries nowhere in the table above is not routed.