fix(beats-voice): agent_id semantics + deterministic test + doc thread cleanup

pmoves/docs/AGENTS/AGNOTE4482.md

@@ -348,6 +348,7 @@ env["RUNNER_ALLOW_RUNNER_REUSE"] = "true"
 - Signed AGNOTE4482_SIGNOFF_CHECKLIST.md sections 1, 3, 7
 
 ### Key Findings
+
 | Finding | Status | Evidence |
 |---------|--------|----------|
 | NATS hotspot dirs (work-marshaling, chat-relay, node-registry, tools) | **RESOLVED** | All production code migrated; 21 files remain in secondary batch |

pmoves/docs/AGENTS/PR_TRIAGE_2026-04-23.md

@@ -72,4 +72,4 @@ Per the feedback note `feedback_rebase_before_merge.md` and the CODEX worktree c
 - `pmoves/docs/AGENTS/AGNOTE4482_SITREP.md` — cold-start context
 - `pmoves/docs/AGENTS/AGNOTE4482_SIGNOFF_CHECKLIST.md` — signoff gate
 - `.claude/CLAUDE.md` section "Submodule working-tree wipe recovery" — for #1370 fix mechanics
-- `~/.claude/projects/.../memory/feedback_rebase_before_merge.md` — rationale for rebasing before merge
+- Rebase-before-merge rationale: stale PRs on shared docs cause silent data loss; always rebase to HEAD before merging agent-authored doc PRs

pmoves/tools/beats_to_voice.py

@@ -159,7 +159,7 @@ async def _handler(msg) -> None:
             try:
                 data = json.loads(msg.data.decode("utf-8"))
                 text = data.get("response_text") or data.get("text", "")
-                aid = data.get("user_id") or agent_id
+                aid = agent_id  # user_id is the request originator, not the processing agent
                 if not text:
                     return
                 sys.stderr.write(

pmoves/tools/test_beats_to_voice_nats.py

@@ -36,18 +36,19 @@ async def test_publish_nats_unavailable(self):
 
 class TestListenHandler(unittest.IsolatedAsyncioTestCase):
     async def test_handler_runs_pipeline_and_publishes(self):
-        """Listen handler calls run_pipeline with response_text and publishes CGP."""
+        """Listen handler calls run_pipeline with response_text and uses configured agent_id."""
         mock_nc = AsyncMock()
         mock_nats = MagicMock()
         mock_nats.connect = AsyncMock(return_value=mock_nc)
 
         msg = MagicMock()
         msg.data = json.dumps({
             "response_text": "Hello from Agent Zero",
-            "user_id": "z890-claude",
+            "user_id": "z890-claude",  # originator — should NOT become agent_id
         }).encode("utf-8")
 
         called = {}
+        subscribed = asyncio.Event()
 
         def fake_pipeline(**kwargs):
             called.update(kwargs)
@@ -57,6 +58,7 @@ def fake_pipeline(**kwargs):
 
         async def fake_subscribe(subject, cb):
             captured_handlers.append(cb)
+            subscribed.set()
 
         mock_nc.subscribe = fake_subscribe
         mock_nc.drain = AsyncMock()
@@ -72,17 +74,16 @@ async def fake_subscribe(subject, cb):
                         "http://localhost:8055",
                     )
                 )
-                await asyncio.sleep(0.05)
-                if captured_handlers:
-                    await captured_handlers[0](msg)
+                await asyncio.wait_for(subscribed.wait(), timeout=2.0)
+                await captured_handlers[0](msg)
                 task.cancel()
                 try:
                     await task
                 except asyncio.CancelledError:
                     pass
 
         assert called.get("text") == "Hello from Agent Zero"
-        assert called.get("agent_id") == "z890-claude"
+        assert called.get("agent_id") == "4090-claude"  # configured agent_id, not user_id
         mock_nc.publish.assert_awaited()