[argus] patch #45: delivery-report callback for scheduled playbook fires

PATCHES.md

@@ -70,6 +70,7 @@ half is upstreamable, the Argus behavior lives in project config).
 | [#42](#patch-42) | subtask card false-"failed" on transient SSE loading gaps | argus-edit | 68a7fd37 |
 | [#43](#patch-43) | per-run allowed-tools from schedule frontmatter | argus-additive | this PR |
 | [#44](#patch-44) | unattended-silence: no blank-final retry, wider narration backstop, no token logging | argus-edit | this PR |
+| [#45](#patch-45) | delivery-report callback for scheduled playbook fires | argus-additive | this PR |
 
 Dropped / deferred / not-carried records are at the bottom, followed by the
 carry budget ledger.
@@ -954,6 +955,45 @@ backstop, no token logging**
 
 ---
 
+## Patch #45
+
+**Patch #45 - delivery-report callback for scheduled playbook fires**
+
+- Class: argus-additive (new `_delivery_report.py`; one field on
+  InboundMessage/PlaybookFireRequest; four call sites at named seams in
+  `manager.py`)
+- Intent: Chronos fires a playbook with a per-run ``report_url`` and has
+  handled a ``delivered|silent|failed`` callback since its patch-#30-era
+  contract (``scheduler/src/routes/api.py::report_run_output``) — but the
+  gateway never called back, so every ``channel_notify`` run sat ``running``
+  for CHRONOS_DELIVERY_REPORT_TIMEOUT_SEC (1800s) and closed as
+  ``ok/unreported``; the dashboard could not tell a delivered briefing from
+  a suppressed-silent tick (patch #31/#32/#34/#44) or a failed turn. The
+  fire endpoint now forwards ``report_url`` onto the InboundMessage and the
+  channel manager POSTs the outcome once the async turn resolves:
+  ``silent`` from both unattended-suppression branches, ``delivered`` after
+  the outbound publish (with the message text), ``failed`` on a captured
+  stream error or a thread-busy rejection. One POST, one retry, never
+  raises — a report failure degrades to exactly the pre-#45 unreported
+  behavior. Interactive turns and fires without ``report_url`` are
+  byte-for-byte unchanged.
+- Files: `backend/app/channels/_delivery_report.py` (NEW),
+  `backend/app/channels/message_bus.py` (EDITED, +field),
+  `backend/app/gateway/routers/playbooks.py` (EDITED, +request field,
+  +forward), `backend/app/channels/manager.py` (EDITED,
+  +`_report_unattended_outcome` + 4 call sites)
+- Tests: `backend/tests/test_delivery_report.py` (NEW: payload shape,
+  internal-token header, retry-once/never-raise, message-text cap, manager
+  gate), `backend/tests/test_playbook_fire.py` (EDITED, +2:
+  report_url flow-through, default-None)
+- Delete-when: the schedule system is re-expressed on a native scheduling
+  API where the scheduler owns the turn lifecycle end-to-end (no async
+  fire-and-deliver seam to report across).
+- Upstream status: none (argus-additive; depends on the argus-only
+  playbook fire endpoint and Chronos).
+
+---
+
 ## Dropped / deferred / re-expressed (v2.0.0 rebase record - do not re-add blindly)
 
 **Dropped as upstream-subsumed (verified during the 2026-06-29/30 rebase):**

backend/app/channels/_delivery_report.py

@@ -0,0 +1,105 @@
+"""[argus patch #45] Delivery-outcome callback for scheduled playbook fires.
+
+Chronos fires a playbook with an optional ``report_url``
+(``{CHRONOS_INTERNAL_URL}/api/runs/{run_id}/output``) and closes the run row
+only when the gateway reports how the unattended turn actually ended. Until
+this patch the gateway never called back, so every ``channel_notify`` run sat
+in ``running`` for CHRONOS_DELIVERY_REPORT_TIMEOUT_SEC (default 1800s) and was
+closed as ``ok/unreported`` — the dashboard could not distinguish a delivered
+briefing from a suppressed-silent tick or a failed turn.
+
+Contract (mirrors Chronos ``scheduler/src/deerflow_client.py::fire_playbook``
+and ``scheduler/src/routes/api.py::report_run_output``):
+
+    POST {report_url}
+    Header: X-DeerFlow-Internal-Token (same global secret Chronos used to fire)
+    Body: {"status": "delivered"|"silent"|"failed",
+           "channel": str?, "chat_id": str?, "message_text": str?,
+           "delivered_at": ISO-8601 str?, "error": str?}
+
+One POST, one retry on any failure, then give up with a log line — a report
+failure must never fail (or delay-loop) the delivery itself. Fire-and-forget
+callers should still ``await``: the timeout budget is small and awaiting keeps
+ordering deterministic for tests and log correlation.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import UTC, datetime
+from typing import Literal
+
+import httpx
+
+logger = logging.getLogger(__name__)
+
+# Chronos is one podman-network hop away; anything slower than this is down.
+_REPORT_TIMEOUT_SECONDS = 10.0
+_RETRY_DELAY_SECONDS = 2.0
+# Defensive cap: message_text is stored in a JSONB column for the dashboard;
+# the run detail page shows the full text but nobody needs a megabyte of it.
+_MESSAGE_TEXT_CAP = 8192
+
+DeliveryStatus = Literal["delivered", "silent", "failed"]
+
+
+async def report_delivery(
+    report_url: str | None,
+    *,
+    status: DeliveryStatus,
+    channel: str | None = None,
+    chat_id: str | None = None,
+    message_text: str | None = None,
+    error: str | None = None,
+) -> bool:
+    """POST the delivery outcome to *report_url*. Returns True when accepted.
+
+    No-op (False) when *report_url* is empty. Never raises: one retry after a
+    short delay, then a warning log — Chronos's poller closes the run as
+    ``ok/unreported`` on its own timeout, exactly the pre-#45 behavior.
+    """
+    if not report_url:
+        return False
+
+    from app.gateway.internal_auth import create_internal_auth_headers
+
+    if message_text and len(message_text) > _MESSAGE_TEXT_CAP:
+        message_text = message_text[:_MESSAGE_TEXT_CAP] + " …[truncated]"
+
+    payload: dict[str, str] = {"status": status}
+    if channel:
+        payload["channel"] = channel
+    if chat_id:
+        payload["chat_id"] = chat_id
+    if message_text:
+        payload["message_text"] = message_text
+    if error:
+        payload["error"] = error
+    if status == "delivered":
+        payload["delivered_at"] = datetime.now(UTC).isoformat()
+
+    headers = create_internal_auth_headers()
+    for attempt in (1, 2):
+        try:
+            async with httpx.AsyncClient(timeout=_REPORT_TIMEOUT_SECONDS) as client:
+                resp = await client.post(report_url, json=payload, headers=headers)
+                resp.raise_for_status()
+            logger.info(
+                "[DeliveryReport] reported %s to %s (attempt %d)", status, report_url, attempt
+            )
+            return True
+        except Exception as exc:  # noqa: BLE001 — a report failure must never propagate
+            if attempt == 1:
+                logger.warning(
+                    "[DeliveryReport] report to %s failed (attempt 1, retrying): %r", report_url, exc
+                )
+                await asyncio.sleep(_RETRY_DELAY_SECONDS)
+            else:
+                logger.warning(
+                    "[DeliveryReport] report to %s failed (attempt 2, giving up; Chronos will "
+                    "close the run as unreported): %r",
+                    report_url,
+                    exc,
+                )
+    return False

backend/app/channels/manager.py

@@ -16,6 +16,7 @@
 import httpx
 from langgraph_sdk.errors import ConflictError
 
+from app.channels._delivery_report import report_delivery
 from app.channels.commands import KNOWN_CHANNEL_COMMANDS
 from app.channels.message_bus import (
     PENDING_CLARIFICATION_METADATA_KEY,
@@ -1587,8 +1588,11 @@ async def _handle_chat(
             if _is_thread_busy_error(exc):
                 logger.warning("[Manager] thread busy (concurrent run rejected): thread_id=%s", thread_id)
                 await self._send_error(msg, THREAD_BUSY_MESSAGE)
+                await self._report_unattended_outcome(msg, status="failed", error=repr(exc))
                 return
             else:
+                # Propagates to the bus dispatcher; an unattended fire without a
+                # report lands as Chronos's ok/unreported timeout fallback.
                 raise
 
         response_text = _extract_response_text(result)
@@ -1615,6 +1619,7 @@ async def _handle_chat(
         # and real errors are untouched.
         if msg.unattended and not attachments and _is_trivial_unattended_text(response_text):
             logger.info("[Manager] unattended turn produced no content; staying silent (channel=%s chat_id=%s)", msg.channel_name, msg.chat_id)
+            await self._report_unattended_outcome(msg, status="silent")
             return
 
         # [argus patch #36] Use _is_blank_text, not `if not response_text`: a
@@ -1640,6 +1645,30 @@ async def _handle_chat(
         )
         logger.info("[Manager] publishing outbound message to bus: channel=%s, chat_id=%s", msg.channel_name, msg.chat_id)
         await self.bus.publish_outbound(outbound)
+        await self._report_unattended_outcome(msg, status="delivered", message_text=response_text)
+
+    async def _report_unattended_outcome(
+        self,
+        msg: InboundMessage,
+        *,
+        status: str,
+        message_text: str | None = None,
+        error: str | None = None,
+    ) -> None:
+        """[argus patch #45] Report a scheduled turn's delivery outcome to
+        Chronos. No-op unless this is an unattended fire that carried a
+        report_url; never raises (see _delivery_report.report_delivery)."""
+        report_url = getattr(msg, "report_url", None)
+        if not (msg.unattended and report_url):
+            return
+        await report_delivery(
+            report_url,
+            status=status,  # type: ignore[arg-type]
+            channel=msg.channel_name,
+            chat_id=msg.chat_id,
+            message_text=message_text,
+            error=error,
+        )
 
     async def _handle_streaming_chat(
         self,
@@ -1803,6 +1832,7 @@ async def _handle_streaming_chat(
 
             if suppress_final:
                 logger.info("[Manager] unattended streaming turn produced no content; staying silent (channel=%s chat_id=%s)", msg.channel_name, msg.chat_id)
+                await self._report_unattended_outcome(msg, status="silent")
             else:
                 logger.info(
                     "[Manager] streaming response completed: thread_id=%s, response_len=%d, artifacts=%d, error=%s",
@@ -1826,6 +1856,12 @@ async def _handle_streaming_chat(
                         metadata=_response_metadata(msg.metadata, pending_clarification=pending_clarification),
                     )
                 )
+                # [argus patch #45] a captured stream error was delivered to the
+                # chat as an error message; the run itself still failed.
+                if stream_error:
+                    await self._report_unattended_outcome(msg, status="failed", error=repr(stream_error))
+                else:
+                    await self._report_unattended_outcome(msg, status="delivered", message_text=response_text)
 
     # -- command handling --------------------------------------------------
 

backend/app/channels/message_bus.py

@@ -78,6 +78,11 @@ class InboundMessage:
     unattended: bool = False
     memory_mode: str | None = None
     allowed_tools: list[str] | None = None
+    # [argus patch #45] Where to POST the delivery outcome
+    # (delivered|silent|failed) once this unattended turn resolves. Set by the
+    # playbook fire endpoint from Chronos's report_url; None for interactive
+    # messages and pre-#45 Chronos fires.
+    report_url: str | None = None
 
 
 @dataclass

backend/app/gateway/routers/playbooks.py

@@ -66,6 +66,11 @@ class PlaybookFireRequest(BaseModel):
     # [argus patch #43] Per-run tool whitelist declared in the schedule
     # frontmatter. Merged with skill allowed-tools by the tool policy filter.
     allowed_tools: list[str] | None = None
+    # [argus patch #45] Chronos's per-run delivery-report callback URL
+    # ({CHRONOS_INTERNAL_URL}/api/runs/{run_id}/output). The channel manager
+    # POSTs the outcome (delivered|silent|failed) there once the async turn
+    # resolves. Optional: a fire without it behaves exactly as before.
+    report_url: str | None = None
 
 
 class PlaybookFireResponse(BaseModel):
@@ -164,6 +169,7 @@ async def fire_playbook(schedule_id: str, body: PlaybookFireRequest, request: Re
             unattended=True,
             memory_mode=memory_mode,
             allowed_tools=body.allowed_tools,
+            report_url=body.report_url,
             metadata={"source": "playbook", "schedule_id": schedule_id},
         )
         await service.bus.publish_inbound(msg)