feat(jobs): add watch command to stream job events - #1031
Open
ironcommit wants to merge 1 commit into
Open
Conversation
Signed-off-by: Ryan S <267728323+ironcommit@users.noreply.github.com>
Contributor
![github-advanced-security[bot]](https://avatars.githubusercontent.com/in/57789?s=60&v=4){kind=small}
Contributor
📝 WalkthroughWalkthroughThe PR adds synchronous and asynchronous job watching with typed events, log pagination, retries, timeouts, CLI rendering, generated command support, tests, and documentation. ChangesJob Watch
Sequence Diagram(s)sequenceDiagram
participant CLI
participant NeMoPlatform
participant watch_job
participant JobsClient
CLI->>NeMoPlatform: watch_job(name, filters, timeout, polling)
NeMoPlatform->>watch_job: delegate watch request
watch_job->>JobsClient: poll status and fetch paginated logs
JobsClient-->>watch_job: status and log responses
watch_job-->>NeMoPlatform: JobWatchEvent stream
NeMoPlatform-->>CLI: rendered event input
Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 4
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/code_generator.py (1)
156-203: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winFix: string-replace corrupts generated code when the resource name contains
--wait.
_render_lifecycle_codebuilds the prelude with the literal text--wait, then doesprelude.replace("--wait", "--watch")forplatform_job. This replace is not scoped to the error message. It also rewrites any occurrence of--waitinside the interpolated resource-name literal. If a job'snamecontains the substring--wait(for examplenightly--wait-job), the generated code's quoted name literal is silently corrupted tonightly--watch-job.Compute the flag name once from
lifecycle_typeand interpolate it directly instead of doing a blind substring replace after the fact.🐛 Proposed fix
resource_name = 'getattr(response, "name", None)' if args.get("name") is not None: resource_name = f"{resource_name} or {_format_python_literal(args['name'])}" + flag_name = "--watch" if lifecycle_type == "platform_job" else "--wait" prelude = dedent( f""" resource_name = {resource_name} if not resource_name: - raise RuntimeError("Unable to determine created resource name for --wait") + raise RuntimeError("Unable to determine created resource name for {flag_name}") """ ).strip() if lifecycle_type == "inference_deployment": ... if lifecycle_type == "platform_job": - prelude = prelude.replace("--wait", "--watch") return "\n\n".join( [ prelude, _render_platform_job_watch_code(args, timeout, poll_interval), ] ) raise ValueError(f"Unsupported lifecycle config type: {lifecycle_type!r}")🤖 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 `@packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/code_generator.py` around lines 156 - 203, Update _render_lifecycle_code to compute the lifecycle flag name from lifecycle_type before constructing prelude, using --watch for platform_job and --wait otherwise, then interpolate that value into the error message. Remove the prelude.replace("--wait", "--watch") call so resource-name literals containing --wait remain unchanged.tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/templates/create_command.py.j2 (1)
32-41: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winReject methods that configure both
waitandwatch.Both configurations are accepted independently. The template then emits duplicate
timeoutandpoll_intervalparameters, causing a PythonSyntaxError.🤖 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 `@tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/templates/create_command.py.j2` around lines 32 - 41, Update the template conditions around the watch_config and wait_config blocks so generated commands cannot enable both configurations simultaneously. Preserve each configuration’s existing parameters when used alone, and ensure timeout and poll_interval are emitted only once to avoid duplicate function parameters.
🧹 Nitpick comments (1)
packages/nemo_platform_ext/src/nemo_platform_ext/client/enhanced.py (1)
199-228: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winMove the function-local imports to module scope.
Line 28 already imports from
nemo_platform_ext.jobs.watchat module level, so the deferred imports on Line 213 and Line 408 cannot break an import cycle. Importwatch_jobandasync_watch_jobnormally.Also add docstrings; every other public method on this client is documented.
As per coding guidelines: "prefer concrete type hints over string-based type hints, and do not import those types only under `TYPE_CHECKING`; import them normally when possible."♻️ Proposed change
-from nemo_platform_ext.jobs.watch import JobWatchEvent +from nemo_platform_ext.jobs.watch import JobWatchEvent, async_watch_job, watch_job) -> Iterator[JobWatchEvent]: - from nemo_platform_ext.jobs.watch import watch_job - return watch_job(🤖 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 `@packages/nemo_platform_ext/src/nemo_platform_ext/client/enhanced.py` around lines 199 - 228, Move watch_job and async_watch_job imports from their methods to module scope alongside the existing nemo_platform_ext.jobs.watch import, then remove the function-local imports. Add concise docstrings to the public watch_job and async_watch_job client methods while preserving their current delegation and parameters.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py`:
- Around line 312-334: Update _drain_logs at
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py:312-334 to track
visited cursors, return when _next_page produces a cursor already seen or
otherwise fails to advance, and enforce the deadline on every loop iteration.
Apply the same cursor-advance guard and per-iteration deadline check in
_async_drain_logs at
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py:352-374.
- Around line 126-150: Update the polling loop around _drain_logs to persist and
reuse the latest next_page cursor across cycles instead of resetting to the
original page_cursor. When the server invalidates that cursor, fall back to the
existing full re-scan behavior; otherwise advance the cursor after each
successful drain while preserving terminal-status handling.
- Around line 126-146: Update the log-draining flow around _drain_logs so
history_seen becomes true immediately after the first page is recorded,
including when pagination later raises a transient error. Preserve suppression
of pre-existing history for include_history=False while allowing newly fetched
lines to emit on the retry, and remove reliance on setting history_seen only
after the entire drain succeeds.
- Around line 102-105: Make both watcher entry points validate eagerly by
converting watch_job at
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py:102-105 and
async_watch_job at
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py:168-171 into
non-generator wrappers. Keep the poll_interval checks and _sync_jobs_client or
_async_jobs_client resolution in each wrapper, then return inner generator or
async-generator functions containing the existing iteration logic.
---
Outside diff comments:
In `@packages/nemo_platform_ext/src/nemo_platform_ext/cli/core/code_generator.py`:
- Around line 156-203: Update _render_lifecycle_code to compute the lifecycle
flag name from lifecycle_type before constructing prelude, using --watch for
platform_job and --wait otherwise, then interpolate that value into the error
message. Remove the prelude.replace("--wait", "--watch") call so resource-name
literals containing --wait remain unchanged.
In
`@tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/templates/create_command.py.j2`:
- Around line 32-41: Update the template conditions around the watch_config and
wait_config blocks so generated commands cannot enable both configurations
simultaneously. Preserve each configuration’s existing parameters when used
alone, and ensure timeout and poll_interval are emitted only once to avoid
duplicate function parameters.
---
Nitpick comments:
In `@packages/nemo_platform_ext/src/nemo_platform_ext/client/enhanced.py`:
- Around line 199-228: Move watch_job and async_watch_job imports from their
methods to module scope alongside the existing nemo_platform_ext.jobs.watch
import, then remove the function-local imports. Add concise docstrings to the
public watch_job and async_watch_job client methods while preserving their
current delegation and parameters.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Enterprise
Run ID: 75f56093-22d4-4101-9ce3-7b89b2a6a52a
⛔ Files ignored due to path filters (11)
sdk/python/nemo-platform/src/nemo_platform/_client.pyis excluded by!sdk/**sdk/python/nemo-platform/src/nemo_platform/cli/commands/api/jobs/__init__.pyis excluded by!sdk/**sdk/python/nemo-platform/src/nemo_platform/cli/core/code_generator.pyis excluded by!sdk/**sdk/python/nemo-platform/src/nemo_platform/cli/core/job_watch_renderer.pyis excluded by!sdk/**sdk/python/nemo-platform/src/nemo_platform/jobs/watch.pyis excluded by!sdk/**sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/commands/test_create_wait.pyis excluded by!sdk/**sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/core/test_code_generator.pyis excluded by!sdk/**sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/core/test_job_watch_renderer.pyis excluded by!sdk/**sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/cli/test_app.pyis excluded by!sdk/**sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/jobs/__init__.pyis excluded by!sdk/**sdk/python/nemo-platform/tests/vendored/nemo_platform_ext/jobs/test_watch.pyis excluded by!sdk/**
📒 Files selected for processing (20)
docs/cli/reference.mdxpackages/nemo_platform_ext/src/nemo_platform_ext/cli/commands/api/jobs/__init__.pypackages/nemo_platform_ext/src/nemo_platform_ext/cli/core/code_generator.pypackages/nemo_platform_ext/src/nemo_platform_ext/cli/core/job_watch_renderer.pypackages/nemo_platform_ext/src/nemo_platform_ext/client/enhanced.pypackages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.pypackages/nemo_platform_ext/tests/cli/commands/test_create_wait.pypackages/nemo_platform_ext/tests/cli/core/test_code_generator.pypackages/nemo_platform_ext/tests/cli/core/test_job_watch_renderer.pypackages/nemo_platform_ext/tests/cli/test_app.pypackages/nemo_platform_ext/tests/jobs/test_watch.pytools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/cli_config.yamltools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/config.pytools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/context_collectors/create_collector.pytools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/overrides/jobs/watch.pytools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/cli_generator/templates/create_command.py.j2tools/nemo-platform-sdk-tools/src/nemo_platform_sdk_tools/sdk/vendor/vendor_package.pytools/nemo-platform-sdk-tools/tests/sdk/cli_generator/test_config.pytools/nemo-platform-sdk-tools/tests/sdk/cli_generator/test_generator.pytools/nemo-platform-sdk-tools/tests/sdk/vendor/test_vendor_package.py
Comment on lines
+102
to
+105
| if poll_interval < 0: | ||
| raise ValueError("poll_interval must be greater than or equal to 0") | ||
|
|
||
| jobs = _sync_jobs_client(client) |
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Argument validation is deferred in both watchers. Both functions are generators, so the poll_interval check and the client resolution run on first iteration rather than at call time. NeMoPlatform.watch_job and AsyncNeMoPlatform.watch_job hand the iterator to callers, so invalid arguments surface far from the call site. Split each into an eager wrapper plus an inner generator.
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py#L102-L105: move thepoll_intervalcheck and_sync_jobs_clientcall into a non-generatorwatch_jobthat returns an inner generator.packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py#L168-L171: move thepoll_intervalcheck and_async_jobs_clientcall into a non-generatorasync_watch_jobthat returns an inner async generator.
📍 Affects 1 file
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py#L102-L105(this comment)packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py#L168-L171
🤖 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 `@packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py` around lines
102 - 105, Make both watcher entry points validate eagerly by converting
watch_job at
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py:102-105 and
async_watch_job at
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py:168-171 into
non-generator wrappers. Keep the poll_interval checks and _sync_jobs_client or
_async_jobs_client resolution in each wrapper, then return inner generator or
async-generator functions containing the existing iteration logic.
Comment on lines
+126
to
+146
| logs_drained = False | ||
| try: | ||
| yield from _drain_logs( | ||
| jobs, | ||
| name, | ||
| workspace=workspace, | ||
| seen_logs=seen_logs, | ||
| emit=history_seen, | ||
| attempt_id=attempt_id, | ||
| step_id=step_id, | ||
| task_id=task_id, | ||
| limit=limit, | ||
| page_cursor=page_cursor, | ||
| ) | ||
| history_seen = True | ||
| logs_drained = True | ||
| except Exception as exc: | ||
| if not _is_transient_error(exc): | ||
| raise | ||
| yield JobWarningEvent(kind="warning", job_name=name, message=f"Transient log check failed: {exc}") | ||
|
|
Contributor
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
A transient log failure can silently drop new log lines when include_history=False.
If the first drain fails part way through pagination, the fetched pages are already recorded in seen_logs but were not emitted. history_seen stays False, so the next drain also suppresses output while marking everything seen. Log lines produced between the two drains are never emitted.
Set the history baseline as soon as the first page is recorded, instead of only after a fully successful drain.
🤖 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 `@packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py` around lines
126 - 146, Update the log-draining flow around _drain_logs so history_seen
becomes true immediately after the first page is recorded, including when
pagination later raises a transient error. Preserve suppression of pre-existing
history for include_history=False while allowing newly fetched lines to emit on
the retry, and remove reliance on setting history_seen only after the entire
drain succeeds.
Comment on lines
+126
to
+150
| logs_drained = False | ||
| try: | ||
| yield from _drain_logs( | ||
| jobs, | ||
| name, | ||
| workspace=workspace, | ||
| seen_logs=seen_logs, | ||
| emit=history_seen, | ||
| attempt_id=attempt_id, | ||
| step_id=step_id, | ||
| task_id=task_id, | ||
| limit=limit, | ||
| page_cursor=page_cursor, | ||
| ) | ||
| history_seen = True | ||
| logs_drained = True | ||
| except Exception as exc: | ||
| if not _is_transient_error(exc): | ||
| raise | ||
| yield JobWarningEvent(kind="warning", job_name=name, message=f"Transient log check failed: {exc}") | ||
|
|
||
| if status_event.terminal and logs_drained: | ||
| return | ||
|
|
||
| _sleep(poll_interval, deadline, name) |
Contributor
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Each poll re-reads the entire log history.
_drain_logs restarts from the original page_cursor on every poll cycle. Total work grows quadratically with job duration, and seen_logs retains one key per log line for the whole watch. A long job with high log volume will spend most of its time re-fetching and re-hashing already-seen pages.
Carry the last observed next_page cursor forward between poll cycles, and keep the full re-scan only when the server invalidates the cursor.
🤖 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 `@packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py` around lines
126 - 150, Update the polling loop around _drain_logs to persist and reuse the
latest next_page cursor across cycles instead of resetting to the original
page_cursor. When the server invalidates that cursor, fall back to the existing
full re-scan behavior; otherwise advance the cursor after each successful drain
while preserving terminal-status handling.
Comment on lines
+312
to
+334
| while True: | ||
| query_params = _log_query_params( | ||
| attempt_id=attempt_id, | ||
| step_id=step_id, | ||
| task_id=task_id, | ||
| limit=limit, | ||
| page_cursor=current_cursor, | ||
| ) | ||
| page_response = jobs.list_job_logs(workspace=workspace, name=name, query_params=query_params) | ||
| page = page_response.page() | ||
| for log in page.items: | ||
| event = _log_event(log, name) | ||
| key = _log_key(log) | ||
| occurrence_counts[key] = occurrence_counts.get(key, 0) + 1 | ||
| if occurrence_counts[key] > seen_logs.get(key, 0): | ||
| seen_logs[key] = occurrence_counts[key] | ||
| if emit: | ||
| yield event | ||
|
|
||
| next_cursor = _next_page(getattr(page, "metadata", None)) | ||
| if next_cursor is None: | ||
| return | ||
| current_cursor = next_cursor |
Contributor
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
Both log drain loops can run forever. Each loop exits only when _next_page returns None. A server that repeats the same next_page cursor produces an unbounded loop. The deadline is enforced only in _sleep, which these loops never reach, so the watch hangs with no timeout.
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py#L312-L334: track visited cursors in_drain_logs, return when the cursor does not advance, and check the deadline each iteration.packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py#L352-L374: apply the same cursor-advance guard and deadline check in_async_drain_logs.
📍 Affects 1 file
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py#L312-L334(this comment)packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py#L352-L374
🤖 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 `@packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py` around lines
312 - 334, Update _drain_logs at
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py:312-334 to track
visited cursors, return when _next_page produces a cursor already seen or
otherwise fails to advance, and enforce the deadline on every loop iteration.
Apply the same cursor-advance guard and per-iteration deadline check in
_async_drain_logs at
packages/nemo_platform_ext/src/nemo_platform_ext/jobs/watch.py:352-374.
Contributor
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary by CodeRabbit
jobs watchto monitor job status, logs, warnings, progress, and completion.--watchbehavior when creating jobs.