feat(shift-crew): NATS push model for beats_to_voice (W6-P3 reactive synthesis)

[POWERFULMOVES]

Summary

• Adds `publish_nats=False` param to `run_pipeline()` — when `True`, publishes CGP packet to `tokenism.prosodic.bpm.v1` after Stage 3
• Adds `listen` subcommand: subscribes to `voice.agent.response.v1`, auto-runs pipeline on each agent response, publishes CGP — transforms Shift Crew from pull model (CLI-only) to push model (reactive pipeline participant)
• nats-py remains optional (lazy `import nats as natspy` pattern from `beats_to_cgp.py:212`)
• 5 unit tests, all mocked (no live NATS required to run)

Data flow (new)

```
voice.agent.response.v1 → beats_to_voice listen → run_pipeline() → tokenism.prosodic.bpm.v1
```

Emperor-CHIT-Humility Disclosure

Have: `beats_to_voice.py` full read, `beats_to_cgp.py` NATS lazy-import pattern, NATS subjects catalog (`voice.agent.response.v1` confirmed), 5/5 tests passing

Missing: Live NATS broker not verified this session; Flute-Gateway at :8055 not health-checked; nats-py availability on target nodes not confirmed (optional dep — graceful fallback)

Files Changed

• `pmoves/tools/beats_to_voice.py` — NATS constants, `_nats_publish_cgp()`, `_listen_loop()`, `_cmd_listen()`, `listen` subparser, `publish_nats` param on `run_pipeline()`
• `pmoves/tools/test_beats_to_voice_nats.py` — 5 mocked unit tests

Village Rule

One scope, one commit, one PR. W6-P3 reactive voice binding gap closed.

Summary by CodeRabbit

• New Features
  • Added event-driven "listen" mode to process voice triggers from external messages
  • New CLI subcommand to activate and manage the listening service
  • Configurable messaging endpoints for flexible deployment
  • Enhanced pipeline with optional message publishing capability

[coderabbitai]

📝 Walkthrough

Walkthrough

The changes add event-driven NATS integration to beats_to_voice.py, enabling the tool to listen for trigger messages from a NATS subject, parse text and agent IDs, execute the audio-to-voice pipeline per message, and publish resulting CGP data back to NATS. New constants, helper functions, and CLI subcommand support configurable NATS connections and message handling.

Changes

Cohort / File(s)	Summary
NATS Event-Driven Integration pmoves/tools/beats_to_voice.py	Adds _nats_publish_cgp helper for publishing CGP JSON with lazy nats-py import; updates run_pipeline signature with publish_nats and nats_url parameters; introduces _listen_loop for persistent NATS subscriptions that parse trigger messages and invoke pipeline per message; adds NATS_URL and NATS_TRIGGER_SUBJECT configuration constants; includes _cmd_listen CLI subcommand to start the subscription loop with command-line overrides.
NATS Functionality Tests pmoves/tools/test_beats_to_voice_nats.py	New test file validating _nats_publish_cgp connectivity and import failure handling; async tests for _listen_loop message parsing and pipeline invocation; synchronous tests for run_pipeline ensuring cgp["nats_published"] state is set correctly based on publish_nats flag.

Sequence Diagram

sequenceDiagram
    participant NATS as NATS Service
    participant Listen as Listen Loop
    participant Pipeline as run_pipeline
    participant Publish as _nats_publish_cgp

    NATS->>Listen: Trigger message (text, user_id)
    activate Listen
    Listen->>Listen: Parse JSON message
    Listen->>Pipeline: Call with text, agent_id
    activate Pipeline
    Pipeline->>Pipeline: Process audio/voice
    Pipeline-->>Listen: Return CGP data
    deactivate Pipeline
    Listen->>Publish: Call with CGP JSON
    activate Publish
    Publish->>NATS: Publish to bpm.v1 subject
    deactivate Publish
    deactivate Listen

Loading

Estimated code review effort

🎯 3 (Moderate) | ⏱️ ~25 minutes

Possibly related PRs

• fix(tests): consolidate NATS fixtures (3 runtime tests rewired) #1201: Updates NATS test fixtures and fake NATS stub (connect returns self, is_connected property) that directly support the new NATS-focused tests and async behavior in beats_to_voice.

