diff --git a/PATCHES.md b/PATCHES.md index ed899a744ef..45dee559dea 100644 --- a/PATCHES.md +++ b/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):** diff --git a/backend/app/channels/_delivery_report.py b/backend/app/channels/_delivery_report.py new file mode 100644 index 00000000000..31ff8444394 --- /dev/null +++ b/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 diff --git a/backend/app/channels/manager.py b/backend/app/channels/manager.py index 3e83ac50916..6d4973bb7e5 100644 --- a/backend/app/channels/manager.py +++ b/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 -------------------------------------------------- diff --git a/backend/app/channels/message_bus.py b/backend/app/channels/message_bus.py index 4d915da26ef..235bb64ed90 100644 --- a/backend/app/channels/message_bus.py +++ b/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 diff --git a/backend/app/gateway/routers/playbooks.py b/backend/app/gateway/routers/playbooks.py index 9042d95b65c..f1f4ded72e3 100644 --- a/backend/app/gateway/routers/playbooks.py +++ b/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) diff --git a/backend/tests/test_delivery_report.py b/backend/tests/test_delivery_report.py new file mode 100644 index 00000000000..276f42da7f2 --- /dev/null +++ b/backend/tests/test_delivery_report.py @@ -0,0 +1,182 @@ +"""[argus patch #45] Delivery-outcome callback (gateway -> Chronos). + +Locks the contract of app.channels._delivery_report.report_delivery — the +POST payload shape, the internal-token header, the retry-once-never-raise +behavior — and the manager's _report_unattended_outcome gate (unattended + +report_url only). +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest + +import app.channels._delivery_report as delivery_report +import app.gateway.internal_auth as internal_auth +from app.channels._delivery_report import report_delivery +from app.channels.manager import ChannelManager +from app.channels.message_bus import InboundMessage + +REPORT_URL = "http://argus-scheduler:8000/api/runs/7/output" +TEST_TOKEN = "test-internal-token" + + +@pytest.fixture(autouse=True) +def _pin_internal_token(): + with patch.object(internal_auth, "_INTERNAL_AUTH_TOKEN", TEST_TOKEN): + yield + + +@pytest.fixture(autouse=True) +def _no_retry_sleep(): + with patch.object(delivery_report.asyncio, "sleep", AsyncMock()): + yield + + +class _CapturingClient: + """Stand-in for httpx.AsyncClient capturing posts; scriptable responses.""" + + def __init__(self, outcomes): + # outcomes: list of "ok", "http500", or an Exception instance + self.outcomes = list(outcomes) + self.posts: list[dict] = [] + + def __call__(self, *args, **kwargs): # constructor stand-in + return self + + async def __aenter__(self): + return self + + async def __aexit__(self, *exc): + return False + + async def post(self, url, *, json=None, headers=None): + self.posts.append({"url": url, "json": json, "headers": headers}) + outcome = self.outcomes.pop(0) + if isinstance(outcome, Exception): + raise outcome + req = httpx.Request("POST", url) + if outcome == "http500": + resp = httpx.Response(500, request=req) + else: + resp = httpx.Response(200, request=req) + return resp + + +class TestReportDelivery: + @pytest.mark.asyncio + async def test_posts_payload_with_internal_token(self): + fake = _CapturingClient(["ok"]) + with patch.object(delivery_report.httpx, "AsyncClient", fake): + ok = await report_delivery( + REPORT_URL, status="delivered", channel="telegram", + chat_id="8726302666", message_text="Meeting brief ...", + ) + assert ok is True + assert len(fake.posts) == 1 + post = fake.posts[0] + assert post["url"] == REPORT_URL + assert post["headers"][internal_auth.INTERNAL_AUTH_HEADER_NAME] == TEST_TOKEN + body = post["json"] + assert body["status"] == "delivered" + assert body["channel"] == "telegram" + assert body["chat_id"] == "8726302666" + assert body["message_text"] == "Meeting brief ..." + assert "delivered_at" in body # ISO stamp only on delivered + + @pytest.mark.asyncio + async def test_silent_payload_is_minimal(self): + fake = _CapturingClient(["ok"]) + with patch.object(delivery_report.httpx, "AsyncClient", fake): + ok = await report_delivery(REPORT_URL, status="silent", channel="telegram", chat_id="1") + assert ok is True + body = fake.posts[0]["json"] + assert body["status"] == "silent" + assert "message_text" not in body + assert "delivered_at" not in body + + @pytest.mark.asyncio + async def test_failed_payload_carries_error(self): + fake = _CapturingClient(["ok"]) + with patch.object(delivery_report.httpx, "AsyncClient", fake): + await report_delivery(REPORT_URL, status="failed", error="RuntimeError('boom')") + body = fake.posts[0]["json"] + assert body["status"] == "failed" + assert body["error"] == "RuntimeError('boom')" + + @pytest.mark.asyncio + async def test_none_url_is_noop(self): + fake = _CapturingClient([]) + with patch.object(delivery_report.httpx, "AsyncClient", fake): + ok = await report_delivery(None, status="silent") + assert ok is False + assert fake.posts == [] + + @pytest.mark.asyncio + async def test_retries_once_then_succeeds(self): + fake = _CapturingClient([httpx.ConnectError("down"), "ok"]) + with patch.object(delivery_report.httpx, "AsyncClient", fake): + ok = await report_delivery(REPORT_URL, status="silent") + assert ok is True + assert len(fake.posts) == 2 + + @pytest.mark.asyncio + async def test_gives_up_after_two_attempts_never_raises(self): + fake = _CapturingClient([httpx.ConnectError("down"), "http500"]) + with patch.object(delivery_report.httpx, "AsyncClient", fake): + ok = await report_delivery(REPORT_URL, status="delivered") + assert ok is False + assert len(fake.posts) == 2 # exactly one retry, no loop + + @pytest.mark.asyncio + async def test_long_message_text_is_capped(self): + fake = _CapturingClient(["ok"]) + with patch.object(delivery_report.httpx, "AsyncClient", fake): + await report_delivery(REPORT_URL, status="delivered", message_text="x" * 20000) + text = fake.posts[0]["json"]["message_text"] + assert len(text) < 20000 + assert text.endswith("…[truncated]") + + +def _msg(*, unattended=True, report_url=REPORT_URL) -> InboundMessage: + return InboundMessage( + channel_name="telegram", chat_id="8726302666", user_id="u1", + text="playbook prompt", unattended=unattended, report_url=report_url, + ) + + +class TestManagerOutcomeGate: + """_report_unattended_outcome only reports for unattended fires that + carried a report_url; interactive turns and pre-#45 fires are no-ops.""" + + def _manager(self) -> ChannelManager: + mgr = ChannelManager.__new__(ChannelManager) # gate needs no manager state + return mgr + + @pytest.mark.asyncio + async def test_reports_for_unattended_with_url(self): + mgr = self._manager() + with patch.object(delivery_report, "report_delivery", AsyncMock()) as rep, \ + patch("app.channels.manager.report_delivery", rep): + await mgr._report_unattended_outcome(_msg(), status="silent") + rep.assert_awaited_once() + kwargs = rep.await_args.kwargs + assert kwargs["status"] == "silent" + assert kwargs["channel"] == "telegram" + assert kwargs["chat_id"] == "8726302666" + + @pytest.mark.asyncio + async def test_noop_without_report_url(self): + mgr = self._manager() + with patch("app.channels.manager.report_delivery", AsyncMock()) as rep: + await mgr._report_unattended_outcome(_msg(report_url=None), status="silent") + rep.assert_not_awaited() + + @pytest.mark.asyncio + async def test_noop_for_interactive_turn(self): + mgr = self._manager() + with patch("app.channels.manager.report_delivery", AsyncMock()) as rep: + await mgr._report_unattended_outcome(_msg(unattended=False), status="delivered") + rep.assert_not_awaited() diff --git a/backend/tests/test_playbook_fire.py b/backend/tests/test_playbook_fire.py index 5eda491a6e6..af98bbcf839 100644 --- a/backend/tests/test_playbook_fire.py +++ b/backend/tests/test_playbook_fire.py @@ -148,6 +148,27 @@ def test_allowed_tools_none_by_default(self, client, playbook_dir): msg = service.bus.publish_inbound.await_args.args[0] assert msg.allowed_tools is None + def test_report_url_flows_through(self, client, playbook_dir): + """[argus patch #45] Chronos's delivery-report callback URL rides the + InboundMessage so the channel manager can POST the outcome.""" + service = _make_service() + url = "http://argus-scheduler:8000/api/runs/123/output" + with patch("app.channels.service.get_channel_service", return_value=service): + resp = _post(client, body={"report_url": url}) + assert resp.status_code == 200 + msg = service.bus.publish_inbound.await_args.args[0] + assert msg.report_url == url + + def test_report_url_none_by_default(self, client, playbook_dir): + """A fire without report_url (pre-#45 Chronos) behaves exactly as + before: msg.report_url is None and nothing is reported.""" + service = _make_service() + with patch("app.channels.service.get_channel_service", return_value=service): + resp = _post(client) + assert resp.status_code == 200 + msg = service.bus.publish_inbound.await_args.args[0] + assert msg.report_url is None + def test_fires_for_every_mapped_chat(self, client, playbook_dir): chats = [ {"channel_name": "telegram", "chat_id": "111", "user_id": "111"},