Implement cross-workflow observability with context propagation and span processing #829
Conversation
…agation and processors Signed-off-by: pira998 <[email protected]>
WalkthroughAdds a cross-workflow observability subsystem: context model and manager, span processors, an observability-aware workflow invoker, Context/Runner plumbing to propagate ObservabilityContext, example scripts and config, and tests. No breaking API removals reported. Changes
Sequence Diagram(s)sequenceDiagram
autonumber
participant Caller
participant Invoker as ObservabilityWorkflowInvoker
participant Workflow
participant Runner
participant Ctx as ContextState
Caller->>Invoker: invoke_workflow_with_context(workflow, message, workflow_name, parent_ctx?)
Note over Invoker: create/derive ObservabilityContext\nmark workflow start_time
Invoker->>Workflow: run(message, observability_context)
Workflow->>Runner: Runner(..., observability_context)
activate Runner
Runner->>Ctx: set observability_context (ContextVar)
Runner-->>Workflow: result
deactivate Runner
Note over Invoker: update end_time, status (success/failure)
Invoker-->>Caller: result
sequenceDiagram
autonumber
participant Runtime as Runtime/Exporter
participant Processor1 as CrossWorkflowProcessor
participant Processor2 as WorkflowRelationshipProcessor
participant Span as Span Item
participant Ctx as Context (with observability_context)
Runtime->>Processor1: process(span)
Processor1->>Ctx: read observability_context
alt context available
Processor1->>Span: enrich with trace_id, span_ids, workflow_chain, timing, tags
else no context
Processor1-->>Runtime: span (unchanged)
end
Processor1-->>Runtime: span
Runtime->>Processor2: process(span)
Processor2->>Ctx: read observability_context
alt context available
Processor2->>Span: add parent/child relationship, nesting, hierarchy path
else
Processor2-->>Runtime: span (unchanged)
end
Processor2-->>Runtime: span (enriched)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Pre-merge checks and finishing touches❌ Failed checks (2 warnings)
✅ Passed checks (3 passed)
✨ Finishing touches
🧪 Generate unit tests
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
coderabbitai
bot
added
feature request
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 10
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/nat/runtime/runner.py (1)
108-168: Reset the observability ContextVar after the run.We call
ContextVar.set(...)in__aenter__but never reset it. That leaves the last workflow’s observability context stuck in the global state and contaminates subsequent runs. Capture the token fromset()and reset it in__aexit__(or finally block).Proposed fix:
- if self._observability_context: - self._context_state.observability_context.set(self._observability_context) + self._observability_context_token = None + if self._observability_context: + self._observability_context_token = ( + self._context_state.observability_context.set(self._observability_context) + ) @@ self._context_state.input_message.reset(self._input_message_token) + if self._observability_context_token is not None: + self._context_state.observability_context.reset(self._observability_context_token) + self._observability_context_token = None
🧹 Nitpick comments (3)
src/nat/observability/__init__.py (1)
22-28: Sort__all__to satisfy Ruff.
ruff(RUF022) asks for the exported names to be sorted; please reorder the list accordingly so linting passes.Updated ordering:
-__all__ = [ - "ObservabilityContext", - "ObservabilityContextManager", - "ObservabilityWorkflowInvoker", - "CrossWorkflowProcessor", - "WorkflowRelationshipProcessor" -] +__all__ = [ + "CrossWorkflowProcessor", + "ObservabilityContext", + "ObservabilityContextManager", + "ObservabilityWorkflowInvoker", + "WorkflowRelationshipProcessor", +]examples/observability/cross_workflow_tracking/integrated_example.py (1)
320-331: Avoid blanketexcept Exceptionin the parallel helper.Catching every exception trips
ruff(BLE001) and hides unexpected failures. Consider letting the task raise and usingasyncio.gather(..., return_exceptions=True)(or catching a specific workflow error type) so you can still surface per-query failures without suppressing unrelated bugs.src/nat/observability/workflow_utils.py (1)
181-187: Don’t blanket-catch every exception when reading the current context.
Context.get()shouldn’t be raising arbitrary exceptions here; catchingExceptionmasks real bugs and triggers BLE001. Either let it propagate or catch only the specific failure you expect (e.g.,LookupError). Given the current implementation, simply drop thetry/exceptand return the property.
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (14)
examples/observability/cross_workflow_tracking/README.md(1 hunks)examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py(1 hunks)examples/observability/cross_workflow_tracking/example.py(1 hunks)examples/observability/cross_workflow_tracking/integrated_example.py(1 hunks)examples/observability/cross_workflow_tracking/observability_config.yml(1 hunks)examples/observability/cross_workflow_tracking/workflow_integration_example.py(1 hunks)src/nat/builder/context.py(4 hunks)src/nat/builder/workflow.py(2 hunks)src/nat/observability/__init__.py(1 hunks)src/nat/observability/context.py(1 hunks)src/nat/observability/processor/cross_workflow_processor.py(1 hunks)src/nat/observability/workflow_utils.py(1 hunks)src/nat/runtime/runner.py(5 hunks)tests/nat/observability/test_cross_workflow_observability.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (11)
**/README.@(md|ipynb)
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
Ensure READMEs follow the naming convention; avoid deprecated names; use “NeMo Agent Toolkit” (capital T) in headings
Files:
examples/observability/cross_workflow_tracking/README.md
**/*
⚙️ CodeRabbit configuration file
**/*: # Code Review Instructions
- Ensure the code follows best practices and coding standards. - For Python code, follow
PEP 20 and
PEP 8 for style guidelines.- Check for security vulnerabilities and potential issues. - Python methods should use type hints for all parameters and return values.
Example:def my_function(param1: int, param2: str) -> bool: pass- For Python exception handling, ensure proper stack trace preservation:
- When re-raising exceptions: use bare
raisestatements to maintain the original stack trace,
and uselogger.error()(notlogger.exception()) to avoid duplicate stack trace output.- When catching and logging exceptions without re-raising: always use
logger.exception()
to capture the full stack trace information.Documentation Review Instructions - Verify that documentation and comments are clear and comprehensive. - Verify that the documentation doesn't contain any TODOs, FIXMEs or placeholder text like "lorem ipsum". - Verify that the documentation doesn't contain any offensive or outdated terms. - Verify that documentation and comments are free of spelling mistakes, ensure the documentation doesn't contain any
words listed in the
ci/vale/styles/config/vocabularies/nat/reject.txtfile, words that might appear to be
spelling mistakes but are listed in theci/vale/styles/config/vocabularies/nat/accept.txtfile are OK.Misc. - All code (except .mdc files that contain Cursor rules) should be licensed under the Apache License 2.0,
and should contain an Apache License 2.0 header comment at the top of each file.
- Confirm that copyright years are up-to date whenever a file is changed.
Files:
examples/observability/cross_workflow_tracking/README.mdexamples/observability/cross_workflow_tracking/observability_config.ymltests/nat/observability/test_cross_workflow_observability.pysrc/nat/observability/__init__.pyexamples/observability/cross_workflow_tracking/integrated_example.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/context.pysrc/nat/builder/workflow.pysrc/nat/runtime/runner.pyexamples/observability/cross_workflow_tracking/example.pysrc/nat/builder/context.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pyexamples/observability/cross_workflow_tracking/workflow_integration_example.pysrc/nat/observability/processor/cross_workflow_processor.py
examples/**/*
⚙️ CodeRabbit configuration file
examples/**/*: - This directory contains example code and usage scenarios for the toolkit, at a minimum an example should
contain a README.md or file README.ipynb.
- If an example contains Python code, it should be placed in a subdirectory named
src/and should
contain apyproject.tomlfile. Optionally, it might also contain scripts in ascripts/directory.- If an example contains YAML files, they should be placed in a subdirectory named
configs/. - If an example contains sample data files, they should be placed in a subdirectory nameddata/, and should
be checked into git-lfs.
Files:
examples/observability/cross_workflow_tracking/README.mdexamples/observability/cross_workflow_tracking/observability_config.ymlexamples/observability/cross_workflow_tracking/integrated_example.pyexamples/observability/cross_workflow_tracking/example.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pyexamples/observability/cross_workflow_tracking/workflow_integration_example.py
**/*.{yaml,yml}
📄 CodeRabbit inference engine (.cursor/rules/nat-test-llm.mdc)
In workflow/config YAML, set llms.._type: nat_test_llm to stub responses.
Files:
examples/observability/cross_workflow_tracking/observability_config.yml
**/*.{py,yaml,yml}
📄 CodeRabbit inference engine (.cursor/rules/nat-test-llm.mdc)
**/*.{py,yaml,yml}: Configure response_seq as a list of strings; values cycle per call, and [] yields an empty string.
Configure delay_ms to inject per-call artificial latency in milliseconds for nat_test_llm.
Files:
examples/observability/cross_workflow_tracking/observability_config.ymltests/nat/observability/test_cross_workflow_observability.pysrc/nat/observability/__init__.pyexamples/observability/cross_workflow_tracking/integrated_example.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/context.pysrc/nat/builder/workflow.pysrc/nat/runtime/runner.pyexamples/observability/cross_workflow_tracking/example.pysrc/nat/builder/context.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pyexamples/observability/cross_workflow_tracking/workflow_integration_example.pysrc/nat/observability/processor/cross_workflow_processor.py
**/*.py
📄 CodeRabbit inference engine (.cursor/rules/nat-test-llm.mdc)
**/*.py: Programmatic use: create TestLLMConfig(response_seq=[...], delay_ms=...), add with builder.add_llm("", cfg).
When retrieving the test LLM wrapper, use builder.get_llm(name, wrapper_type=LLMFrameworkEnum.) and call the framework’s method (e.g., ainvoke, achat, call).
**/*.py: In code comments/identifiers use NAT abbreviations as specified: nat for API namespace/CLI, nvidia-nat for package name, NAT for env var prefixes; do not use these abbreviations in documentation
Follow PEP 20 and PEP 8; run yapf with column_limit=120; use 4-space indentation; end files with a single trailing newline
Run ruff check --fix as linter (not formatter) using pyproject.toml config; fix warnings unless explicitly ignored
Respect naming: snake_case for functions/variables, PascalCase for classes, UPPER_CASE for constants
Treat pyright warnings as errors during development
Exception handling: use bare raise to re-raise; log with logger.error() when re-raising to avoid duplicate stack traces; use logger.exception() when catching without re-raising
Provide Google-style docstrings for every public module, class, function, and CLI command; first line concise and ending with a period; surround code entities with backticks
Validate and sanitize all user input, especially in web or CLI interfaces
Prefer httpx with SSL verification enabled by default and follow OWASP Top-10 recommendations
Use async/await for I/O-bound work; profile CPU-heavy paths with cProfile or mprof before optimizing; cache expensive computations with functools.lru_cache or external cache; leverage NumPy vectorized operations when beneficial
Files:
tests/nat/observability/test_cross_workflow_observability.pysrc/nat/observability/__init__.pyexamples/observability/cross_workflow_tracking/integrated_example.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/context.pysrc/nat/builder/workflow.pysrc/nat/runtime/runner.pyexamples/observability/cross_workflow_tracking/example.pysrc/nat/builder/context.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pyexamples/observability/cross_workflow_tracking/workflow_integration_example.pysrc/nat/observability/processor/cross_workflow_processor.py
tests/**/*.py
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
Unit tests reside under tests/ and should use markers defined in pyproject.toml (e.g., integration)
Files:
tests/nat/observability/test_cross_workflow_observability.py
⚙️ CodeRabbit configuration file
tests/**/*.py: - Ensure that tests are comprehensive, cover edge cases, and validate the functionality of the code. - Test functions should be named using thetest_prefix, using snake_case. - Any frequently repeated code should be extracted into pytest fixtures. - Pytest fixtures should define the name argument when applying the pytest.fixture decorator. The fixture
function being decorated should be named using thefixture_prefix, using snake_case. Example:
@pytest.fixture(name="my_fixture")
def fixture_my_fixture():
pass
Files:
tests/nat/observability/test_cross_workflow_observability.py
{tests/**/*.py,examples/*/tests/**/*.py}
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
{tests/**/*.py,examples/*/tests/**/*.py}: Use pytest (with pytest-asyncio for async); name test files test_*.py; test functions start with test_; extract repeated code into fixtures; fixtures must set name in decorator and be named with fixture_ prefix
Mock external services with pytest_httpserver or unittest.mock; do not hit live endpoints
Mark expensive tests with @pytest.mark.slow or @pytest.mark.integration
Files:
tests/nat/observability/test_cross_workflow_observability.py
src/**/*.py
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
All importable Python code must live under src/ (or packages//src/)
Files:
src/nat/observability/__init__.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/context.pysrc/nat/builder/workflow.pysrc/nat/runtime/runner.pysrc/nat/builder/context.pysrc/nat/observability/processor/cross_workflow_processor.py
src/nat/**/*
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
Changes in src/nat should prioritize backward compatibility
Files:
src/nat/observability/__init__.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/context.pysrc/nat/builder/workflow.pysrc/nat/runtime/runner.pysrc/nat/builder/context.pysrc/nat/observability/processor/cross_workflow_processor.py
⚙️ CodeRabbit configuration file
This directory contains the core functionality of the toolkit. Changes should prioritize backward compatibility.
Files:
src/nat/observability/__init__.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/context.pysrc/nat/builder/workflow.pysrc/nat/runtime/runner.pysrc/nat/builder/context.pysrc/nat/observability/processor/cross_workflow_processor.py
{src/**/*.py,packages/*/src/**/*.py}
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
All public APIs must have Python 3.11+ type hints on parameters and return values; prefer typing/collections.abc abstractions; use typing.Annotated when useful
Files:
src/nat/observability/__init__.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/context.pysrc/nat/builder/workflow.pysrc/nat/runtime/runner.pysrc/nat/builder/context.pysrc/nat/observability/processor/cross_workflow_processor.py
🧬 Code graph analysis (12)
tests/nat/observability/test_cross_workflow_observability.py (6)
src/nat/runtime/runner.py (5)
context(90-91)result(130-131)result(134-135)result(137-168)result_stream(170-200)src/nat/observability/context.py (18)
ObservabilityContext(39-141)ObservabilityContextManager(144-210)create_root_context(54-67)create_root_context(166-170)create_child_context(69-84)create_child_context(173-181)create_span_context(86-92)add_attribute(94-96)get_current_workflow(98-100)get_workflow_depth(102-104)to_dict(106-122)from_dict(125-141)get_current_context(156-158)clear_context(189-191)propagate_context(184-186)get_trace_id(194-197)get_current_span_id(200-203)add_workflow_attribute(206-210)src/nat/observability/workflow_utils.py (5)
ObservabilityWorkflowInvoker(33-281)invoke_workflow_with_context(42-105)invoke_workflow_stream_with_context(108-171)create_observability_context(190-212)get_current_observability_context(174-187)src/nat/observability/processor/cross_workflow_processor.py (4)
CrossWorkflowProcessor(21-105)WorkflowRelationshipProcessor(108-165)process(31-105)process(117-165)src/nat/builder/context.py (4)
Context(122-329)get(118-119)get(319-329)set_observability_context(309-316)src/nat/builder/workflow.py (1)
run(98-115)
src/nat/observability/__init__.py (3)
src/nat/observability/context.py (2)
ObservabilityContext(39-141)ObservabilityContextManager(144-210)src/nat/observability/processor/cross_workflow_processor.py (2)
CrossWorkflowProcessor(21-105)WorkflowRelationshipProcessor(108-165)src/nat/observability/workflow_utils.py (1)
ObservabilityWorkflowInvoker(33-281)
examples/observability/cross_workflow_tracking/integrated_example.py (5)
src/nat/runtime/loader.py (1)
load_workflow(97-117)src/nat/observability/context.py (10)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)get_workflow_depth(102-104)add_attribute(94-96)to_dict(106-122)from_dict(125-141)src/nat/builder/context.py (6)
Context(122-329)ContextState(66-119)observability_context(114-115)observability_context(296-307)get(118-119)get(319-329)src/nat/observability/processor/cross_workflow_processor.py (4)
CrossWorkflowProcessor(21-105)WorkflowRelationshipProcessor(108-165)process(31-105)process(117-165)src/nat/builder/workflow.py (1)
run(98-115)
src/nat/observability/workflow_utils.py (4)
src/nat/runtime/runner.py (5)
context(90-91)result(130-131)result(134-135)result(137-168)result_stream(170-200)src/nat/observability/context.py (6)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)src/nat/builder/workflow.py (3)
Workflow(42-169)run(98-115)result_with_steps(117-134)src/nat/builder/context.py (6)
Context(122-329)get(118-119)get(319-329)observability_context(114-115)observability_context(296-307)set_observability_context(309-316)
src/nat/observability/context.py (2)
src/nat/builder/context.py (2)
get(118-119)get(319-329)src/nat/runtime/runner.py (1)
context(90-91)
src/nat/builder/workflow.py (3)
src/nat/runtime/runner.py (2)
context(90-91)Runner(46-200)src/nat/observability/context.py (1)
ObservabilityContext(39-141)src/nat/builder/context.py (3)
observability_context(114-115)observability_context(296-307)input_message(128-139)
src/nat/runtime/runner.py (3)
src/nat/observability/context.py (1)
ObservabilityContext(39-141)src/nat/observability/exporter_manager.py (1)
ExporterManager(26-335)src/nat/builder/context.py (2)
observability_context(114-115)observability_context(296-307)
examples/observability/cross_workflow_tracking/example.py (1)
src/nat/observability/context.py (8)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)get_workflow_depth(102-104)add_attribute(94-96)
src/nat/builder/context.py (1)
src/nat/observability/context.py (1)
ObservabilityContext(39-141)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py (5)
src/nat/runtime/runner.py (4)
context(90-91)result(130-131)result(134-135)result(137-168)src/nat/observability/context.py (9)
ObservabilityContext(39-141)create_root_context(54-67)create_root_context(166-170)add_attribute(94-96)create_child_context(69-84)create_child_context(173-181)get_workflow_depth(102-104)to_dict(106-122)from_dict(125-141)src/nat/runtime/loader.py (1)
load_workflow(97-117)src/nat/builder/context.py (2)
observability_context(114-115)observability_context(296-307)src/nat/builder/workflow.py (1)
run(98-115)
examples/observability/cross_workflow_tracking/workflow_integration_example.py (5)
src/nat/runtime/runner.py (4)
context(90-91)result(130-131)result(134-135)result(137-168)src/nat/observability/context.py (8)
ObservabilityContext(39-141)create_root_context(54-67)create_root_context(166-170)add_attribute(94-96)create_child_context(69-84)create_child_context(173-181)get_current_workflow(98-100)get_workflow_depth(102-104)src/nat/runtime/loader.py (1)
load_workflow(97-117)src/nat/builder/context.py (2)
observability_context(114-115)observability_context(296-307)src/nat/builder/workflow.py (1)
run(98-115)
src/nat/observability/processor/cross_workflow_processor.py (3)
src/nat/observability/processor/processor.py (1)
Processor(27-74)src/nat/utils/type_utils.py (1)
override(56-57)src/nat/builder/context.py (5)
Context(122-329)get(118-119)get(319-329)observability_context(114-115)observability_context(296-307)
🪛 Ruff (0.13.2)
tests/nat/observability/test_cross_workflow_observability.py
349-349: Unused function argument: to_type
(ARG001)
src/nat/observability/__init__.py
22-28: __all__ is not sorted
Apply an isort-style sorting to __all__
(RUF022)
examples/observability/cross_workflow_tracking/integrated_example.py
1-1: Shebang is present but file is not executable
(EXE001)
98-98: Consider moving this statement to an else block
(TRY300)
177-189: f-string without any placeholders
Remove extraneous f prefix
(F541)
196-208: f-string without any placeholders
Remove extraneous f prefix
(F541)
215-227: f-string without any placeholders
Remove extraneous f prefix
(F541)
302-302: Do not catch blind exception: Exception
(BLE001)
329-329: Consider moving this statement to an else block
(TRY300)
330-330: Do not catch blind exception: Exception
(BLE001)
src/nat/observability/workflow_utils.py
75-75: Do not catch blind exception: Exception
(BLE001)
142-142: Do not catch blind exception: Exception
(BLE001)
185-185: Consider moving this statement to an else block
(TRY300)
186-186: Do not catch blind exception: Exception
(BLE001)
209-210: try-except-pass detected, consider logging the exception
(S110)
209-209: Do not catch blind exception: Exception
(BLE001)
248-248: Do not catch blind exception: Exception
(BLE001)
271-271: Consider moving this statement to an else block
(TRY300)
src/nat/observability/context.py
78-78: Consider [*self.workflow_chain, child_workflow] instead of concatenation
Replace with [*self.workflow_chain, child_workflow]
(RUF005)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
1-1: Shebang is present but file is not executable
(EXE001)
examples/observability/cross_workflow_tracking/workflow_integration_example.py
1-1: Shebang is present but file is not executable
(EXE001)
79-79: Do not catch blind exception: Exception
(BLE001)
src/nat/observability/processor/cross_workflow_processor.py
101-101: Do not catch blind exception: Exception
(BLE001)
161-161: Do not catch blind exception: Exception
(BLE001)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
Outdated
Show resolved
Hide resolved
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
Show resolved
Hide resolved
examples/observability/cross_workflow_tracking/integrated_example.py
Outdated
Show resolved
Hide resolved
examples/observability/cross_workflow_tracking/observability_config.yml
Outdated
Show resolved
Hide resolved
examples/observability/cross_workflow_tracking/workflow_integration_example.py
Show resolved
Hide resolved
examples/observability/cross_workflow_tracking/workflow_integration_example.py
Show resolved
Hide resolved
There was a problem hiding this comment.
Actionable comments posted: 4
♻️ Duplicate comments (5)
examples/observability/cross_workflow_tracking/integrated_example.py (1)
80-87: Use the public observability API (repeat).We still poke
workflow._context_stateand then callworkflow.run(...)without passing the context, so the runner wipes the setting and spans miss the cross-workflow data. Please drop the private mutation and callworkflow.run(input_data, observability_context=obs_context)instead.- async with load_workflow(config_file) as workflow: - # Set observability context in the workflow's context state - context_state = workflow._context_state - context_state.observability_context.set(obs_context) - - # Run the workflow without the observability_context parameter - async with workflow.run(input_data) as runner: + async with load_workflow(config_file) as workflow: + # Run the workflow with the propagated observability context + async with workflow.run(input_data, observability_context=obs_context) as runner: result = await runner.result(to_type=str)examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py (1)
22-34: Stub the example withnat_test_llminstead of NIM.Example configs must never call real services. Replace the
nvidia_llmblock with anat_test_llmstub (includeresponse_seqanddelay_ms) and pointworkflow.llm_nameat it so the sample stays offline. Based on learnings-llms: - nvidia_llm: - _type: nim - model_name: meta/llama-3.1-8b-instruct - temperature: 0.7 - max_tokens: 1024 +llms: + demo_llm: + _type: nat_test_llm + response_seq: + - "Stubbed support reply." + delay_ms: 0 workflow: _type: chat_completion - llm_name: nvidia_llm + llm_name: demo_llmsrc/nat/observability/workflow_utils.py (3)
66-111: Clean up context propagation error handling.Catching
Exceptionand doing per-branch bookkeeping breaks our lint config (BLE001) and reintroduces the pattern the previous review asked to refactor. Switch totry/finally, captureexc = sys.exc_info()[1], update status/tags in thefinallyblock, and log withlogger.error(...)whenexcexists before re-raising. This also lets you drop the blanket catch entirely.- try: - async with workflow.run(message, observability_context=obs_context) as runner: - result = await runner.result(to_type=to_type) - - # Update workflow metadata on completion - if obs_context: - current_workflow = obs_context.get_current_workflow() - if current_workflow: - current_workflow.end_time = time.time() - current_workflow.status = "completed" - - return result - - except Exception as e: - # Update workflow metadata on failure and log error - if obs_context: - current_workflow = obs_context.get_current_workflow() - if current_workflow: - current_workflow.end_time = time.time() - current_workflow.status = "failed" - current_workflow.tags["error"] = str(e) - - logger.error(f"Workflow '{workflow_name}' failed with error: {e}", exc_info=True) - raise + result = None + try: + async with workflow.run(message, observability_context=obs_context) as runner: + result = await runner.result(to_type=to_type) + return result + finally: + exc = sys.exc_info()[1] + if obs_context: + current_workflow = obs_context.get_current_workflow() + if current_workflow: + current_workflow.end_time = time.time() + if exc is None: + current_workflow.status = "completed" + else: + current_workflow.status = "failed" + current_workflow.tags["error"] = str(exc) + logger.error("Workflow '%s' failed: %s", workflow_name, exc)
148-179: Apply the samefinallypattern to streaming runs.The streaming helper still swallows BLE001 with a broad
except. Mirror thetry/finally+sys.exc_info()structure so metadata updates and logging run exactly once and lint passes.- try: - async with workflow.run(message, observability_context=obs_context) as runner: - async for item in runner.result_stream(to_type=to_type): - yield item - - # Update workflow metadata on completion - if obs_context: - current_workflow = obs_context.get_current_workflow() - if current_workflow: - current_workflow.end_time = time.time() - current_workflow.status = "completed" - - except Exception as e: - # Update workflow metadata on failure and log error - if obs_context: - current_workflow = obs_context.get_current_workflow() - if current_workflow: - current_workflow.end_time = time.time() - current_workflow.status = "failed" - current_workflow.tags["error"] = str(e) - - logger.error(f"Streaming workflow '{workflow_name}' failed with error: {e}", exc_info=True) - raise + try: + async with workflow.run(message, observability_context=obs_context) as runner: + async for item in runner.result_stream(to_type=to_type): + yield item + finally: + exc = sys.exc_info()[1] + if obs_context: + current_workflow = obs_context.get_current_workflow() + if current_workflow: + current_workflow.end_time = time.time() + if exc is None: + current_workflow.status = "completed" + else: + current_workflow.status = "failed" + current_workflow.tags["error"] = str(exc) + logger.error("Streaming workflow '%s' failed: %s", workflow_name, exc)
198-291: Likewise fixinvoke_with_steps_and_context.We still have blanket
except Exceptionhere. Convert to the sharedtry/finallystructure so status bookkeeping and logging stay consistent and lint stays clean.- try: - result, steps = await workflow.result_with_steps( - message, - to_type=to_type, - observability_context=obs_context - ) - - # Update workflow metadata on completion - if obs_context: - current_workflow = obs_context.get_current_workflow() - if current_workflow: - current_workflow.end_time = time.time() - current_workflow.status = "completed" - - return result, steps - - except Exception as e: - # Update workflow metadata on failure and log error - if obs_context: - current_workflow = obs_context.get_current_workflow() - if current_workflow: - current_workflow.end_time = time.time() - current_workflow.status = "failed" - current_workflow.tags["error"] = str(e) - - logger.error(f"Workflow with steps '{workflow_name}' failed with error: {e}", exc_info=True) - raise + result: SingleOutputT | None = None + steps: Any = None + try: + result, steps = await workflow.result_with_steps( + message, + to_type=to_type, + observability_context=obs_context + ) + return result, steps + finally: + exc = sys.exc_info()[1] + if obs_context: + current_workflow = obs_context.get_current_workflow() + if current_workflow: + current_workflow.end_time = time.time() + if exc is None: + current_workflow.status = "completed" + else: + current_workflow.status = "failed" + current_workflow.tags["error"] = str(exc) + logger.error("Workflow with steps '%s' failed: %s", workflow_name, exc)
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (6)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py(1 hunks)examples/observability/cross_workflow_tracking/integrated_example.py(1 hunks)examples/observability/cross_workflow_tracking/observability_config.yml(1 hunks)external/nat-ui(1 hunks)src/nat/observability/processor/cross_workflow_processor.py(1 hunks)src/nat/observability/workflow_utils.py(1 hunks)
✅ Files skipped from review due to trivial changes (1)
- external/nat-ui
🚧 Files skipped from review as they are similar to previous changes (1)
- examples/observability/cross_workflow_tracking/observability_config.yml
🧰 Additional context used
📓 Path-based instructions (7)
**/*.{py,yaml,yml}
📄 CodeRabbit inference engine (.cursor/rules/nat-test-llm.mdc)
**/*.{py,yaml,yml}: Configure response_seq as a list of strings; values cycle per call, and [] yields an empty string.
Configure delay_ms to inject per-call artificial latency in milliseconds for nat_test_llm.
Files:
examples/observability/cross_workflow_tracking/integrated_example.pysrc/nat/observability/processor/cross_workflow_processor.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pysrc/nat/observability/workflow_utils.py
**/*.py
📄 CodeRabbit inference engine (.cursor/rules/nat-test-llm.mdc)
**/*.py: Programmatic use: create TestLLMConfig(response_seq=[...], delay_ms=...), add with builder.add_llm("", cfg).
When retrieving the test LLM wrapper, use builder.get_llm(name, wrapper_type=LLMFrameworkEnum.) and call the framework’s method (e.g., ainvoke, achat, call).
**/*.py: In code comments/identifiers use NAT abbreviations as specified: nat for API namespace/CLI, nvidia-nat for package name, NAT for env var prefixes; do not use these abbreviations in documentation
Follow PEP 20 and PEP 8; run yapf with column_limit=120; use 4-space indentation; end files with a single trailing newline
Run ruff check --fix as linter (not formatter) using pyproject.toml config; fix warnings unless explicitly ignored
Respect naming: snake_case for functions/variables, PascalCase for classes, UPPER_CASE for constants
Treat pyright warnings as errors during development
Exception handling: use bare raise to re-raise; log with logger.error() when re-raising to avoid duplicate stack traces; use logger.exception() when catching without re-raising
Provide Google-style docstrings for every public module, class, function, and CLI command; first line concise and ending with a period; surround code entities with backticks
Validate and sanitize all user input, especially in web or CLI interfaces
Prefer httpx with SSL verification enabled by default and follow OWASP Top-10 recommendations
Use async/await for I/O-bound work; profile CPU-heavy paths with cProfile or mprof before optimizing; cache expensive computations with functools.lru_cache or external cache; leverage NumPy vectorized operations when beneficial
Files:
examples/observability/cross_workflow_tracking/integrated_example.pysrc/nat/observability/processor/cross_workflow_processor.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pysrc/nat/observability/workflow_utils.py
**/*
⚙️ CodeRabbit configuration file
**/*: # Code Review Instructions
- Ensure the code follows best practices and coding standards. - For Python code, follow
PEP 20 and
PEP 8 for style guidelines.- Check for security vulnerabilities and potential issues. - Python methods should use type hints for all parameters and return values.
Example:def my_function(param1: int, param2: str) -> bool: pass- For Python exception handling, ensure proper stack trace preservation:
- When re-raising exceptions: use bare
raisestatements to maintain the original stack trace,
and uselogger.error()(notlogger.exception()) to avoid duplicate stack trace output.- When catching and logging exceptions without re-raising: always use
logger.exception()
to capture the full stack trace information.Documentation Review Instructions - Verify that documentation and comments are clear and comprehensive. - Verify that the documentation doesn't contain any TODOs, FIXMEs or placeholder text like "lorem ipsum". - Verify that the documentation doesn't contain any offensive or outdated terms. - Verify that documentation and comments are free of spelling mistakes, ensure the documentation doesn't contain any
words listed in the
ci/vale/styles/config/vocabularies/nat/reject.txtfile, words that might appear to be
spelling mistakes but are listed in theci/vale/styles/config/vocabularies/nat/accept.txtfile are OK.Misc. - All code (except .mdc files that contain Cursor rules) should be licensed under the Apache License 2.0,
and should contain an Apache License 2.0 header comment at the top of each file.
- Confirm that copyright years are up-to date whenever a file is changed.
Files:
examples/observability/cross_workflow_tracking/integrated_example.pysrc/nat/observability/processor/cross_workflow_processor.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pysrc/nat/observability/workflow_utils.py
examples/**/*
⚙️ CodeRabbit configuration file
examples/**/*: - This directory contains example code and usage scenarios for the toolkit, at a minimum an example should
contain a README.md or file README.ipynb.
- If an example contains Python code, it should be placed in a subdirectory named
src/and should
contain apyproject.tomlfile. Optionally, it might also contain scripts in ascripts/directory.- If an example contains YAML files, they should be placed in a subdirectory named
configs/. - If an example contains sample data files, they should be placed in a subdirectory nameddata/, and should
be checked into git-lfs.
Files:
examples/observability/cross_workflow_tracking/integrated_example.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
src/**/*.py
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
All importable Python code must live under src/ (or packages//src/)
Files:
src/nat/observability/processor/cross_workflow_processor.pysrc/nat/observability/workflow_utils.py
src/nat/**/*
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
Changes in src/nat should prioritize backward compatibility
Files:
src/nat/observability/processor/cross_workflow_processor.pysrc/nat/observability/workflow_utils.py
⚙️ CodeRabbit configuration file
This directory contains the core functionality of the toolkit. Changes should prioritize backward compatibility.
Files:
src/nat/observability/processor/cross_workflow_processor.pysrc/nat/observability/workflow_utils.py
{src/**/*.py,packages/*/src/**/*.py}
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
All public APIs must have Python 3.11+ type hints on parameters and return values; prefer typing/collections.abc abstractions; use typing.Annotated when useful
Files:
src/nat/observability/processor/cross_workflow_processor.pysrc/nat/observability/workflow_utils.py
🧠 Learnings (2)
📚 Learning: 2025-09-09T20:32:39.016Z
Learnt from: CR
PR: NVIDIA/NeMo-Agent-Toolkit#0
File: .cursor/rules/nat-test-llm.mdc:0-0
Timestamp: 2025-09-09T20:32:39.016Z
Learning: When stubbing deterministic LLM responses in tests/workflows, use the NAT Test LLM (nat_test_llm).
Applied to files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
📚 Learning: 2025-09-09T20:32:39.016Z
Learnt from: CR
PR: NVIDIA/NeMo-Agent-Toolkit#0
File: .cursor/rules/nat-test-llm.mdc:0-0
Timestamp: 2025-09-09T20:32:39.016Z
Learning: Applies to **/*.{yaml,yml} : In workflow/config YAML, set llms.<name>._type: nat_test_llm to stub responses.
Applied to files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
🧬 Code graph analysis (4)
examples/observability/cross_workflow_tracking/integrated_example.py (5)
src/nat/runtime/loader.py (1)
load_workflow(97-117)src/nat/observability/context.py (10)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)get_workflow_depth(102-104)add_attribute(94-96)to_dict(106-122)from_dict(125-141)src/nat/builder/context.py (6)
Context(122-329)ContextState(66-119)observability_context(114-115)observability_context(296-307)get(118-119)get(319-329)src/nat/observability/processor/cross_workflow_processor.py (4)
CrossWorkflowProcessor(25-110)WorkflowRelationshipProcessor(113-171)process(35-110)process(122-171)src/nat/builder/workflow.py (1)
run(98-115)
src/nat/observability/processor/cross_workflow_processor.py (3)
src/nat/observability/processor/processor.py (1)
Processor(27-74)src/nat/utils/type_utils.py (1)
override(56-57)src/nat/builder/context.py (5)
Context(122-329)get(118-119)get(319-329)observability_context(114-115)observability_context(296-307)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py (5)
src/nat/runtime/runner.py (4)
context(90-91)result(130-131)result(134-135)result(137-168)src/nat/observability/context.py (9)
ObservabilityContext(39-141)create_root_context(54-67)create_root_context(166-170)add_attribute(94-96)create_child_context(69-84)create_child_context(173-181)get_workflow_depth(102-104)to_dict(106-122)from_dict(125-141)src/nat/runtime/loader.py (1)
load_workflow(97-117)src/nat/builder/context.py (2)
observability_context(114-115)observability_context(296-307)src/nat/builder/workflow.py (1)
run(98-115)
src/nat/observability/workflow_utils.py (4)
src/nat/runtime/runner.py (5)
context(90-91)result(130-131)result(134-135)result(137-168)result_stream(170-200)src/nat/observability/context.py (6)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)src/nat/builder/workflow.py (3)
Workflow(42-169)run(98-115)result_with_steps(117-134)src/nat/builder/context.py (6)
Context(122-329)get(118-119)get(319-329)observability_context(114-115)observability_context(296-307)set_observability_context(309-316)
🪛 Ruff (0.13.2)
examples/observability/cross_workflow_tracking/integrated_example.py
1-1: Shebang is present but file is not executable
(EXE001)
98-98: Consider moving this statement to an else block
(TRY300)
302-302: Do not catch blind exception: Exception
(BLE001)
329-329: Consider moving this statement to an else block
(TRY300)
330-330: Do not catch blind exception: Exception
(BLE001)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
1-1: Shebang is present but file is not executable
(EXE001)
src/nat/observability/workflow_utils.py
79-79: Do not catch blind exception: Exception
(BLE001)
148-148: Do not catch blind exception: Exception
(BLE001)
193-193: Consider moving this statement to an else block
(TRY300)
194-194: Do not catch blind exception: Exception
(BLE001)
217-218: try-except-pass detected, consider logging the exception
(S110)
217-217: Do not catch blind exception: Exception
(BLE001)
256-256: Do not catch blind exception: Exception
(BLE001)
279-279: Consider moving this statement to an else block
(TRY300)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
Show resolved
Hide resolved
There was a problem hiding this comment.
Actionable comments posted: 1
♻️ Duplicate comments (1)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py (1)
78-82: Use the publicobservability_contextparameter instead of private attribute mutation.Setting
workflow._context_state.observability_contextis overwritten whenworkflow.run()resets the context var. Passquery_contextvia the public parameter so the context reaches the runner.Apply this diff:
- # Set observability context in the workflow's context state - workflow._context_state.observability_context.set(query_context) - - # Execute workflow - async with workflow.run(query) as runner: + # Execute workflow with propagated observability context + async with workflow.run(query, observability_context=query_context) as runner:</review_comment_end -->
🧹 Nitpick comments (6)
src/nat/observability/processor/cross_workflow_processor.py (2)
107-107: Remove redundant exception object fromlogger.exception()call.
logger.exception()automatically includes the exception details, so passingein the format string is redundant.Apply this diff:
- logger.exception("Error processing cross-workflow observability data: %s", e) + logger.exception("Error processing cross-workflow observability data")</review_comment_end -->
168-168: Remove redundant exception object fromlogger.exception()call.Same issue as above—drop the
%s, esincelogger.exception()captures it automatically.Apply this diff:
- logger.exception("Error processing workflow relationship data: %s", e) + logger.exception("Error processing workflow relationship data")</review_comment_end -->
examples/observability/cross_workflow_tracking/example.py (1)
127-127: Add missing return type hint.All public async functions should have type hints per Python guidelines. Add
-> Nonetomain().Apply this diff:
-async def main(): +async def main() -> None:</review_comment_end -->
src/nat/observability/workflow_utils.py (3)
70-80: Log exceptions when falling back to root context.The blanket
Exceptioncatch is acceptable here since observability is optional and shouldn't break workflows. However, log the exception so unexpected failures are visible.Apply this diff:
try: # Import here to avoid circular import from nat.builder.context import Context current_context = Context.get() existing_obs_context = current_context.observability_context if existing_obs_context: obs_context = existing_obs_context.create_child_context(workflow_name) else: obs_context = ObservabilityContext.create_root_context(workflow_name) - except Exception: + except Exception as e: + logger.debug("Failed to get existing observability context, creating root: %s", e) obs_context = ObservabilityContext.create_root_context(workflow_name)Apply the same pattern at lines 131-141 and 232-242.
</review_comment_end -->
174-180: Log when context retrieval fails.Returning
Noneon failure is appropriate, but log at debug level so developers can diagnose issues.Apply this diff:
try: # Import here to avoid circular import from nat.builder.context import Context current_context = Context.get() return current_context.observability_context - except Exception: + except Exception as e: + logger.debug("Failed to get current observability context: %s", e) return None</review_comment_end -->
197-203: Log when setting context fails.The silent
passon exception is flagged by S110. Log at debug level so failures are visible during troubleshooting.Apply this diff:
# Set it in the current context if available try: # Import here to avoid circular import from nat.builder.context import Context current_context = Context.get() current_context.set_observability_context(context) - except Exception: - pass # Context might not be available + except Exception as e: + logger.debug("Failed to set observability context in execution context: %s", e)</review_comment_end -->
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py(1 hunks)examples/observability/cross_workflow_tracking/example.py(1 hunks)src/nat/observability/processor/cross_workflow_processor.py(1 hunks)src/nat/observability/workflow_utils.py(1 hunks)
🧰 Additional context used
📓 Path-based instructions (7)
**/*.{py,yaml,yml}
📄 CodeRabbit inference engine (.cursor/rules/nat-test-llm.mdc)
**/*.{py,yaml,yml}: Configure response_seq as a list of strings; values cycle per call, and [] yields an empty string.
Configure delay_ms to inject per-call artificial latency in milliseconds for nat_test_llm.
Files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pyexamples/observability/cross_workflow_tracking/example.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
**/*.py
📄 CodeRabbit inference engine (.cursor/rules/nat-test-llm.mdc)
**/*.py: Programmatic use: create TestLLMConfig(response_seq=[...], delay_ms=...), add with builder.add_llm("", cfg).
When retrieving the test LLM wrapper, use builder.get_llm(name, wrapper_type=LLMFrameworkEnum.) and call the framework’s method (e.g., ainvoke, achat, call).
**/*.py: In code comments/identifiers use NAT abbreviations as specified: nat for API namespace/CLI, nvidia-nat for package name, NAT for env var prefixes; do not use these abbreviations in documentation
Follow PEP 20 and PEP 8; run yapf with column_limit=120; use 4-space indentation; end files with a single trailing newline
Run ruff check --fix as linter (not formatter) using pyproject.toml config; fix warnings unless explicitly ignored
Respect naming: snake_case for functions/variables, PascalCase for classes, UPPER_CASE for constants
Treat pyright warnings as errors during development
Exception handling: use bare raise to re-raise; log with logger.error() when re-raising to avoid duplicate stack traces; use logger.exception() when catching without re-raising
Provide Google-style docstrings for every public module, class, function, and CLI command; first line concise and ending with a period; surround code entities with backticks
Validate and sanitize all user input, especially in web or CLI interfaces
Prefer httpx with SSL verification enabled by default and follow OWASP Top-10 recommendations
Use async/await for I/O-bound work; profile CPU-heavy paths with cProfile or mprof before optimizing; cache expensive computations with functools.lru_cache or external cache; leverage NumPy vectorized operations when beneficial
Files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pyexamples/observability/cross_workflow_tracking/example.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
**/*
⚙️ CodeRabbit configuration file
**/*: # Code Review Instructions
- Ensure the code follows best practices and coding standards. - For Python code, follow
PEP 20 and
PEP 8 for style guidelines.- Check for security vulnerabilities and potential issues. - Python methods should use type hints for all parameters and return values.
Example:def my_function(param1: int, param2: str) -> bool: pass- For Python exception handling, ensure proper stack trace preservation:
- When re-raising exceptions: use bare
raisestatements to maintain the original stack trace,
and uselogger.error()(notlogger.exception()) to avoid duplicate stack trace output.- When catching and logging exceptions without re-raising: always use
logger.exception()
to capture the full stack trace information.Documentation Review Instructions - Verify that documentation and comments are clear and comprehensive. - Verify that the documentation doesn't contain any TODOs, FIXMEs or placeholder text like "lorem ipsum". - Verify that the documentation doesn't contain any offensive or outdated terms. - Verify that documentation and comments are free of spelling mistakes, ensure the documentation doesn't contain any
words listed in the
ci/vale/styles/config/vocabularies/nat/reject.txtfile, words that might appear to be
spelling mistakes but are listed in theci/vale/styles/config/vocabularies/nat/accept.txtfile are OK.Misc. - All code (except .mdc files that contain Cursor rules) should be licensed under the Apache License 2.0,
and should contain an Apache License 2.0 header comment at the top of each file.
- Confirm that copyright years are up-to date whenever a file is changed.
Files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pyexamples/observability/cross_workflow_tracking/example.pysrc/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
examples/**/*
⚙️ CodeRabbit configuration file
examples/**/*: - This directory contains example code and usage scenarios for the toolkit, at a minimum an example should
contain a README.md or file README.ipynb.
- If an example contains Python code, it should be placed in a subdirectory named
src/and should
contain apyproject.tomlfile. Optionally, it might also contain scripts in ascripts/directory.- If an example contains YAML files, they should be placed in a subdirectory named
configs/. - If an example contains sample data files, they should be placed in a subdirectory nameddata/, and should
be checked into git-lfs.
Files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.pyexamples/observability/cross_workflow_tracking/example.py
src/**/*.py
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
All importable Python code must live under src/ (or packages//src/)
Files:
src/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
src/nat/**/*
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
Changes in src/nat should prioritize backward compatibility
Files:
src/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
⚙️ CodeRabbit configuration file
This directory contains the core functionality of the toolkit. Changes should prioritize backward compatibility.
Files:
src/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
{src/**/*.py,packages/*/src/**/*.py}
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
All public APIs must have Python 3.11+ type hints on parameters and return values; prefer typing/collections.abc abstractions; use typing.Annotated when useful
Files:
src/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
🧠 Learnings (2)
📚 Learning: 2025-09-09T20:32:39.016Z
Learnt from: CR
PR: NVIDIA/NeMo-Agent-Toolkit#0
File: .cursor/rules/nat-test-llm.mdc:0-0
Timestamp: 2025-09-09T20:32:39.016Z
Learning: When stubbing deterministic LLM responses in tests/workflows, use the NAT Test LLM (nat_test_llm).
Applied to files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
📚 Learning: 2025-09-09T20:32:39.016Z
Learnt from: CR
PR: NVIDIA/NeMo-Agent-Toolkit#0
File: .cursor/rules/nat-test-llm.mdc:0-0
Timestamp: 2025-09-09T20:32:39.016Z
Learning: Applies to **/*.{yaml,yml} : In workflow/config YAML, set llms.<name>._type: nat_test_llm to stub responses.
Applied to files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
🧬 Code graph analysis (4)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py (5)
src/nat/runtime/runner.py (4)
context(90-91)result(130-131)result(134-135)result(137-168)src/nat/observability/context.py (9)
ObservabilityContext(39-141)create_root_context(54-67)create_root_context(166-170)add_attribute(94-96)create_child_context(69-84)create_child_context(173-181)get_workflow_depth(102-104)to_dict(106-122)from_dict(125-141)src/nat/runtime/loader.py (1)
load_workflow(97-117)src/nat/builder/context.py (2)
observability_context(114-115)observability_context(296-307)src/nat/builder/workflow.py (1)
run(98-115)
examples/observability/cross_workflow_tracking/example.py (2)
src/nat/observability/context.py (8)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)get_workflow_depth(102-104)add_attribute(94-96)src/nat/builder/workflow.py (1)
run(98-115)
src/nat/observability/workflow_utils.py (4)
src/nat/runtime/runner.py (5)
context(90-91)result(130-131)result(134-135)result(137-168)result_stream(170-200)src/nat/observability/context.py (6)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)src/nat/builder/workflow.py (3)
Workflow(42-169)run(98-115)result_with_steps(117-134)src/nat/builder/context.py (6)
Context(122-329)get(118-119)get(319-329)observability_context(114-115)observability_context(296-307)set_observability_context(309-316)
src/nat/observability/processor/cross_workflow_processor.py (3)
src/nat/observability/processor/processor.py (1)
Processor(27-74)src/nat/utils/type_utils.py (1)
override(56-57)src/nat/builder/context.py (5)
Context(122-329)get(118-119)get(319-329)observability_context(114-115)observability_context(296-307)
🪛 Ruff (0.13.2)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
1-1: Shebang is present but file is not executable
(EXE001)
src/nat/observability/workflow_utils.py
79-79: Do not catch blind exception: Exception
(BLE001)
140-140: Do not catch blind exception: Exception
(BLE001)
178-178: Consider moving this statement to an else block
(TRY300)
179-179: Do not catch blind exception: Exception
(BLE001)
202-203: try-except-pass detected, consider logging the exception
(S110)
202-202: Do not catch blind exception: Exception
(BLE001)
241-241: Do not catch blind exception: Exception
(BLE001)
src/nat/observability/processor/cross_workflow_processor.py
107-107: Redundant exception object included in logging.exception call
(TRY401)
168-168: Redundant exception object included in logging.exception call
(TRY401)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
Show resolved
Hide resolved
Signed-off-by: pira998 <[email protected]>
Signed-off-by: pira998 <[email protected]>
pira998
force-pushed
the
feature/cross-workflow-observability
branch
from
184ef8a to
9012d26
Compare
There was a problem hiding this comment.
Actionable comments posted: 0
♻️ Duplicate comments (4)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py (2)
78-82: Use the public API to propagate observability context.The code mutates the private
workflow._context_state.observability_contextdirectly, butworkflow.run()resets the ContextVar from itsobservability_contextparameter. This means the manually set context is lost and cross-workflow data never reaches the runner.Apply this diff to use the public API:
- # Set observability context in the workflow's context state - workflow._context_state.observability_context.set(query_context) - - # Execute workflow - async with workflow.run(query) as runner: + # Execute workflow with propagated observability context + async with workflow.run(query, observability_context=query_context) as runner:
1-1: Make the file executable or remove the shebang.The shebang is present but the file lacks execute permissions.
Either mark it executable (
chmod +x) or remove the shebang line if direct execution isn't intended.examples/observability/cross_workflow_tracking/integrated_example.py (2)
80-86: Use the public API to propagate observability context.The code directly mutates
workflow._context_state.observability_context, butworkflow.run()resets the ContextVar based on itsobservability_contextparameter. PassingNone(the default) wipes the value you just set, so spans emitted during execution won't carry the cross-workflow data.Apply this diff:
async with load_workflow(config_file) as workflow: - # Set observability context in the workflow's context state - context_state = workflow._context_state - context_state.observability_context.set(obs_context) - - # Run the workflow without the observability_context parameter - async with workflow.run(input_data) as runner: + # Run the workflow with the propagated observability context + async with workflow.run(input_data, observability_context=obs_context) as runner:
1-1: Make the file executable or remove the shebang.The shebang is present but the file lacks execute permissions.
🧹 Nitpick comments (3)
src/nat/observability/workflow_utils.py (1)
177-180: Consider moving the return to an else block.The return statement at line 178 follows an except block without re-raising. Moving it into an
elseblock would make the control flow clearer and satisfy the TRY300 guideline.Apply this diff:
try: # Import here to avoid circular import from nat.builder.context import Context current_context = Context.get() - return current_context.observability_context + obs = current_context.observability_context except Exception: return None + else: + return obssrc/nat/observability/processor/cross_workflow_processor.py (1)
105-108: Remove redundant exception object from logger.exception calls.
logger.exception()automatically captures the exception's stack trace, so including the exception object in the message format string (%s", e) is redundant and triggers TRY401.Apply this diff:
except (AttributeError, KeyError, TypeError, ValueError) as e: # If there's any error in processing, log it but don't fail the span - logger.exception("Error processing cross-workflow observability data: %s", e) + logger.exception("Error processing cross-workflow observability data") item.set_attribute("observability.processing_error", str(e))And:
except (AttributeError, IndexError, TypeError) as e: # If there's any error in processing, log it but don't fail the span - logger.exception("Error processing workflow relationship data: %s", e) + logger.exception("Error processing workflow relationship data") item.set_attribute("relationship.processing_error", str(e))Also applies to: 166-169
examples/observability/cross_workflow_tracking/integrated_example.py (1)
98-98: Consider moving return statements to else blocks.The return statements at lines 98 and 329 follow except blocks without re-raising. Moving them into
elseblocks would make the control flow clearer and satisfy the TRY300 guideline.For line 98, apply this diff:
try: # Load and run the actual NAT workflow async with load_workflow(config_file) as workflow: # ... + result = None # Update completion status if current_workflow: current_workflow.end_time = time.time() current_workflow.status = "completed" if current_workflow.start_time: duration = current_workflow.end_time - current_workflow.start_time print(f" - Duration: {duration:.3f}s") print(f" - Result: {result}") - - return result - except Exception as e: # Update error status if current_workflow: current_workflow.end_time = time.time() current_workflow.status = "failed" current_workflow.tags["error"] = str(e) print(f" - Error: {e}") raise + else: + return resultSimilar pattern for line 329 in the nested function.
Also applies to: 329-329
📜 Review details
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (7)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py(1 hunks)examples/observability/cross_workflow_tracking/example.py(1 hunks)examples/observability/cross_workflow_tracking/integrated_example.py(1 hunks)examples/observability/cross_workflow_tracking/observability_config.yml(1 hunks)external/nat-ui(1 hunks)src/nat/observability/processor/cross_workflow_processor.py(1 hunks)src/nat/observability/workflow_utils.py(1 hunks)
🚧 Files skipped from review as they are similar to previous changes (2)
- external/nat-ui
- examples/observability/cross_workflow_tracking/observability_config.yml
🧰 Additional context used
📓 Path-based instructions (7)
**/*.{py,yaml,yml}
📄 CodeRabbit inference engine (.cursor/rules/nat-test-llm.mdc)
**/*.{py,yaml,yml}: Configure response_seq as a list of strings; values cycle per call, and [] yields an empty string.
Configure delay_ms to inject per-call artificial latency in milliseconds for nat_test_llm.
Files:
src/nat/observability/workflow_utils.pyexamples/observability/cross_workflow_tracking/integrated_example.pyexamples/observability/cross_workflow_tracking/example.pysrc/nat/observability/processor/cross_workflow_processor.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
**/*.py
📄 CodeRabbit inference engine (.cursor/rules/nat-test-llm.mdc)
**/*.py: Programmatic use: create TestLLMConfig(response_seq=[...], delay_ms=...), add with builder.add_llm("", cfg).
When retrieving the test LLM wrapper, use builder.get_llm(name, wrapper_type=LLMFrameworkEnum.) and call the framework’s method (e.g., ainvoke, achat, call).
**/*.py: In code comments/identifiers use NAT abbreviations as specified: nat for API namespace/CLI, nvidia-nat for package name, NAT for env var prefixes; do not use these abbreviations in documentation
Follow PEP 20 and PEP 8; run yapf with column_limit=120; use 4-space indentation; end files with a single trailing newline
Run ruff check --fix as linter (not formatter) using pyproject.toml config; fix warnings unless explicitly ignored
Respect naming: snake_case for functions/variables, PascalCase for classes, UPPER_CASE for constants
Treat pyright warnings as errors during development
Exception handling: use bare raise to re-raise; log with logger.error() when re-raising to avoid duplicate stack traces; use logger.exception() when catching without re-raising
Provide Google-style docstrings for every public module, class, function, and CLI command; first line concise and ending with a period; surround code entities with backticks
Validate and sanitize all user input, especially in web or CLI interfaces
Prefer httpx with SSL verification enabled by default and follow OWASP Top-10 recommendations
Use async/await for I/O-bound work; profile CPU-heavy paths with cProfile or mprof before optimizing; cache expensive computations with functools.lru_cache or external cache; leverage NumPy vectorized operations when beneficial
Files:
src/nat/observability/workflow_utils.pyexamples/observability/cross_workflow_tracking/integrated_example.pyexamples/observability/cross_workflow_tracking/example.pysrc/nat/observability/processor/cross_workflow_processor.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
src/**/*.py
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
All importable Python code must live under src/ (or packages//src/)
Files:
src/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
src/nat/**/*
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
Changes in src/nat should prioritize backward compatibility
Files:
src/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
⚙️ CodeRabbit configuration file
This directory contains the core functionality of the toolkit. Changes should prioritize backward compatibility.
Files:
src/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
{src/**/*.py,packages/*/src/**/*.py}
📄 CodeRabbit inference engine (.cursor/rules/general.mdc)
All public APIs must have Python 3.11+ type hints on parameters and return values; prefer typing/collections.abc abstractions; use typing.Annotated when useful
Files:
src/nat/observability/workflow_utils.pysrc/nat/observability/processor/cross_workflow_processor.py
**/*
⚙️ CodeRabbit configuration file
**/*: # Code Review Instructions
- Ensure the code follows best practices and coding standards. - For Python code, follow
PEP 20 and
PEP 8 for style guidelines.- Check for security vulnerabilities and potential issues. - Python methods should use type hints for all parameters and return values.
Example:def my_function(param1: int, param2: str) -> bool: pass- For Python exception handling, ensure proper stack trace preservation:
- When re-raising exceptions: use bare
raisestatements to maintain the original stack trace,
and uselogger.error()(notlogger.exception()) to avoid duplicate stack trace output.- When catching and logging exceptions without re-raising: always use
logger.exception()
to capture the full stack trace information.Documentation Review Instructions - Verify that documentation and comments are clear and comprehensive. - Verify that the documentation doesn't contain any TODOs, FIXMEs or placeholder text like "lorem ipsum". - Verify that the documentation doesn't contain any offensive or outdated terms. - Verify that documentation and comments are free of spelling mistakes, ensure the documentation doesn't contain any
words listed in the
ci/vale/styles/config/vocabularies/nat/reject.txtfile, words that might appear to be
spelling mistakes but are listed in theci/vale/styles/config/vocabularies/nat/accept.txtfile are OK.Misc. - All code (except .mdc files that contain Cursor rules) should be licensed under the Apache License 2.0,
and should contain an Apache License 2.0 header comment at the top of each file.
- Confirm that copyright years are up-to date whenever a file is changed.
Files:
src/nat/observability/workflow_utils.pyexamples/observability/cross_workflow_tracking/integrated_example.pyexamples/observability/cross_workflow_tracking/example.pysrc/nat/observability/processor/cross_workflow_processor.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
examples/**/*
⚙️ CodeRabbit configuration file
examples/**/*: - This directory contains example code and usage scenarios for the toolkit, at a minimum an example should
contain a README.md or file README.ipynb.
- If an example contains Python code, it should be placed in a subdirectory named
src/and should
contain apyproject.tomlfile. Optionally, it might also contain scripts in ascripts/directory.- If an example contains YAML files, they should be placed in a subdirectory named
configs/. - If an example contains sample data files, they should be placed in a subdirectory nameddata/, and should
be checked into git-lfs.
Files:
examples/observability/cross_workflow_tracking/integrated_example.pyexamples/observability/cross_workflow_tracking/example.pyexamples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
🧠 Learnings (2)
📚 Learning: 2025-09-09T20:32:39.016Z
Learnt from: CR
PR: NVIDIA/NeMo-Agent-Toolkit#0
File: .cursor/rules/nat-test-llm.mdc:0-0
Timestamp: 2025-09-09T20:32:39.016Z
Learning: When stubbing deterministic LLM responses in tests/workflows, use the NAT Test LLM (nat_test_llm).
Applied to files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
📚 Learning: 2025-09-09T20:32:39.016Z
Learnt from: CR
PR: NVIDIA/NeMo-Agent-Toolkit#0
File: .cursor/rules/nat-test-llm.mdc:0-0
Timestamp: 2025-09-09T20:32:39.016Z
Learning: Applies to **/*.{yaml,yml} : In workflow/config YAML, set llms.<name>._type: nat_test_llm to stub responses.
Applied to files:
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
🧬 Code graph analysis (5)
src/nat/observability/workflow_utils.py (4)
src/nat/runtime/runner.py (5)
context(90-91)result(130-131)result(134-135)result(137-168)result_stream(170-200)src/nat/observability/context.py (6)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)src/nat/builder/workflow.py (3)
Workflow(42-169)run(98-115)result_with_steps(117-134)src/nat/builder/context.py (6)
Context(122-329)get(118-119)get(319-329)observability_context(114-115)observability_context(296-307)set_observability_context(309-316)
examples/observability/cross_workflow_tracking/integrated_example.py (6)
src/nat/runtime/loader.py (1)
load_workflow(97-117)src/nat/runtime/runner.py (4)
context(90-91)result(130-131)result(134-135)result(137-168)src/nat/observability/context.py (10)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)get_workflow_depth(102-104)add_attribute(94-96)to_dict(106-122)from_dict(125-141)src/nat/builder/context.py (6)
Context(122-329)ContextState(66-119)observability_context(114-115)observability_context(296-307)get(118-119)get(319-329)src/nat/observability/processor/cross_workflow_processor.py (4)
CrossWorkflowProcessor(25-110)WorkflowRelationshipProcessor(113-171)process(35-110)process(122-171)src/nat/builder/workflow.py (1)
run(98-115)
examples/observability/cross_workflow_tracking/example.py (2)
src/nat/observability/context.py (8)
ObservabilityContext(39-141)create_child_context(69-84)create_child_context(173-181)create_root_context(54-67)create_root_context(166-170)get_current_workflow(98-100)get_workflow_depth(102-104)add_attribute(94-96)src/nat/builder/workflow.py (1)
run(98-115)
src/nat/observability/processor/cross_workflow_processor.py (3)
src/nat/observability/processor/processor.py (1)
Processor(27-74)src/nat/utils/type_utils.py (1)
override(56-57)src/nat/builder/context.py (5)
Context(122-329)get(118-119)get(319-329)observability_context(114-115)observability_context(296-307)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py (5)
src/nat/runtime/runner.py (4)
context(90-91)result(130-131)result(134-135)result(137-168)src/nat/observability/context.py (9)
ObservabilityContext(39-141)create_root_context(54-67)create_root_context(166-170)add_attribute(94-96)create_child_context(69-84)create_child_context(173-181)get_workflow_depth(102-104)to_dict(106-122)from_dict(125-141)src/nat/runtime/loader.py (1)
load_workflow(97-117)src/nat/builder/context.py (2)
observability_context(114-115)observability_context(296-307)src/nat/builder/workflow.py (1)
run(98-115)
🪛 Ruff (0.13.2)
src/nat/observability/workflow_utils.py
79-79: Do not catch blind exception: Exception
(BLE001)
140-140: Do not catch blind exception: Exception
(BLE001)
178-178: Consider moving this statement to an else block
(TRY300)
179-179: Do not catch blind exception: Exception
(BLE001)
202-203: try-except-pass detected, consider logging the exception
(S110)
202-202: Do not catch blind exception: Exception
(BLE001)
241-241: Do not catch blind exception: Exception
(BLE001)
examples/observability/cross_workflow_tracking/integrated_example.py
1-1: Shebang is present but file is not executable
(EXE001)
98-98: Consider moving this statement to an else block
(TRY300)
302-302: Do not catch blind exception: Exception
(BLE001)
329-329: Consider moving this statement to an else block
(TRY300)
330-330: Do not catch blind exception: Exception
(BLE001)
src/nat/observability/processor/cross_workflow_processor.py
107-107: Redundant exception object included in logging.exception call
(TRY401)
168-168: Redundant exception object included in logging.exception call
(TRY401)
examples/observability/cross_workflow_tracking/cross_workflow_tracking_example.py
1-1: Shebang is present but file is not executable
(EXE001)
🔇 Additional comments (3)
src/nat/observability/workflow_utils.py (1)
79-80: Blind Exception catches are justified here.The static analysis tool flags these as BLE001 violations, but all instances correctly implement graceful fallback patterns where catching broad
Exceptionis appropriate:
- Lines 79-80, 140-141, 241-242: Defensive fallback to create a root context when
Context.get()or accessingobservability_contextfails- Lines 179-180: Returns
Nonewhen the execution context is unavailable- Lines 202-203: Silent pass for optional context mutation that may not be available
These patterns enable the observability utilities to work in contexts where the NAT execution environment may not be fully initialized, which is essential for cross-workflow propagation.
Also applies to: 140-141, 179-180, 202-203, 241-242
examples/observability/cross_workflow_tracking/integrated_example.py (1)
302-307: Broad Exception catches are acceptable in example code.The static analysis tool flags these as BLE001 violations, but for demonstration/example code, catching broad
Exceptionto display error messages and continue the example flow is a reasonable pattern. The examples are designed to be resilient and educational rather than production-grade.Also applies to: 329-331
examples/observability/cross_workflow_tracking/example.py (1)
1-283: LGTM! Example demonstrates cross-workflow observability well.The mock workflow orchestration correctly:
- Creates and propagates
ObservabilityContexthierarchically- Uses context to track workflow chains, depths, and timing
- Demonstrates sequential, parallel, and nested workflow patterns
- Shows practical usage of context serialization
The conditional logic for workflow name matching (lines 102-109) is now correct after the previous review's fix. No new issues found.
Closes #760
Implement cross-workflow observability with context propagation and span processing
Fixes
This PR introduces a comprehensive cross-workflow
observability feature that enables tracking and monitoring
execution across multiple interconnected NAT workflows.
Key Features
for trace propagation across workflow boundaries
with workflow hierarchy and relationship data
processing with context preservation
observability integration
for observability support
Implementation Details
Core Components
ObservabilityContextclass for trace ID management andworkflow hierarchy tracking
CrossWorkflowProcessorandWorkflowRelationshipProcessorfor span enhancementpatterns
Observability Data Added
Files Changed
Core Implementation
src/nat/observability/context.py- NewObservabilityContext implementation
src/nat/observability/processor/cross_workflow_processor .py- Span processorssrc/nat/observability/workflow_utils.py- Utilityfunctions
src/nat/builder/context.py- Context integrationsrc/nat/builder/workflow.py- Workflow integrationsrc/nat/runtime/runner.py- Runtime integrationsrc/nat/observability/__init__.py- Module exportsExamples and Documentation
examples/observability/cross_workflow_tracking/-Complete example suite
tests/nat/observability/test_cross_workflow_observabilit y.py- Test coverageTesting
Comprehensive test suite covers:
Usage
This feature enables end-to-end visibility into complex workflow chains, performance analysis, and debugging across workflow boundaries.
Summary by CodeRabbit
New Features
Documentation
Examples
Tests