Poem

🐰 A rabbit listens to NATS so keen,
Messages hop through a subscription scene,
Parse the text and the agent's name,
Run the pipeline, publish CGP—what a game!
Event-driven beats, asynchronously blessed,
Our voice tool hops faster than the rest! 🎵

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 68.75% 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 accurately summarizes the main change: adding a NATS push model for beats_to_voice to enable reactive synthesis (W6-P3), which aligns with the core objective of converting from pull to push.
Description check	✅ Passed	The description provides a clear summary of changes and includes most required sections, but is missing the Testing section with commands/output and does not explicitly mark the Required Checks checkboxes.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

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

✨ Finishing Touches

📝 Generate docstrings

• Create stacked PR
• Commit on current branch

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch feat/shift-crew-nats-push

---

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.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: aed3396d89

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[coderabbitai]

Actionable comments posted: 3

🧹 Nitpick comments (4)

pmoves/tools/test_beats_to_voice_nats.py (2)

30-42: Simplify: the builtins.__import__ patch is redundant.

Setting sys.modules["nats"] = None already causes import nats to raise ImportError per CPython's import machinery, so the wrapper around builtins.__import__ (and the __builtins__ dict-vs-module contortion on Line 32) adds no behavior. Dropping it shortens the test and removes a global hook that's easy to misuse in future tests.

🔧 Proposed fix

     async def test_publish_nats_unavailable(self):
         """_nats_publish_cgp returns False when nats-py raises on import."""
-        original_import = __builtins__.__import__ if hasattr(__builtins__, "__import__") else __import__
-
-        def _failing_import(name, *args, **kwargs):
-            if name == "nats":
-                raise ImportError("no nats")
-            return original_import(name, *args, **kwargs)
-
         with patch.dict("sys.modules", {"nats": None}):
-            with patch("builtins.__import__", side_effect=_failing_import):
-                result = await beats_to_voice._nats_publish_cgp({"spec": "chit.cgp.v0.2"})
+            result = await beats_to_voice._nats_publish_cgp({"spec": "chit.cgp.v0.2"})
         assert result is False

🤖 Prompt for AI Agents

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

In `@pmoves/tools/test_beats_to_voice_nats.py` around lines 30 - 42, The test
test_publish_nats_unavailable should be simplified by removing the redundant
builtins.__import__ patch and its helpers (original_import and _failing_import);
keep only the patch.dict("sys.modules", {"nats": None}) context so that
importing "nats" fails naturally, then call await
beats_to_voice._nats_publish_cgp({"spec":"chit.cgp.v0.2"}) and assert the result
is False; remove any code that patches builtins.__import__ or references
__builtins__ to avoid unnecessary global import hooks.

---

106-113: Prefer patching _nats_publish_cgp over asyncio.run.

Patching asyncio.run globally is broader than needed and couples the test to the current implementation choice (sync run_pipeline running an async helper). Patch the helper directly so the test states intent and stays robust if run_pipeline is later refactored to await directly or to use a runner.

🔧 Proposed fix

     def test_publish_nats_true_sets_key(self):
         """run_pipeline with publish_nats=True sets nats_published in cgp stage."""
         with patch.object(beats_to_voice, "_check_flute_health", return_value=False):
-            with patch("asyncio.run", return_value=True):
+            async def _fake_publish(*_a, **_kw):
+                return True
+            with patch.object(beats_to_voice, "_nats_publish_cgp", side_effect=_fake_publish):
                 results = beats_to_voice.run_pipeline(
                     text="test text", bpm=90, publish_nats=True
                 )
         assert results["stages"]["cgp"]["nats_published"] is True

🤖 Prompt for AI Agents

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

In `@pmoves/tools/test_beats_to_voice_nats.py` around lines 106 - 113, The test
test_publish_nats_true_sets_key should patch the internal helper
_nats_publish_cgp instead of asyncio.run; update the test to use
patch.object(beats_to_voice, "_nats_publish_cgp", return_value=True) (keep the
existing patch for _check_flute_health) so run_pipeline(text="test text",
bpm=90, publish_nats=True) exercises the real control flow and the helper
returns True, then assert results["stages"]["cgp"]["nats_published"] is True.

