From e616eb0234c1375a6733fe56cdaf0033b566ee87 Mon Sep 17 00:00:00 2001
From: briandevans <252620095+briandevans@users.noreply.github.com>
Date: Thu, 14 May 2026 17:17:59 -0700
Subject: [PATCH 1/4] feat(cron): expose structured cron context via adapter
 send metadata
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Plugin platform adapters that route cron output to external inboxes
need job_id/job_name/schedule/deliver/origin to render proper source
labels and chain follow-up turns. Today the only structured side-channel
on adapter.send() is thread_id; adapters recover the rest by regex-parsing
the "Cronjob Response: <name>\n(job_id: <id>)" envelope, which is brittle
and breaks entirely when cron.wrap_response=false (issue #26004).

Enrich the live-adapter send_metadata with a nested "cron" dict carrying
the fields already available in _deliver_result: job_id, job_name (falls
back to job_id), schedule, deliver, and the already-computed origin.
Adapters that ignore unknown metadata keys are unaffected — the dict
already received {"thread_id": ...} today.

Scoped to the issue's "Step 1 (minimal, fully backwards-compatible)".
The response_id / session_id fields the issue lists are deferred: today
they would require widening run_job's return tuple, which is out of
scope for a minimal patch.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 cron/scheduler.py            |  20 ++++-
 tests/cron/test_scheduler.py | 164 +++++++++++++++++++++++++++++++++++
 2 files changed, 183 insertions(+), 1 deletion(-)

diff --git a/cron/scheduler.py b/cron/scheduler.py
index a51ade8efe65..df7d65d393e1 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -711,12 +711,30 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
             delivery_errors.append(msg)
             continue
 
+        # Structured side-channel so plugin adapters can recover cron context
+        # without regex-parsing the envelope text. `thread_id` was the only
+        # existing key; nested under "cron" we expose job_id/job_name/schedule/
+        # deliver/origin. Adapters that ignore unknown keys are unaffected.
+        cron_meta = {
+            "job_id": job.get("id", ""),
+            "job_name": job.get("name") or job.get("id", ""),
+            "schedule": job.get("schedule"),
+            "deliver": job.get("deliver"),
+            "origin": origin or None,
+        }
+        cron_meta = {k: v for k, v in cron_meta.items() if v}
+
+        send_metadata: Optional[dict] = None
+        if thread_id:
+            send_metadata = {"thread_id": thread_id}
+        if cron_meta:
+            send_metadata = {**(send_metadata or {}), "cron": cron_meta}
+
         # Prefer the live adapter when the gateway is running — this supports E2EE
         # rooms (e.g. Matrix) where the standalone HTTP path cannot encrypt.
         runtime_adapter = (adapters or {}).get(platform)
         delivered = False
         if runtime_adapter is not None and loop is not None and getattr(loop, "is_running", lambda: False)():
-            send_metadata = {"thread_id": thread_id} if thread_id else None
             try:
                 # Send cleaned text (MEDIA tags stripped) — not the raw content
                 text_to_send = cleaned_delivery_content.strip()
diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py
index 38da3fe40875..f6a0badbb486 100644
--- a/tests/cron/test_scheduler.py
+++ b/tests/cron/test_scheduler.py
@@ -837,6 +837,170 @@ def test_origin_delivery_preserves_thread_id(self):
         assert send_mock.call_args.kwargs["thread_id"] == "17585"
 
 
+class TestDeliverResultCronMetadata:
+    """Live adapter sends should receive structured cron context in metadata.
+
+    Without this, plugin adapters have to regex-parse the "Cronjob Response: ..."
+    envelope to recover job_id/job_name, which is brittle and breaks entirely
+    when cron.wrap_response=false. The structured side-channel under
+    metadata["cron"] makes job_id/job_name/schedule/deliver/origin available
+    via the existing metadata kwarg, alongside thread_id.
+    """
+
+    def _build_adapter_send_mock(self):
+        from concurrent.futures import Future
+
+        adapter = AsyncMock()
+        adapter.send.return_value = MagicMock(success=True)
+
+        def fake_run_coro(coro, _loop):
+            future = Future()
+            future.set_result(MagicMock(success=True))
+            coro.close()
+            return future
+
+        loop = MagicMock()
+        loop.is_running.return_value = True
+        return adapter, loop, fake_run_coro
+
+    def test_live_adapter_send_receives_cron_metadata(self):
+        from gateway.config import Platform
+
+        adapter, loop, fake_run_coro = self._build_adapter_send_mock()
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        job = {
+            "id": "abc123",
+            "name": "daily-summary",
+            "schedule": "0 9 * * *",
+            "deliver": "origin",
+            "origin": {"platform": "telegram", "chat_id": "999"},
+        }
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
+            _deliver_result(
+                job,
+                "hello",
+                adapters={Platform.TELEGRAM: adapter},
+                loop=loop,
+            )
+
+        adapter.send.assert_called_once()
+        metadata = adapter.send.call_args.kwargs.get("metadata")
+        assert metadata is not None, "send_metadata should be populated with cron context"
+        assert "cron" in metadata, f"cron context missing: {metadata!r}"
+        cron_ctx = metadata["cron"]
+        assert cron_ctx["job_id"] == "abc123"
+        assert cron_ctx["job_name"] == "daily-summary"
+        assert cron_ctx["schedule"] == "0 9 * * *"
+        assert cron_ctx["deliver"] == "origin"
+        assert cron_ctx["origin"] == {"platform": "telegram", "chat_id": "999"}
+
+    def test_cron_metadata_falls_back_to_job_id_when_name_missing(self):
+        from gateway.config import Platform
+
+        adapter, loop, fake_run_coro = self._build_adapter_send_mock()
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        job = {
+            "id": "noname-job",
+            "deliver": "origin",
+            "origin": {"platform": "telegram", "chat_id": "5"},
+        }
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
+            _deliver_result(
+                job,
+                "hi",
+                adapters={Platform.TELEGRAM: adapter},
+                loop=loop,
+            )
+
+        cron_ctx = adapter.send.call_args.kwargs["metadata"]["cron"]
+        assert cron_ctx["job_name"] == "noname-job"
+
+    def test_cron_metadata_coexists_with_thread_id(self):
+        from gateway.config import Platform
+
+        adapter, loop, fake_run_coro = self._build_adapter_send_mock()
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        job = {
+            "id": "topic-job",
+            "name": "topic-job",
+            "deliver": "origin",
+            "origin": {
+                "platform": "telegram",
+                "chat_id": "-1001",
+                "thread_id": "17585",
+            },
+        }
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
+            _deliver_result(
+                job,
+                "hi",
+                adapters={Platform.TELEGRAM: adapter},
+                loop=loop,
+            )
+
+        metadata = adapter.send.call_args.kwargs["metadata"]
+        assert metadata["thread_id"] == "17585"
+        assert metadata["cron"]["job_id"] == "topic-job"
+
+    def test_cron_metadata_omits_empty_fields(self):
+        """Optional fields (schedule, deliver) absent on the job should be omitted."""
+        from gateway.config import Platform
+
+        adapter, loop, fake_run_coro = self._build_adapter_send_mock()
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        job = {
+            "id": "min-job",
+            "deliver": "origin",
+            "origin": {"platform": "telegram", "chat_id": "5"},
+        }  # no name, no schedule
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
+            _deliver_result(
+                job,
+                "hi",
+                adapters={Platform.TELEGRAM: adapter},
+                loop=loop,
+            )
+
+        cron_ctx = adapter.send.call_args.kwargs["metadata"]["cron"]
+        assert "schedule" not in cron_ctx
+        # job_id is present (required), job_name falls back to it,
+        # deliver and origin are present.
+        assert cron_ctx == {
+            "job_id": "min-job",
+            "job_name": "min-job",
+            "deliver": "origin",
+            "origin": {"platform": "telegram", "chat_id": "5"},
+        }
+
+
 class TestDeliverResultErrorReturns:
     """Verify _deliver_result returns error strings on failure, None on success."""
 

From c68c234c8d80201abb3104793e8916e0f9f0d1ac Mon Sep 17 00:00:00 2001
From: briandevans <252620095+briandevans@users.noreply.github.com>
Date: Thu, 14 May 2026 18:09:32 -0700
Subject: [PATCH 2/4] fixup(cron): hoist invariant cron_meta above target loop;
 filter on is not None
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Addresses Copilot inline review on #26012:

1. cron_meta and origin are invariant across `for target in targets:` —
   every field reads `job.*` and a per-job-resolved origin. Hoist the
   construction once above the loop so we don't rebuild it on every
   target.

2. Switch the strip-empty filter from truthiness (`if v`) to explicit
   `if v is not None` so falsy-but-meaningful values (e.g. an
   explicitly-empty schedule string a future job type might set) still
   reach adapters. Only true absence (None) is dropped.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
---
 cron/scheduler.py | 29 +++++++++++++++--------------
 1 file changed, 15 insertions(+), 14 deletions(-)

diff --git a/cron/scheduler.py b/cron/scheduler.py
index df7d65d393e1..0fd99b22f5f9 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -674,13 +674,27 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
 
     delivery_errors = []
 
+    # Structured side-channel so plugin adapters can recover cron context
+    # without regex-parsing the envelope text. Invariant across targets, so
+    # built once before the loop; only thread_id varies per target.
+    # Filter on `is not None` rather than truthiness so falsy-but-meaningful
+    # values (e.g. an explicitly empty schedule) reach adapters.
+    origin = _resolve_origin(job) or {}
+    cron_meta_full = {
+        "job_id": job.get("id", ""),
+        "job_name": job.get("name") or job.get("id", ""),
+        "schedule": job.get("schedule"),
+        "deliver": job.get("deliver"),
+        "origin": origin or None,
+    }
+    cron_meta = {k: v for k, v in cron_meta_full.items() if v is not None}
+
     for target in targets:
         platform_name = target["platform"]
         chat_id = target["chat_id"]
         thread_id = target.get("thread_id")
 
         # Diagnostic: log thread_id for topic-aware delivery debugging
-        origin = _resolve_origin(job) or {}
         origin_thread = origin.get("thread_id")
         if origin_thread and not thread_id:
             logger.warning(
@@ -711,19 +725,6 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
             delivery_errors.append(msg)
             continue
 
-        # Structured side-channel so plugin adapters can recover cron context
-        # without regex-parsing the envelope text. `thread_id` was the only
-        # existing key; nested under "cron" we expose job_id/job_name/schedule/
-        # deliver/origin. Adapters that ignore unknown keys are unaffected.
-        cron_meta = {
-            "job_id": job.get("id", ""),
-            "job_name": job.get("name") or job.get("id", ""),
-            "schedule": job.get("schedule"),
-            "deliver": job.get("deliver"),
-            "origin": origin or None,
-        }
-        cron_meta = {k: v for k, v in cron_meta.items() if v}
-
         send_metadata: Optional[dict] = None
         if thread_id:
             send_metadata = {"thread_id": thread_id}

From 7f41d69de2009526d05d751759ea2387c32df691 Mon Sep 17 00:00:00 2001
From: briandevans <252620095+briandevans@users.noreply.github.com>
Date: Wed, 20 May 2026 09:09:42 -0700
Subject: [PATCH 3/4] fixup(cron): add status + ran_at fields to structured
 cron metadata
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Incorporates @deestax's suggested extensions from #26012 review:

- `status`: "ok" | "error" — set by `tick()` based on the run's `success` bool,
  threaded through `_deliver_result` as a `status_hint` kwarg (defaults to "ok"
  for non-tick call sites). Lets plugin adapters distinguish failed-run
  deliveries from successful ones without re-parsing the wrap envelope, which
  also doesn't work when `cron.wrap_response: false`.
- `ran_at`: ISO timestamp from `_hermes_now()`. Trivial to compute and lets
  adapters annotate cron deliveries with a verifiable run time independent of
  message-receive time.

Both are unconditional fields (the truthy-only filter only strips `None`s, so
both reach adapters without the falsy-filter problem fixed in 16fc509d).

Tests:
- New failure-path test asserts `status == "error"` and full context survives
  when `wrap_response=false` (the brittle path Step 2 of #26004 cares about).
- Updated the omits-empty-fields exact-match assertion to pop `ran_at` (varies
  by clock) and expect `status: "ok"` on the default path.
---
 cron/scheduler.py            | 18 ++++++++++++--
 tests/cron/test_scheduler.py | 46 ++++++++++++++++++++++++++++++++++++
 2 files changed, 62 insertions(+), 2 deletions(-)

diff --git a/cron/scheduler.py b/cron/scheduler.py
index 0fd99b22f5f9..b483af0c2f35 100644
--- a/cron/scheduler.py
+++ b/cron/scheduler.py
@@ -615,7 +615,13 @@ def _send_media_via_adapter(
             logger.warning("Job '%s': failed to send media %s: %s", job.get("id", "?"), media_path, e)
 
 
-def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Optional[str]:
+def _deliver_result(
+    job: dict,
+    content: str,
+    adapters=None,
+    loop=None,
+    status_hint: str = "ok",
+) -> Optional[str]:
     """
     Deliver job output to the configured target(s) (origin chat, specific platform, etc.).
 
@@ -686,6 +692,8 @@ def _deliver_result(job: dict, content: str, adapters=None, loop=None) -> Option
         "schedule": job.get("schedule"),
         "deliver": job.get("deliver"),
         "origin": origin or None,
+        "status": status_hint,
+        "ran_at": _hermes_now().isoformat(),
     }
     cron_meta = {k: v for k, v in cron_meta_full.items() if v is not None}
 
@@ -1971,7 +1979,13 @@ def _process_job(job: dict) -> bool:
                 delivery_error = None
                 if should_deliver:
                     try:
-                        delivery_error = _deliver_result(job, deliver_content, adapters=adapters, loop=loop)
+                        delivery_error = _deliver_result(
+                            job,
+                            deliver_content,
+                            adapters=adapters,
+                            loop=loop,
+                            status_hint="ok" if success else "error",
+                        )
                     except Exception as de:
                         delivery_error = str(de)
                         logger.error("Delivery failed for job %s: %s", job["id"], de)
diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py
index f6a0badbb486..434a8a9b7c98 100644
--- a/tests/cron/test_scheduler.py
+++ b/tests/cron/test_scheduler.py
@@ -993,13 +993,59 @@ def test_cron_metadata_omits_empty_fields(self):
         assert "schedule" not in cron_ctx
         # job_id is present (required), job_name falls back to it,
         # deliver and origin are present.
+        # status defaults to "ok"; ran_at is always set (ISO timestamp).
+        ran_at = cron_ctx.pop("ran_at", None)
+        assert isinstance(ran_at, str) and "T" in ran_at
         assert cron_ctx == {
             "job_id": "min-job",
             "job_name": "min-job",
             "deliver": "origin",
             "origin": {"platform": "telegram", "chat_id": "5"},
+            "status": "ok",
         }
 
+    def test_cron_metadata_attached_on_failed_run_with_wrap_disabled(self):
+        """Failed run + wrap_response=false must still surface status=error and full context.
+
+        Regression guard for #26004 step 2: without this, adapters that rely on
+        `metadata["cron"]["status"]` to distinguish success from failure (e.g.
+        for routing failure summaries to a different chat) would always see
+        the default "ok" even when the agent raised.
+        """
+        from gateway.config import Platform
+
+        adapter, loop, fake_run_coro = self._build_adapter_send_mock()
+        pconfig = MagicMock()
+        pconfig.enabled = True
+        mock_cfg = MagicMock()
+        mock_cfg.platforms = {Platform.TELEGRAM: pconfig}
+
+        job = {
+            "id": "fail-job",
+            "name": "nightly-report",
+            "schedule": "0 3 * * *",
+            "deliver": "origin",
+            "origin": {"platform": "telegram", "chat_id": "999"},
+        }
+
+        with patch("gateway.config.load_gateway_config", return_value=mock_cfg), \
+             patch("cron.scheduler.load_config", return_value={"cron": {"wrap_response": False}}), \
+             patch("asyncio.run_coroutine_threadsafe", side_effect=fake_run_coro):
+            _deliver_result(
+                job,
+                "Job hit an exception",
+                adapters={Platform.TELEGRAM: adapter},
+                loop=loop,
+                status_hint="error",
+            )
+
+        cron_ctx = adapter.send.call_args.kwargs["metadata"]["cron"]
+        assert cron_ctx["status"] == "error"
+        assert cron_ctx["job_id"] == "fail-job"
+        assert cron_ctx["job_name"] == "nightly-report"
+        assert cron_ctx["schedule"] == "0 3 * * *"
+        assert "ran_at" in cron_ctx and "T" in cron_ctx["ran_at"]
+
 
 class TestDeliverResultErrorReturns:
     """Verify _deliver_result returns error strings on failure, None on success."""

From 2f1560d649f1f0fffe9e83acecd7bbb9cee91f7a Mon Sep 17 00:00:00 2001
From: briandevans <252620095+briandevans@users.noreply.github.com>
Date: Wed, 20 May 2026 11:11:16 -0700
Subject: [PATCH 4/4] test(cron): update thread_fallback assertion for
 structured cron metadata
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The test added in d81b88880 asserts the exact metadata payload sent to
the live adapter. After this PR enriches send_metadata with a "cron"
context block, the exact-equality assertion no longer holds — but the
load-bearing invariant (thread_id is preserved on fallback) still does.

Rewrite the assertion to check call count, positional args, and the
thread_id field directly, plus a couple of cron context fields to keep
the structured-metadata path covered.
---
 tests/cron/test_scheduler.py | 14 +++++++++-----
 1 file changed, 9 insertions(+), 5 deletions(-)

diff --git a/tests/cron/test_scheduler.py b/tests/cron/test_scheduler.py
index 434a8a9b7c98..11bbe0c1866c 100644
--- a/tests/cron/test_scheduler.py
+++ b/tests/cron/test_scheduler.py
@@ -2736,11 +2736,15 @@ def fake_run_coro(coro, _loop):
             "configured thread_id 7072 for telegram:226252250 was not found; "
             "delivered without thread_id"
         )
-        adapter.send.assert_called_once_with(
-            "226252250",
-            "Hello world",
-            metadata={"thread_id": "7072"},
-        )
+        assert adapter.send.call_count == 1
+        args, kwargs = adapter.send.call_args
+        assert args == ("226252250", "Hello world")
+        metadata = kwargs["metadata"]
+        # thread_id is the load-bearing field for this test — it must survive
+        # alongside the structured cron context added in #26004.
+        assert metadata["thread_id"] == "7072"
+        assert metadata["cron"]["job_id"] == "thread-fallback-job"
+        assert metadata["cron"]["deliver"] == "telegram:226252250:7072"
 
 
 class TestSendMediaTimeoutCancelsFuture: