feat(jobs): make NATS queue activity visible in async ML job logs

[mihow]

Summary

Closes #1220. Makes NATS queue lifecycle events visible in the per-job log that users see in the UI, so async ML jobs no longer have a silent gap where their NATS activity used to be invisible.

TaskQueueManager now accepts an optional job_logger. When set, lifecycle events are mirrored to both the module logger (unchanged — still feeds stdout / container logs / observability pipelines) and the per-job logger:

Setup

• Created NATS stream job_<id> / Reusing NATS stream job_<id> (messages=N, last_seq=N)
• Created NATS consumer job-<id>-consumer (max_deliver=5, ack_wait=30.0s, max_ack_pending=1000, deliver_policy=all, ack_policy=explicit) — config snapshot read from the ConsumerInfo returned by the server
• Reusing NATS consumer job-<id>-consumer (delivered=N ack_floor=N num_pending=N num_ack_pending=N num_redelivered=N)

Cleanup (forensic)

• Finalizing NATS consumer job-<id>-consumer before deletion (delivered=N ack_floor=N num_pending=N num_ack_pending=N num_redelivered=N) — logged before the delete calls. This is the single most useful line for post-mortem investigations of jobs that ended in a weird state; without it, the consumer is already gone by the time anyone asks.
• Deleted NATS consumer job-<id>-consumer / Deleted NATS stream job_<id> and matching failure lines

Error paths

• Failed to publish task to stream for job '<id>': <error> — previously only on the module logger, now also on the job logger with traceback preserved on the module side via exc_info=True

What's deliberately NOT logged to job_logger

Per-message and per-poll success events stay on the module logger only — a 10k-image job would otherwise drown its own log:

• publish_task success (one per image)
• reserve_tasks (one per worker poll, every few seconds)
• acknowledge_task success (one per result)

Per-image queueing is routed through manager.log_async at DEBUG level, so it's available in the job log when debug logging is intentionally enabled (off by default). These might be worth surfacing more permanently when we can identify each processing service (#1194).

Implementation notes

Async bridging. Logging from inside the async block goes through the manager's log_async(level, msg, exc_info=False) helper — one entry point that:

1. Always emits to the module logger (synchronous, fine on the event loop)
2. Mirrors to job.logger via sync_to_async when one was passed in

The bridge is needed because Django's JobLogHandler.emit() does ORM writes (refresh_from_db + save), which are forbidden from the event loop. Without it, every lifecycle line was silently dropped — the handler's broad except swallowed SynchronousOnlyOperation and routed a warning to the module logger that nobody was watching.

Centralizing this in log_async means the bridge lives in one place instead of being re-implemented at every call site (an earlier draft of this PR had three different logging styles in the publish loop alone — module-direct, module-with-traceback, and a hand-rolled sync_to_async(job.logger.error) bridge).

Future intent — route, not mirror. log_async currently fans granular per-job lifecycle out to both loggers. The longer-term preference (documented inline in the log_async docstring) is to route — granular lifecycle lives on job.logger only, matching the convention in ami.jobs.tasks.save_results and friends, where job.logger.propagate = False and per-job detail is kept out of stdout / NR. The module logger then becomes a clean ops channel for true infra signals (connection failures, NATS-side errors) plus an automatic mirror at WARNING+ so error signals always reach ops dashboards. Kept symmetric for now because async ML processing is still being stabilized and the extra stdout visibility is actively helping debug. Switching the fan-out direction is a follow-up once the per-job UI log is trusted as the canonical inspection surface.

Dedup and early-return. _ensure_stream and _ensure_consumer are called on every publish_task (once per image). After the first call per manager session, subsequent calls skip the NATS round-trip entirely via an early return keyed on two set[int] fields (_streams_logged, _consumers_logged).

Concurrency note: Job.cancel() can trigger cleanup_async_job_resources (running in the request thread) while queue_images_to_nats is still in its publish loop (running in the Celery worker), so the stream/consumer can be deleted mid-flight from a different manager session. The early-return stays safe in that case — subsequent publishes in the queue loop fail loudly (publish_task returns False and logs an ERROR) rather than silently recreating an orphan stream without a consumer. That's actually better than the non-deduped baseline behavior, where _ensure_stream would silently recreate the stream on every iteration and the subsequent publishes would succeed to a stream that no worker will ever consume from. Not exercised by a concurrent-cancel test yet; worth adding if we lean harder on this claim.

Exception narrowing. _ensure_consumer previously caught broad Exception on the consumer_info() call, masking auth/API/transient errors as "consumer missing." Narrowed to nats.js.errors.NotFoundError to match the pattern already used in _ensure_stream and _stream_exists.

Caller wiring. cleanup_async_job_resources(job_id) now takes just job_id and resolves the Job (and its per-job logger) internally — matching the save_results pattern in ami/jobs/tasks.py. If the Job row is gone (e.g. the Job.DoesNotExist path in _fail_job), it falls back to the module logger. queue_images_to_nats passes job.logger directly to TaskQueueManager. Callers without a job context (_ack_task_via_nats, the worker-poll view, the DLQ management command) construct TaskQueueManager() without a logger.

Tests

Added TestTaskQueueManagerJobLogger in ami/ml/orchestration/tests/test_nats_queue.py (7 new tests):

1. First publish logs stream + consumer creation to job_logger, including config snapshot
2. Second publish in the same session emits no new lifecycle lines (dedup)
3. Stream/consumer reuse logs include state and stats
4. cleanup_job_resources logs the forensic snapshot before delete calls
5. cleanup_job_resources tolerates a missing consumer without raising
6. publish_task failures surface on job_logger at ERROR level
7. TaskQueueManager() without a job_logger still works (module logger fallback)

Tests use a real logging.Logger with a capture handler (not MagicMock), so the sync_to_async bridge in log_async() is exercised on the real code path. All 18 tests in test_nats_queue.py pass. Not covered: concurrent-cancel-during-queue (see Dedup and early-return above).

Manual verification

Deployed to a staging environment and ran async_api jobs end-to-end against real ML workers.

Small smoke test (18 images) — two jobs completed the full lifecycle. A third was canceled mid-flight; its UI log showed:

Finalizing NATS consumer job-<id>-consumer before deletion
  (delivered=50 ack_floor=0 num_pending=0 num_ack_pending=16 num_redelivered=16)

Larger test (636 images) — canceled after ~16 minutes (staging server backlogged on result ingestion). The forensic snapshot:

Queuing 636 images to NATS stream for job '<id>'
Created NATS stream job_<id>
Created NATS consumer job-<id>-consumer (max_deliver=5, ack_wait=30s, ...)
Successfully queued 636/636 images to stream for job '<id>'
... (per-result lines, pre-existing behavior) ...
Finalizing NATS consumer job-<id>-consumer before deletion
  (delivered=920 ack_floor=169 num_pending=366 num_ack_pending=43 num_redelivered=141)
Deleted NATS consumer job-<id>-consumer for job '<id>'
Deleted NATS stream job_<id> for job '<id>'
Cleaned up NATS resources for job <id>

920 delivery attempts for 636 tasks (~1.45× factor from redeliveries), 169 acked, 43 in flight, 141 redelivered. Workers were pulling but result POSTs were timing out, causing ack-before-ack_wait failures. Without this line you'd be guessing.

What this PR is NOT