pmoves/tools/beats_to_voice.py (2)

156-184: Drain in finally so the connection is cleaned up on any exit, not just cancel/KeyboardInterrupt.

If nc.subscribe(...) raises, or any non-cancel exception bubbles out of the sleep loop, the connection is leaked. Move the drain into a finally to make shutdown deterministic.

🔧 Proposed fix

     nc = await natspy.connect(nats_url)
-
-    async def _handler(msg) -> None:
-        ...
-
-    await nc.subscribe(trigger_subject, cb=_handler)
-    sys.stderr.write(
-        f"[beats_to_voice] Listening on {trigger_subject} → publishes to {NATS_SUBJECT}\n"
-    )
     try:
-        while True:
-            await asyncio.sleep(1)
-    except (KeyboardInterrupt, asyncio.CancelledError):
+        async def _handler(msg) -> None:
+            ...
+        await nc.subscribe(trigger_subject, cb=_handler)
+        sys.stderr.write(
+            f"[beats_to_voice] Listening on {trigger_subject} → publishes to {NATS_SUBJECT}\n"
+        )
+        while True:
+            await asyncio.sleep(1)
+    finally:
         await nc.drain()

🤖 Prompt for AI Agents

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

In `@pmoves/tools/beats_to_voice.py` around lines 156 - 184, The nc NATS
connection is only drained on KeyboardInterrupt/CancelledError, leaking the
connection if subscribe() or other errors occur; wrap the subscribe + listening
loop in a try/finally and move await nc.drain() into the finally block (guarding
that nc is truthy/connected) so the connection is always cleaned up on any exit;
update the scope around nc, the async _handler, await
nc.subscribe(trigger_subject, cb=_handler) and the sleep loop to ensure drain
runs even if subscribe or the loop raises.

---

343-345: Foot-gun: run_pipeline(publish_nats=True) cannot be called from an async context.

asyncio.run() raises RuntimeError when invoked inside a running event loop. The current _listen_loop handler avoids this by passing publish_nats=False and publishing manually, but that contract is implicit. Consider either (a) guarding with asyncio.get_running_loop() and falling back to scheduling on the live loop, or (b) documenting in the docstring that publish_nats=True is only valid when run_pipeline is called from sync code.

🤖 Prompt for AI Agents

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

In `@pmoves/tools/beats_to_voice.py` around lines 343 - 345, The current call to
asyncio.run() in run_pipeline when publish_nats=True will raise if called inside
an existing event loop; change the logic around the call to _nats_publish_cgp so
that you first check for a running loop (use asyncio.get_running_loop() in a
try/except), and if no running loop keep using
asyncio.run(_nats_publish_cgp(...)), but if a loop is running submit the
coroutine to that loop (use
asyncio.run_coroutine_threadsafe(_nats_publish_cgp(cgp_packet, nats_url), loop)
and wait on the returned Future.result()) and then set
results["stages"]["cgp"]["nats_published"] accordingly; update run_pipeline and
any callers (e.g., _listen_loop) to rely on this guarded behavior rather than
implicitly passing publish_nats=False.

🤖 Prompt for all review comments with AI agents

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

Inline comments:
In `@pmoves/tools/beats_to_voice.py`:
- Around line 158-174: The handler in _handler currently reads response_text at
the top level and falls back to user_id as agent_id, which causes
envelope-wrapped messages to be ignored and misattributes agent identity; update
_handler to first unwrap the possible envelope/payload shape (reuse the
dual-shape logic from voice_follow_agent.py::_extract_text or import that
helper) so you extract text from either top-level response_text/text or
payload.response_text/text, and stop using user_id as a fallback for
agent_id—keep the CLI-provided agent_id unless an explicit agent field (e.g.,
meta.model_used or meta.platform) is present in the message metadata, then pass
the correct agent_id into run_pipeline and preserve existing publishing/error
behavior.
- Around line 127-137: In _nats_publish_cgp ensure the NATS connection is always
drained on success or error to avoid leaking sockets: after connecting (nc =
await natspy.connect(...)) wrap the publish call in a try/finally so that await
nc.drain() is executed in the finally block if nc was created; do not add a
separate close() after drain() since drain() already closes the connection per
nats.py docs, and preserve the existing exception handling to return False on
error.

