feat(cron): pass origin user_id to delivery mirror (send_message parity)

Multi-participant parity with interactive send_message, which passes
HERMES_SESSION_USER_ID to gateway.mirror.mirror_to_session so the mirror
lands in the exact participant's session.

- cronjob_tools._origin_from_env now captures user_id from the session
  context at job-create time (alongside platform/chat_id/thread_id).
- _maybe_mirror_cron_delivery forwards user_id to mirror_to_session.
- _deliver_result threads origin.user_id through for the origin target.

Effect: in a per-user-isolated group chat (group_sessions_per_user=True,
the default), the mirror resolves to the member who scheduled the job
instead of conservatively no-op'ing on ambiguous candidates. DMs and
shared group/thread sessions are unaffected (single candidate). Default
still OFF.

Tests: helper forwards user_id; E2E _deliver_result forwards origin
user_id. 17/17 in TestCronDeliveryMirror; 527 cron tests pass (4 failures
pre-existing: croniter-not-installed + TZ, identical on baseline).

cron/scheduler.py

@@ -415,6 +415,7 @@ def _maybe_mirror_cron_delivery(
     chat_id: str,
     mirror_text: str,
     thread_id: Optional[str] = None,
+    user_id: Optional[str] = None,
     *,
     enabled: bool = False,
 ) -> None:
@@ -423,9 +424,11 @@ def _maybe_mirror_cron_delivery(
     No-op unless ``enabled`` (resolved once by the caller, and already scoped to
     the origin target — see ``_target_matches_origin``). Reuses the shipped
     ``mirror_to_session`` so cron rides exactly the same path that interactive
-    ``send_message`` mirroring already uses. All failures are swallowed — a
-    delivery that succeeded must never be reported as failed because the
-    transcript mirror hit a problem.
+    ``send_message`` mirroring already uses, including passing ``user_id`` so a
+    per-user-isolated group chat resolves to the exact member who scheduled the
+    job (parity with ``send_message``). All failures are swallowed — a delivery
+    that succeeded must never be reported as failed because the transcript
+    mirror hit a problem.
 
     Because the caller only enables this for the target that equals the job's
     origin conversation, the session is expected to exist (the job was born in
@@ -447,6 +450,7 @@ def _maybe_mirror_cron_delivery(
             text,
             source_label="cron",
             thread_id=thread_id,
+            user_id=user_id,
         )
         if ok:
             logger.info(
@@ -1001,6 +1005,9 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
         mirror_this_target = mirror_enabled and _target_matches_origin(
             origin, platform_name, chat_id, thread_id
         )
+        # Pass the origin's user_id so a per-user-isolated group chat resolves to
+        # the exact member who scheduled the job — parity with send_message.
+        origin_user_id = origin.get("user_id") if mirror_this_target else None
 
         # Built-in names resolve to their enum member; plugin platform names
         # create dynamic members via Platform._missing_().
@@ -1238,7 +1245,8 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
                     delivered = True
                     _maybe_mirror_cron_delivery(
                         job, platform_name, chat_id, mirror_text,
-                        thread_id=thread_id, enabled=mirror_this_target,
+                        thread_id=thread_id, user_id=origin_user_id,
+                        enabled=mirror_this_target,
                     )
             except Exception as e:
                 err_msg = f"live adapter delivery to {platform_name}:{chat_id} failed: {e}"
@@ -1280,7 +1288,8 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
             logger.info("Job '%s': delivered to %s:%s", job["id"], platform_name, chat_id)
             _maybe_mirror_cron_delivery(
                 job, platform_name, chat_id, mirror_text,
-                thread_id=thread_id, enabled=mirror_this_target,
+                thread_id=thread_id, user_id=origin_user_id,
+                enabled=mirror_this_target,
             )
 
     if delivery_errors:

tests/cron/test_scheduler.py

@@ -3710,3 +3710,44 @@ class TestCronDeliveryMirror:
         # Exactly one mirror, and it is the origin chat (123) — not 999.
         mirror_mock.assert_called_once()
         assert mirror_mock.call_args[0][1] == "123"
+
+    # --- multi-participant parity with send_message (user_id passthrough) ---
+
+    def test_mirror_passes_user_id_through(self):
+        """The helper forwards user_id to mirror_to_session so a per-user-
+        isolated group resolves to the exact member who scheduled the job —
+        parity with interactive send_message."""
+        from cron.scheduler import _maybe_mirror_cron_delivery
+
+        with patch("gateway.mirror.mirror_to_session", return_value=True) as m:
+            _maybe_mirror_cron_delivery(
+                {"id": "j1"}, "telegram", "123", "brief",
+                thread_id=None, user_id="U999", enabled=True,
+            )
+        m.assert_called_once()
+        assert m.call_args.kwargs.get("user_id") == "U999"
+
+    def test_delivery_forwards_origin_user_id(self):
+        """End-to-end: a job whose origin carries user_id mirrors with that
+        user_id, so multi-participant resolution matches send_message."""
+        from gateway.config import Platform
+
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("tools.send_message_tool._send_to_platform", new=AsyncMock(return_value={"success": True})), \
+             patch("gateway.mirror.mirror_to_session", return_value=True) as mirror_mock:
+            job = {
+                "id": "test-job",
+                "name": "daily-report",
+                "deliver": "origin",
+                "origin": {"platform": "telegram", "chat_id": "123", "user_id": "U42"},
+                "attach_to_session": True,
+            }
+            _deliver_result(job, "Here is today's summary.")
+
+        mirror_mock.assert_called_once()
+        assert mirror_mock.call_args.kwargs.get("user_id") == "U42"

tools/cronjob_tools.py

@@ -294,6 +294,12 @@ def _origin_from_env() -> Optional[Dict[str, str]]:
             "chat_id": origin_chat_id,
             "chat_name": get_session_env("HERMES_SESSION_CHAT_NAME") or None,
             "thread_id": thread_id,
+            # Captured so an opt-in delivery mirror (cron.mirror_delivery /
+            # attach_to_session) can resolve the exact participant's session in
+            # per-user-isolated group chats — parity with interactive
+            # send_message, which passes HERMES_SESSION_USER_ID to
+            # gateway.mirror.mirror_to_session. Harmless for DMs/shared sessions.
+            "user_id": get_session_env("HERMES_SESSION_USER_ID") or None,
         }
     return None