• Does not change NATS semantics or retry behavior
• Does not touch reserve_tasks or the worker-poll path
• Does not add a job_logger to the NATS advisory listener (follow-up)
• Does not add per-worker identity to the job log (follow-up, depends on feat(ui): API key management and client info for processing services #1201)
• Does not fix async_api jobs killed by transient Redis errors during update_state — RedisError and "state actually missing" are conflated into a single fatal path #1219 (transient Redis error conflation) — but makes those diagnostics visible in the job log once async_api jobs killed by transient Redis errors during update_state — RedisError and "state actually missing" are conflated into a single fatal path #1219 lands
• Does not flip log_async from "mirror to both" to "route, then auto-mirror at WARNING+" — see Future intent above

Test plan

• pytest ami/ml/orchestration/tests/test_nats_queue.py — 18 tests pass
• pytest ami/ml/orchestration/tests/test_cleanup.py — integration tests pass (real NATS + Redis)
• black, isort, flake8, pyupgrade clean on all touched files (via pre-commit)
• Manual verification in staging — lifecycle and cleanup lines land in job.logs.stdout end-to-end

🤖 Generated with Claude Code

Summary by CodeRabbit

• New Features

  • Per‑job logging for async queueing and NATS lifecycle events, including one‑time lifecycle messages and a finalizing stats snapshot; cleanup accepts an optional job logger.

• Bug Fixes

  • Publish failures and cleanup errors now surface in the job log with safe fallback; reduced duplicate lifecycle noise and lowered routine success verbosity.

• Tests

  • Added tests for job‑scoped logging, lifecycle deduplication, finalizing stats ordering, error reporting, and logger fallback.

[netlify]

✅ Deploy Preview for antenna-preview canceled.

Name	Link
🔨 Latest commit	766137c
🔍 Latest deploy log	https://app.netlify.com/projects/antenna-preview/deploys/69dd97a5e57a2900084855d7

[coderabbitai]

Note

Reviews paused

It looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the reviews.auto_review.auto_pause_after_reviewed_commits setting.

Use the following commands to manage reviews:

• @coderabbitai resume to resume automatic reviews.
• @coderabbitai review to trigger a single review.

Use the checkboxes below for quick actions:

• ▶️ Resume reviews
• 🔍 Trigger review

📝 Walkthrough

Walkthrough

Extended orchestration and TaskQueueManager to accept an optional per-job logger, mirror selected NATS lifecycle and error messages into that job logger, add per-session deduplication of lifecycle messages, propagate job.logger from callers, adjust logging levels and async sync-to-async bridging, and add tests validating job-logger behavior.

Changes

Cohort / File(s)	Summary
Orchestration callers ami/ml/orchestration/jobs.py, ami/jobs/tasks.py	cleanup_async_job_resources signature now accepts `job_logger: logging.Logger
NATS queue manager ami/ml/orchestration/nats_queue.py	Added optional job_logger to TaskQueueManager; implemented async _log(level, msg) to mirror lifecycle/error messages to the per-job logger (safely via sync_to_async), added per-session _streams_logged/_consumers_logged dedupe, rewrote _ensure_stream/_ensure_consumer to probe info and emit one-time create/reuse logs with stats/config, mirror publish/delete results/errors to job logger, and added _log_final_consumer_stats.
Tests: job-logger behavior ami/ml/orchestration/tests/test_nats_queue.py	New async TestTaskQueueManagerJobLogger suite verifying creation vs reuse lifecycle messages (with stats/config) are emitted once to provided job logger, publish failures are mirrored to job logger, final consumer stats appear before deletion during cleanup, missing consumer tolerated, and job_logger=None remains safe.
Minor callsite adjustment ami/ml/orchestration/jobs.py (call sites)	Callers updated to forward job.logger where appropriate; simplified one cleanup_async_job_resources(job_id) fallback path when job lookup fails.

Sequence Diagram(s)

sequenceDiagram
    participant Caller as Job / Caller
    participant TQM as TaskQueueManager
    participant NATS as NATS JetStream
    participant ModLog as Module Logger
    participant JobLog as Job Logger

    Caller->>TQM: __init__(..., job_logger=job.logger)
    Caller->>TQM: publish_task(job_id, payload)
    TQM->>NATS: stream_info(job_<id>)
    alt stream exists
        TQM->>ModLog: log(stream reuse)
        TQM->>JobLog: _log(stream reuse + stats)
    else stream missing
        TQM->>NATS: add_stream(job_<id>)
        TQM->>ModLog: log(stream created)
        TQM->>JobLog: _log(stream created)
    end
    TQM->>NATS: consumer_info(job_<id>)
    alt consumer exists
        TQM->>ModLog: log(consumer reuse)
        TQM->>JobLog: _log(consumer reuse + stats)
    else consumer missing
        TQM->>NATS: add_consumer(job_<id>)
        TQM->>ModLog: log(consumer created)
        TQM->>JobLog: _log(consumer created + config)
    end
    TQM->>NATS: publish()
    alt publish succeeds
        TQM->>ModLog: debug(success)
    else publish fails
        TQM->>ModLog: error(...)
        TQM->>JobLog: _log(ERROR, ...)
    end
    Caller->>TQM: cleanup_job_resources(job_id)
    TQM->>NATS: consumer_info(job_<id>)
    TQM->>JobLog: _log_final_consumer_stats(job_id)
    TQM->>NATS: delete_consumer / delete_stream
    TQM->>ModLog: deletion logs
    TQM->>JobLog: _log(deletion logs)

Loading

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~50 minutes

Possibly related PRs

• Distributed Worker Architecture for ML Processing (Processing Service V2) pt. 1 #987 — Prior TaskQueueManager and orchestration changes touching the same lifecycle and logging surfaces.
• PSv2: Implement queue clean-up upon job completion #1113 — Overlaps on cleanup_async_job_resources and related NATS cleanup/logging flow.
• Support for NATS dead letter queue #1175 — Related edits to TaskQueueManager consumer/stream lifecycle handling that intersect with these changes.

Poem

🐇 I hopped through streams and mirrored each cue,

Whispered lifecycle traces so jobs could view.
Created, reused, then finalized with care,
Errors echoed gently into job logs there.
A tiny hop for clarity — logs bloom everywhere.

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 78.95% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title clearly and concisely summarizes the main objective: making NATS queue activity visible in async ML job logs, which aligns with the primary change in the changeset.
Linked Issues check	✅ Passed	The PR successfully implements the primary coding objective from #1220 (surface NATS lifecycle events in job logger with config/stats snapshots, forensic consumer logging before deletion, and publish error visibility). #1219 (distinguish transient Redis errors) is explicitly noted as out of scope for this PR.
Out of Scope Changes check	✅ Passed	All changes are directly scoped to #1220 objectives: TaskQueueManager now accepts job_logger, lifecycle events are mirrored, dedup logic prevents redundant NATS calls, and callers are wired appropriately. No unrelated refactoring or scope creep detected.
Description check	✅ Passed	The PR description comprehensively addresses all required template sections with substantial detail and context.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/nats-queue-job-logger

---

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.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[netlify]

✅ Deploy Preview for antenna-ssec canceled.

Name	Link
🔨 Latest commit	766137c
🔍 Latest deploy log	https://app.netlify.com/projects/antenna-ssec/deploys/69dd97a5f8d78f00088d2dc4

[Copilot]

Pull request overview

This PR makes NATS JetStream queue lifecycle events (stream/consumer setup, cleanup snapshots, and publish failures) visible in the per-job UI log by mirroring selected module-level logs to an optional job_logger, with async-safe bridging for Django ORM-backed log handlers.

Changes:

• Add optional job_logger to TaskQueueManager and mirror key lifecycle/error logs via an async _log() helper using sync_to_async.
• Wire job.logger into NATS publishing and cleanup paths so job logs include stream/consumer setup, delete events, and forensic pre-delete consumer stats.
• Add unit tests validating mirroring, deduping, reuse stats formatting, cleanup snapshot ordering, and failure surfacing.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
ami/ml/orchestration/nats_queue.py	Adds job_logger support, async-safe log mirroring, lifecycle dedupe, and pre-delete consumer stats logging.
ami/ml/orchestration/jobs.py	Passes per-job logger into TaskQueueManager for publish/cleanup and bridges one async error log write via sync_to_async.
ami/ml/orchestration/tests/test_nats_queue.py	Adds unit tests covering job-logger mirroring behavior and dedupe/forensics expectations.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[coderabbitai]

Actionable comments posted: 1

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

ami/ml/orchestration/nats_queue.py (1)

238-254: ⚠️ Potential issue | 🟠 Major

Only catch NotFoundError to detect missing consumers.

The current code catches all Exception exceptions, treating any consumer_info() failure as "consumer missing". If JetStream returns an auth, API, or transient error here, the code falls through to add_consumer(), which hides the real failure and can emit misleading lifecycle logs.

Change the broad except Exception to except nats.js.errors.NotFoundError so other exceptions propagate correctly. This requires adding the import: from nats.js.errors import NotFoundError.

Suggested fix

         try:
             info = await asyncio.wait_for(
                 self.js.consumer_info(stream_name, consumer_name),
                 timeout=NATS_JETSTREAM_TIMEOUT,
             )
             if job_id not in self._consumers_logged:
                 await self._log(
                     logging.INFO,
                     f"Reusing NATS consumer {consumer_name} ({self._format_consumer_stats(info)})",
                 )
                 self._consumers_logged.add(job_id)
             return
         except asyncio.TimeoutError:
             raise  # NATS unreachable — let caller handle it
-        except Exception:
+        except nats.js.errors.NotFoundError:
             # Consumer doesn't exist, fall through to create it.
             pass

Also add to imports near the top: from nats.js.errors import NotFoundError

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@ami/ml/orchestration/nats_queue.py` around lines 238 - 254, The try/except in
the block calling self.js.consumer_info incorrectly swallows all exceptions
(masking auth/API/transient errors) — replace the broad "except Exception" with
"except NotFoundError" (from nats.js.errors) so only missing-consumer errors
fall through to add_consumer; add the import "from nats.js.errors import
NotFoundError" near the top and keep the existing asyncio.TimeoutError handling
and logging using self._log, consumer_name, _format_consumer_stats, and
_consumers_logged unchanged.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@ami/ml/orchestration/jobs.py`:
- Around line 47-50: The cleanup function currently forwards whatever _logger it
receives into TaskQueueManager; change cleanup_async_job_resources to accept
job_logger: logging.Logger | None = None and only forward an actual per-job
logger (job.logger) from callers — update callers (e.g., where Job lookup
succeeds) to pass job.logger, and where Job lookup fails pass None; inside
cleanup_async_job_resources use a conditional context: if job_logger is not None
then "async with TaskQueueManager(job_logger=job_logger) as manager" else "async
with TaskQueueManager() as manager" so unrelated module loggers are not used for
per-job NATS lifecycle logging.

---

Outside diff comments:
In `@ami/ml/orchestration/nats_queue.py`:
- Around line 238-254: The try/except in the block calling self.js.consumer_info
incorrectly swallows all exceptions (masking auth/API/transient errors) —
replace the broad "except Exception" with "except NotFoundError" (from
nats.js.errors) so only missing-consumer errors fall through to add_consumer;
add the import "from nats.js.errors import NotFoundError" near the top and keep
the existing asyncio.TimeoutError handling and logging using self._log,
consumer_name, _format_consumer_stats, and _consumers_logged unchanged.

🪄 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: defaults

Review profile: CHILL

Plan: Pro

Run ID: ef374171-e1d2-4120-8bd7-e1dce1735926

📥 Commits

Reviewing files that changed from the base of the PR and between 8f8b177 and 7e8ce28.

📒 Files selected for processing (3)

• ami/ml/orchestration/jobs.py
• ami/ml/orchestration/nats_queue.py
• ami/ml/orchestration/tests/test_nats_queue.py

[coderabbitai]

🧹 Nitpick comments (1)

ami/ml/orchestration/nats_queue.py (1)

375-395: Consider passing job_logger in _ack_task_via_nats for consistency.

The _ack_task_via_nats function in ami/jobs/tasks.py (lines 204-211) has access to job_logger but doesn't forward it to TaskQueueManager. While acknowledge_task only logs at debug/error level and doesn't involve lifecycle events, passing job_logger would ensure any acknowledgment failures surface in the job log.

This is a minor consistency improvement and can be deferred if desired.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@ami/ml/orchestration/nats_queue.py` around lines 375 - 395, Update
TaskQueueManager.acknowledge_task to accept an optional job_logger parameter and
use it instead of the module-level logger for debug/error outputs; then update
the caller _ack_task_via_nats in ami/jobs/tasks.py to forward its job_logger
when calling TaskQueueManager.acknowledge_task so acknowledgment failures are
logged to the job-specific logger. Ensure the new parameter is optional
(defaulting to None) to maintain backward compatibility and only use job_logger
if provided.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Nitpick comments:
In `@ami/ml/orchestration/nats_queue.py`:
- Around line 375-395: Update TaskQueueManager.acknowledge_task to accept an
optional job_logger parameter and use it instead of the module-level logger for
debug/error outputs; then update the caller _ack_task_via_nats in
ami/jobs/tasks.py to forward its job_logger when calling
TaskQueueManager.acknowledge_task so acknowledgment failures are logged to the
job-specific logger. Ensure the new parameter is optional (defaulting to None)
to maintain backward compatibility and only use job_logger if provided.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: 3d05cb6e-5e33-404c-926f-5e2a45ac59d3

📥 Commits

Reviewing files that changed from the base of the PR and between 7e8ce28 and 439300a.

📒 Files selected for processing (3)

• ami/jobs/tasks.py
• ami/ml/orchestration/jobs.py
• ami/ml/orchestration/nats_queue.py

[coderabbitai]

Actionable comments posted: 1

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@ami/ml/orchestration/jobs.py`:
- Around line 120-127: The except block that currently awaits
sync_to_async(job.logger.error)(...) should also emit the full traceback to the
module logger so ops logs keep stack traces; add a module-level
logger.exception(...) call (using the module-level logger variable) inside the
except handler (e.g., in the same except Exception as e in publish/publish_task
path) before or immediately after the sync_to_async(job.logger.error) call so
the exception and traceback are recorded, while keeping the existing
job.logger.error call to write the message to job.logs.stdout.

🪄 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: defaults

Review profile: CHILL

Plan: Pro

Run ID: 8803dbbe-b042-4439-a8ce-bba8e51ce8c0

📥 Commits

Reviewing files that changed from the base of the PR and between 439300a and 9007197.

📒 Files selected for processing (2)

• ami/jobs/tasks.py
• ami/ml/orchestration/jobs.py

🚧 Files skipped from review as they are similar to previous changes (1)

• ami/jobs/tasks.py

[mihow]

Very helpful!

[coderabbitai]

Actionable comments posted: 1

🧹 Nitpick comments (1)

ami/ml/orchestration/nats_queue.py (1)

217-226: Consider defensive access for info.state fields.

While StreamInfo.state should always be present when stream_info() succeeds, the code accesses info.state.messages and info.state.last_seq directly. For consistency with the defensive pattern used in _format_consumer_stats(), you might want to handle potential None values.

♻️ Optional defensive access

             if job_id not in self._streams_logged:
+                state = info.state
+                messages = state.messages if state else "?"
+                last_seq = state.last_seq if state else "?"
                 await self._log(
                     logging.INFO,
                     f"Reusing NATS stream {stream_name} "
-                    f"(messages={info.state.messages}, last_seq={info.state.last_seq})",
+                    f"(messages={messages}, last_seq={last_seq})",
                 )

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In `@ami/ml/orchestration/nats_queue.py` around lines 217 - 226, The code accesses
info.state.messages and info.state.last_seq directly; make this defensive like
_format_consumer_stats by first grabbing state = info.state or a default, then
use getattr(state, "messages", 0) and getattr(state, "last_seq", 0) (or sensible
string defaults) when building the log message in the block that calls
self.js.stream_info(stream_name), and keep using self._log and
self._streams_logged as before so the log prints safe values even if info.state
is None.

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Inline comments:
In `@ami/ml/orchestration/nats_queue.py`:
- Around line 121-127: The ack_wait value from ConsumerInfo.config (ack_wait) is
in nanoseconds and is being logged directly, producing misleading values; update
the string construction that builds the consumer config summary (the return that
references cfg.ack_wait) to convert ack_wait from nanoseconds to seconds by
dividing by 1e9 when present so the log shows human-readable seconds (e.g., 30s)
instead of nanoseconds.

---

Nitpick comments:
In `@ami/ml/orchestration/nats_queue.py`:
- Around line 217-226: The code accesses info.state.messages and
info.state.last_seq directly; make this defensive like _format_consumer_stats by
first grabbing state = info.state or a default, then use getattr(state,
"messages", 0) and getattr(state, "last_seq", 0) (or sensible string defaults)
when building the log message in the block that calls
self.js.stream_info(stream_name), and keep using self._log and
self._streams_logged as before so the log prints safe values even if info.state
is None.

🪄 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: defaults

Review profile: CHILL

Plan: Pro

Run ID: d5c4336b-33b5-4131-ae23-a5487ad2a802

📥 Commits

Reviewing files that changed from the base of the PR and between 9007197 and f8d09ac.

📒 Files selected for processing (2)

• ami/ml/orchestration/nats_queue.py
• ami/ml/orchestration/tests/test_nats_queue.py

✅ Files skipped from review due to trivial changes (1)

• ami/ml/orchestration/tests/test_nats_queue.py

[coderabbitai]

♻️ Duplicate comments (1)

ami/ml/orchestration/nats_queue.py (1)

126-132: ⚠️ Potential issue | 🟡 Minor

Verify ack_wait units before rendering as seconds.

At Line 128, ack_wait is logged with an s suffix directly from ConsumerInfo.config. If this value is nanoseconds in your nats-py version, the UI log will be misleading.

In the exact nats.py version used by this repository, what unit is `ConsumerInfo.config.ack_wait` returned in by `consumer_info()` / `add_consumer()` responses: seconds or nanoseconds?

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: defaults

Review profile: CHILL

Plan: Pro

Run ID: b5e8d277-43a6-40ab-88b0-fce7f830e46d

📥 Commits

Reviewing files that changed from the base of the PR and between f8d09ac and 4c62d34.

📒 Files selected for processing (1)

• ami/ml/orchestration/nats_queue.py