feat(intake): log each router's own model call#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. |
d47ff00 to
857f20f
Compare
|
@sir-merge-a-lot groom |
e7c259e to
0c1e1a7
Compare
Signed-off-by: Elyas Mehtabuddin <emehtabuddin@nvidia.com>
0c1e1a7 to
9e8ff05
Compare
Signed-off-by: Elyas Mehtabuddin <emehtabuddin@nvidia.com>
elyasmnvidian
changed the title
|
Two changes since the earlier version of this PR — I rewrote the body to match. The router-call logging moved into the Rust intake sink. Earlier it was a separate Python processor with its own The posting target is now a configurable URL. Proof is in the body: a live Claude Haiku 4.5 run through our inference gateway, real classifier + routed-turn token counts landing in two flat documents on the configurable target, with distinct |
|
Signed-off-by: Elyas Mehtabuddin <emehtabuddin@nvidia.com>
Signed-off-by: Elyas Mehtabuddin <emehtabuddin@nvidia.com>
|
Validated end to end against a live NVDataflow sandbox — HTTP 201. To confirm the old behavior still works, I ran a real model call through the new sink with Same flat, type-prefixed document and same unauthenticated POST as before — the wire behavior is unchanged. The one migration step is that you pass the full posting URL instead of a project name: Use whichever path your project is registered under (the sandbox and registered projects differ). Because the URL is fully configurable now, moving between the sandbox, a registered project, or a different store is config, not a code change — which is the point of the refactor. Also pushed since the last update:
|
2 participants
What this does
Switchyard's routers make an extra 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 sink — which saved 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 PR saves each router's own model call as its own record, next to the routed turn, through the same transport, so the numbers reflect the full cost of routing.
How it works
ctx.record_submodel_call(model, prompt_tokens, completion_tokens, cached_tokens, router_type, routed_to). It appends to a list, so a chain that runs more than one router (classifier + planner) records each, not just the last.IntakeResponseProcessorsends one record per recorded call, right after the routed turn, reusing the same payload builder. A routing-call record looks the same as a routed turn to the store, minus the message text._id({session}-{ts}for the turn,{session}-{ts}-{routed_to}-{index}for each routing call), so records that share a session and timestamp never overwrite each other.Both the routed turn and the routing-call records now go out through the same Rust intake sink. That closes the two gaps an earlier draft left open:
IntakeRequestState.skip), which combines thex-switchyard-intake-enabledheader opt-in with the OpenAIstore: truebody opt-in. A raw client that only setsstore: truenow gets its routing records too. A turn that did not opt in sends nothing, so a routing record is never sent for a caller who opted out.The posting target is configurable
The sink no longer hardcodes a destination. It selects one with
IntakeSinkConfig.target(IntakeTarget { url, format, authenticated }):--intake-target-url <full URL>(envSWITCHYARD_INTAKE_TARGET_URL) posts a flat, top-level document straight to that URL — any data-lake posting endpoint — unauthenticated. The caller passes the full URL; no host or path is baked into the sink.This replaces the old
--intake-nvdataflow-projectflag, which built one fixed URL against one fixed host. A data-lake store is now just a URL you point at.Proof: real token usage lands in the records
A live run against Claude Haiku 4.5 through our inference gateway, with the target pointed at a local capture server. Two real calls — a routing classifier call and the routed turn — driven through the real request and response processors with
store: trueas the only opt-in:The routing-call record (real classifier tokens, its own route label, no latency because a routing call's latency is not the turn's):
{ "_id": "live-proof-1-1784589046905-classifier-0", "b_switchyard_routed": true, "l_switchyard_input_tokens": 24, "l_switchyard_output_tokens": 5, "l_switchyard_total_tokens": 29, "s_switchyard_router_type": "deterministic", "s_switchyard_routed_to": "classifier", "s_switchyard_served_model": "claude-haiku-4-5", "s_switchyard_session_id": "live-proof-1" }The routed-turn record (real turn tokens, distinct
_id, its own latency):{ "_id": "live-proof-1-1784589046905", "b_switchyard_routed": false, "l_switchyard_input_tokens": 18, "l_switchyard_output_tokens": 11, "l_switchyard_total_tokens": 29, "l_switchyard_latency_ms": 1, "s_switchyard_served_model": "claude-haiku-4-5", "s_switchyard_session_id": "live-proof-1" }Both posted to the configurable target URL, unauthenticated, with distinct
_ids and real token counts. Records are metadata-only by default — no message text.What stays the same
SWITCHYARD_INTAKE_CAPTURE_CONTENTis set.