In `@pmoves/tools/test_beats_to_voice_nats.py`:
- Around line 64-94: The test uses a fragile sleep to wait for
beats_to_voice._listen_loop to call mock_nc.subscribe; replace this timing
dependency by having fake_subscribe set an asyncio.Event when invoked and
awaiting that event instead of asyncio.sleep; specifically modify fake_subscribe
(which is assigned to mock_nc.subscribe) to accept (subject, cb), append cb to
captured_handlers, call event.set(), and in the test await event.wait() (with a
timeout) before invoking captured_handlers[0](msg) and cancelling the task so
the test is deterministic.

---

Nitpick comments:
In `@pmoves/tools/beats_to_voice.py`:
- Around line 156-184: The nc NATS connection is only drained on
KeyboardInterrupt/CancelledError, leaking the connection if subscribe() or other
errors occur; wrap the subscribe + listening loop in a try/finally and move
await nc.drain() into the finally block (guarding that nc is truthy/connected)
so the connection is always cleaned up on any exit; update the scope around nc,
the async _handler, await nc.subscribe(trigger_subject, cb=_handler) and the
sleep loop to ensure drain runs even if subscribe or the loop raises.
- Around line 343-345: The current call to asyncio.run() in run_pipeline when
publish_nats=True will raise if called inside an existing event loop; change the
logic around the call to _nats_publish_cgp so that you first check for a running
loop (use asyncio.get_running_loop() in a try/except), and if no running loop
keep using asyncio.run(_nats_publish_cgp(...)), but if a loop is running submit
the coroutine to that loop (use
asyncio.run_coroutine_threadsafe(_nats_publish_cgp(cgp_packet, nats_url), loop)
and wait on the returned Future.result()) and then set
results["stages"]["cgp"]["nats_published"] accordingly; update run_pipeline and
any callers (e.g., _listen_loop) to rely on this guarded behavior rather than
implicitly passing publish_nats=False.

In `@pmoves/tools/test_beats_to_voice_nats.py`:
- Around line 30-42: The test test_publish_nats_unavailable should be simplified
by removing the redundant builtins.__import__ patch and its helpers
(original_import and _failing_import); keep only the patch.dict("sys.modules",
{"nats": None}) context so that importing "nats" fails naturally, then call
await beats_to_voice._nats_publish_cgp({"spec":"chit.cgp.v0.2"}) and assert the
result is False; remove any code that patches builtins.__import__ or references
__builtins__ to avoid unnecessary global import hooks.
- Around line 106-113: The test test_publish_nats_true_sets_key should patch the
internal helper _nats_publish_cgp instead of asyncio.run; update the test to use
patch.object(beats_to_voice, "_nats_publish_cgp", return_value=True) (keep the
existing patch for _check_flute_health) so run_pipeline(text="test text",
bpm=90, publish_nats=True) exercises the real control flow and the helper
returns True, then assert results["stages"]["cgp"]["nats_published"] is True.

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

Run ID: 5cf21817-16bb-4ff8-8df1-a84f0fbed451

📥 Commits

Reviewing files that changed from the base of the PR and between 115fb30 and aed3396.

📒 Files selected for processing (2)

• pmoves/tools/beats_to_voice.py
• pmoves/tools/test_beats_to_voice_nats.py

[POWERFULMOVES]

Threads vcp7 (P1 envelope unwrapping), vcqE (P2 user_id/agent_id), verN (Major duplicate), and verP (race-prone sleep) remain open. Tracking as separate follow-up issues — these require a deeper look at the voice.agent.response.v1 envelope schema and how agent_id vs user_id is intended to flow through the listen handler. Will address in a follow-up PR once the envelope format is confirmed against the NATS subject catalog.

[POWERFULMOVES]

Resolved P1 (vcp7) — NATS subject catalog at .claude/context/nats-subjects.md confirms voice.agent.response.v1 payload has response_text at the top level (not nested under an envelope key). Handler is correct as written. Closing this thread.