feat: stream NAT OpenAI chunks natively - #202
Conversation
WalkthroughThis PR adds native OpenAI Chat Completions streaming across Fabric Core, the Python SDK, adapter lifecycle hosts, NAT workflows, schemas, tests, and API documentation. It uses authenticated loopback HTTP with correlated chunk and terminal records. ChangesNative OpenAI streaming
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Client
participant Runtime
participant Stream as OpenAIInvokeStream
participant Host
participant Adapter
Client->>Runtime: invoke_openai_stream()
Runtime->>Stream: create authenticated stream
Runtime->>Host: dispatch stream invocation
Host->>Adapter: invoke_openai_stream(payload, emit)
Adapter->>Stream: emit OpenAI chunks
Stream->>Client: yield validated chunks
Adapter->>Host: return terminal result
Stream->>Client: return RunResult
Possibly related PRs
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
|
Fern docs preview: https://nvidia-preview-pull-request-202.docs.buildwithfern.com/nemo/fabric |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 15
🤖 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 `@adapters/common/src/nemo_fabric_adapters/common/lifecycle.py`:
- Around line 577-612: Update _handle_invoke_openai_stream so adapter transport
or cancellation failures remain the primary exception when writer.finish() also
fails. Preserve and re-raise adapter_error, while logging the cleanup error
through the existing error-reporting mechanism instead of raising it; if only
cleanup fails, continue raising that cleanup error.
- Around line 233-241: Update the HTTP status parsing near status_line so it
does not require a reason phrase: split with a maximum of two splits, then read
the status token by index and convert it to int. Preserve LifecycleError
handling for malformed or undecodable responses while accepting valid lines such
as “HTTP/1.1 200\r\n”.
- Around line 298-370: Confirm the repository Ruff configuration treats PLR0912
and SIM102 as errors, then reduce _validated_openai_chunk’s branch count by
extracting each choice’s validation into a dedicated helper. Within that helper,
collapse the nested checks for function_call, tool_calls, finish_reason, and
logprobs into combined conditions while preserving all existing validation
behavior and error messages.
- Around line 122-136: Consolidate the duplicated writer cleanup in the
lifecycle operation around the `LifecycleError` and generic `Exception` handlers
into a `finally` block, preserving the existing error propagation and wrapping
behavior. Guard cleanup with the existing writer-local check and change
`suppress(ConnectionError)` to `suppress(OSError)` so `wait_closed()` failures
do not replace the intended error. Apply the same suppression widening in
`finish()`.
In `@crates/fabric-core/src/runtime.rs`:
- Around line 829-834: Update validate_adapter_compatibility to validate
streaming support using the resolved adapter descriptor’s capabilities, rather
than only plan.capabilities.streaming. Reject streaming when the requested plan
and adapter descriptor are incompatible, while preserving the existing
UnsupportedRuntimeCapability error context. Add a regression test covering a
stale mismatched RunPlan that attempts streaming with an adapter lacking
streaming support.
In `@docs/reference/api/python-library-reference/nemo_fabric.openai_streaming.md`:
- Around line 20-22: Update the source docstring or documentation generator that
produces the OpenAI streaming API reference so :meth:`result` and :meth:`aclose`
become valid Markdown links, and convert any emitted :class: roles similarly.
Regenerate the API references afterward; do not modify docs/reference/api/
directly.
In
`@docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaichatcompletionchunkobject.mdx`:
- Line 2: Update the reference-title generator to preserve the official “OpenAI”
capitalization, then run “just docs” to regenerate the affected pages. The
generated titles must be “Enum OpenAI Chat Completion Chunk Object” in
docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaichatcompletionchunkobject.mdx:2-2,
“Enum OpenAI Stream Host” in
docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamhost.mdx:2-2,
“Enum OpenAI Stream Profile” in
docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamprofile.mdx:2-2,
“Enum OpenAI Stream Protocol Version” in
docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamprotocolversion.mdx:2-2,
and “Enum OpenAI Stream Record” in
docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdx:2-2;
do not edit these generated pages directly.
In
`@docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaichatcompletionchunk.mdx`:
- Line 2: Update the generator’s identifier display mapping so the OpenAI prefix
retains official capitalization and renders “OpenAI Chat Completion” rather than
“Open AiChat”; then regenerate the affected pages:
docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaichatcompletionchunk.mdx:2,
struct-openaichatcompletionchunkchoice.mdx:2, and
struct-openaichatcompletionchunkdelta.mdx:2. Do not edit the generated reference
pages directly; each must produce the specified OpenAI Chat Completion Chunk,
Choice, and Delta titles.
In `@docs/sdk/python.mdx`:
- Around line 463-468: Update the stream lifecycle guidance to require awaiting
stream.result() after fully consuming the iterator and before starting another
turn, ensuring terminal RunResult completion. For abandoned streams, require
awaiting stream.aclose() instead; clarify that either operation must complete
before reusing the runtime.
In `@external/nat/README.md`:
- Line 18: Update the prose references in the README to consistently use “NeMo
Fabric” instead of standalone “Fabric,” including the references to supported
intent, portable agent intent, and terminal result. Preserve the configuration
value `fabric.agent.react` unchanged.
In `@schemas/adapter-contract/legacy/openai-stream-invocation.schema.json`:
- Around line 170-178: Introduce a new adapter contract version for
InvokeOpenaiStream so OpenAiStreamSink.protocol_version can negotiate the
lifecycle operation. Update OpenAiStreamProtocolVersion, all related
descriptors, and snapshots to use the new version, while rejecting
fabric.adapter/v1alpha2 before lifecycle dispatch rather than routing by
streaming alone. Add a regression test covering rejection of the previous
contract version.
In `@tests/adapters/test_external_nat_adapter.py`:
- Around line 307-338: Pin the NAT dependency used by
test_installed_nat_chat_response_chunk_serializes_to_openai_mapping to a
compatible 1.8 release, or add an explicit version check before invoking
ChatResponseChunk.from_string so incompatible installations are skipped instead
of failing with a signature error.
In `@tests/python/test_openai_streaming.py`:
- Around line 810-826: Rename the test’s openai_streaming parameter and its
parametrization tuple entry to avoid shadowing the imported openai_streaming
module, and update the _runtime_wrapper argument and assertion in
test_native_and_relay_streaming_capabilities_are_independent to use the new
parameter name.
- Around line 766-781: Initialize connections before entering the try block,
then update the cleanup loop to access connections directly instead of using
locals().get(). Preserve the existing socket-closing and listener-closing
behavior while allowing missing connection data to raise clearly.
- Around line 482-487: Bound the polling loop in the streaming cancellation test
around stream._end_observed with a finite deadline or timeout, and fail with a
clear assertion message if the flag is not observed in time. Preserve the
existing iterator cancellation and asyncio.CancelledError assertions after
successful observation.
🪄 Autofix
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: ASSERTIVE
Plan: Enterprise
Run ID: a43c8fb4-4207-44c2-b4a2-ad7fd15e4b0a
📒 Files selected for processing (78)
README.mdadapters/common/README.mdadapters/common/src/nemo_fabric_adapters/common/lifecycle.pycrates/fabric-core/src/error.rscrates/fabric-core/src/lib.rscrates/fabric-core/src/runtime.rscrates/fabric-core/src/schema.rscrates/fabric-python/src/lib.rsdocs/adapter-contract/conformance.mddocs/adapter-contract/execution.mddocs/index.ymldocs/reference/api/python-library-reference/index.mddocs/reference/api/python-library-reference/nemo_fabric.openai_streaming.mddocs/reference/api/python-library-reference/nemo_fabric.runtime.mddocs/reference/api/python-library-reference/nemo_fabric.streaming.mddocs/reference/api/rust-library-reference/nemo-fabric-core/adapter-contract/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/agent-config/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/agent-execution/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/doctor/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/fn-version.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/constant-openai-chat-completions-chunk-profile.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/constant-openai-stream-host.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/constant-openai-stream-protocol-version.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-errorstage.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaichatcompletionchunkobject.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamhost.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamprofile.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamprotocolversion.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-runstatus.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/fn-invoke-openai-stream.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/fn-invoke-runtime.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/fn-prepare-environment.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/fn-run-plan.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/fn-start-runtime.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/fn-stop-runtime.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaichatcompletionchunk.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaichatcompletionchunkchoice.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaichatcompletionchunkdelta.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreaminvocation.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamtransport.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-runrequest.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-runresult.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-runtimecontext.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-runtimehandle.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-runtimetelemetrycontext.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-telemetryref.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/schema/enum-schemaname.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/schema/index.mdxdocs/sdk/python.mdxexternal/nat/README.mdexternal/nat/fabric-adapter.jsonexternal/nat/src/nemo_fabric_adapters/nat/adapter.pypython/src/nemo_fabric/__init__.pypython/src/nemo_fabric/_native.pyipython/src/nemo_fabric/openai_streaming.pypython/src/nemo_fabric/runtime.pyschemas/SCHEMA.mdschemas/adapter-contract/legacy/openai-stream-invocation.schema.jsonschemas/adapter-contract/legacy/openai-stream-record.schema.jsonscripts/docs/enhance_python_api_reference.pyscripts/generate_api_docs.shskills/nemo-fabric-build-adapter/SKILL.mdskills/nemo-fabric-integrate/SKILL.mdskills/nemo-fabric-integrate/references/sdk-api-inventory.mdtests/adapters/test_adapters_common_lifecycle.pytests/adapters/test_external_nat_adapter.pytests/docs/test_python_api_docs.pytests/fixtures/hermes-shim-agent/adapters/hermes-shim/fabric-adapter.jsontests/fixtures/hermes-shim-agent/adapters/hermes-shim/src/nemo_fabric_test_adapters/hermes_shim/adapter.pytests/python/test_native_sdk.pytests/python/test_openai_streaming.py
| status_line = await reader.readline() | ||
| try: | ||
| _version, raw_status, _reason = status_line.decode("ascii").split(" ", 2) | ||
| status = int(raw_status) | ||
| except (UnicodeDecodeError, ValueError) as error: | ||
| raise LifecycleError( | ||
| "lifecycle_stream_transport_failed", | ||
| "OpenAI stream listener returned an invalid HTTP response", | ||
| ) from error |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Parse the HTTP status line without requiring a reason phrase.
split(" ", 2) unpacks into exactly three values. A response line such as HTTP/1.1 200\r\n produces two parts and raises ValueError. The code then reports lifecycle_stream_transport_failed even though the listener returned a valid status. RFC 9112 allows an empty reason phrase. Split with maxsplit=2 and read the status by index instead.
🐛 Proposed fix
status_line = await reader.readline()
try:
- _version, raw_status, _reason = status_line.decode("ascii").split(" ", 2)
- status = int(raw_status)
+ parts = status_line.decode("ascii").split(" ", 2)
+ if len(parts) < 2:
+ raise ValueError("missing HTTP status code")
+ status = int(parts[1])
except (UnicodeDecodeError, ValueError) as error:📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| status_line = await reader.readline() | |
| try: | |
| _version, raw_status, _reason = status_line.decode("ascii").split(" ", 2) | |
| status = int(raw_status) | |
| except (UnicodeDecodeError, ValueError) as error: | |
| raise LifecycleError( | |
| "lifecycle_stream_transport_failed", | |
| "OpenAI stream listener returned an invalid HTTP response", | |
| ) from error | |
| status_line = await reader.readline() | |
| try: | |
| parts = status_line.decode("ascii").split(" ", 2) | |
| if len(parts) < 2: | |
| raise ValueError("missing HTTP status code") | |
| status = int(parts[1]) | |
| except (UnicodeDecodeError, ValueError) as error: | |
| raise LifecycleError( | |
| "lifecycle_stream_transport_failed", | |
| "OpenAI stream listener returned an invalid HTTP response", | |
| ) from error |
🤖 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 `@adapters/common/src/nemo_fabric_adapters/common/lifecycle.py` around lines
233 - 241, Update the HTTP status parsing near status_line so it does not
require a reason phrase: split with a maximum of two splits, then read the
status token by index and convert it to int. Preserve LifecycleError handling
for malformed or undecodable responses while accepting valid lines such as
“HTTP/1.1 200\r\n”.
| iterator = asyncio.create_task(anext(stream)) | ||
| while not stream._end_observed: | ||
| await asyncio.sleep(0) | ||
| iterator.cancel() | ||
| with pytest.raises(asyncio.CancelledError): | ||
| await iterator |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick win
Bound the wait on _end_observed.
Lines 483-484 spin on stream._end_observed with await asyncio.sleep(0). asyncio.sleep(0) yields without blocking, so this loop saturates the event loop. If a regression prevents _end_observed from becoming true, the test hangs instead of failing. Add a deadline so the test fails with a clear message.
♻️ Proposed fix to add a deadline to the wait loop
iterator = asyncio.create_task(anext(stream))
- while not stream._end_observed:
- await asyncio.sleep(0)
+ deadline = time.monotonic() + 5
+ while not stream._end_observed:
+ assert time.monotonic() < deadline, "stream never observed its end record"
+ await asyncio.sleep(0.001)
iterator.cancel()🧰 Tools
🪛 Ruff (0.16.1)
[warning] 483-484: Use asyncio.Event instead of awaiting asyncio.sleep in a while loop
(ASYNC110)
🤖 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/python/test_openai_streaming.py` around lines 482 - 487, Bound the
polling loop in the streaming cancellation test around stream._end_observed with
a finite deadline or timeout, and fail with a clear assertion message if the
flag is not observed in time. Preserve the existing iterator cancellation and
asyncio.CancelledError assertions after successful observation.
Source: Linters/SAST tools
40721d7 to
e551aa0
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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 `@external/nat/src/nemo_fabric_adapters/nat/adapter.py`:
- Around line 835-959: Extract the inner stream-draining and response
aggregation logic from invoke_openai_stream into a focused private helper,
including chunk serialization, emit ordering, content collection, and stream
cleanup semantics. Have invoke_openai_stream call the helper while preserving
cancellation and lifecycle error propagation; leave all existing LOGGER.error
calls unchanged.
In `@tests/adapters/test_external_nat_adapter.py`:
- Around line 461-465: Add an explicit boolean parameter to the relevant test
cases indicating whether the workflow is expected to be ReAct, and use that
parameter for both the assertion near result["workflow"]["_type"] and the
fixture setup near the second occurrence. Remove the duplicated
expected_type.rsplit("/", 1)[-1] membership checks so the test does not
reimplement _is_react_agent.
🪄 Autofix
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: ASSERTIVE
Plan: Enterprise
Run ID: 229fe51f-4228-467f-bd5c-e82e6d02c28c
📒 Files selected for processing (4)
external/nat/README.mdexternal/nat/fabric-adapter.jsonexternal/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
📜 Review details
⏰ Context from checks skipped due to timeout. (16)
- GitHub Check: Preview docs
- GitHub Check: Test (Python 3.11, linux-amd64)
- GitHub Check: Test (Python 3.12, linux-arm64)
- GitHub Check: Test (Python 3.11, macos-arm64)
- GitHub Check: Test (Python 3.12, macos-arm64)
- GitHub Check: Test (Python 3.11, linux-arm64)
- GitHub Check: Test (Python 3.13, linux-amd64)
- GitHub Check: Test (Python 3.14, macos-arm64)
- GitHub Check: Test (Python 3.13, macos-arm64)
- GitHub Check: Test (Python 3.14, linux-amd64)
- GitHub Check: Test (Python 3.11, windows-amd64)
- GitHub Check: Test (Python 3.14, windows-amd64)
- GitHub Check: Test (Python 3.14, linux-arm64)
- GitHub Check: Test (Python 3.13, windows-amd64)
- GitHub Check: Test (Python 3.12, linux-amd64)
- GitHub Check: Pre-commit
🧰 Additional context used
📓 Path-based instructions (21)
**/*.{rs,py,pyi,json,yaml,yml}
📄 CodeRabbit inference engine (.agents/skills/contribute-api/SKILL.md)
Determine and update every affected public surface, including the CLI, PyO3 bindings, Python SDK, type stubs, schemas, and adapter contract, so they remain in parity.
Files:
external/nat/fabric-adapter.jsonexternal/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
**/*
📄 CodeRabbit inference engine (.agents/skills/karpathy-guidelines/SKILL.md)
**/*: Before implementing, explicitly state assumptions, surface ambiguity and tradeoffs, present multiple interpretations when relevant, and ask for clarification rather than silently deciding or proceeding when requirements are unclear.
Prefer the minimum code needed to solve the requested problem: avoid speculative features, unnecessary abstractions, unrequested flexibility, and handling of impossible scenarios; simplify overcomplicated solutions.
When editing existing code, make surgical changes only: do not modify unrelated code, comments, formatting, or pre-existing dead code; match the existing style, and remove only unused imports, variables, or functions introduced by your changes.
Define verifiable success criteria for each task, such as writing regression tests for bugs and invalid-input tests for validation, then verify the implementation against those criteria. For multi-step work, state a brief plan with a verification check for each step.
**/*: Always spellNVIDIAin all caps; do not useNvidia,nvidia,nVidia,nVIDIA, orNV.
Usean NVIDIAbefore a noun, because the name begins with an “en” sound.
Do not add a registered trademark symbol afterNVIDIAwhen referring to the company; use trademark symbols with product names only when required by the document type or legal guidance.
Verify official capitalization, spacing, hyphenation, and spelling for NVIDIA and third-party product names; do not rewrite official product names for grammar or title-case rules.
Precede NVIDIA product names withNVIDIAon first mention when natural and accurate, and link the first mention when the destination helps the reader.
On first use, include the company name and full model qualifier when it helps identify the model; preserve official capitalization and punctuation, and use shorter family names only after establishing the full name.
For learning-oriented and developer content, do not force trademark symbols unless explicitly required; for press, ...
Files:
external/nat/fabric-adapter.jsonexternal/nat/README.mdexternal/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
**/*.{json,jsonschema}
📄 CodeRabbit inference engine (CONTRIBUTING.md)
Public contract changes must keep checked-in JSON Schema snapshots synchronized.
Files:
external/nat/fabric-adapter.json
**/*.{md,rst}
📄 CodeRabbit inference engine (.agents/skills/contribute-api/SKILL.md)
Update documentation and examples in the same branch as the public API change.
Files:
external/nat/README.md
**/*.{md,mdx,rst}
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/assets/nvidia-style-guide.md)
**/*.{md,mdx,rst}: For NeMo Fabric documentation, verify technical claims against the current repository, public API, or documented command before reviewing style.
Always spellNVIDIAin all caps; do not useNvidia,nvidia, orNV.
Format commands, code elements, expressions, package names, file names, and paths as inline code.
Use descriptive link text; avoid raw URLs and weak anchors such ashereorread more.
Use title case consistently for technical documentation headings.
Introduce code blocks, lists, tables, and images with complete sentences.
Write procedures as imperative, parallel steps; split long procedures into smaller tasks.
Prefer active voice, present tense, short sentences, contractions, and plain English while preserving necessary technical precision.
Usecanfor possibility and reservemayfor permission.
Useafterfor temporal relationships instead ofonce, and preferrefer tooverseewhen directing readers to another resource.
Avoid culture-specific idioms, unnecessary Latinisms, jokes, and marketing exaggeration in technical documentation.
Spell out months in body text, avoid ordinal dates, and use clear time zones.
Spell out whole numbers from zero through nine unless they are technical values, parameters, versions, or UI values; use numerals for 10 or greater and commas in thousands.
Do not add trademark symbols to learning-oriented documentation unless the source, platform, or legal guidance explicitly requires them.
Do not replace precise technical terms with simpler words when doing so would lose precision.
Do not flag passive voice when the actor is unknown or the action is the important part.
Do not rewrite API names, package names, command flags, or code literals for style.
**/*.{md,mdx,rst}: Use consistent title case for technical-document headings and table headers; avoid quotation marks, ampersands, and exclamation marks in headings, while preserving official product, event, research, and whitepaper title ...
Files:
external/nat/README.md
**/*.{md,rst,txt,adoc}
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/assets/nvidia-style-language-mechanics.md)
**/*.{md,rst,txt,adoc}: For technical documentation, use professional, active, conversational, engaging, precise, and plain-English prose. Prefer active voice, present tense, short sentences, and scannable paragraphs. Avoid casual or imprecise language, swearing, threats, insults, jokes, puns, culture-specific idioms, marketing exaggeration, and unsupported third-party comparisons.
Usecanfor possibility and reservemayfor permission; useafterfor temporal order; userefer tofor cross-references; prefer short direct sentences and specific verbs; avoid unnecessarypleasein technical documentation.
Prefer active voice when the actor matters. Passive voice is acceptable when the actor is unknown or irrelevant, when the action or result is the focus, or in programmer documentation.
Use natural contractions in conversational technical prose, but do not force them in formal legal copy, API references, or generated text.
Prefer simpler English over Latinisms: usefor exampleorsuch asinstead ofe.g.,and so oninstead ofetc.,that isinstead ofi.e.,compared toinstead ofvs., andby,through, orusinginstead ofvia. Use industry-standard terms such as in silico, in vitro, and in vivo when appropriate, and italicize them in running text.
Usethatwithout commas for essential clauses, andwhichwith commas for nonessential clauses.
Format dates and times clearly: spell out months in body text; use forms such asJune 12, 2025; avoid numeric or ordinal dates; capitalize days; use 12-hour time when appropriate; include a space beforea.m.orp.m.; useETandPTfor needed time zones; avoid24/7; and preferfrom 12:30 to 1:00 p.m.for prose ranges.
Format numbers consistently: spell out zero through nine in body text, use numerals for 10 or greater and for technical values, use commas in thousands, do not begin a sentence with a numeral, spell out ordinals, and use numerals consistently within a category wh...
Files:
external/nat/README.md
**/*.{md,mdx,yml,py,rs,sh}
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/SKILL.md)
Keep documentation aligned with current NeMo Fabric behavior, repository layout, entry points, commands, package names, APIs, bindings, and support claims.
Files:
external/nat/README.mdexternal/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
**/*.{md,mdx}
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/SKILL.md)
**/*.{md,mdx}: Use the full product nameNVIDIA NeMo Fabricon first use, typically in the title and H1; useNeMo Fabricthereafter. Usefabricalone only for the CLI tool and surround it with backticks.
Treat incorrect or stale commands, package names, paths, APIs, support claims, procedures, examples, terminology, or public behavior documentation as blocking issues.
CapitalizeNVIDIAcorrectly and format code, commands, paths, and filenames as inline code where needed.
Use title case for technical-documentation headings.
Introduce code blocks, tables, and lists with complete lead-in sentences; ensure examples match current APIs and build commands.
Use descriptive anchor text, avoid raw URLs and generic labels such ashere, and use repository-relative.mdxpaths for links withindocs/.
Prefer active voice, present tense, short sentences, plain English, consistent terminology, and imperative, parallel, scannable procedures.
Useafterinstead ofoncewhen expressing temporal sequence, and usecanrather thanmaywhen describing possibility rather than permission.
Avoid ambiguous numeric dates and ordinal dates in body text.
For learning-oriented documentation, do not force trademark symbols unless the source document explicitly requires them.
When reporting documentation-review findings, lead withMust fix,Should fix, andNice to havecategories; include file path, line reference, current problem, rationale, and a concrete rewrite or direction.
Files:
external/nat/README.md
**/*.md
📄 CodeRabbit inference engine (CONTRIBUTING.md)
Update relevant SDK, API reference, adapter, example, integration, and support documentation when the corresponding public surface changes.
Files:
external/nat/README.md
**/*.{html,md}
📄 CodeRabbit inference engine (CONTRIBUTING.md)
HTML and Markdown files must use the specified SPDX HTML-comment header.
Files:
external/nat/README.md
{*.md,**/*.md,**/*.mdx,**/*.ipynb}
⚙️ CodeRabbit configuration file
{*.md,**/*.md,**/*.mdx,**/*.ipynb}: Enforce the product name in user-facing prose: use "NVIDIA NeMo Fabric" on first use and "NeMo Fabric" thereafter. Flag standalone capitalized "Fabric" when it refers to the product. Do not flag the lowercasefabricCLI command, package/import/crate names, code identifiers, API symbols, configuration keys, file paths, or unrelated generic uses of the word.
Files:
external/nat/README.md
**/*.{rs,py}
📄 CodeRabbit inference engine (.agents/skills/prepare-pr/SKILL.md)
For native binding changes, run
cargo check -p fabric-python --locked.Use
snake_casefor functions and variables; usePascalCasefor Rust types and Python classes.
Files:
external/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
**/*.{py,pyi}
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
If Python code or a Python-facing adapter changes, run
just test-python.In Python SDK, adapters, examples, and tests, follow the existing style, use type annotations for public APIs, and keep native binding declarations synchronized with their Rust implementations.
Files:
external/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
**/*.{rs,py,pyi}
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
**/*.{rs,py,pyi}: If public configuration types change, confirm schema snapshot tests injust test-rustpass and review generated schema diffs.
For schema or public contract changes, run both language suites and review changes underschemas/and generated API references.
Files:
external/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
**/*.{py,pyi,rs}
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
For Python SDK or PyO3 binding changes, use
python-tests, run focused pytest tests first, thenjust test-python; rebuild withjust build-pythonwhen native code or packaging changes.Public contract changes must keep native Python binding declarations synchronized with their Rust implementations.
Files:
external/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
**/*.{rs,py,toml}
📄 CodeRabbit inference engine (.agents/skills/update-project-version/SKILL.md)
When editing version helpers, verify every
nemo-fabric-*workspace package through Cargo metadata and reject a static version inpython/pyproject.toml.
Files:
external/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
**/*.{toml,rs,py}
📄 CodeRabbit inference engine (.agents/skills/update-project-version/SKILL.md)
Avoid blind repository-wide replacement of version-like strings; distinguish package-version references from examples and unrelated dependency versions.
Files:
external/nat/src/nemo_fabric_adapters/nat/adapter.pytests/adapters/test_external_nat_adapter.py
tests/adapters/**/*.py
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
tests/adapters/**/*.py: If an adapter or integration changes, run its focused tests.
For adapter behavior changes, run focused adapter tests undertests/adapters, then runjust test-python.
Files:
tests/adapters/test_external_nat_adapter.py
tests/**/*.{rs,py}
📄 CodeRabbit inference engine (CONTRIBUTING.md)
When adding functionality, include tests in the corresponding Rust crate or the relevant area under
tests/.
Files:
tests/adapters/test_external_nat_adapter.py
tests/**/*.py
📄 CodeRabbit inference engine (.agents/skills/python-tests/SKILL.md)
tests/**/*.py: Use pytest to run Python tests.
Do not add@pytest.mark.asyncioto tests; async tests are automatically detected by the async runner.
Do not add-> Nonereturn type annotations to test functions.
When mocking a class, useunittest.mock.MagicMockorAsyncMock, using thespecargument when necessary, rather than defining a new class.
Prefix mocked class names withmock, notfake.
Prefer pytest fixtures over helper methods.
If a fixture is needed in multiple test files, define it once inconftest.pyrather than repeating it.
Define fixtures using@pytest.fixture(name="<fixture_name>"[, scope="<scope>"])and a<fixture_name>_fixturefunction; specifyscopeonly when it is notfunction.
Preferpytest.mark.parametrizeover separate tests for different input types.
Use@pytest.mark.usefixtureswhen a fixture is needed but its returned value is unused or it returns no value.
Avoid defensive programming in tests; access expected values directly so missing data raises a clear failure, such as usingresults["data"]instead ofresults.get("data").
When adapter installation metadata changes, packaging metadata tests must directly assert that the root project depends unconditionally on the exact-versionnemo-fabric-runtimedistribution.
Packaging metadata tests must verify that each root harness extra delegates to the matching version of the leaf adapter'sharnessextra.
Packaging metadata tests must verify that bare leaf dependencies remain adapter-owned and that the rootadapter-testsdependency group installs each leaf through itsharnessextra.
Packaging metadata tests must verify that every leaf providesfull; only adapters importing NeMo Relay Python APIs providerelay, while adapters using an external Relay executable havefullequal toharness.
Files:
tests/adapters/test_external_nat_adapter.py
{tests/**,python/tests/**}
⚙️ CodeRabbit configuration file
{tests/**,python/tests/**}: Tests should cover the behavior promised by the changed API surface, including error paths, lifecycle cleanup, and SDK/native parity where relevant.
Files:
tests/adapters/test_external_nat_adapter.py
🧠 Learnings (1)
📚 Learning: 2026-06-29T22:34:52.407Z
Learnt from: AjayThorve
Repo: NVIDIA/NeMo-Fabric PR: 27
File: adapters/codex-cli/fabric-adapter.json:13-15
Timestamp: 2026-06-29T22:34:52.407Z
Learning: In NeMo-Fabric adapter manifest files (e.g., `*/fabric-adapter.json`), keep `config.accepts` limited to the top-level Fabric capability sections that `resolve_capability_plan` consumes (such as `models`, `tools`, `mcp`, `skills`, `telemetry`). Do not add adapter-owned `harness.settings` keys to `config.accepts`; `harness.settings` should remain adapter-owned and be passed through unchanged.
Applied to files:
external/nat/fabric-adapter.json
🪛 ast-grep (0.45.1)
tests/adapters/test_external_nat_adapter.py
[info] 1336-1336: use jsonify instead of json.dumps for JSON output
Context: json.dumps(result)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1374-1374: use jsonify instead of json.dumps for JSON output
Context: json.dumps(result)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
🪛 Ruff (0.16.1)
external/nat/src/nemo_fabric_adapters/nat/adapter.py
[warning] 656-659: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 835-835: Too many return statements (7 > 6)
(PLR0911)
[warning] 835-835: Too many branches (21 > 12)
(PLR0912)
[warning] 835-835: Too many statements (65 > 50)
(PLR0915)
[warning] 854-854: Do not catch blind exception: Exception
(BLE001)
[warning] 855-858: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 883-887: Use a single with statement with multiple contexts instead of nested with statements
(SIM117)
[warning] 909-909: Abstract raise to an inner function
(TRY301)
[warning] 926-926: Do not catch blind exception: Exception
(BLE001)
[warning] 927-931: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 940-943: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 948-948: Do not catch blind exception: Exception
(BLE001)
[warning] 949-952: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
tests/adapters/test_external_nat_adapter.py
[warning] 45-45: Missing return type annotation for special method __aiter__
(ANN204)
[warning] 48-48: Missing return type annotation for special method __anext__
(ANN204)
[warning] 1350-1350: Dynamically typed expressions (typing.Any) are disallowed in serialization_failure
(ANN401)
🔇 Additional comments (9)
external/nat/README.md (2)
56-68: UseNeMo Fabricfor the product name in prose.Line 56 reads "separate Fabric runtimes" and line 68 reads "The terminal Fabric result". Both use a standalone capitalized
Fabricfor the product. The rest of the file usesNeMo Fabric.📝 Proposed wording
-builder, different users remain isolated, and separate Fabric runtimes own -separate session managers. +builder, different users remain isolated, and separate NeMo Fabric runtimes own +separate session managers. @@ -serialized OpenAI Chat Completions chunks in order. The terminal Fabric result +serialized OpenAI Chat Completions chunks in order. The terminal NeMo Fabric result contains the concatenated string `delta.content` values for choice index `0`;As per path instructions: "Enforce the product name in user-facing prose: use "NVIDIA NeMo Fabric" on first use and "NeMo Fabric" thereafter. Flag standalone capitalized "Fabric" when it refers to the product."
Source: Path instructions
59-73: LGTM!Also applies to: 89-92
tests/adapters/test_external_nat_adapter.py (4)
310-343: Pin or guard the NAT version used by this installed-NAT test.
pytest.importorskipskips when NAT is absent. It does not skip when an incompatible NAT version is installed.ChatResponseChunk.from_string(..., id_=..., model=..., finish_reason=...)then fails with a signature error instead of skipping. Pin the compatible NAT release or add an explicit version guard.
32-62: LGTM!Also applies to: 157-157, 169-169, 190-190, 203-207, 229-229, 289-292
468-507: LGTM!Also applies to: 575-620, 668-789
1085-1304: LGTM!Also applies to: 1306-1423, 1426-1574
external/nat/fabric-adapter.json (1)
24-31: LGTM!Also applies to: 91-91
external/nat/src/nemo_fabric_adapters/nat/adapter.py (2)
9-11: LGTM!Also applies to: 34-34, 52-55, 111-113, 214-217
638-670: LGTM!Also applies to: 721-745, 787-795
| async def invoke_openai_stream( | ||
| self, | ||
| payload: dict[str, Any], | ||
| emit: lifecycle.OpenAIChunkEmitter, | ||
| ) -> dict[str, Any]: | ||
| """Forward one NAT result stream declared as OpenAI chunks.""" | ||
|
|
||
| request, failure = self._invocation_request(payload) | ||
| if failure is not None: | ||
| return failure | ||
| assert request is not None | ||
| assert self._sessions is not None | ||
|
|
||
| try: | ||
| from nat.data_models.api_server import ChatResponseChunk | ||
|
|
||
| streaming_output_schema = ( | ||
| self._sessions.get_workflow_streaming_output_schema() | ||
| ) | ||
| except Exception as error: | ||
| LOGGER.error( | ||
| "NAT workflow streaming schema lookup failed (error_type=%s)", | ||
| type(error).__name__, | ||
| ) | ||
| return _failure_output( | ||
| "nat_workflow_stream_failed", | ||
| "NAT workflow streaming failed; inspect adapter stderr for details", | ||
| ) | ||
|
|
||
| if streaming_output_schema is not ChatResponseChunk: | ||
| return _failure_output( | ||
| "nat_openai_stream_unsupported_schema", | ||
| "NAT native OpenAI streaming requires a ChatResponseChunk output schema", | ||
| ) | ||
|
|
||
| try: | ||
| session_kwargs = _session_kwargs( | ||
| request, | ||
| require_user_id=self._sessions.is_workflow_per_user, | ||
| ) | ||
| except ValueError as error: | ||
| return _failure_output("nat_invalid_request", str(error)) | ||
|
|
||
| response_parts: list[str] = [] | ||
| try: | ||
| from nat.data_models.runtime_enum import RuntimeTypeEnum | ||
| from pydantic_core import to_jsonable_python | ||
|
|
||
| async with self._sessions.session(**session_kwargs) as session: | ||
| async with session.run( | ||
| request.get("input", ""), | ||
| runtime_type=RuntimeTypeEnum.RUN_OR_SERVE, | ||
| ) as runner: | ||
| stream = runner.result_stream(to_type=ChatResponseChunk) | ||
|
|
||
| async def close_stream() -> None: | ||
| close = getattr(stream, "aclose", None) | ||
| if callable(close): | ||
| await close() | ||
|
|
||
| try: | ||
| async for chunk in stream: | ||
| try: | ||
| serialized = to_jsonable_python( | ||
| chunk, | ||
| serialize_unknown=False, | ||
| ) | ||
| except asyncio.CancelledError: | ||
| raise | ||
| except lifecycle.LifecycleError: | ||
| raise | ||
| except Exception as error: | ||
| raise _NatStreamSerializationError from error | ||
| if not isinstance(serialized, dict): | ||
| raise _NatStreamSerializationError | ||
|
|
||
| await emit(serialized) | ||
| for choice in serialized.get("choices", []): | ||
| if not isinstance(choice, dict): | ||
| continue | ||
| if choice.get("index") != 0: | ||
| continue | ||
| delta = choice.get("delta") | ||
| if not isinstance(delta, dict): | ||
| continue | ||
| content = delta.get("content") | ||
| if isinstance(content, str): | ||
| response_parts.append(content) | ||
| except BaseException: | ||
| try: | ||
| await close_stream() | ||
| except Exception as error: | ||
| LOGGER.error( | ||
| "NAT workflow stream cleanup failed while preserving " | ||
| "the primary failure (error_type=%s)", | ||
| type(error).__name__, | ||
| ) | ||
| raise | ||
| else: | ||
| await close_stream() | ||
| except asyncio.CancelledError: | ||
| raise | ||
| except lifecycle.LifecycleError: | ||
| raise | ||
| except _NatStreamSerializationError as error: | ||
| LOGGER.error( | ||
| "NAT workflow returned a non-JSON stream chunk (error_type=%s)", | ||
| type(error.__cause__ or error).__name__, | ||
| ) | ||
| return _failure_output( | ||
| "nat_stream_chunk_not_json_serializable", | ||
| "NAT workflow returned a stream chunk that cannot be represented as JSON", | ||
| ) | ||
| except Exception as error: | ||
| LOGGER.error( | ||
| "NAT workflow streaming failed (error_type=%s)", | ||
| type(error).__name__, | ||
| ) | ||
| return _failure_output( | ||
| "nat_workflow_stream_failed", | ||
| "NAT workflow streaming failed; inspect adapter stderr for details", | ||
| ) | ||
|
|
||
| return _success_output("".join(response_parts)) | ||
|
|
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Determine whether Ruff enforces PLR0911/PLR0912/PLR0915 and how per-file ignores apply.
set -uo pipefail
echo "== Ruff configuration files =="
fd -H -t f 'ruff.toml|.ruff.toml|pyproject.toml' | while IFS= read -r file; do
echo "--- $file ---"
rg -n -A 40 '\[tool\.ruff|^\[lint\]|^\[tool\.ruff\.lint\]' "$file"
done
echo
echo "== Explicit references to the reported rules =="
rg -n 'PLR0911|PLR0912|PLR0915|"PL"|"TRY"|"BLE"|per-file-ignores' --glob '*.toml'
echo
echo "== Lint recipes =="
fd -H -t f 'justfile|Justfile|.justfile' --exec rg -n -A 6 'lint|ruff' {}Repository: NVIDIA/NeMo-Fabric
Length of output: 947
🏁 Script executed:
set -u
echo "== Ruff-related files and recipes =="
git ls-files | rg '(^|/)(pyproject\.toml|ruff\.toml|\.ruff\.toml|Justfile|justfile|\.justfile)$' |
while IFS= read -r file; do
echo "--- $file ---"
rg -n -A 12 -B 4 'ruff|lint|PLR0911|PLR0912|PLR0915|per-file-ignores' "$file" || true
done
echo
echo "== Adapter project configuration =="
fd -H -t f 'pyproject.toml|ruff.toml|.ruff.toml' external/nat --exec sh -c '
for file do
echo "--- $file ---"
cat -n "$file"
done
' shRepository: NVIDIA/NeMo-Fabric
Length of output: 1003
🏁 Script executed:
set -u
echo "== All Ruff invocations and lint entry points =="
rg -n -S -g '!*.lock' -g '!*.sum' '(^|[^[:alnum:]_])ruff([^[:alnum:]_]|$)|just lint|validate-change' . || true
echo
echo "== Justfile lint sections =="
rg -n -A 20 -B 5 '(^|[[:space:]])lint([[:space:]]|:)|ruff' justfile || true
echo
echo "== Target method complexity =="
python3 - <<'PY'
from pathlib import Path
import ast
path = Path("external/nat/src/nemo_fabric_adapters/nat/adapter.py")
tree = ast.parse(path.read_text())
for node in ast.walk(tree):
if isinstance(node, (ast.AsyncFunctionDef, ast.FunctionDef)) and node.name == "invoke_openai_stream":
returns = sum(isinstance(n, ast.Return) for n in ast.walk(node))
branches = sum(isinstance(n, (ast.If, ast.For, ast.AsyncFor, ast.While, ast.Try, ast.ExceptHandler, ast.With, ast.AsyncWith, ast.BoolOp)) for n in ast.walk(node))
statements = sum(isinstance(n, ast.stmt) for n in ast.walk(node))
print(f"function={node.name} lines={node.lineno}-{node.end_lineno} returns={returns} branches={branches} statements={statements}")
PYRepository: NVIDIA/NeMo-Fabric
Length of output: 385
Consider extracting the stream-drain helper.
invoke_openai_stream handles schema validation, session setup, serialization, emission, aggregation, and cleanup. Extracting the drain and aggregation logic would improve maintainability while preserving ordering and cleanup semantics. Ruff does not enforce PLR0911, PLR0912, or PLR0915 in this repository.
Keep the LOGGER.error calls unchanged. LOGGER.exception could expose secrets through traceback output.
🧰 Tools
🪛 Ruff (0.16.1)
[warning] 835-835: Too many return statements (7 > 6)
(PLR0911)
[warning] 835-835: Too many branches (21 > 12)
(PLR0912)
[warning] 835-835: Too many statements (65 > 50)
(PLR0915)
[warning] 854-854: Do not catch blind exception: Exception
(BLE001)
[warning] 855-858: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 883-887: Use a single with statement with multiple contexts instead of nested with statements
(SIM117)
[warning] 909-909: Abstract raise to an inner function
(TRY301)
[warning] 926-926: Do not catch blind exception: Exception
(BLE001)
[warning] 927-931: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 940-943: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 948-948: Do not catch blind exception: Exception
(BLE001)
[warning] 949-952: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
🤖 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 `@external/nat/src/nemo_fabric_adapters/nat/adapter.py` around lines 835 - 959,
Extract the inner stream-draining and response aggregation logic from
invoke_openai_stream into a focused private helper, including chunk
serialization, emit ordering, content collection, and stream cleanup semantics.
Have invoke_openai_stream call the helper while preserving cancellation and
lifecycle error propagation; leave all existing LOGGER.error calls unchanged.
Source: Linters/SAST tools
e551aa0 to
7ccb8c8
Compare
There was a problem hiding this comment.
Actionable comments posted: 12
♻️ Duplicate comments (1)
adapters/common/src/nemo_fabric_adapters/common/lifecycle.py (1)
236-245: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winParse the HTTP status line without requiring a reason phrase.
split(" ", 2)unpacks into exactly three names. A response line such asHTTP/1.1 200\r\nyields two parts and raisesValueError, so a valid response is reported aslifecycle_stream_transport_failed. RFC 9112 allows an empty reason phrase. Read the status by index.🐛 Proposed fix
status_line = await reader.readline() try: - _version, raw_status, _reason = status_line.decode("ascii").split(" ", 2) - status = int(raw_status) + parts = status_line.decode("ascii").split(" ", 2) + if len(parts) < 2: + raise ValueError("missing HTTP status code") + status = int(parts[1]) except (UnicodeDecodeError, ValueError) as error:🤖 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 `@adapters/common/src/nemo_fabric_adapters/common/lifecycle.py` around lines 236 - 245, Update _read_http_response to parse the HTTP version and status from indexed fields rather than requiring a third reason-phrase field from split(" ", 2). Preserve integer status conversion and LifecycleError handling for malformed or non-ASCII status lines, while accepting valid responses with no reason phrase.
🤖 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 `@adapters/common/src/nemo_fabric_adapters/common/lifecycle.py`:
- Around line 302-374: The chunk validation logic is duplicated between
_validated_openai_chunk and _validate_openai_chunk, allowing the two OpenAI
streaming contracts to drift. Consolidate the shared field and bounds validation
into one importable predicate used by both validators, preserving each
validator’s existing exception type; if the distribution boundary prevents
sharing, add parity tests covering the same invalid-chunk corpus and rejection
behavior.
In `@crates/fabric-core/src/runtime.rs`:
- Around line 3072-3141: Update openai_stream_listener to configure a read
timeout on the accepted TcpStream before constructing the BufReader, using the
test’s expected timeout and asserting that set_read_timeout succeeds. Keep the
existing accept behavior and ensure subsequent read operations fail promptly
when the client does not connect fully.
In
`@docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdx`:
- Around line 24-68: Update the enum struct-variant field heading generation in
generate_rust_library_reference.py so fields are emitted one level below their
variant’s Fields heading, rather than at the variant heading level. Preserve the
existing variant headings and apply the correction to all enum struct variants,
then run just docs to regenerate the affected reference page; do not edit
generated docs directly.
In `@external/nat/src/nemo_fabric_adapters/nat/adapter.py`:
- Around line 872-878: Extract the duplicated session-kwargs derivation,
including the is_workflow_per_user getattr probe, into a private helper on the
adapter. Update both invoke and invoke_openai_stream to call that helper,
preserving the existing request input and require_user_id behavior.
In `@python/src/nemo_fabric/openai_streaming.py`:
- Around line 142-158: The pending-connection limit in _accept allows
unauthenticated idle clients to consume every slot until
_OPENAI_STREAM_HEADER_TIMEOUT expires. Update the connection admission or
header-timeout handling around _accept and _handle_client so unauthenticated
connections cannot exhaust _MAX_PENDING_CONNECTIONS, while preserving the
existing rejection behavior for genuinely pending authenticated work.
- Around line 647-661: Add a narrow public error-setting method such as
set_error(message) to _OpenAIStreamListener that records the error without
setting the completion event, then update _validate_and_accept_result to use it
instead of accessing _set_error directly. Preserve the existing invocation-ID
validation and protocol-failure flow.
- Around line 271-285: Update the _emit_line and stream-record size error
messages to interpolate the configured self._max_record_bytes value instead of
the fixed “1 MiB” text, and update the corresponding test assertion in
test_openai_streaming.py to expect the configured limit.
In `@python/src/nemo_fabric/runtime.py`:
- Around line 116-121: Update Runtime.supports_openai_streaming in
python/src/nemo_fabric/runtime.py:116-121 to require both
plan.capabilities.streaming and
plan.adapter_descriptor.descriptor.capabilities.streaming, so unsupported stale
plans raise FabricCapabilityError before dispatch. In
tests/python/test_openai_streaming.py:846-853, add coverage for a
streaming-enabled plan with a non-streaming descriptor and assert the native
module is not called.
In `@tests/adapters/test_adapters_common_lifecycle.py`:
- Around line 132-136: Update the three listener.records.get() calls in the
lifecycle test to use asyncio.wait_for with the existing one-second timeout
pattern, ensuring each read fails promptly rather than hanging when fewer than
three records are emitted.
In `@tests/adapters/test_external_nat_adapter.py`:
- Around line 1342-1358: Update
test_openai_stream_normalizes_chunk_serialization_failure so the non-exception
serialization_failure parameter overrides the mock_nat["to_jsonable"] side
effect before assigning return_value, ensuring the test actually receives
["not", "a", "mapping"] and exercises the serializer-returned non-mapping path.
In `@tests/python/test_openai_streaming.py`:
- Around line 846-853: Add a test alongside
test_invoke_openai_stream_rejects_an_unsupported_adapter where _plan()
advertises streaming but adapter_descriptor.descriptor omits the streaming
capability. Assert runtime.invoke_openai_stream raises FabricCapabilityError
with the appropriate unavailable code and mock_native.invoke_openai_stream is
not called, verifying rejection occurs before the native module.
- Around line 484-489: Update the wait_for_end polling loop to replace
asyncio.sleep(0) with a short positive sleep interval, allowing the worker
thread to make progress while preserving the existing asyncio.wait_for timeout
and _end_observed termination condition.
---
Duplicate comments:
In `@adapters/common/src/nemo_fabric_adapters/common/lifecycle.py`:
- Around line 236-245: Update _read_http_response to parse the HTTP version and
status from indexed fields rather than requiring a third reason-phrase field
from split(" ", 2). Preserve integer status conversion and LifecycleError
handling for malformed or non-ASCII status lines, while accepting valid
responses with no reason phrase.
🪄 Autofix
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: ASSERTIVE
Plan: Enterprise
Run ID: 9de3831d-9e7c-49b0-9104-83888862c04f
📒 Files selected for processing (27)
adapters/common/src/nemo_fabric_adapters/common/lifecycle.pycrates/fabric-core/src/runtime.rscrates/fabric-core/src/schema.rsdocs/reference/api/python-library-reference/nemo_fabric.openai_streaming.mddocs/reference/api/python-library-reference/nemo_fabric.runtime.mddocs/reference/api/python-library-reference/nemo_fabric.streaming.mddocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaichatcompletionchunkobject.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamhost.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamprofile.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamprotocolversion.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaichatcompletionchunk.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaichatcompletionchunkchoice.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaichatcompletionchunkdelta.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreaminvocation.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamtransport.mdxexternal/nat/README.mdexternal/nat/src/nemo_fabric_adapters/nat/adapter.pypython/src/nemo_fabric/openai_streaming.pypython/src/nemo_fabric/runtime.pypython/src/nemo_fabric/streaming.pyschemas/adapter-contract/legacy/openai-stream-record.schema.jsonscripts/docs/generate_rust_library_reference.pytests/adapters/test_adapters_common_lifecycle.pytests/adapters/test_external_nat_adapter.pytests/python/test_openai_streaming.py
| def _validated_openai_chunk(value: dict[str, Any]) -> dict[str, Any]: | ||
| def invalid(message: str) -> LifecycleError: | ||
| return LifecycleError("lifecycle_invalid_openai_stream_event", message) | ||
|
|
||
| if value.get("object") != "chat.completion.chunk": | ||
| raise invalid("OpenAI stream events must use object 'chat.completion.chunk'") | ||
| identifier = value.get("id") | ||
| model = value.get("model") | ||
| created = value.get("created") | ||
| choices = value.get("choices") | ||
| if not isinstance(identifier, str) or not identifier.strip(): | ||
| raise invalid( | ||
| "OpenAI stream event id must be a non-empty string containing " | ||
| "a non-whitespace character" | ||
| ) | ||
| if not isinstance(model, str) or not model.strip(): | ||
| raise invalid( | ||
| "OpenAI stream event model must be a non-empty string containing " | ||
| "a non-whitespace character" | ||
| ) | ||
| if ( | ||
| isinstance(created, bool) | ||
| or not isinstance(created, int) | ||
| or not 0 <= created <= _UINT64_MAX | ||
| ): | ||
| raise invalid("OpenAI stream event created must be an unsigned 64-bit integer") | ||
| if not isinstance(choices, list): | ||
| raise invalid("OpenAI stream event choices must be a list") | ||
| for choice in choices: | ||
| if not isinstance(choice, dict): | ||
| raise invalid("OpenAI stream choices must be mappings") | ||
| index = choice.get("index") | ||
| delta = choice.get("delta") | ||
| if ( | ||
| isinstance(index, bool) | ||
| or not isinstance(index, int) | ||
| or not 0 <= index <= _UINT32_MAX | ||
| ): | ||
| raise invalid("OpenAI stream choice index must be an unsigned 32-bit integer") | ||
| if not isinstance(delta, dict): | ||
| raise invalid("OpenAI stream choice delta must be a mapping") | ||
| for name in ("content", "refusal", "role"): | ||
| if name in delta and delta[name] is not None and not isinstance( | ||
| delta[name], str | ||
| ): | ||
| raise invalid( | ||
| f"OpenAI stream choice delta {name} must be a string or null" | ||
| ) | ||
| if "function_call" in delta and delta["function_call"] is not None: | ||
| if not isinstance(delta["function_call"], dict): | ||
| raise invalid( | ||
| "OpenAI stream choice delta function_call must be a mapping or null" | ||
| ) | ||
| if "tool_calls" in delta and delta["tool_calls"] is not None: | ||
| tool_calls = delta["tool_calls"] | ||
| if not isinstance(tool_calls, list) or not all( | ||
| isinstance(tool_call, dict) for tool_call in tool_calls | ||
| ): | ||
| raise invalid( | ||
| "OpenAI stream choice delta tool_calls must be a list of mappings or null" | ||
| ) | ||
| if "finish_reason" in choice and choice["finish_reason"] is not None: | ||
| if not isinstance(choice["finish_reason"], str): | ||
| raise invalid( | ||
| "OpenAI stream choice finish_reason must be a string or null" | ||
| ) | ||
| if "logprobs" in choice and choice["logprobs"] is not None: | ||
| if not isinstance(choice["logprobs"], dict): | ||
| raise invalid("OpenAI stream choice logprobs must be a mapping or null") | ||
| if "usage" in value and value["usage"] is not None: | ||
| if not isinstance(value["usage"], dict): | ||
| raise invalid("OpenAI stream event usage must be a mapping or null") | ||
| return value |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Two copies of the chunk-profile validation now exist.
_validated_openai_chunk duplicates _validate_openai_chunk in python/src/nemo_fabric/openai_streaming.py lines 368-445, field for field, including the _UINT32_MAX and _UINT64_MAX bounds and the message wording. The two differ only in the exception type. The openai.chat_completions.chunk/v1 profile is one contract; two hand-maintained validators will drift, and a drift makes the adapter emit records that the SDK listener rejects mid-stream.
Generate both validators from the committed schema snapshot, or move the shared predicate into one module that both distributions import. If the packaging boundary blocks sharing, add a test that asserts both implementations reject the same corpus of invalid chunks.
🧰 Tools
🪛 Ruff (0.16.1)
[warning] 302-302: Too many branches (21 > 12)
(PLR0912)
[warning] 307-307: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 313-316: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 318-321: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 327-327: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 329-329: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 332-332: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 340-340: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 342-342: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 347-349: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 350-351: Use a single if statement instead of nested if statements
(SIM102)
[warning] 352-354: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 360-362: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 363-364: Use a single if statement instead of nested if statements
(SIM102)
[warning] 365-367: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 368-369: Use a single if statement instead of nested if statements
(SIM102)
[warning] 370-370: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 371-372: Use a single if statement instead of nested if statements
(SIM102)
[warning] 373-373: Avoid specifying long messages outside the exception class
(TRY003)
🤖 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 `@adapters/common/src/nemo_fabric_adapters/common/lifecycle.py` around lines
302 - 374, The chunk validation logic is duplicated between
_validated_openai_chunk and _validate_openai_chunk, allowing the two OpenAI
streaming contracts to drift. Consolidate the shared field and bounds validation
into one importable predicate used by both validators, preserving each
validator’s existing exception type; if the distribution boundary prevents
sharing, add parity tests covering the same invalid-chunk corpus and rejection
behavior.
| fn openai_stream_listener( | ||
| token: &str, | ||
| ) -> (OpenAiStreamTransport, thread::JoinHandle<Vec<Value>>) { | ||
| let listener = TcpListener::bind((OPENAI_STREAM_HOST, 0)).expect("bind stream listener"); | ||
| let port = listener.local_addr().expect("listener address").port(); | ||
| let expected_token = token.to_string(); | ||
| let capture = thread::spawn(move || { | ||
| let (mut stream, _) = listener.accept().expect("accept stream connection"); | ||
| let mut reader = BufReader::new(stream.try_clone().expect("clone stream")); | ||
| let mut headers = Vec::new(); | ||
| loop { | ||
| let mut line = String::new(); | ||
| reader.read_line(&mut line).expect("read HTTP header"); | ||
| if line == "\r\n" { | ||
| break; | ||
| } | ||
| headers.push(line); | ||
| } | ||
| assert_eq!( | ||
| headers.first().map(String::as_str), | ||
| Some("POST /openai-stream HTTP/1.1\r\n") | ||
| ); | ||
| assert!(headers.iter().any(|header| { | ||
| header == &format!("Authorization: Bearer {expected_token}\r\n") | ||
| })); | ||
| stream | ||
| .write_all(b"HTTP/1.1 100 Continue\r\n\r\n") | ||
| .expect("accept stream request"); | ||
| stream.flush().expect("flush continue response"); | ||
|
|
||
| let mut records = Vec::new(); | ||
| loop { | ||
| let mut size_line = String::new(); | ||
| reader.read_line(&mut size_line).expect("read chunk size"); | ||
| let size = usize::from_str_radix(size_line.trim(), 16).expect("hex chunk size"); | ||
| if size == 0 { | ||
| let mut terminator = String::new(); | ||
| reader | ||
| .read_line(&mut terminator) | ||
| .expect("read chunk terminator"); | ||
| assert_eq!(terminator, "\r\n"); | ||
| break; | ||
| } | ||
| let mut encoded = vec![0; size]; | ||
| reader.read_exact(&mut encoded).expect("read chunk body"); | ||
| let mut terminator = [0; 2]; | ||
| reader | ||
| .read_exact(&mut terminator) | ||
| .expect("read chunk terminator"); | ||
| assert_eq!(&terminator, b"\r\n"); | ||
| records.push( | ||
| serde_json::from_slice(encoded.strip_suffix(b"\n").unwrap_or(&encoded)) | ||
| .expect("parse stream record"), | ||
| ); | ||
| } | ||
| stream | ||
| .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") | ||
| .expect("complete stream request"); | ||
| stream.flush().expect("flush final response"); | ||
| records | ||
| }); | ||
| ( | ||
| OpenAiStreamTransport { | ||
| port, | ||
| token: token.to_string(), | ||
| }, | ||
| capture, | ||
| ) | ||
| } | ||
|
|
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick win
Add socket timeouts to the test listener helper.
openai_stream_listener spawns a thread that blocks on listener.accept() and then on reader.read_line. No timeout is set. If a future change makes the adapter fail before it connects, or truncates the chunked body, capture.join() at the call site blocks forever and the test hangs instead of failing.
Set a read timeout on the accepted stream and assert on failure.
♻️ Proposed change
let capture = thread::spawn(move || {
let (mut stream, _) = listener.accept().expect("accept stream connection");
+ stream
+ .set_read_timeout(Some(Duration::from_secs(10)))
+ .expect("set stream read timeout");
let mut reader = BufReader::new(stream.try_clone().expect("clone stream"));TcpListener::accept itself cannot take a timeout directly; if you also want to bound the accept, set the listener non-blocking or keep the current behavior and rely on the harness timeout.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| fn openai_stream_listener( | |
| token: &str, | |
| ) -> (OpenAiStreamTransport, thread::JoinHandle<Vec<Value>>) { | |
| let listener = TcpListener::bind((OPENAI_STREAM_HOST, 0)).expect("bind stream listener"); | |
| let port = listener.local_addr().expect("listener address").port(); | |
| let expected_token = token.to_string(); | |
| let capture = thread::spawn(move || { | |
| let (mut stream, _) = listener.accept().expect("accept stream connection"); | |
| let mut reader = BufReader::new(stream.try_clone().expect("clone stream")); | |
| let mut headers = Vec::new(); | |
| loop { | |
| let mut line = String::new(); | |
| reader.read_line(&mut line).expect("read HTTP header"); | |
| if line == "\r\n" { | |
| break; | |
| } | |
| headers.push(line); | |
| } | |
| assert_eq!( | |
| headers.first().map(String::as_str), | |
| Some("POST /openai-stream HTTP/1.1\r\n") | |
| ); | |
| assert!(headers.iter().any(|header| { | |
| header == &format!("Authorization: Bearer {expected_token}\r\n") | |
| })); | |
| stream | |
| .write_all(b"HTTP/1.1 100 Continue\r\n\r\n") | |
| .expect("accept stream request"); | |
| stream.flush().expect("flush continue response"); | |
| let mut records = Vec::new(); | |
| loop { | |
| let mut size_line = String::new(); | |
| reader.read_line(&mut size_line).expect("read chunk size"); | |
| let size = usize::from_str_radix(size_line.trim(), 16).expect("hex chunk size"); | |
| if size == 0 { | |
| let mut terminator = String::new(); | |
| reader | |
| .read_line(&mut terminator) | |
| .expect("read chunk terminator"); | |
| assert_eq!(terminator, "\r\n"); | |
| break; | |
| } | |
| let mut encoded = vec![0; size]; | |
| reader.read_exact(&mut encoded).expect("read chunk body"); | |
| let mut terminator = [0; 2]; | |
| reader | |
| .read_exact(&mut terminator) | |
| .expect("read chunk terminator"); | |
| assert_eq!(&terminator, b"\r\n"); | |
| records.push( | |
| serde_json::from_slice(encoded.strip_suffix(b"\n").unwrap_or(&encoded)) | |
| .expect("parse stream record"), | |
| ); | |
| } | |
| stream | |
| .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") | |
| .expect("complete stream request"); | |
| stream.flush().expect("flush final response"); | |
| records | |
| }); | |
| ( | |
| OpenAiStreamTransport { | |
| port, | |
| token: token.to_string(), | |
| }, | |
| capture, | |
| ) | |
| } | |
| fn openai_stream_listener( | |
| token: &str, | |
| ) -> (OpenAiStreamTransport, thread::JoinHandle<Vec<Value>>) { | |
| let listener = TcpListener::bind((OPENAI_STREAM_HOST, 0)).expect("bind stream listener"); | |
| let port = listener.local_addr().expect("listener address").port(); | |
| let expected_token = token.to_string(); | |
| let capture = thread::spawn(move || { | |
| let (mut stream, _) = listener.accept().expect("accept stream connection"); | |
| stream | |
| .set_read_timeout(Some(Duration::from_secs(10))) | |
| .expect("set stream read timeout"); | |
| let mut reader = BufReader::new(stream.try_clone().expect("clone stream")); | |
| let mut headers = Vec::new(); | |
| loop { | |
| let mut line = String::new(); | |
| reader.read_line(&mut line).expect("read HTTP header"); | |
| if line == "\r\n" { | |
| break; | |
| } | |
| headers.push(line); | |
| } | |
| assert_eq!( | |
| headers.first().map(String::as_str), | |
| Some("POST /openai-stream HTTP/1.1\r\n") | |
| ); | |
| assert!(headers.iter().any(|header| { | |
| header == &format!("Authorization: Bearer {expected_token}\r\n") | |
| })); | |
| stream | |
| .write_all(b"HTTP/1.1 100 Continue\r\n\r\n") | |
| .expect("accept stream request"); | |
| stream.flush().expect("flush continue response"); | |
| let mut records = Vec::new(); | |
| loop { | |
| let mut size_line = String::new(); | |
| reader.read_line(&mut size_line).expect("read chunk size"); | |
| let size = usize::from_str_radix(size_line.trim(), 16).expect("hex chunk size"); | |
| if size == 0 { | |
| let mut terminator = String::new(); | |
| reader | |
| .read_line(&mut terminator) | |
| .expect("read chunk terminator"); | |
| assert_eq!(terminator, "\r\n"); | |
| break; | |
| } | |
| let mut encoded = vec![0; size]; | |
| reader.read_exact(&mut encoded).expect("read chunk body"); | |
| let mut terminator = [0; 2]; | |
| reader | |
| .read_exact(&mut terminator) | |
| .expect("read chunk terminator"); | |
| assert_eq!(&terminator, b"\r\n"); | |
| records.push( | |
| serde_json::from_slice(encoded.strip_suffix(b"\n").unwrap_or(&encoded)) | |
| .expect("parse stream record"), | |
| ); | |
| } | |
| stream | |
| .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n") | |
| .expect("complete stream request"); | |
| stream.flush().expect("flush final response"); | |
| records | |
| }); | |
| ( | |
| OpenAiStreamTransport { | |
| port, | |
| token: token.to_string(), | |
| }, | |
| capture, | |
| ) | |
| } |
🤖 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 `@crates/fabric-core/src/runtime.rs` around lines 3072 - 3141, Update
openai_stream_listener to configure a read timeout on the accepted TcpStream
before constructing the BufReader, using the test’s expected timeout and
asserting that set_read_timeout succeeds. Keep the existing accept behavior and
ensure subsequent read operations fail promptly when the client does not connect
fully.
| def _accept( | ||
| self, | ||
| reader: asyncio.StreamReader, | ||
| writer: asyncio.StreamWriter, | ||
| ) -> None: | ||
| if len(self._tasks) >= _MAX_PENDING_CONNECTIONS: | ||
| writer.close() | ||
| return | ||
| task = asyncio.create_task(self._handle_client(reader, writer)) | ||
| self._tasks.add(task) | ||
| task.add_done_callback(self._task_done) | ||
|
|
||
| def _task_done(self, task: asyncio.Task[None]) -> None: | ||
| self._tasks.discard(task) | ||
| if not task.cancelled(): | ||
| task.exception() | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🔵 Trivial | 💤 Low value
Unauthenticated clients can occupy every accept slot.
_accept rejects new connections after _MAX_PENDING_CONNECTIONS tasks exist. A client that opens a socket and sends nothing holds a slot for _OPENAI_STREAM_HEADER_TIMEOUT (10 s). Eight such sockets block the adapter connection until the timeouts expire, and the invocation then fails with a protocol error.
The listener binds to 127.0.0.1, so the exposure is local only. Consider lowering the header timeout or excluding unauthenticated connections from the slot count.
🤖 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 `@python/src/nemo_fabric/openai_streaming.py` around lines 142 - 158, The
pending-connection limit in _accept allows unauthenticated idle clients to
consume every slot until _OPENAI_STREAM_HEADER_TIMEOUT expires. Update the
connection admission or header-timeout handling around _accept and
_handle_client so unauthenticated connections cannot exhaust
_MAX_PENDING_CONNECTIONS, while preserving the existing rejection behavior for
genuinely pending authenticated work.
| @pytest.mark.parametrize( | ||
| "serialization_failure", | ||
| [TypeError("secret-object-repr"), ["not", "a", "mapping"]], | ||
| ) | ||
| async def test_openai_stream_normalizes_chunk_serialization_failure( | ||
| make_payload, | ||
| make_invocation_payload, | ||
| mock_nat, | ||
| caplog, | ||
| serialization_failure: Any, | ||
| ): | ||
| stream = _AsyncChunkStream([object()]) | ||
| mock_nat["runner"].result_stream.return_value = stream | ||
| if isinstance(serialization_failure, BaseException): | ||
| mock_nat["to_jsonable"].side_effect = serialization_failure | ||
| else: | ||
| mock_nat["to_jsonable"].return_value = serialization_failure |
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
The non-exception parameter does not exercise the intended path.
The mock_nat fixture creates to_jsonable with side_effect=lambda value, **_kwargs: value. side_effect takes precedence over return_value unless it returns mock.DEFAULT. For the ["not", "a", "mapping"] parameter, Line 1358 sets return_value, but the mock still returns the original object() instance. The test passes only because object() is also not a dict, so the intended "serializer returned a non-mapping" case is never asserted.
Clear the side_effect so the parameter takes effect.
🐛 Proposed fix
if isinstance(serialization_failure, BaseException):
mock_nat["to_jsonable"].side_effect = serialization_failure
else:
+ mock_nat["to_jsonable"].side_effect = None
mock_nat["to_jsonable"].return_value = serialization_failure📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| @pytest.mark.parametrize( | |
| "serialization_failure", | |
| [TypeError("secret-object-repr"), ["not", "a", "mapping"]], | |
| ) | |
| async def test_openai_stream_normalizes_chunk_serialization_failure( | |
| make_payload, | |
| make_invocation_payload, | |
| mock_nat, | |
| caplog, | |
| serialization_failure: Any, | |
| ): | |
| stream = _AsyncChunkStream([object()]) | |
| mock_nat["runner"].result_stream.return_value = stream | |
| if isinstance(serialization_failure, BaseException): | |
| mock_nat["to_jsonable"].side_effect = serialization_failure | |
| else: | |
| mock_nat["to_jsonable"].return_value = serialization_failure | |
| `@pytest.mark.parametrize`( | |
| "serialization_failure", | |
| [TypeError("secret-object-repr"), ["not", "a", "mapping"]], | |
| ) | |
| async def test_openai_stream_normalizes_chunk_serialization_failure( | |
| make_payload, | |
| make_invocation_payload, | |
| mock_nat, | |
| caplog, | |
| serialization_failure: Any, | |
| ): | |
| stream = _AsyncChunkStream([object()]) | |
| mock_nat["runner"].result_stream.return_value = stream | |
| if isinstance(serialization_failure, BaseException): | |
| mock_nat["to_jsonable"].side_effect = serialization_failure | |
| else: | |
| mock_nat["to_jsonable"].side_effect = None | |
| mock_nat["to_jsonable"].return_value = serialization_failure |
🧰 Tools
🪛 Ruff (0.16.1)
[warning] 1351-1351: Dynamically typed expressions (typing.Any) are disallowed in serialization_failure
(ANN401)
🤖 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/adapters/test_external_nat_adapter.py` around lines 1342 - 1358, Update
test_openai_stream_normalizes_chunk_serialization_failure so the non-exception
serialization_failure parameter overrides the mock_nat["to_jsonable"] side
effect before assigning return_value, ensuring the test actually receives
["not", "a", "mapping"] and exercises the serializer-returned non-mapping path.
| async def wait_for_end() -> None: | ||
| while not stream._end_observed: | ||
| await asyncio.sleep(0) | ||
|
|
||
| try: | ||
| await asyncio.wait_for(wait_for_end(), timeout=1) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Replace the zero-delay spin with a small sleep.
asyncio.sleep(0) yields without delay, so the loop saturates the event loop for up to the full one-second timeout and starves the worker thread that feeds the listener. The asyncio.wait_for bound already prevents a hang. Use a short sleep interval.
♻️ Proposed fix
async def wait_for_end() -> None:
while not stream._end_observed:
- await asyncio.sleep(0)
+ await asyncio.sleep(0.001)Based on static analysis hints from Ruff 0.16.1 (ASYNC110).
🧰 Tools
🪛 Ruff (0.16.1)
[warning] 485-486: Use asyncio.Event instead of awaiting asyncio.sleep in a while loop
(ASYNC110)
🤖 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/python/test_openai_streaming.py` around lines 484 - 489, Update the
wait_for_end polling loop to replace asyncio.sleep(0) with a short positive
sleep interval, allowing the worker thread to make progress while preserving the
existing asyncio.wait_for timeout and _end_observed termination condition.
Source: Linters/SAST tools
Signed-off-by: Ajay Thorve <athorve@nvidia.com>
Signed-off-by: Ajay Thorve <athorve@nvidia.com>
Signed-off-by: Ajay Thorve <athorve@nvidia.com>
7ccb8c8 to
99174ab
Compare
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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 `@tests/adapters/test_adapters_common_lifecycle.py`:
- Around line 30-51: Replace the hand-written _BackpressuredStreamWriter,
CancellingWriter, and FailingWriter doubles with unittest.mock
AsyncMock/MagicMock instances, using spec=asyncio.StreamWriter where
appropriate. Preserve backpressure coordination through the backpressured mock’s
drain side_effect and retain the cancellation and failure behavior through
side_effect callbacks, naming any mock variables with the mock prefix.
In `@tests/adapters/test_external_nat_adapter.py`:
- Around line 791-796: Update the test around SessionManager.create to require
NAT version 1.5.0 or skip unsupported releases before accessing
is_workflow_per_user and get_workflow_streaming_output_schema. Preserve the
existing missing-module skip behavior and ensure older versions do not fail with
AttributeError.
🪄 Autofix
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: ASSERTIVE
Plan: Enterprise
Run ID: d6b14f0a-d82d-4d6b-9747-3589b37d461e
📒 Files selected for processing (14)
crates/fabric-core/src/runtime.rsdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxexternal/nat/src/nemo_fabric_adapters/nat/adapter.pypython/src/nemo_fabric/runtime.pyschemas/adapter-contract/legacy/openai-stream-invocation.schema.jsonscripts/docs/generate_rust_library_reference.pytests/adapters/test_adapters_common_lifecycle.pytests/adapters/test_external_nat_adapter.pytests/python/test_openai_streaming.py
📜 Review details
⏰ Context from checks skipped due to timeout. (17)
- GitHub Check: Preview docs
- GitHub Check: Pre-commit
- GitHub Check: Test (Python 3.14, linux-arm64)
- GitHub Check: Test (Python 3.14, windows-amd64)
- GitHub Check: Test (Python 3.14, linux-amd64)
- GitHub Check: Test (Python 3.13, macos-arm64)
- GitHub Check: Test (Python 3.12, windows-amd64)
- GitHub Check: Test (Python 3.13, linux-amd64)
- GitHub Check: Test (Python 3.13, linux-arm64)
- GitHub Check: Test (Python 3.12, macos-arm64)
- GitHub Check: Test (Python 3.11, linux-amd64)
- GitHub Check: Test (Python 3.11, macos-arm64)
- GitHub Check: Test (Python 3.11, linux-arm64)
- GitHub Check: Test (Python 3.14, macos-arm64)
- GitHub Check: Test (Python 3.12, linux-amd64)
- GitHub Check: Test (Python 3.11, windows-amd64)
- GitHub Check: Test (Python 3.12, linux-arm64)
🧰 Additional context used
📓 Path-based instructions (36)
{README.md,docs/**/*.{md,mdx,yml},examples/**/*.{md,mdx,yml}}
📄 CodeRabbit inference engine (.agents/skills/contribute-docs/SKILL.md)
Keep package names, repository references, and build commands current in documentation and examples.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
{docs/**/*.{md,mdx,yml},examples/**/*.{md,mdx,yml}}
📄 CodeRabbit inference engine (.agents/skills/contribute-docs/SKILL.md)
Update relevant getting-started, reference, adapter, and example documentation when the corresponding examples or adapters change.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
**/*.mdx
📄 CodeRabbit inference engine (.agents/skills/contribute-docs/SKILL.md)
In MDX files, use JSX comment delimiters (
{/*and*/}) for top-of-file comments, including SPDX headers; do not use HTML comments.
**/*.mdx: For documentation site changes, runjust docsto regenerate Python and Rust API references and validate Fern configuration.
MDX files must use the specified JSX-comment SPDX header.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
docs/**/*.{md,mdx,yml}
📄 CodeRabbit inference engine (.agents/skills/contribute-docs/SKILL.md)
Run
just docswhen the documentation site changes.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
**/*
📄 CodeRabbit inference engine (.agents/skills/karpathy-guidelines/SKILL.md)
**/*: Before implementing, explicitly state assumptions, surface ambiguity and tradeoffs, present multiple interpretations when relevant, and ask for clarification rather than silently deciding or proceeding when requirements are unclear.
Prefer the minimum code needed to solve the requested problem: avoid speculative features, unnecessary abstractions, unrequested flexibility, and handling of impossible scenarios; simplify overcomplicated solutions.
When editing existing code, make surgical changes only: do not modify unrelated code, comments, formatting, or pre-existing dead code; match the existing style, and remove only unused imports, variables, or functions introduced by your changes.
Define verifiable success criteria for each task, such as writing regression tests for bugs and invalid-input tests for validation, then verify the implementation against those criteria. For multi-step work, state a brief plan with a verification check for each step.
**/*: Always spellNVIDIAin all caps; do not useNvidia,nvidia,nVidia,nVIDIA, orNV.
Usean NVIDIAbefore a noun, because the name begins with an “en” sound.
Do not add a registered trademark symbol afterNVIDIAwhen referring to the company; use trademark symbols with product names only when required by the document type or legal guidance.
Verify official capitalization, spacing, hyphenation, and spelling for NVIDIA and third-party product names; do not rewrite official product names for grammar or title-case rules.
Precede NVIDIA product names withNVIDIAon first mention when natural and accurate, and link the first mention when the destination helps the reader.
On first use, include the company name and full model qualifier when it helps identify the model; preserve official capitalization and punctuation, and use shorter family names only after establishing the full name.
For learning-oriented and developer content, do not force trademark symbols unless explicitly required; for press, ...
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxscripts/docs/generate_rust_library_reference.pydocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxtests/adapters/test_adapters_common_lifecycle.pydocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxschemas/adapter-contract/legacy/openai-stream-invocation.schema.jsonpython/src/nemo_fabric/runtime.pytests/python/test_openai_streaming.pydocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxtests/adapters/test_external_nat_adapter.pyexternal/nat/src/nemo_fabric_adapters/nat/adapter.pydocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdxcrates/fabric-core/src/runtime.rs
**/*.{md,mdx,rst}
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/assets/nvidia-style-guide.md)
**/*.{md,mdx,rst}: For NeMo Fabric documentation, verify technical claims against the current repository, public API, or documented command before reviewing style.
Always spellNVIDIAin all caps; do not useNvidia,nvidia, orNV.
Format commands, code elements, expressions, package names, file names, and paths as inline code.
Use descriptive link text; avoid raw URLs and weak anchors such ashereorread more.
Use title case consistently for technical documentation headings.
Introduce code blocks, lists, tables, and images with complete sentences.
Write procedures as imperative, parallel steps; split long procedures into smaller tasks.
Prefer active voice, present tense, short sentences, contractions, and plain English while preserving necessary technical precision.
Usecanfor possibility and reservemayfor permission.
Useafterfor temporal relationships instead ofonce, and preferrefer tooverseewhen directing readers to another resource.
Avoid culture-specific idioms, unnecessary Latinisms, jokes, and marketing exaggeration in technical documentation.
Spell out months in body text, avoid ordinal dates, and use clear time zones.
Spell out whole numbers from zero through nine unless they are technical values, parameters, versions, or UI values; use numerals for 10 or greater and commas in thousands.
Do not add trademark symbols to learning-oriented documentation unless the source, platform, or legal guidance explicitly requires them.
Do not replace precise technical terms with simpler words when doing so would lose precision.
Do not flag passive voice when the actor is unknown or the action is the important part.
Do not rewrite API names, package names, command flags, or code literals for style.
**/*.{md,mdx,rst}: Use consistent title case for technical-document headings and table headers; avoid quotation marks, ampersands, and exclamation marks in headings, while preserving official product, event, research, and whitepaper title ...
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
docs/reference/api/**/*
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/SKILL.md)
Treat all files under
docs/reference/api/as generated output and do not modify them directly.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
docs/**/*.mdx
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/SKILL.md)
docs/**/*.mdx: Use source-relative links with the target.mdxextension for links between files underdocs/; do not use Fern site-root paths.
Use{/* ... */}delimiters for top-of-file MDX SPDX comments, not HTML comment delimiters.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
**/*.{md,mdx,yml,py,rs,sh}
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/SKILL.md)
Keep documentation aligned with current NeMo Fabric behavior, repository layout, entry points, commands, package names, APIs, bindings, and support claims.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxscripts/docs/generate_rust_library_reference.pydocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxtests/adapters/test_adapters_common_lifecycle.pydocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxpython/src/nemo_fabric/runtime.pytests/python/test_openai_streaming.pydocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxtests/adapters/test_external_nat_adapter.pyexternal/nat/src/nemo_fabric_adapters/nat/adapter.pydocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdxcrates/fabric-core/src/runtime.rs
**/*.{md,mdx}
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/SKILL.md)
**/*.{md,mdx}: Use the full product nameNVIDIA NeMo Fabricon first use, typically in the title and H1; useNeMo Fabricthereafter. Usefabricalone only for the CLI tool and surround it with backticks.
Treat incorrect or stale commands, package names, paths, APIs, support claims, procedures, examples, terminology, or public behavior documentation as blocking issues.
CapitalizeNVIDIAcorrectly and format code, commands, paths, and filenames as inline code where needed.
Use title case for technical-documentation headings.
Introduce code blocks, tables, and lists with complete lead-in sentences; ensure examples match current APIs and build commands.
Use descriptive anchor text, avoid raw URLs and generic labels such ashere, and use repository-relative.mdxpaths for links withindocs/.
Prefer active voice, present tense, short sentences, plain English, consistent terminology, and imperative, parallel, scannable procedures.
Useafterinstead ofoncewhen expressing temporal sequence, and usecanrather thanmaywhen describing possibility rather than permission.
Avoid ambiguous numeric dates and ordinal dates in body text.
For learning-oriented documentation, do not force trademark symbols unless the source document explicitly requires them.
When reporting documentation-review findings, lead withMust fix,Should fix, andNice to havecategories; include file path, line reference, current problem, rationale, and a concrete rewrite or direction.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
docs/**
📄 CodeRabbit inference engine (AGENTS.md)
Run just docs after changing the documentation site.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
{docs/**,README.md,AGENTS.md}
⚙️ CodeRabbit configuration file
{docs/**,README.md,AGENTS.md}: Review documentation for technical accuracy against the current API, command correctness, and consistency with generated schemas.
For links between files under docs/, require paths relative to the source file with the target file's .mdx extension so they work in both Fern builds and repository browsers. Flag Fern site-root links such as NeMo Fabric overview; use the repository-relative equivalent, such as NeMo Fabric overview.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
{*.md,**/*.md,**/*.mdx,**/*.ipynb}
⚙️ CodeRabbit configuration file
{*.md,**/*.md,**/*.mdx,**/*.ipynb}: Enforce the product name in user-facing prose: use "NVIDIA NeMo Fabric" on first use and "NeMo Fabric" thereafter. Flag standalone capitalized "Fabric" when it refers to the product. Do not flag the lowercasefabricCLI command, package/import/crate names, code identifiers, API symbols, configuration keys, file paths, or unrelated generic uses of the word.
Files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
**/*.{rs,py,pyi,json,yaml,yml}
📄 CodeRabbit inference engine (.agents/skills/contribute-api/SKILL.md)
Determine and update every affected public surface, including the CLI, PyO3 bindings, Python SDK, type stubs, schemas, and adapter contract, so they remain in parity.
Files:
scripts/docs/generate_rust_library_reference.pytests/adapters/test_adapters_common_lifecycle.pyschemas/adapter-contract/legacy/openai-stream-invocation.schema.jsonpython/src/nemo_fabric/runtime.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.pyexternal/nat/src/nemo_fabric_adapters/nat/adapter.pycrates/fabric-core/src/runtime.rs
**/*.{rs,py}
📄 CodeRabbit inference engine (.agents/skills/prepare-pr/SKILL.md)
For native binding changes, run
cargo check -p fabric-python --locked.Use
snake_casefor functions and variables; usePascalCasefor Rust types and Python classes.
Files:
scripts/docs/generate_rust_library_reference.pytests/adapters/test_adapters_common_lifecycle.pypython/src/nemo_fabric/runtime.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.pyexternal/nat/src/nemo_fabric_adapters/nat/adapter.pycrates/fabric-core/src/runtime.rs
**/*.{py,pyi}
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
If Python code or a Python-facing adapter changes, run
just test-python.In Python SDK, adapters, examples, and tests, follow the existing style, use type annotations for public APIs, and keep native binding declarations synchronized with their Rust implementations.
Files:
scripts/docs/generate_rust_library_reference.pytests/adapters/test_adapters_common_lifecycle.pypython/src/nemo_fabric/runtime.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.pyexternal/nat/src/nemo_fabric_adapters/nat/adapter.py
**/*.{rs,py,pyi}
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
**/*.{rs,py,pyi}: If public configuration types change, confirm schema snapshot tests injust test-rustpass and review generated schema diffs.
For schema or public contract changes, run both language suites and review changes underschemas/and generated API references.
Files:
scripts/docs/generate_rust_library_reference.pytests/adapters/test_adapters_common_lifecycle.pypython/src/nemo_fabric/runtime.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.pyexternal/nat/src/nemo_fabric_adapters/nat/adapter.pycrates/fabric-core/src/runtime.rs
**/*.{py,pyi,rs}
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
For Python SDK or PyO3 binding changes, use
python-tests, run focused pytest tests first, thenjust test-python; rebuild withjust build-pythonwhen native code or packaging changes.Public contract changes must keep native Python binding declarations synchronized with their Rust implementations.
Files:
scripts/docs/generate_rust_library_reference.pytests/adapters/test_adapters_common_lifecycle.pypython/src/nemo_fabric/runtime.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.pyexternal/nat/src/nemo_fabric_adapters/nat/adapter.pycrates/fabric-core/src/runtime.rs
**/*.{rs,py,toml}
📄 CodeRabbit inference engine (.agents/skills/update-project-version/SKILL.md)
When editing version helpers, verify every
nemo-fabric-*workspace package through Cargo metadata and reject a static version inpython/pyproject.toml.
Files:
scripts/docs/generate_rust_library_reference.pytests/adapters/test_adapters_common_lifecycle.pypython/src/nemo_fabric/runtime.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.pyexternal/nat/src/nemo_fabric_adapters/nat/adapter.pycrates/fabric-core/src/runtime.rs
**/*.{toml,rs,py}
📄 CodeRabbit inference engine (.agents/skills/update-project-version/SKILL.md)
Avoid blind repository-wide replacement of version-like strings; distinguish package-version references from examples and unrelated dependency versions.
Files:
scripts/docs/generate_rust_library_reference.pytests/adapters/test_adapters_common_lifecycle.pypython/src/nemo_fabric/runtime.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.pyexternal/nat/src/nemo_fabric_adapters/nat/adapter.pycrates/fabric-core/src/runtime.rs
scripts/docs/generate_rust_library_reference.py
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/SKILL.md)
For Rust API reference changes, update
scripts/docs/generate_rust_library_reference.pywhen the generator itself must change.
Files:
scripts/docs/generate_rust_library_reference.py
tests/adapters/**/*.py
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
tests/adapters/**/*.py: If an adapter or integration changes, run its focused tests.
For adapter behavior changes, run focused adapter tests undertests/adapters, then runjust test-python.
Files:
tests/adapters/test_adapters_common_lifecycle.pytests/adapters/test_external_nat_adapter.py
tests/**/*.{rs,py}
📄 CodeRabbit inference engine (CONTRIBUTING.md)
When adding functionality, include tests in the corresponding Rust crate or the relevant area under
tests/.
Files:
tests/adapters/test_adapters_common_lifecycle.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.py
tests/**/*.py
📄 CodeRabbit inference engine (.agents/skills/python-tests/SKILL.md)
tests/**/*.py: Use pytest to run Python tests.
Do not add@pytest.mark.asyncioto tests; async tests are automatically detected by the async runner.
Do not add-> Nonereturn type annotations to test functions.
When mocking a class, useunittest.mock.MagicMockorAsyncMock, using thespecargument when necessary, rather than defining a new class.
Prefix mocked class names withmock, notfake.
Prefer pytest fixtures over helper methods.
If a fixture is needed in multiple test files, define it once inconftest.pyrather than repeating it.
Define fixtures using@pytest.fixture(name="<fixture_name>"[, scope="<scope>"])and a<fixture_name>_fixturefunction; specifyscopeonly when it is notfunction.
Preferpytest.mark.parametrizeover separate tests for different input types.
Use@pytest.mark.usefixtureswhen a fixture is needed but its returned value is unused or it returns no value.
Avoid defensive programming in tests; access expected values directly so missing data raises a clear failure, such as usingresults["data"]instead ofresults.get("data").
When adapter installation metadata changes, packaging metadata tests must directly assert that the root project depends unconditionally on the exact-versionnemo-fabric-runtimedistribution.
Packaging metadata tests must verify that each root harness extra delegates to the matching version of the leaf adapter'sharnessextra.
Packaging metadata tests must verify that bare leaf dependencies remain adapter-owned and that the rootadapter-testsdependency group installs each leaf through itsharnessextra.
Packaging metadata tests must verify that every leaf providesfull; only adapters importing NeMo Relay Python APIs providerelay, while adapters using an external Relay executable havefullequal toharness.
Files:
tests/adapters/test_adapters_common_lifecycle.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.py
{tests/**,python/tests/**}
⚙️ CodeRabbit configuration file
{tests/**,python/tests/**}: Tests should cover the behavior promised by the changed API surface, including error paths, lifecycle cleanup, and SDK/native parity where relevant.
Files:
tests/adapters/test_adapters_common_lifecycle.pytests/python/test_openai_streaming.pytests/adapters/test_external_nat_adapter.py
**/*.{json,jsonschema}
📄 CodeRabbit inference engine (CONTRIBUTING.md)
Public contract changes must keep checked-in JSON Schema snapshots synchronized.
Files:
schemas/adapter-contract/legacy/openai-stream-invocation.schema.json
schemas/**/*
⚙️ CodeRabbit configuration file
schemas/**/*: Schemas are generated public contract snapshots. Check that schema diffs correspond to intentional Rust type changes and are covered by core tests.
Files:
schemas/adapter-contract/legacy/openai-stream-invocation.schema.json
python/src/nemo_fabric/**/*.py
📄 CodeRabbit inference engine (.agents/skills/review-doc-style/SKILL.md)
For Python API reference changes, update source docstrings under
python/src/nemo_fabric/instead of generated API reference files.
Files:
python/src/nemo_fabric/runtime.py
python/src/nemo_fabric/**/*.{py,rs}
📄 CodeRabbit inference engine (.agents/skills/maintain-packaging/SKILL.md)
Ensure native extension naming and placement remain compatible with downstream consumers, including the editable maturin build producing
nemo_fabric._native.
Files:
python/src/nemo_fabric/runtime.py
python/src/nemo_fabric/**/*
⚙️ CodeRabbit configuration file
python/src/nemo_fabric/**/*: Review Python SDK changes for typed API consistency, import-time dependency neutrality, async/session behavior, and parity with the native extension.
Stubs and runtime implementations should stay aligned.
Files:
python/src/nemo_fabric/runtime.py
**/*.rs
📄 CodeRabbit inference engine (.agents/skills/contribute-api/SKILL.md)
Implement new runtime or binding behavior in the shared Rust core first.
Files:
crates/fabric-core/src/runtime.rs
**/*.{rs,toml}
📄 CodeRabbit inference engine (.agents/skills/prepare-pr/SKILL.md)
For any Rust change, run
just test-rustandcargo fmt --all -- --check.For Rust core, CLI, or shared runtime semantic changes, run Rust formatting and tests, and add Python tests when behavior is exposed through the SDK.
Use Rust stable tooling; format Rust code with
cargo fmt --all, verify formatting withcargo fmt --all -- --check, and compile withcargo check --workspace --locked.
Files:
crates/fabric-core/src/runtime.rs
crates/fabric-core/**/*.{rs,py}
📄 CodeRabbit inference engine (.agents/skills/prepare-pr/SKILL.md)
Changes under
crates/fabric-coremust run both the Rust and Python test suites.
Files:
crates/fabric-core/src/runtime.rs
**/*.{rs,rmeta}
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
If Rust code changes, run
cargo fmt --all -- --checkandjust test-rust.
Files:
crates/fabric-core/src/runtime.rs
crates/fabric-core/**/*.rs
📄 CodeRabbit inference engine (.agents/skills/validate-change/SKILL.md)
If
crates/fabric-corechanges in a way exposed through Python, run both the Rust and Python suites.For Rust API reference changes, update Rust documentation comments under
crates/fabric-core/instead of generated API reference files.
Files:
crates/fabric-core/src/runtime.rs
crates/fabric-core/src/**/*.rs
⚙️ CodeRabbit configuration file
crates/fabric-core/src/**/*.rs: Review the Rust core for runtime lifecycle correctness, handle validation, capability routing accuracy, schema stability, and error semantics.
Public API changes should match committed schemas, tests, and documentation.
Files:
crates/fabric-core/src/runtime.rs
🧠 Learnings (3)
📚 Learning: 2026-07-24T16:07:22.255Z
Learnt from: AjayThorve
Repo: NVIDIA/NeMo-Fabric PR: 118
File: docs/reference/api/rust-library-reference/nemo-fabric-core/config/struct-adapterdescriptor.mdx:5-5
Timestamp: 2026-07-24T16:07:22.255Z
Learning: In this repo, files generated under `docs/reference/api/**` are NVIDIA NeMo Fabric API reference output. When reviewing changes to these generated pages, do not treat sidebar `position`/ordering updates as direct manual edits—these can be regenerated by running `just docs` after adding public types. Only flag substantive content changes that are not explained by generation.
Applied to files:
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdxdocs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx
📚 Learning: 2026-08-07T07:15:33.918Z
Learnt from: AnuradhaKaruppiah
Repo: NVIDIA/NeMo-Fabric PR: 186
File: schemas/adapter-contract/legacy/adapter-invocation.schema.json:176-176
Timestamp: 2026-08-07T07:15:33.918Z
Learning: For the NeMo Fabric v1alpha southbound adapter contract, treat the adapter descriptor's `contract_version` as the version of the complete contract, including `RuntimeContext`. Keep `RuntimeContext` strict by rejecting unknown properties, and require a negotiated contract-version change for additive shape changes.
Applied to files:
schemas/adapter-contract/legacy/openai-stream-invocation.schema.json
📚 Learning: 2026-06-28T04:03:32.877Z
Learnt from: AjayThorve
Repo: NVIDIA/NeMo-Fabric PR: 26
File: python/tests/smoke_typed_config.py:163-177
Timestamp: 2026-06-28T04:03:32.877Z
Learning: In NVIDIA NeMo Fabric Python SDK serialization of `RuntimeCapabilities` (to satisfy the “parity contract” with Rust core and the CLI), do not emit metadata keys when the corresponding metadata is absent. Instead, omit those fields entirely so the produced JSON matches the Rust/CLI output (e.g., avoid `null`, empty objects, or placeholder metadata). During review, verify the serializer/builders follow this omission rule and that Python outputs/parity tests reflect the same shape.
Applied to files:
python/src/nemo_fabric/runtime.py
🪛 ast-grep (0.45.1)
python/src/nemo_fabric/runtime.py
[info] 194-194: use jsonify instead of json.dumps for JSON output
Context: json.dumps(self._runtime.to_mapping())
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 195-195: use jsonify instead of json.dumps for JSON output
Context: json.dumps(payload)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 207-207: use jsonify instead of json.dumps for JSON output
Context: json.dumps(dict(openai_stream_transport))
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
tests/python/test_openai_streaming.py
[info] 206-206: use jsonify instead of json.dumps for JSON output
Context: json.dumps(record, separators=(",", ":"))
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 224-224: use jsonify instead of json.dumps for JSON output
Context: json.dumps(record, separators=(",", ":"))
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 253-253: use jsonify instead of json.dumps for JSON output
Context: json.dumps(_result(request, runtime, invocation_id=invocation_id))
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 263-269: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_result(
request,
runtime,
invocation_id=f"invocation-{len(mock_native.requests)}",
)
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 327-329: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_result(request, runtime, invocation_id="invocation-empty")
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 365-367: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_result(request, runtime, invocation_id="invocation-many")
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 385-387: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_result(request, runtime, invocation_id="invocation-without-stream")
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 417-419: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_result(request, runtime, invocation_id="invocation-after-probe")
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 453-455: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_result(request, runtime, invocation_id="invocation-late-stream")
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 483-489: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_result(
request,
runtime,
invocation_id="invocation-cancel-after-end",
)
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 556-563: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_result(
request,
runtime,
invocation_id="invocation-failed",
failed=True,
)
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 586-588: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_result(request, runtime, invocation_id="invocation-terminal")
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 675-675: use jsonify instead of json.dumps for JSON output
Context: json.dumps(record)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 689-689: use jsonify instead of json.dumps for JSON output
Context: json.dumps(nonfinite)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 703-703: use jsonify instead of json.dumps for JSON output
Context: json.dumps(_record())
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 706-712: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_record(
sequence=1,
invocation_id="invocation-other",
chunk=_chunk("chunk-2", "other"),
)
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 796-798: use jsonify instead of json.dumps for JSON output
Context: json.dumps(
_record(record_type="end"), separators=(",", ":")
)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
tests/adapters/test_external_nat_adapter.py
[info] 1367-1367: use jsonify instead of json.dumps for JSON output
Context: json.dumps(result)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
[info] 1405-1405: use jsonify instead of json.dumps for JSON output
Context: json.dumps(result)
Note: [CWE-116] Improper Encoding or Escaping of Output.
(use-jsonify)
🪛 LanguageTool
docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdx
[grammar] ~26-~26: Ensure spelling is correct
Context: ...reaminvocation.mdx): One adapter-native OpenAI streaming invocation. - [OpenAiStreamSink](struct-openaistrea...
(QB_NEW_EN_ORTHOGRAPHY_ERROR_IDS_1)
[grammar] ~41-~41: Ensure spelling is correct
Context: ...num-openaistreamprofile.mdx): Supported OpenAI-compatible chunk profile. - [OpenAiStreamProtocolVersion](enum-openai...
(QB_NEW_EN_ORTHOGRAPHY_ERROR_IDS_1)
[style] ~50-~50: Three successive sentences begin with the same word. Consider rewording the sentence or use a thesaurus to find a synonym.
Context: ...for adapter-native OpenAI streaming. - [OPENAI_STREAM_PROTOCOL_VERSION](constant-opena...
(ENGLISH_WORD_REPEAT_BEGINNING_RULE)
[grammar] ~50-~50: Ensure spelling is correct
Context: ...und protocol version for adapter-native OpenAI streaming. ## Functions - [invoke_openai_stream](fn-i...
(QB_NEW_EN_ORTHOGRAPHY_ERROR_IDS_1)
🪛 Ruff (0.16.1)
tests/adapters/test_adapters_common_lifecycle.py
[warning] 94-94: Missing return type annotation for private function start
Add return type annotation: None
(ANN202)
[warning] 97-97: Missing return type annotation for private function invoke
Add return type annotation: Never
(ANN202)
[warning] 98-98: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 100-100: Missing return type annotation for private function invoke_openai_stream
(ANN202)
[warning] 122-122: Missing return type annotation for private function stop
Add return type annotation: None
(ANN202)
[warning] 251-251: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 255-255: Missing return type annotation for private function open_connection
(ANN202)
[warning] 255-255: Missing type annotation for *_args
(ANN002)
[warning] 255-255: Missing type annotation for **_kwargs
(ANN003)
[warning] 290-290: Missing return type annotation for private function connect
(ANN202)
[warning] 299-299: Missing return type annotation for private function invoke_openai_stream
Add return type annotation: Never
(ANN202)
[warning] 341-341: Missing return type annotation for private function start
Add return type annotation: None
(ANN202)
[warning] 344-344: Missing return type annotation for private function invoke
(ANN202)
[warning] 347-347: Missing return type annotation for private function stop
Add return type annotation: None
(ANN202)
[warning] 432-432: Missing return type annotation for private function start
Add return type annotation: None
(ANN202)
[warning] 435-435: Missing return type annotation for private function invoke
Add return type annotation: None
(ANN202)
[warning] 438-438: Missing return type annotation for private function stop
Add return type annotation: None
(ANN202)
python/src/nemo_fabric/runtime.py
[error] 307-307: Function argument input is shadowing a Python builtin
(A002)
[warning] 307-307: Dynamically typed expressions (typing.Any) are disallowed in input
(ANN401)
[warning] 324-329: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 331-334: Avoid specifying long messages outside the exception class
(TRY003)
tests/python/test_openai_streaming.py
[warning] 19-19: Use from nemo_fabric import openai_streaming in lieu of alias
Replace with from nemo_fabric import openai_streaming
(PLR0402)
[warning] 125-125: Dynamically typed expressions (typing.Any) are disallowed in sequence
(ANN401)
[warning] 143-143: Dynamically typed expressions (typing.Any) are disallowed in stream
(ANN401)
[warning] 185-185: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 318-318: Missing return type annotation for private function invoke_empty
(ANN202)
[warning] 356-356: Missing return type annotation for private function invoke_many
(ANN202)
[warning] 383-383: Missing return type annotation for private function invoke_without_stream
(ANN202)
[warning] 406-406: Missing return type annotation for private function invoke_after_probe
(ANN202)
[error] 410-410: Possible hardcoded password assigned to argument: "token"
(S106)
[warning] 436-436: Missing return type annotation for private function invoke_before_stream
(ANN202)
[warning] 473-473: Missing return type annotation for private function invoke_after_end
(ANN202)
[warning] 498-499: Use asyncio.Event instead of awaiting asyncio.sleep in a while loop
(ASYNC110)
[warning] 547-547: Missing return type annotation for private function invoke_failed
(ANN202)
[warning] 577-577: Missing return type annotation for private function invoke_mismatch
(ANN202)
[warning] 718-718: Missing return type annotation for private function invoke_without_end
Add return type annotation: Never
(ANN202)
[warning] 731-731: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 743-743: Missing return type annotation for private function invoke_invalid_chunk
Add return type annotation: Never
(ANN202)
[warning] 759-759: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 781-781: Missing return type annotation for private function candidate
(ANN202)
[warning] 846-846: Boolean-typed positional argument in function definition
(FBT001)
[warning] 847-847: Boolean-typed positional argument in function definition
(FBT001)
[warning] 887-887: Missing return type annotation for private function fail_create_task
Add return type annotation: Never
(ANN202)
[warning] 889-889: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 891-891: Missing return type annotation for private function invoke
Add return type annotation: Never
(ANN202)
[warning] 892-892: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 920-920: Missing return type annotation for private function invoke_with_bad_token
Add return type annotation: Never
(ANN202)
[error] 929-929: Possible hardcoded password assigned to argument: "token"
(S106)
[warning] 931-931: Avoid specifying long messages outside the exception class
(TRY003)
tests/adapters/test_external_nat_adapter.py
[warning] 45-45: Missing return type annotation for special method __aiter__
(ANN204)
[warning] 48-48: Missing return type annotation for special method __anext__
(ANN204)
[warning] 466-466: Boolean-typed positional argument in function definition
(FBT001)
[warning] 1381-1381: Dynamically typed expressions (typing.Any) are disallowed in serialization_failure
(ANN401)
external/nat/src/nemo_fabric_adapters/nat/adapter.py
[warning] 663-666: Avoid specifying long messages outside the exception class
(TRY003)
[warning] 844-844: Too many return statements (7 > 6)
(PLR0911)
[warning] 844-844: Too many branches (21 > 12)
(PLR0912)
[warning] 844-844: Too many statements (65 > 50)
(PLR0915)
[warning] 863-863: Do not catch blind exception: Exception
(BLE001)
[warning] 864-867: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 894-898: Use a single with statement with multiple contexts instead of nested with statements
(SIM117)
[warning] 920-920: Abstract raise to an inner function
(TRY301)
[warning] 937-937: Do not catch blind exception: Exception
(BLE001)
[warning] 938-942: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 951-954: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
[warning] 959-959: Do not catch blind exception: Exception
(BLE001)
[warning] 960-963: Use logging.exception instead of logging.error
Replace with exception
(TRY400)
🔇 Additional comments (29)
crates/fabric-core/src/runtime.rs (5)
3076-3145: Set a read timeout on the accepted stream.
openai_stream_listenerblocks onaccept()andread_linewith no timeout. If a regression prevents the adapter from completing the chunked body,capture.join()blocks forever and the test hangs instead of failing.♻️ Proposed change
let capture = thread::spawn(move || { let (mut stream, _) = listener.accept().expect("accept stream connection"); + stream + .set_read_timeout(Some(Duration::from_secs(10))) + .expect("set stream read timeout"); let mut reader = BufReader::new(stream.try_clone().expect("clone stream"));
40-46: LGTM!Also applies to: 349-560, 576-587, 618-630, 675-681
823-869: LGTM!
1227-1297: LGTM!Also applies to: 1332-1362, 2130-2160
2840-2841: LGTM!Also applies to: 2886-2886, 2912-2957, 2972-2972, 2983-3009, 3029-3029, 3237-3394
tests/python/test_openai_streaming.py (3)
497-499: Replace the zero-delay spin with a short sleep.
asyncio.sleep(0)yields without delay, so this loop saturates the event loop for up to the full one-second timeout and starves the worker thread that feeds the listener. Useawait asyncio.sleep(0.001).
31-62: LGTM!Also applies to: 231-295, 772-811, 840-881
112-228: LGTM!Also applies to: 298-476, 517-714, 717-769, 884-941
python/src/nemo_fabric/runtime.py (2)
25-25: LGTM!Also applies to: 76-76, 176-178, 194-216, 250-251, 269-270, 304-352, 382-383
116-133: 🩺 Stability & AvailabilityRetain the mapping lookup for
adapter_descriptor;RunPlaninheritsMapping.get()and preserves this extension field.> Likely an incorrect or invalid review comment.tests/adapters/test_adapters_common_lifecycle.py (2)
15-15: LGTM!Also applies to: 54-157
160-228: LGTM!Also applies to: 231-311, 314-451
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatofsinkconfig.mdx (1)
26-38: LGTM!Also applies to: 50-78
docs/reference/api/rust-library-reference/nemo-fabric-core/error/enum-fabricerror.mdx (2)
12-12: LGTM!Also applies to: 288-327
26-276: LGTM!Also applies to: 336-514
scripts/docs/generate_rust_library_reference.py (2)
399-401: LGTM!
514-514: 🎯 Functional CorrectnessKeep the existing title normalization. Generated page titles already use
OpenAIcasing.> Likely an incorrect or invalid review comment.tests/adapters/test_external_nat_adapter.py (3)
1385-1388: 🎯 Functional Correctness | ⚡ Quick winClear the mock side effect before setting
return_value.The
mock_natfixture setsto_jsonablewithside_effect=lambda value, **_kwargs: value. Aside_effectcallable takes precedence overreturn_valueunless it returnsmock.DEFAULT. For the["not", "a", "mapping"]parameter, the mock still returns the originalobject(). The test passes only becauseobject()is also not adict, so the "serializer returned a non-mapping" path is never exercised.🐛 Proposed fix
if isinstance(serialization_failure, BaseException): mock_nat["to_jsonable"].side_effect = serialization_failure else: + mock_nat["to_jsonable"].side_effect = None mock_nat["to_jsonable"].return_value = serialization_failure
32-62: LGTM!Also applies to: 157-229, 289-343
1115-1200: LGTM!Also applies to: 1203-1295, 1298-1334, 1337-1370, 1410-1454, 1457-1512, 1515-1547, 1550-1566, 1569-1604
external/nat/src/nemo_fabric_adapters/nat/adapter.py (3)
879-885: 📐 Maintainability & Code Quality | 💤 Low valueDeduplicate the per-user session-kwargs derivation.
Lines 880-885 repeat the block at Lines 799-804, including the
getattr(self._sessions, "is_workflow_per_user", False)probe. Extract one private method and call it frominvokeandinvoke_openai_stream.
9-11: LGTM!Also applies to: 34-44, 62-65, 121-126, 223-224, 235-235, 645-677, 728-752, 794-806
844-970: LGTM!schemas/adapter-contract/legacy/openai-stream-invocation.schema.json (2)
170-179: 🗄️ Data Integrity & Integration | 🏗️ Heavy liftContract-version negotiation for
InvokeOpenaiStreamremains unresolved.
OpenAiStreamProtocolVersiondescribes only the stream wire protocol. It cannot negotiate support for the new lifecycle operation itself, because descriptors still declare the previous adapter contract version and the core dispatches on thestreamingcapability alone. Bump the adapter contract version, update descriptors and snapshots, and reject the older version before lifecycle dispatch.
180-239: LGTM!Also applies to: 344-368
docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/struct-openaistreamsink.mdx (1)
1-123: LGTM!docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/enum-openaistreamrecord.mdx (1)
1-143: LGTM!docs/reference/api/rust-library-reference/nemo-fabric-core/runtime/index.mdx (1)
5-5: LGTM!Also applies to: 23-28, 39-54
docs/reference/api/rust-library-reference/nemo-fabric-core/config/enum-relayatifstorageconfig.mdx (1)
26-42: LGTM!Also applies to: 54-86
| class _BackpressuredStreamWriter: | ||
| def __init__(self) -> None: | ||
| self.parts: list[bytes] = [] | ||
| self.drain_calls = 0 | ||
| self.first_drain_started = asyncio.Event() | ||
| self.release_first_drain = asyncio.Event() | ||
| self.closed = False | ||
|
|
||
| def write(self, data: bytes) -> None: | ||
| self.parts.append(data) | ||
|
|
||
| async def drain(self) -> None: | ||
| self.drain_calls += 1 | ||
| if self.drain_calls == 1: | ||
| self.first_drain_started.set() | ||
| await self.release_first_drain.wait() | ||
|
|
||
| def close(self) -> None: | ||
| self.closed = True | ||
|
|
||
| async def wait_closed(self) -> None: | ||
| pass |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low value
Prefer AsyncMock/MagicMock doubles over hand-written stub classes.
_BackpressuredStreamWriter (lines 30-51), CancellingWriter (lines 237-251), and FailingWriter (lines 283-288) each define a new class to stand in for a mocked collaborator. The test path instructions require unittest.mock.MagicMock or AsyncMock with spec for this, and require the mock prefix for mocked class names. _BackpressuredStreamWriter needs real coordination state, so an AsyncMock(spec=asyncio.StreamWriter) with side_effect functions for drain keeps the behavior and satisfies the rule. CancellingWriter and FailingWriter map directly onto AsyncMock with side_effect.
As per path instructions: "When mocking a class, use unittest.mock.MagicMock or AsyncMock, using the spec argument when necessary, rather than defining a new class" and "Prefix mocked class names with mock, not fake."
🤖 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/adapters/test_adapters_common_lifecycle.py` around lines 30 - 51,
Replace the hand-written _BackpressuredStreamWriter, CancellingWriter, and
FailingWriter doubles with unittest.mock AsyncMock/MagicMock instances, using
spec=asyncio.StreamWriter where appropriate. Preserve backpressure coordination
through the backpressured mock’s drain side_effect and retain the cancellation
and failure behavior through side_effect callbacks, naming any mock variables
with the mock prefix.
Source: Path instructions
| manager = await session_module.SessionManager.create( | ||
| config=config, | ||
| shared_builder=shared_builder, | ||
| ) | ||
| assert manager.is_workflow_per_user is True | ||
| assert manager.get_workflow_streaming_output_schema() is None |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Confirm SessionManager exposes is_workflow_per_user and get_workflow_streaming_output_schema.
set -uo pipefail
repo="NVIDIA/NeMo-Agent-Toolkit"
for tag in v1.8.0 develop; do
echo "== $tag =="
curl -fsSL "https://raw.githubusercontent.com/$repo/$tag/packages/nvidia_nat_core/src/nat/runtime/session.py" 2>/dev/null \
| rg -n 'is_workflow_per_user|get_workflow_streaming_output_schema|def create'
doneRepository: NVIDIA/NeMo-Fabric
Length of output: 1661
🏁 Script executed:
#!/bin/bash
set -uo pipefail
echo "== Repository NAT references =="
rg -n -i 'nemo.?agent|nat|importorskip|SessionManager|is_workflow_per_user|get_workflow_streaming_output_schema' \
tests/adapters/test_external_nat_adapter.py pyproject.toml python 2>/dev/null | head -300
echo "== NAT API history across release tags =="
repo="NVIDIA/NeMo-Agent-Toolkit"
for tag in v1.0.0 v1.1.0 v1.2.0 v1.3.0 v1.4.0 v1.5.0 v1.6.0 v1.7.0 v1.8.0; do
url="https://raw.githubusercontent.com/$repo/$tag/packages/nvidia_nat_core/src/nat/runtime/session.py"
body="$(curl -fsSL "$url" 2>/dev/null || true)"
if [ -z "$body" ]; then
echo "$tag: unavailable"
else
printf '%s: ' "$tag"
python3 -c 'import sys; s=sys.stdin.read(); print("is_workflow_per_user=" + str("def is_workflow_per_user" in s) + ", get_workflow_streaming_output_schema=" + str("def get_workflow_streaming_output_schema" in s) + ", create=" + str("async def create" in s))' <<<"$body"
fi
doneRepository: NVIDIA/NeMo-Fabric
Length of output: 31930
🏁 Script executed:
#!/bin/bash
set -uo pipefail
echo "== Test and adapter context =="
sed -n '700,810p' tests/adapters/test_external_nat_adapter.py
printf '\n== NAT-related project metadata ==\n'
rg -n -i 'nemo.?agent|nat|adapter-tests|harness' \
pyproject.toml external tests README.md docs 2>/dev/null | head -250
echo "== NAT v1.5.0 package metadata and API declarations =="
for path in \
packages/nvidia_nat_core/pyproject.toml \
packages/nvidia_nat_core/src/nat/runtime/session.py \
packages/nvidia_nat_core/src/nat/builder/per_user_workflow_builder.py; do
echo "--- $path ---"
curl -fsSL "https://raw.githubusercontent.com/NVIDIA/NeMo-Agent-Toolkit/v1.5.0/$path" 2>/dev/null \
| rg -n 'version|requires-python|def is_workflow_per_user|def get_workflow_streaming_output_schema|class SessionManager|class PerUserWorkflowBuilder' \
| head -80
done
echo "== Exact API tag/ref availability =="
for tag in v1.4.0 v1.5.0; do
curl -fsSL "https://api.github.com/repos/NVIDIA/NeMo-Agent-Toolkit/git/ref/tags/$tag" 2>/dev/null \
| jq -r '[.ref, .object.type, .object.sha] | `@tsv`' || true
doneRepository: NVIDIA/NeMo-Fabric
Length of output: 31610
Declare the required NAT API version or guard unsupported releases. The test skips missing modules but does not check either SessionManager member. NAT v1.5.0 provides both members. Enforce that minimum version or skip releases that lack them instead of raising AttributeError.
🤖 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/adapters/test_external_nat_adapter.py` around lines 791 - 796, Update
the test around SessionManager.create to require NAT version 1.5.0 or skip
unsupported releases before accessing is_workflow_per_user and
get_workflow_streaming_output_schema. Preserve the existing missing-module skip
behavior and ensure older versions do not fail with AttributeError.
1 participant
Overview
Important
Depends on #200 and #201. This draft targets
mainbecause both dependency branches live only on the contributor fork and cannot be selected as base branches inNVIDIA/NeMo-Fabric. The current GitHub diff temporarily includes both prerequisites. Review the NAT integration asgit diff 02b2f860..99174abb;02b2f860is the corrected #200 commit replayed on top of #201. After both prerequisites merge, this branch will be rebased ontoupstream/mainand the PR will reduce to its intended four-file delta.Implements the NAT adapter slice of FABRIC-162 by forwarding NVIDIA NeMo Agent Toolkit
ChatResponseChunkvalues through Fabric's adapter-native OpenAI streaming contract. FABRIC-192 remains a duplicate of FABRIC-162 and is not treated as a separate contract.Breaking changes: none.
Details
capabilities.streaming: trueonly with a complete chunk and terminal-result path.SessionManager.get_workflow_streaming_output_schema() is ChatResponseChunk; it does not infer support from a registry reference or from shared versus per-user lifecycle.SessionManager.is_workflow_per_useronly to validate and forward identity. A future per-user NAT registration that declaresChatResponseChunkcan stream without an adapter workflow-name change.runner.result_stream(to_type=ChatResponseChunk)exactly once, and never callsrunner.result()or replays the request.to_jsonable_python(..., serialize_unknown=False)boundary and awaits each Fabric emit for backpressure.delta.contentinto the normalized terminal response.[DONE], finish chunks, or token usage.ChatResponseChunkbefore opening a session. NAT 1.8's current per-user ReAct registration falls into this category because it has no stream function.Validation
just test-python— 804 passed, 22 skipped on the combined branch.ChatResponseChunkserialization passed the frozen Fabric OpenAI chunk validator, and the installed per-user registration reports no streaming schema.git diff --checkpassed.Where should the reviewer start?
Until #200 and #201 merge, review only
external/nat/src/nemo_fabric_adapters/nat/adapter.py,external/nat/fabric-adapter.json,external/nat/README.md, and the streaming cases intests/adapters/test_external_nat_adapter.py. The dependency-relative range is02b2f860..99174abb; the key design decision is the output-schema gate inNatRuntime.invoke_openai_stream.Related Issues: (use one of the action keywords Closes / Fixes / Resolves / Relates to)
Relates to FABRIC-162
Relates to feat: support NAT registry workflows #200
Relates to feat: add native OpenAI streaming #201
I confirm this contribution is my own work, or I have the right to submit it under this project's license.
I searched existing issues and open pull requests, and this does not duplicate existing work.
Summary by CodeRabbit
New Features
Documentation
Tests