feat(intake): log each router's own model call to NVDataflow#74
Conversation
131988a to
f6e898c
Compare
elyasmnvidian
changed the title
WalkthroughChangesSubmodel Intake Capture
Estimated code review effort: 4 (Complex) | ~45 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (3)
tests/test_planning_request_processor.py (1)
387-402: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winUse
pytest-mockfor the new general test doubles.
tests/test_planning_request_processor.py#L387-L402: replace_UsagePlannerClientwith amocker.AsyncMockconfigured to returnPlannerCompletion.tests/test_submodel_intake_response_processor.py#L26-L34: replace_FakeIntakeClientwith amocker.Mockand anenqueue_backgroundside effect that captures the produced payload.As per coding guidelines, tests must “use
pytest-mockfor general mocking.”🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_planning_request_processor.py` around lines 387 - 402, Replace _UsagePlannerClient in tests/test_planning_request_processor.py:387-402 with a mocker.AsyncMock configured to return PlannerCompletion with the required content and usage. Replace _FakeIntakeClient in tests/test_submodel_intake_response_processor.py:26-34 with a mocker.Mock, configuring enqueue_background with a side effect that captures the produced payload.Source: Coding guidelines
tests/test_submodel_intake_response_processor.py (1)
173-193: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winCover the actual fail-open exception paths.
The malformed non-dict case is skipped before emission; add cases where client initialization and one dict record’s payload build/enqueue fail, verifying the response remains successful and later valid calls still emit.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@tests/test_submodel_intake_response_processor.py` around lines 173 - 193, Extend the response-processor tests around SubModelIntakeResponseProcessor.process to cover fail-open errors during client initialization and while building or enqueueing a dict record’s payload. Assert each failure leaves response processing successful and does not prevent a later valid call from emitting its background payload, while preserving the existing malformed-entry coverage.switchyard/lib/processors/llm_classifier/request_processor.py (1)
461-469: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winKeep new token extraction within the async execution path.
These three new synchronous helpers violate the async-only requirement. Inline extraction in the existing async methods, or replace them with an awaitable shared helper.
switchyard/lib/processors/llm_classifier/request_processor.py#L461-L469: remove or make_classifier_token_countsasync.switchyard/lib/processors/plan_execute/request_processor.py#L719-L727: remove or make_planner_token_countsasync.switchyard/lib/processors/stage_router/classifier.py#L264-L272: remove or make_tier_token_countsasync.As per coding guidelines,
switchyard/**/*.pyrequires “async-only code.”🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@switchyard/lib/processors/llm_classifier/request_processor.py` around lines 461 - 469, Keep token extraction within async execution paths by removing or converting the synchronous helpers _classifier_token_counts in switchyard/lib/processors/llm_classifier/request_processor.py:461-469, _planner_token_counts in switchyard/lib/processors/plan_execute/request_processor.py:719-727, and _tier_token_counts in switchyard/lib/processors/stage_router/classifier.py:264-272 to awaitable helpers; update their call sites accordingly while preserving the existing token-count results.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@switchyard/lib/processors/submodel_intake_response_processor.py`:
- Around line 97-105: Validate each mapping in the calls loop before invoking
_build_submodel_payload, requiring valid model, token, and routing fields with
non-negative token counts; skip invalid entries rather than coercing them into
zero-token or blank-field payloads. Keep the existing fail-open logging behavior
for rejected records and only call client.enqueue_background for validated
entries.
- Around line 85-90: Update the submodel intake gating around
RequestMetadata.intake.enabled so requests with body-level store=true also enter
the intake path. Propagate the body opt-in into RequestMetadata or reuse the
primary intake gate, while preserving the existing behavior for requests without
either opt-in.
In `@tests/test_stage_router_classifier.py`:
- Around line 231-232: Remove the explicit `@pytest.mark.asyncio` decorator from
test_classify_stashes_submodel_call_for_intake, relying on the configured
asyncio_mode = "auto" to execute the async test.
---
Nitpick comments:
In `@switchyard/lib/processors/llm_classifier/request_processor.py`:
- Around line 461-469: Keep token extraction within async execution paths by
removing or converting the synchronous helpers _classifier_token_counts in
switchyard/lib/processors/llm_classifier/request_processor.py:461-469,
_planner_token_counts in
switchyard/lib/processors/plan_execute/request_processor.py:719-727, and
_tier_token_counts in
switchyard/lib/processors/stage_router/classifier.py:264-272 to awaitable
helpers; update their call sites accordingly while preserving the existing
token-count results.
In `@tests/test_planning_request_processor.py`:
- Around line 387-402: Replace _UsagePlannerClient in
tests/test_planning_request_processor.py:387-402 with a mocker.AsyncMock
configured to return PlannerCompletion with the required content and usage.
Replace _FakeIntakeClient in
tests/test_submodel_intake_response_processor.py:26-34 with a mocker.Mock,
configuring enqueue_background with a side effect that captures the produced
payload.
In `@tests/test_submodel_intake_response_processor.py`:
- Around line 173-193: Extend the response-processor tests around
SubModelIntakeResponseProcessor.process to cover fail-open errors during client
initialization and while building or enqueueing a dict record’s payload. Assert
each failure leaves response processing successful and does not prevent a later
valid call from emitting its background payload, while preserving the existing
malformed-entry coverage.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 5b2ccee4-f5b4-489b-81b2-c155963863a2
📒 Files selected for processing (13)
switchyard/__init__.pyswitchyard/cli/launchers/launch_intake_config.pyswitchyard/cli/switchyard_cli.pyswitchyard/lib/processors/__init__.pyswitchyard/lib/processors/intake_payload_builder.pyswitchyard/lib/processors/llm_classifier/request_processor.pyswitchyard/lib/processors/plan_execute/request_processor.pyswitchyard/lib/processors/stage_router/classifier.pyswitchyard/lib/processors/submodel_intake_response_processor.pytests/test_llm_classifier_request_processor.pytests/test_planning_request_processor.pytests/test_stage_router_classifier.pytests/test_submodel_intake_response_processor.py
| # ponytail: mirrors the header opt-in only; the OpenAI `store: true` | ||
| # body opt-in isn't mirrored, so a raw client using only `store` | ||
| # under-emits its routing-call record — the safe direction (a missing | ||
| # record, never a leaked one). | ||
| if metadata.intake.enabled is not True: | ||
| return response |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
ast-grep outline switchyard/lib/processors/intake_payload_builder.py --items all
rg -n -C3 'store|intake\.enabled|_request_metadata' \
switchyard/lib/processors/intake_payload_builder.py \
switchyard/lib/processors/submodel_intake_response_processor.py \
switchyard/cli/switchyard_cli.py \
tests/test_submodel_intake_response_processor.pyRepository: NVIDIA-NeMo/Switchyard
Length of output: 20934
🏁 Script executed:
#!/bin/bash
set -euo pipefail
sed -n '1,220p' switchyard/lib/processors/submodel_intake_response_processor.py
printf '\n---\n'
sed -n '1,260p' switchyard/lib/request_metadata.py
printf '\n---\n'
sed -n '1,220p' tests/test_submodel_intake_response_processor.pyRepository: NVIDIA-NeMo/Switchyard
Length of output: 20754
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C4 'store|intake\.enabled|RequestMetadata|attach_request_metadata|x-switchyard-intake-enabled' \
switchyard/lib/request_metadata.py \
switchyard/lib/processors/submodel_intake_response_processor.py \
switchyard/lib/processors/intake_payload_builder.py \
tests/test_submodel_intake_response_processor.py \
tests/test_request_metadata.py \
switchyard/cli/switchyard_cli.pyRepository: NVIDIA-NeMo/Switchyard
Length of output: 29966
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C3 '"store"|\'store\'|store:' switchyard | sed -n '1,220p'Repository: NVIDIA-NeMo/Switchyard
Length of output: 220
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C3 'intake\.enabled|x-switchyard-intake-enabled|store' switchyard/lib switchyard/cli tests | sed -n '1,260p'Repository: NVIDIA-NeMo/Switchyard
Length of output: 17182
🏁 Script executed:
#!/bin/bash
set -euo pipefail
rg -n -C4 'attach_request_metadata|RequestMetadata\(|intake\.enabled|store=true|x-switchyard-intake-enabled|store' switchyard/lib switchyard/cli tests | sed -n '1,300p'Repository: NVIDIA-NeMo/Switchyard
Length of output: 20479
🏁 Script executed:
#!/bin/bash
set -euo pipefail
wc -l switchyard/lib/request_metadata.py switchyard/lib/processors/submodel_intake_response_processor.pyRepository: NVIDIA-NeMo/Switchyard
Length of output: 283
Honor store=true on the submodel intake path. RequestMetadata.intake.enabled is the only gate here, so store=true-only requests can still drop routing-call records even though the Intake sink advertises store=true as an opt-in. Either propagate that body opt-in into RequestMetadata or share the primary intake gate.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@switchyard/lib/processors/submodel_intake_response_processor.py` around lines
85 - 90, Update the submodel intake gating around RequestMetadata.intake.enabled
so requests with body-level store=true also enter the intake path. Propagate the
body opt-in into RequestMetadata or reuse the primary intake gate, while
preserving the existing behavior for requests without either opt-in.
There was a problem hiding this comment.
Good catch — this is the known gap, and you've named the right fix ("share the primary intake gate").
The primary gate is the Rust IntakeRequestState.skip flag, which already ORs in the store: true body toggle. That flag lives in the Rust typed context slot, not in Python ctx.metadata, and this is a response processor with no request in hand to re-read store from — so honoring it means either a new PyO3 getter on the context (plus a rebuild) or moving emission into the Rust sink.
I've folded it into the already-noted follow-up that moves sub-model emission into the Rust sink: that inherits the sink's exact opt-in gate (header or store) for free, and also fixes the separate NVDataflow-direct transport split. Until then the behavior is safe — a store-only opt-in under-emits (a missing record); it never leaks a record for an opted-out caller. Sharpened the code comment to say exactly this.
There was a problem hiding this comment.
@elyasmnvidian, that's a clear explanation — makes sense that this is a Rust/Python boundary constraint rather than something fixable in this processor alone, and folding it into the already-planned move to the Rust sink is the right call since that fix subsumes this one anyway. The fail-open direction (under-emit, never leak) is the correct tradeoff in the meantime, and the sharpened comment should help future readers understand why the gate is intentionally partial here.
Would you like me to open a follow-up issue to track moving sub-model emission into the Rust sink (covering this store-only gap, the NVDataflow-direct transport consistency, and the OpenAI store: true opt-in handling mentioned in the PR notes)? Happy to file it if useful.
✏️ Learnings added
Learnt from: elyasmnvidian
Repo: NVIDIA-NeMo/Switchyard PR: 74
File: switchyard/lib/processors/submodel_intake_response_processor.py:0-0
Timestamp: 2026-07-16T07:49:35.850Z
Learning: In the Switchyard repo (switchyard/lib/processors/submodel_intake_response_processor.py), the primary intake capture gate is the Rust `IntakeRequestState.skip` flag, which already ORs in the request body's `store: true` toggle. That flag lives in a Rust typed context slot, not in Python `ctx.metadata`, so Python response processors (like `SubModelIntakeResponseProcessor`) cannot read the body-level `store` opt-in directly — only `RequestMetadata.intake.enabled` (the header-based opt-in) is visible from Python. Honoring `store: true` from a Python response processor requires either a new PyO3 getter on the context or moving emission into the Rust intake sink. The known tradeoff is considered safe: a `store`-only opt-in under-emits (missing record) but never leaks a record for an opted-out caller.
Note: Learnings are effective only in the context of similar code segments. To apply general coding guidelines, please configure review instructions. You can manage existing learnings in the app.
There was a problem hiding this comment.
Thanks — no need to open a public issue for it. This follow-up is already tracked on our side and written up in the PR's "Open questions / follow-ups" section, so I'll leave it there rather than adding a third place to track the same thing. Appreciate the review.
There was a problem hiding this comment.
@elyasmnvidian, understood — keeping the follow-up in the PR avoids duplicating tracking. Thanks for the clarification and for documenting the scope and safe interim behavior so clearly.
🐇👍
🧠 Learnings used
Learnt from: elyasmnvidian
Repo: NVIDIA-NeMo/Switchyard PR: 74
File: switchyard/lib/processors/submodel_intake_response_processor.py:0-0
Timestamp: 2026-07-16T07:49:35.850Z
Learning: In Switchyard Python response processors (e.g., submodel/intake processors), the body-level `store: true` toggle is not exposed via Python `ctx.metadata` because it is represented by the Rust `IntakeRequestState.skip` flag in a Rust typed context slot. Python processors should only rely on the header-based opt-in `RequestMetadata.intake.enabled` for whether intake emission is enabled. If a Python processor must honor body-level `store: true`, add an explicit PyO3 getter to expose the needed value on the context (or move the emission decision into the Rust intake sink). This design is intentionally safe: a `store`-only opt-in may under-emit (missing record) but must not leak a record for callers who opted out.
Live check with real tokens ✅Ran the real classifier through the pipeline against a live endpoint ( A real router call produces one anonymous record with real token counts, real cost, and the session id; a turn that didn't opt in emits nothing. (POSTing to a live intake endpoint needs the NeMo Platform SDK + auth, which isn't in this sandbox — the model call above is real, and the record is captured from the real sink code instead of being posted.) NVDataflow-direct open question — resolved (it's a real gap)NeMo Platform intake and NVDataflow are two separate stores: the Rust sink posts to NVDataflow instead of the NeMo Platform chat-completions ingest (unauthenticated vs bearer-auth). That direct mode ( So in the default mode both the routed turn and these records go to the same place (NeMo Platform intake). But with |
f6e898c to
d47ff00
Compare
|
Pushed Fixed
Kept as-is, with reasons
ruff + mypy (strict, 164 files) clean; 58 targeted tests pass. |
Signed-off-by: Elyas Mehtabuddin <emehtabuddin@nvidia.com>
d47ff00 to
857f20f
Compare
|
@sir-merge-a-lot groom |
2 participants
What this does
Switchyard's routers make an extra, hidden model call on every turn to decide where to send the request. The deterministic LLM classifier, the stage-router tier classifier, and the plan/execute planner each call a small model first, before the real answer is generated. That call uses tokens — but it never reaches the routed backend, so the intake response processor, which saves one record per routed turn, never saved it. The router's own spend was missing from the usage logs, which made reported cost look lower and savings-vs-baseline look higher than they really were.
This saves each router's own model call as its own record, next to the routed turn, so the numbers reflect the full cost of routing.
Quick orientation: intake and NVDataflow
So the change is: make intake also send the router's own call, so it shows up in NVDataflow next to the routed turn.
What changed
ctx.metadata[CTX_SUBMODEL_CALLS]list —{model, prompt_tokens, completion_tokens, cached_tokens, router_type, routed_to}. It's a list, so a chain that runs more than one (e.g. classifier + planner) adds to it rather than overwriting what's there.SubModelIntakeResponseProcessorreads that list and sends one intake record per entry, reusing the existing cost/version/timestamp helpers so a routing-call record looks the same as a routed turn to NVDataflow.Who gets logged
A routing-call record is sent only when the turn opted into capture — the same per-request opt-in the routed turn uses (
RequestMetadata.intake.enabled, set from thex-switchyard-intake-enabledheader the launcher andserve --intake-enabledsend). The session id comes from that sameRequestMetadata. A turn that did not opt in sends nothing, so a routing-call record is never sent for a caller who opted out.The request-side intake state — session id, opt-in, inbound format — is Rust-owned and lives in a typed context slot, not in Python
ctx.metadata. Reading it throughRequestMetadatais the Python-visible source; the older_intake_*metadata keys are never filled in on this path.Ownership and safety
The new processor builds and owns its own
IntakeClient(the same client class the routed-turn path uses, a separate instance) and closes it inshutdown(), which the server lifespan calls. It never breaks a turn: a build failure turns it off for the rest of the run, and any per-record error is logged and skipped — a routing record can never turn a successful response into a 500.Open questions / follow-ups
IntakeClient, which posts on the NeMo Platform ingest route. The direct-to-NVDataflow flat-document mode (--intake-nvdataflow-project) is written only in the Rust sink, so in that mode the routed turn goes straight to NVDataflow while these records take the NeMo Platform route — fine if that route ends up in the same place, a gap if it doesn't. Worth confirming, or moving the sub-model logging into the Rust sink so there's one path out.store: truebody opt-in isn't mirrored, so a raw client using onlystoresends fewer routing records than it should — a missing record, never one sent by mistake.IntakeClienttransport test are follow-ups.Summary by CodeRabbit
New Features
Bug Fixes