From b13f7eee56f94448c61788f3d0ea4114658336c0 Mon Sep 17 00:00:00 2001 From: Asaf Varon Date: Fri, 16 Oct 2026 15:47:45 +0300 Subject: [PATCH 1/2] feat(manager): log channel status and webhook rejections The worker logs one `channel status:` line per configured interval with queue counts, the oldest unfinished and unsent ages, and unknown, failed, and blocked reply counts. The read is bounded to five seconds so monitoring cannot stall scheduling or a graceful drain. Every rejected webhook request logs a fixed reason. No line contains message text, sender IDs, tokens, or provider IDs. Adds EXTRA_WHATSAPP_STATUS_LOG_SECONDS (default 60). Co-Authored-By: Claude Opus 5.5 --- src/agent_manager/api/routes/whatsapp.py | 26 +- .../application/channels/worker.py | 45 ++++ .../application/channels/worker_options.py | 2 + src/agent_manager/composition.py | 12 +- src/agent_manager/config.py | 1 + src/agent_manager/domain/channels/status.py | 29 +++ .../persistence/channels/status.py | 104 ++++++++ tests/agent_manager/test_channel_status.py | 230 ++++++++++++++++++ tests/agent_manager/test_whatsapp_webhook.py | 36 +++ 9 files changed, 474 insertions(+), 11 deletions(-) create mode 100644 src/agent_manager/domain/channels/status.py create mode 100644 src/agent_manager/infrastructure/persistence/channels/status.py create mode 100644 tests/agent_manager/test_channel_status.py diff --git a/src/agent_manager/api/routes/whatsapp.py b/src/agent_manager/api/routes/whatsapp.py index 32d21aa2..978c2e61 100644 --- a/src/agent_manager/api/routes/whatsapp.py +++ b/src/agent_manager/api/routes/whatsapp.py @@ -18,6 +18,12 @@ router = APIRouter(prefix="/channels/whatsapp", tags=["whatsapp"]) +def _rejected(status_code: int, detail: str, reason: str) -> HTTPException: + """Log a fixed reason only; request content never reaches ordinary logs.""" + logger.warning("whatsapp webhook rejected: reason=%s", reason) + return HTTPException(status_code=status_code, detail=detail) + + def _webhook(request: Request) -> WhatsAppWebhook: webhook = getattr(request.app.state, "whatsapp_webhook", None) if webhook is None: @@ -30,7 +36,7 @@ async def verify_subscription(request: Request) -> PlainTextResponse: webhook = _webhook(request) names = ("hub.mode", "hub.verify_token", "hub.challenge") if any(len(request.query_params.getlist(name)) != 1 for name in names): - raise HTTPException(status_code=403, detail="invalid webhook subscription") + raise _rejected(403, "invalid webhook subscription", "subscription") try: challenge = subscription_challenge( request.query_params.get("hub.mode"), @@ -39,7 +45,7 @@ async def verify_subscription(request: Request) -> PlainTextResponse: expected=webhook.config.verify_token.get_secret_value(), ) except WebhookAuthenticationError: - raise HTTPException(status_code=403, detail="invalid webhook subscription") from None + raise _rejected(403, "invalid webhook subscription", "subscription") from None return PlainTextResponse(challenge) @@ -48,16 +54,16 @@ async def receive_notification(request: Request) -> dict[str, str]: webhook = _webhook(request) signatures = request.headers.getlist("X-Hub-Signature-256") if len(signatures) != 1: - raise HTTPException(status_code=401, detail="invalid webhook signature") + raise _rejected(401, "invalid webhook signature", "signature") if request.headers.get("content-encoding", "identity") != "identity": - raise HTTPException(status_code=415, detail="encoded webhook bodies are unsupported") + raise _rejected(415, "encoded webhook bodies are unsupported", "encoding") if request.headers.get("content-type", "").split(";", 1)[0].lower() != "application/json": - raise HTTPException(status_code=415, detail="webhook requires application/json") + raise _rejected(415, "webhook requires application/json", "content_type") body = await _bounded_body(request, webhook.config.max_body_bytes) try: await webhook.receive(body, signatures[0], datetime.now(UTC)) except WebhookAuthenticationError: - raise HTTPException(status_code=401, detail="invalid webhook signature") from None + raise _rejected(401, "invalid webhook signature", "signature") from None except Exception as exc: # The failure boundary must never reflect payloads, identifiers, or DB # exception text into provider responses or ordinary logs. @@ -73,15 +79,15 @@ async def _bounded_body(request: Request, limit: int) -> bytes: if len(lengths) > 1 or ( lengths and (len(lengths[0]) > 20 or not lengths[0].isascii() or not lengths[0].isdecimal()) ): - raise HTTPException(status_code=400, detail="invalid content length") + raise _rejected(400, "invalid content length", "length") if lengths and int(lengths[0]) > limit: - raise HTTPException(status_code=413, detail="webhook body too large") + raise _rejected(413, "webhook body too large", "size") body = bytearray() try: async for chunk in request.stream(): if len(body) + len(chunk) > limit: - raise HTTPException(status_code=413, detail="webhook body too large") + raise _rejected(413, "webhook body too large", "size") body.extend(chunk) except ClientDisconnect: - raise HTTPException(status_code=400, detail="incomplete webhook body") from None + raise _rejected(400, "incomplete webhook body", "incomplete") from None return bytes(body) diff --git a/src/agent_manager/application/channels/worker.py b/src/agent_manager/application/channels/worker.py index 002f2d60..3bf8fb8f 100644 --- a/src/agent_manager/application/channels/worker.py +++ b/src/agent_manager/application/channels/worker.py @@ -6,6 +6,7 @@ import logging from collections import deque from contextlib import AsyncExitStack +from datetime import UTC, datetime from types import TracebackType from agent_manager.application.channels.delivery import ChannelDeliveryService @@ -17,8 +18,10 @@ ChannelWorkerOwnershipLost, ) from agent_manager.domain.channels.repository import ChannelRepository +from agent_manager.domain.channels.status import ChannelStatusReader logger = logging.getLogger(__name__) +STATUS_READ_TIMEOUT_SECONDS = 5.0 class ChannelWorker: @@ -30,12 +33,15 @@ def __init__( options: ChannelWorkerOptions, *, delivery: ChannelDeliveryService | None = None, + status: ChannelStatusReader | None = None, ) -> None: self._repository = repository self._processor = processor self._ownership = ownership self._options = options self._delivery = delivery + self._status = status + self._status_due = 0.0 self._stack = AsyncExitStack() self._stop = asyncio.Event() self._jobs: dict[str, asyncio.Task[None]] = {} @@ -110,6 +116,8 @@ async def _serve(self) -> None: raise except Exception as exc: logger.warning("channel scheduling failed: type=%s", type(exc).__name__) + if self._status is not None and not self._stop.is_set(): + await self._log_status(self._status) stop_waiter = asyncio.create_task(self._stop.wait()) try: await asyncio.wait( @@ -163,6 +171,39 @@ async def _schedule(self) -> None: if item.event_id not in self._jobs: self._jobs[item.event_id] = asyncio.create_task(self._processor.process(item)) + async def _log_status(self, reader: ChannelStatusReader) -> None: + due = asyncio.get_running_loop().time() + if due < self._status_due: + return + self._status_due = due + self._options.status_seconds + try: + # Monitoring must never stall scheduling or a graceful drain. + async with asyncio.timeout(STATUS_READ_TIMEOUT_SECONDS): + status = await reader.read( + self._options.connection_id, self._options.system_id, datetime.now(UTC) + ) + except Exception as exc: + logger.warning("channel status unavailable: type=%s", type(exc).__name__) + return + logger.info( + "channel status: connection=%s inbox_queued=%d inbox_running=%d " + "inbox_answered=%d inbox_oldest_s=%s needs_review=%d inbox_failed=%d " + "outbox_pending=%d outbox_oldest_s=%s delivery_unknown=%d " + "delivery_failed=%d delivery_blocked=%d", + self._options.connection_id, + status.inbox_queued, + status.inbox_running, + status.inbox_answered, + _seconds(status.inbox_oldest_seconds), + status.inbox_needs_review, + status.inbox_failed, + status.outbox_pending, + _seconds(status.outbox_oldest_seconds), + status.delivery_unknown, + status.delivery_failed, + status.delivery_blocked, + ) + def _reap(self) -> None: for event_id, task in tuple(self._jobs.items()): if task.done(): @@ -189,3 +230,7 @@ async def _drain(self, *, immediate: bool) -> None: def _observe_failure(task: asyncio.Task[None]) -> None: if not task.cancelled() and task.exception() is not None: logger.error("channel supervisor stopped: type=%s", type(task.exception()).__name__) + + +def _seconds(value: float | None) -> str: + return "none" if value is None else str(int(value)) diff --git a/src/agent_manager/application/channels/worker_options.py b/src/agent_manager/application/channels/worker_options.py index fafd3f26..79ac7ef7 100644 --- a/src/agent_manager/application/channels/worker_options.py +++ b/src/agent_manager/application/channels/worker_options.py @@ -14,6 +14,7 @@ class ChannelWorkerOptions: poll_seconds: float = 0.5 turn_seconds: float = 120 drain_seconds: float = 15 + status_seconds: float = 60 def __post_init__(self) -> None: bounded_identifier(self.connection_id, limit=128) @@ -25,5 +26,6 @@ def __post_init__(self) -> None: 0 < self.poll_seconds <= 60 and 0 < self.turn_seconds <= 3600 and 0 < self.drain_seconds <= 60 + and 0 < self.status_seconds <= 3600 ): raise ValueError("channel worker time bounds must be finite and positive") diff --git a/src/agent_manager/composition.py b/src/agent_manager/composition.py index 4ffda5d2..609fbd36 100644 --- a/src/agent_manager/composition.py +++ b/src/agent_manager/composition.py @@ -32,6 +32,7 @@ from agent_manager.domain.channels.delivery import DeliveryRepository from agent_manager.domain.channels.identity import ChannelIdentityResolver from agent_manager.domain.channels.repository import ChannelRepository +from agent_manager.domain.channels.status import ChannelStatusReader from agent_manager.domain.channels.webhooks import WebhookRepository from agent_manager.infrastructure.auth.anonymous_identity_source import AnonymousIdentitySource from agent_manager.infrastructure.auth.claim_mapping import ClaimMapping @@ -50,6 +51,7 @@ from agent_manager.infrastructure.persistence.channels.delivery import SqlDeliveryRepository from agent_manager.infrastructure.persistence.channels.ownership import SqlChannelWorkerOwnership from agent_manager.infrastructure.persistence.channels.repository import SqlChannelRepository +from agent_manager.infrastructure.persistence.channels.status import SqlChannelStatusReader from agent_manager.infrastructure.persistence.channels.webhooks import SqlWebhookRepository from agent_manager.infrastructure.persistence.database import create_db_engine, session_factory from agent_manager.infrastructure.persistence.memory_repository import MemoryRepository @@ -72,6 +74,7 @@ class ApplicationRepositories: webhooks: WebhookRepository | None = None deliveries: DeliveryRepository | None = None controls: ControlRepository | None = None + status: ChannelStatusReader | None = None def build_whatsapp_webhook( @@ -123,6 +126,7 @@ async def channel_worker_lifespan( poll_seconds=settings.extra_whatsapp_worker_poll_seconds, turn_seconds=settings.extra_whatsapp_worker_turn_seconds, drain_seconds=settings.extra_whatsapp_worker_drain_seconds, + status_seconds=settings.extra_whatsapp_status_log_seconds, ) identities = load_channel_identity_resolver(settings.extra_whatsapp_identity_resolver) sender: WhatsAppSender | None = None @@ -169,7 +173,12 @@ async def channel_worker_lifespan( settings.effective_database_url, options.connection_id ) async with ChannelWorker( - repositories.channels, processor, ownership, options, delivery=delivery + repositories.channels, + processor, + ownership, + options, + delivery=delivery, + status=repositories.status, ) as worker: yield worker finally: @@ -305,6 +314,7 @@ async def application_repositories( webhooks=SqlWebhookRepository(sessions), deliveries=SqlDeliveryRepository(sessions), controls=SqlControlRepository(sessions), + status=SqlChannelStatusReader(sessions), ) finally: await db_engine.dispose() diff --git a/src/agent_manager/config.py b/src/agent_manager/config.py index 98ccb451..e7592ccb 100644 --- a/src/agent_manager/config.py +++ b/src/agent_manager/config.py @@ -176,6 +176,7 @@ class Settings(BaseSettings): default=120, gt=0, le=3600, allow_inf_nan=False ) extra_whatsapp_worker_drain_seconds: float = Field(default=15, gt=0, le=60, allow_inf_nan=False) + extra_whatsapp_status_log_seconds: float = Field(default=60, gt=0, le=3600, allow_inf_nan=False) extra_whatsapp_sender_enabled: bool = False extra_whatsapp_api_version: str | None = None extra_whatsapp_access_token: SecretStr | None = Field(default=None, repr=False, exclude=True) diff --git a/src/agent_manager/domain/channels/status.py b/src/agent_manager/domain/channels/status.py new file mode 100644 index 00000000..4528e984 --- /dev/null +++ b/src/agent_manager/domain/channels/status.py @@ -0,0 +1,29 @@ +"""Operational counts for one connection, without message content or identifiers.""" + +from abc import ABC, abstractmethod +from dataclasses import dataclass +from datetime import datetime + + +@dataclass(frozen=True) +class ChannelStatus: + inbox_queued: int + inbox_running: int + inbox_answered: int + inbox_needs_review: int + inbox_failed: int + # Seconds since the oldest unfinished inbound message was received. + inbox_oldest_seconds: float | None + # Unsent reply parts; one reply can be split into several parts. + outbox_pending: int + # Seconds since the inbound message behind the oldest unsent part arrived. + outbox_oldest_seconds: float | None + # Replies, not parts, whose delivery ended in each outcome. + delivery_unknown: int + delivery_failed: int + delivery_blocked: int + + +class ChannelStatusReader(ABC): + @abstractmethod + async def read(self, connection_id: str, system_id: str, now: datetime) -> ChannelStatus: ... diff --git a/src/agent_manager/infrastructure/persistence/channels/status.py b/src/agent_manager/infrastructure/persistence/channels/status.py new file mode 100644 index 00000000..3b7ebfff --- /dev/null +++ b/src/agent_manager/infrastructure/persistence/channels/status.py @@ -0,0 +1,104 @@ +"""Grouped counts over channel queues; never selects text or sender identifiers.""" + +from collections.abc import Iterable +from datetime import datetime + +from sqlalchemy import func +from sqlalchemy.ext.asyncio import async_sessionmaker +from sqlmodel import col, select +from sqlmodel.ext.asyncio.session import AsyncSession + +from agent_manager.domain.channels.models import InboxState +from agent_manager.domain.channels.status import ChannelStatus, ChannelStatusReader +from agent_manager.domain.pagination import ensure_utc +from agent_manager.infrastructure.persistence.channels.tables import ( + ChannelDeliveryBatchRow as Batch, +) +from agent_manager.infrastructure.persistence.channels.tables import ( + ChannelDeliveryPartRow as Part, +) +from agent_manager.infrastructure.persistence.channels.tables import ( + ChannelInboxRow as Inbox, +) + +_RUNNING = (InboxState.PREPARING, InboxState.PREPARED, InboxState.RUNNING) +_UNFINISHED = (InboxState.QUEUED, *_RUNNING, InboxState.ANSWERED) +_UNSENT = ("ready", "retry", "sending") + + +class SqlChannelStatusReader(ChannelStatusReader): + def __init__(self, sessions: async_sessionmaker[AsyncSession]) -> None: + self._sessions = sessions + + async def read(self, connection_id: str, system_id: str, now: datetime) -> ChannelStatus: + async with self._sessions() as session: + inbox = ( + await session.exec( + select( + Inbox.state, + func.count(), + func.min(Inbox.received_at), + ) + .where( + col(Inbox.connection_id) == connection_id, + col(Inbox.system_id) == system_id, + ) + .group_by(col(Inbox.state)) + ) + ).all() + parts = ( + await session.exec( + select( + Part.state, + func.count(), + func.min(Inbox.received_at), + ) + .join( + Batch, + col(Batch.event_id) == col(Part.event_id), + ) + .join( + Inbox, + col(Inbox.event_id) == col(Batch.event_id), + ) + .where( + col(Batch.connection_id) == connection_id, + col(Batch.system_id) == system_id, + ) + .group_by(col(Part.state)) + ) + ).all() + # Outcomes count replies: a reply mirrors its part's failure state. + replies = ( + await session.exec( + select(Batch.state, func.count()) + .where( + col(Batch.connection_id) == connection_id, + col(Batch.system_id) == system_id, + ) + .group_by(col(Batch.state)) + ) + ).all() + inbox_counts = {state: count for state, count, _ in inbox} + part_counts = {state: count for state, count, _ in parts} + reply_counts = dict(replies) + return ChannelStatus( + inbox_queued=inbox_counts.get(InboxState.QUEUED, 0), + inbox_running=sum(inbox_counts.get(state, 0) for state in _RUNNING), + inbox_answered=inbox_counts.get(InboxState.ANSWERED, 0), + inbox_needs_review=inbox_counts.get(InboxState.NEEDS_REVIEW, 0), + inbox_failed=inbox_counts.get(InboxState.FAILED, 0), + inbox_oldest_seconds=_age(now, (o for s, _, o in inbox if s in _UNFINISHED)), + outbox_pending=sum(part_counts.get(state, 0) for state in _UNSENT), + outbox_oldest_seconds=_age(now, (o for s, _, o in parts if s in _UNSENT)), + delivery_unknown=reply_counts.get("delivery_unknown", 0), + delivery_failed=reply_counts.get("failed", 0), + delivery_blocked=sum( + count for state, count in reply_counts.items() if state.startswith("blocked_") + ), + ) + + +def _age(now: datetime, oldest: Iterable[datetime | None]) -> float | None: + times = [value for value in map(ensure_utc, oldest) if value is not None] + return max(0.0, (now - min(times)).total_seconds()) if times else None diff --git a/tests/agent_manager/test_channel_status.py b/tests/agent_manager/test_channel_status.py new file mode 100644 index 00000000..d3703e37 --- /dev/null +++ b/tests/agent_manager/test_channel_status.py @@ -0,0 +1,230 @@ +"""Operational status: bounded counts and ages, logged without message identifiers.""" + +import asyncio +import logging +import time +from collections.abc import AsyncIterator +from dataclasses import replace +from datetime import datetime, timedelta +from pathlib import Path +from typing import Any, cast + +import pytest +from pydantic import ValidationError +from sqlmodel import SQLModel + +from agent_manager.application.channels.processing import ChannelTurnProcessor +from agent_manager.application.channels.worker import ChannelWorker +from agent_manager.config import Settings +from agent_manager.domain.channels.repository import ChannelRepository +from agent_manager.domain.channels.status import ChannelStatus, ChannelStatusReader +from agent_manager.infrastructure.persistence.channels.status import SqlChannelStatusReader +from agent_manager.infrastructure.persistence.channels.tables import ( + ChannelDeliveryBatchRow, + ChannelDeliveryPartRow, + ChannelInboxRow, +) +from agent_manager.infrastructure.persistence.database import create_db_engine, session_factory +from tests.agent_manager.test_channel_worker import NOW, OPTIONS, Ownership + +SENDER = "+15550001111" + + +@pytest.fixture +async def reader(tmp_path: Path) -> AsyncIterator[tuple[SqlChannelStatusReader, Any]]: + database = create_db_engine(f"sqlite+aiosqlite:///{tmp_path / 'status.db'}") + async with database.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + sessions = session_factory(database) + try: + yield SqlChannelStatusReader(sessions), sessions + finally: + await database.dispose() + + +def inbox(event_id: str, state: str, age: int, *, connection: str = "c") -> ChannelInboxRow: + return ChannelInboxRow( + event_id=event_id, + connection_id=connection, + system_id="s", + sender_id=SENDER, + provider_message_id=f"wamid.{event_id}", + kind="text", + content="private message text", + received_at=NOW - timedelta(seconds=age), + state=state, + ) + + +def delivery( + event_id: str, state: str, sequence: int, *, parts: int = 1, connection: str = "c" +) -> list[ChannelDeliveryBatchRow | ChannelDeliveryPartRow]: + return [ + ChannelDeliveryBatchRow( + event_id=event_id, + connection_id=connection, + system_id="s", + sender_id=SENDER, + sequence=sequence, + state=state, + ), + *( + ChannelDeliveryPartRow( + delivery_id=f"d-{event_id}-{index}", + event_id=event_id, + connection_id=connection, + part_index=index, + text="private reply text", + state=state, + ) + for index in range(parts) + ), + ] + + +async def test_reader_counts_attention_states_and_ages_for_one_connection(reader): + status_reader, sessions = reader + async with sessions() as session, session.begin(): + session.add_all( + [ + inbox("queued", "queued", 100), + inbox("preparing", "preparing", 50), + inbox("answered", "answered", 30), + inbox("unknown", "done", 1000), + inbox("failed-send", "done", 900), + inbox("blocked", "done", 800), + inbox("review", "needs_review", 700), + inbox("failed", "failed", 600), + inbox("other", "queued", 5000, connection="other"), + ] + ) + async with sessions() as session, session.begin(): + session.add_all( + [ + # Pending work counts parts; outcomes count replies. + *delivery("answered", "ready", 3, parts=2), + *delivery("unknown", "delivery_unknown", 4), + *delivery("failed-send", "failed", 5, parts=2), + *delivery("blocked", "blocked_window", 6), + *delivery("other", "ready", 1, connection="other"), + ] + ) + + status = await status_reader.read("c", "s", NOW) + + assert status == ChannelStatus( + inbox_queued=1, + inbox_running=1, + inbox_answered=1, + inbox_needs_review=1, + inbox_failed=1, + inbox_oldest_seconds=100.0, + outbox_pending=2, + outbox_oldest_seconds=30.0, + delivery_unknown=1, + delivery_failed=1, + delivery_blocked=1, + ) + + +async def test_empty_connection_reports_zero_counts_and_no_ages(reader): + status_reader, _ = reader + status = await status_reader.read("c", "s", NOW) + assert status == ChannelStatus(0, 0, 0, 0, 0, None, 0, None, 0, 0, 0) + + +class IdleRepository: + async def list_active(self, *args: Any, **kwargs: Any) -> list[Any]: + return [] + + async def claim_next(self, *args: Any, **kwargs: Any) -> None: + return None + + +class FixedReader(ChannelStatusReader): + def __init__(self, status: ChannelStatus | Exception) -> None: + self.status = status + self.reads = 0 + + async def read(self, connection_id: str, system_id: str, now: datetime) -> ChannelStatus: + self.reads += 1 + if isinstance(self.status, Exception): + raise self.status + return self.status + + +async def run_worker(reader: ChannelStatusReader, *, status_seconds: float) -> bool: + worker = ChannelWorker( + cast(ChannelRepository, IdleRepository()), + cast(ChannelTurnProcessor, None), + Ownership(), + replace(OPTIONS, status_seconds=status_seconds), + status=reader, + ) + async with worker: + await asyncio.sleep(0.15) + return worker.running + + +STATUS = ChannelStatus(2, 1, 0, 1, 0, 42.9, 3, None, 1, 0, 2) + + +async def test_worker_logs_one_bounded_status_line_per_interval(caplog): + caplog.set_level(logging.INFO, logger="agent_manager.application.channels.worker") + reader = FixedReader(STATUS) + + assert await run_worker(reader, status_seconds=60) + + lines = [r.getMessage() for r in caplog.records if r.getMessage().startswith("channel status:")] + assert lines == [ + "channel status: connection=connection inbox_queued=2 inbox_running=1 " + "inbox_answered=0 inbox_oldest_s=42 needs_review=1 inbox_failed=0 " + "outbox_pending=3 outbox_oldest_s=none delivery_unknown=1 " + "delivery_failed=0 delivery_blocked=2" + ] + assert reader.reads == 1 + + +async def test_short_interval_logs_repeatedly(caplog): + caplog.set_level(logging.INFO, logger="agent_manager.application.channels.worker") + reader = FixedReader(STATUS) + assert await run_worker(reader, status_seconds=0.02) + assert reader.reads >= 2 + + +async def test_status_failure_is_logged_by_type_and_never_stops_the_worker(caplog): + caplog.set_level(logging.INFO, logger="agent_manager.application.channels.worker") + reader = FixedReader(RuntimeError(f"database said {SENDER}")) + + assert await run_worker(reader, status_seconds=60) + + assert "channel status unavailable: type=RuntimeError" in caplog.text + assert SENDER not in caplog.text + + +class HangingReader(ChannelStatusReader): + async def read(self, connection_id: str, system_id: str, now: datetime) -> ChannelStatus: + await asyncio.Event().wait() + raise AssertionError("unreachable") + + +async def test_hanging_status_read_neither_stalls_the_worker_nor_its_shutdown(caplog, monkeypatch): + caplog.set_level(logging.INFO, logger="agent_manager.application.channels.worker") + monkeypatch.setattr( + "agent_manager.application.channels.worker.STATUS_READ_TIMEOUT_SECONDS", 0.05 + ) + started = time.monotonic() + + assert await run_worker(HangingReader(), status_seconds=60) + + # Without a bound, shutdown waits drain_seconds + 1 and then cancels. + assert time.monotonic() - started < OPTIONS.drain_seconds + 1 + assert "channel status unavailable: type=TimeoutError" in caplog.text + + +@pytest.mark.parametrize("seconds", [0, -1, 3601, float("inf")]) +def test_status_interval_is_bounded(seconds): + with pytest.raises(ValueError): + replace(OPTIONS, status_seconds=seconds) + with pytest.raises(ValidationError): + Settings.from_values(extra_whatsapp_status_log_seconds=seconds) diff --git a/tests/agent_manager/test_whatsapp_webhook.py b/tests/agent_manager/test_whatsapp_webhook.py index 79816430..3a112b4a 100644 --- a/tests/agent_manager/test_whatsapp_webhook.py +++ b/tests/agent_manager/test_whatsapp_webhook.py @@ -6,6 +6,7 @@ import hashlib import hmac import json +import logging from collections.abc import Iterator from datetime import UTC, datetime from pathlib import Path @@ -381,3 +382,38 @@ def test_duplicate_subscription_parameters_are_rejected(channel): ("hub.challenge", "two"), ] assert client.get(ENDPOINT, params=params).status_code == 403 + + +@pytest.mark.parametrize( + ("request_kind", "reason"), + [ + ("bad_signature", "signature"), + ("missing_signature", "signature"), + ("content_type", "content_type"), + ("encoding", "encoding"), + ("subscription", "subscription"), + ], +) +def test_rejections_log_a_fixed_reason_without_request_content( + channel, caplog, request_kind, reason +): + client, _, _ = channel + caplog.set_level(logging.WARNING, logger="agent_manager.api.routes.whatsapp") + body = encoded(payload(change(message(text={"body": "private question"})))) + forged = "sha256=" + "0" * 64 + if request_kind == "subscription": + params = {"hub.mode": "subscribe", "hub.verify_token": "wrong-token", "hub.challenge": "c"} + client.get(ENDPOINT, params=params) + else: + headers = { + "bad_signature": {"Content-Type": "application/json", "X-Hub-Signature-256": forged}, + "missing_signature": {"Content-Type": "application/json"}, + "content_type": {**signed(body), "Content-Type": "text/plain"}, + "encoding": {**signed(body), "Content-Encoding": "gzip"}, + }[request_kind] + client.post(ENDPOINT, content=body, headers=headers) + + rejected = [r.getMessage() for r in caplog.records if "webhook rejected" in r.getMessage()] + assert rejected == [f"whatsapp webhook rejected: reason={reason}"] + for private in ("private question", forged, "wrong-token", APP_SECRET): + assert private not in caplog.text From 675c8741abb19dea4c5e6cbf67f7a6fa95d34b51 Mon Sep 17 00:00:00 2001 From: Asaf Varon Date: Fri, 16 Oct 2026 15:47:51 +0300 Subject: [PATCH 2/2] feat(examples): add WhatsApp pilot example, runbook, and rehearsal The runbook covers preflight, deployment, staged enablement and pausing, health and status logs, restart recovery, rollback, retention, and the real-number smoke-test checklist. The example runs agent-manager on durable storage with a tool-free assistant and an allowlist identity resolver; its port is bound to loopback because the web conversation API is anonymous by default. A fake-provider rehearsal proves that paused intake survives restarts, is answered and sent exactly once, projects a delivered callback, and is not resent after a further restart. Co-Authored-By: Claude Opus 5.5 --- docs/README.md | 10 +- docs/WHATSAPP_INTEGRATION_PLAN.md | 9 +- docs/WHATSAPP_RUNBOOK.md | 239 ++++++++++++++++++ examples/whatsapp-pilot/.env.example | 34 +++ examples/whatsapp-pilot/.gitignore | 1 + examples/whatsapp-pilot/README.md | 42 +++ examples/whatsapp-pilot/agents.yml | 23 ++ examples/whatsapp-pilot/docker-compose.yml | 28 ++ examples/whatsapp-pilot/prompts/assistant.md | 8 + examples/whatsapp-pilot/senders.example.json | 7 + examples/whatsapp-pilot/whatsapp_identity.py | 52 ++++ tasks/0021-whatsapp-pilot-packaging.md | 105 ++++++++ tasks/README.md | 1 + .../test_whatsapp_pilot_rehearsal.py | 192 ++++++++++++++ 14 files changed, 744 insertions(+), 7 deletions(-) create mode 100644 docs/WHATSAPP_RUNBOOK.md create mode 100644 examples/whatsapp-pilot/.env.example create mode 100644 examples/whatsapp-pilot/.gitignore create mode 100644 examples/whatsapp-pilot/README.md create mode 100644 examples/whatsapp-pilot/agents.yml create mode 100644 examples/whatsapp-pilot/docker-compose.yml create mode 100644 examples/whatsapp-pilot/prompts/assistant.md create mode 100644 examples/whatsapp-pilot/senders.example.json create mode 100644 examples/whatsapp-pilot/whatsapp_identity.py create mode 100644 tasks/0021-whatsapp-pilot-packaging.md create mode 100644 tests/agent_manager/test_whatsapp_pilot_rehearsal.py diff --git a/docs/README.md b/docs/README.md index b2ba5f59..39bd9a9b 100644 --- a/docs/README.md +++ b/docs/README.md @@ -37,9 +37,13 @@ implemented, so protected-node access control is not actually enforced today. channel for one organization-owned number, with durable delivery, conversation reuse, identity boundaries, and scoped implementation tasks. Generic turn preparation, SQL channel storage, authenticated webhook intake, an - opt-in turn worker with conservative recovery, and opt-in provider delivery - with text controls are implemented. Customer account linking and live pilot - packaging remain planned. + opt-in turn worker with conservative recovery, opt-in provider delivery + with text controls, and pilot packaging are implemented. Customer account + linking, data retention, and the real-number smoke test remain planned. +- [WhatsApp pilot runbook](WHATSAPP_RUNBOOK.md) — preflight, deployment, + staged enablement and pausing, health and status logs, restart recovery, + rollback, retention, and the smoke-test checklist + ([task 0021](../tasks/0021-whatsapp-pilot-packaging.md)). [ADR 0006](adr/0006-idempotent-conversation-turn-preparation.md) records the implemented provider-independent preparation prerequisite for channels. diff --git a/docs/WHATSAPP_INTEGRATION_PLAN.md b/docs/WHATSAPP_INTEGRATION_PLAN.md index 57f2ec78..a0810212 100644 --- a/docs/WHATSAPP_INTEGRATION_PLAN.md +++ b/docs/WHATSAPP_INTEGRATION_PLAN.md @@ -2,9 +2,10 @@ - **Status:** WA-01 preparation, WA-02 durable channel storage/identity boundary, WA-03 authenticated webhook intake, WA-04 turn worker/recovery, - and WA-05 outbound delivery/controls are implemented. Organization account - linking and live pilot setup remain pending. Intake, worker and sender are - separately disabled by default. + WA-05 outbound delivery/controls, and WA-06 pilot packaging (runbook, + example, status logs, fake-provider rehearsal) are implemented. Organization + account linking, data retention, and the authorized real-number smoke test + remain pending. Intake, worker and sender are separately disabled by default. - **Date:** 2026-10-15. - **Updated:** 2026-10-16. - **Confirmed scope:** One organization-owned WhatsApp Business number. @@ -457,7 +458,7 @@ ships with tests, and each implementation task must pass `make check`. | WA-03: Cloud API intake boundary — implemented | Task 0015; `api/routes/whatsapp.py`, `infrastructure/channels/whatsapp/`, settings/composition, domain/SQL callback port, migration 0009, and tests. | Challenge and byte-level signature verification, account checks, complete batch parsing, separate status observations, safe quarantine, atomic commit-before-ack, and no agent work. Disabled by default. | | WA-04: Turn worker and recovery — implemented | Task 0016; channel processor/supervisor, active-work scan, authorized service response lookup, client identity loader, ownership adapter, settings/lifespan, and tests. | Per-sender order and independent concurrency; fresh identity; one execution claim; saved-answer recovery; ambiguous work held for review; approvals cancelled; streams awaited before releasing ownership; engine reused. | | WA-05: Outbound delivery and controls — implemented | Task 0019; sender/rendering adapters, delivery/control ports and services, SQL plans/projection, migration 0010, settings/supervisor wiring and tests. | Ordered Unicode/grapheme-safe splitting; accepted/delivered distinction; bounded retry; unknown-delivery review; fresh original-owner/window/opt-out checks; unsupported media/reset/help/stop/start without model calls. Disabled by default, verified locally with a fake provider. | -| WA-06: Pilot packaging and operation | New WhatsApp example/runbook, docs index, deployment command/configuration and fake-provider end-to-end tests. | Documented administrator setup and rollback; durable restart rehearsal; metrics/redaction; full fake-provider round trip; then an explicitly authorized real-number smoke test. | +| WA-06: Pilot packaging and operation — implemented except the smoke test | Task 0021; [runbook](WHATSAPP_RUNBOOK.md), `examples/whatsapp-pilot/`, `channel status:` and webhook-rejection log lines, and a fake-provider restart rehearsal. | Documented administrator setup and rollback; durable restart rehearsal; metrics/redaction; full fake-provider round trip; then an explicitly authorized real-number smoke test. | WA-00 → WA-01 → WA-02 → WA-03 → WA-04 → WA-05 → WA-06 is the delivery order. The identity/account-linking prerequisite is required for the confirmed internal diff --git a/docs/WHATSAPP_RUNBOOK.md b/docs/WHATSAPP_RUNBOOK.md new file mode 100644 index 00000000..683d7155 --- /dev/null +++ b/docs/WHATSAPP_RUNBOOK.md @@ -0,0 +1,239 @@ +# WhatsApp pilot runbook + +How an administrator deploys, operates, pauses, and rolls back the WhatsApp +channel for one organization-owned number. It describes the implemented +behavior from the [integration plan](WHATSAPP_INTEGRATION_PLAN.md) (WA-01 to +WA-06). The deployable example is in +[`examples/whatsapp-pilot/`](../examples/whatsapp-pilot/README.md). + +No real account or number has been activated yet. The real-number smoke test +(§10) needs explicit authorization from the organization. + +## 1. Preflight + +Confirm all of these before enabling anything: + +- [ ] The organization owns the number. If it already uses the WhatsApp Business + app, check current coexistence or migration eligibility in Meta's + documentation first. This channel does not import history. +- [ ] Business account ID and phone number ID for the Cloud API. +- [ ] Meta app secret, used to verify webhook signatures. +- [ ] A long-lived system-user access token with messaging permission. It must + differ from the app secret and the verify token. +- [ ] A Graph API version the account supports, pinned explicitly (`vNN.0`). +- [ ] A public HTTPS URL that routes to `/channels/whatsapp/webhook` on the + manager, and a webhook subscription to the `messages` field. +- [ ] Billing is ready for the account. +- [ ] A real support contact for `/help` and escalation. +- [ ] Identity: who may use the pilot and how senders are linked (§3). +- [ ] Durable storage: a SQLite file on a persistent volume for one host, or + PostgreSQL. +- [ ] A retention decision for stored messages (§9). + +## 2. Deploy + +Run the conversation manager, not the engine API. The image's default +`agentctl serve` does not host conversations or the channel. + +```bash +make build +``` + +Then, from `examples/whatsapp-pilot/` with a filled-in `.env`: + +```bash +docker compose up -d +``` + +- **Expose only the webhook.** The manager also serves the web conversation API, + which accepts anonymous visitors by default. Anyone who reaches that API can + talk to the assistant without the WhatsApp identity check. The compose file + therefore binds port 8100 to `127.0.0.1`. Your HTTPS reverse proxy on the + same host should forward only `/channels/whatsapp/webhook` to it. If the proxy + runs in a container, attach it to the compose network instead of publishing + the port. +- `agent-manager` applies database migrations at startup (`--migrate`, the + default). Migrations 0007–0010 add the channel tables. +- Storage defaults to SQLite on the `pilot-data` volume. `EXTRA_DB_BACKEND` and + `EXTRA_DB_URL` in `.env` override it, for example for PostgreSQL. +- The queue and outbox live in the database. Keep its volume across restarts. +- Only one instance may run the worker for a connection. A second instance with + the worker enabled fails at startup with `ChannelWorkerAlreadyRunning`. + Other instances may run intake only. +- `docker compose stop` sends SIGTERM. The worker drains active turns for up to + `EXTRA_WHATSAPP_WORKER_DRAIN_SECONDS`, so the compose file allows 30 seconds + before the process is killed. + +## 3. Identity + +The worker resolves every sender through the configured identity plugin +(`module:Class`) at preparation, before execution, and before each send. +Message text never chooses the user, organization, or roles, and the principal's +organization must equal `EXTRA_WHATSAPP_ORGANIZATION_ID`. + +The example plugin, `whatsapp_identity:AllowlistIdentityResolver`, reads a JSON +allowlist on every lookup. Keys are the sender IDs exactly as Meta sends them in +`from` (international digits without `+`). Removing an entry revokes access on +the next message. Possessing the phone is the only proof, so use it only for a +closed pilot. Replace it with the organization's account-linking resolver before +the assistant reaches internal information. + +## 4. Configuration + +All settings are environment variables, read at startup. Changing one requires +a restart. + +| Variable | Stage | Meaning | +| --- | --- | --- | +| `EXTRA_WHATSAPP_ENABLED` | 1 | Mount the webhook and store incoming messages. | +| `EXTRA_WHATSAPP_CONNECTION_ID` | 1 | Internal name of this number (default `whatsapp`). | +| `EXTRA_WHATSAPP_ACCOUNT_ID` | 1 | Business account ID; entries for other accounts are ignored. | +| `EXTRA_WHATSAPP_PHONE_NUMBER_ID` | 1 | Receiving and sending number ID. | +| `EXTRA_WHATSAPP_APP_SECRET` | 1 | Verifies `X-Hub-Signature-256`. | +| `EXTRA_WHATSAPP_VERIFY_TOKEN` | 1 | Answers Meta's subscription challenge. | +| `EXTRA_WHATSAPP_MAX_BODY_BYTES` | 1 | Webhook body limit (default 262144). | +| `EXTRA_WHATSAPP_WORKER_ENABLED` | 2 | Run the turn worker. Requires stage 1. | +| `EXTRA_WHATSAPP_ORGANIZATION_ID` | 2 | Organization every sender must belong to. | +| `EXTRA_WHATSAPP_IDENTITY_RESOLVER` | 2 | Identity plugin, `module:Class`. | +| `EXTRA_WHATSAPP_WORKER_CONCURRENCY` | 2 | Senders processed in parallel (1–16, default 4). | +| `EXTRA_WHATSAPP_WORKER_POLL_SECONDS` | 2 | Queue polling interval (default 0.5). | +| `EXTRA_WHATSAPP_WORKER_TURN_SECONDS` | 2 | Time limit per turn (default 120). | +| `EXTRA_WHATSAPP_WORKER_DRAIN_SECONDS` | 2 | Shutdown drain limit (default 15). | +| `EXTRA_WHATSAPP_STATUS_LOG_SECONDS` | 2 | Interval of `channel status:` lines (default 60). | +| `EXTRA_WHATSAPP_SENDER_ENABLED` | 3 | Send answers. Requires stage 2. | +| `EXTRA_WHATSAPP_API_VERSION` | 3 | Pinned Graph API version, `vNN.0`. | +| `EXTRA_WHATSAPP_ACCESS_TOKEN` | 3 | System-user token for sending. | +| `EXTRA_WHATSAPP_SUPPORT_CONTACT` | 3 | Contact returned by `/help`. | + +Startup fails if a stage is enabled without its prerequisites or required +values, or if storage is in-memory. Secrets are excluded from settings output. + +## 5. Enablement order and pause states + +Enable one stage at a time, and check health and logs before the next: + +| State | Flags on | New messages | Pending work | +| --- | --- | --- | --- | +| Off | none | Webhook returns 404; nothing is stored. | Preserved. | +| Intake | 1 | Verified and stored as `queued`. | Waits. No status lines yet (they come from the worker). | +| Answering | 1, 2 | Answered and saved, not sent. | Answers wait as `answered`. | +| Live | 1, 2, 3 | Answered and sent. | Waiting answers are sent. | + +Moving between states never deletes accepted work. To pause, set the flags for +the target state and restart. While intake is off, Meta's own retry policy +applies to callbacks it could not deliver; check its current documentation +before a long outage. + +Free-form replies are allowed only within 24 hours of the user's latest +message. An answer that waits longer than that is blocked rather than sent, and +is counted in `delivery_blocked`. + +## 6. Health and monitoring + +`GET /health` returns 200 with `"status": "ok"`. When the worker is enabled but +has stopped, for example after losing exclusive ownership, it returns 503 with +`"status": "degraded"`. + +The worker logs one line per `EXTRA_WHATSAPP_STATUS_LOG_SECONDS`: + +```text +channel status: connection=whatsapp inbox_queued=0 inbox_running=1 inbox_answered=0 + inbox_oldest_s=4 needs_review=0 inbox_failed=0 outbox_pending=0 outbox_oldest_s=none + delivery_unknown=0 delivery_failed=0 delivery_blocked=0 +``` + +| Field | Meaning | +| --- | --- | +| `inbox_queued`, `inbox_running`, `inbox_answered` | Messages waiting, in progress, or answered but not yet sent. | +| `inbox_oldest_s` | Age of the oldest unfinished message; `none` when there is none. | +| `needs_review`, `inbox_failed` | Turns held for an administrator, or failed. | +| `outbox_pending`, `outbox_oldest_s` | Unsent reply parts (a long reply is split into several), and how long ago their message arrived. | +| `delivery_unknown` | Replies with an uncertain send outcome; never retried automatically. | +| `delivery_failed` | Replies the provider rejected. | +| `delivery_blocked` | Replies blocked by identity, the 24-hour window, or format. | + +`needs_review`, `inbox_failed`, and the `delivery_*` fields are totals since +the database was created, so alert when they increase. Lines contain no message +text, sender IDs, tokens, or provider message IDs. + +Every rejected webhook request logs `whatsapp webhook rejected: reason=`, +where the reason is `signature`, `subscription`, `encoding`, `content_type`, +`length`, `size`, or `incomplete`. A burst of `signature` rejections means a +wrong app secret or forged traffic. + +Suggested alerts: health 503; `inbox_oldest_s` or `outbox_oldest_s` above a few +minutes; any increase in `needs_review`, `delivery_unknown`, or +`delivery_failed`; repeated `reason=signature`; `webhook intake failed` or +`channel supervisor stopped` warnings. + +## 7. Restarts and recovery + +A restart preserves every accepted message. On the next start the worker: + +- continues `queued` and prepared messages; +- sends an answer that was saved but not yet sent, without generating it again; +- holds a turn that was running without a saved answer for review instead of + running it twice; +- marks a send interrupted mid-request as `delivery_unknown` for review instead + of risking a duplicate message. + +Rehearse this before the pilot, as the automated +`test_whatsapp_pilot_rehearsal.py` does with a fake provider: + +1. Start with intake only and send a test message: it is stored as `queued`. +2. Restart with all three stages: it is answered and sent once. +3. Restart again: nothing is sent twice, and the status line shows no pending + work. + +## 8. Rollback + +To stop the channel but keep the current version, disable the flags (§5) and +restart. Data is kept. + +To run an application version older than the channel migrations, downgrade the +schema first. **This deletes all channel data** (queue, bindings, outbox, +delivery history) while keeping conversations. Back up the database and stop +the manager. Then open a shell in the same image, volume, and settings; the +image's entrypoint would otherwise pass `sh` to `agentctl`: + +```bash +docker compose run --rm --entrypoint sh manager +``` + +Inside that shell: + +```bash +cd "$(python -c 'import agent_manager, pathlib; print(pathlib.Path(agent_manager.__file__).parent / "infrastructure/persistence/migrations")')" +``` + +```bash +alembic -c alembic.ini downgrade 0006 +``` + +Without Docker, run the last two commands in the manager's Python environment +with the same `EXTRA_DB_URL`. + +## 9. Data and retention + +Stored channel data: normalized message text and sender IDs, reply text, sender +bindings and opt-out state, delivery observations, and webhook receipts (body +digest and issue codes, not raw bodies). Ordinary logs exclude message text, +sender IDs, and tokens. + +There is no automatic deletion of channel data yet. Agree on a retention period +with the organization, and restrict database access, before the pilot. Automated +retention is a follow-up task. + +## 10. Real-number smoke test + +Only with explicit authorization, on the pilot number, with one allowlisted +tester: + +1. Deploy with intake only. Confirm the subscription challenge succeeds in Meta. +2. Send a message: a `queued` row exists and no signature rejections are logged. +3. Enable the worker and restart: an answer is saved, not sent. +4. Enable sending and restart: the tester receives the answer once. +5. Send `/help`: the support contact arrives without a model call. +6. Send a message from a non-allowlisted number: it is marked `denied` and no + answer is sent. +7. Stop and start: no duplicate messages. Record the results and the API version. diff --git a/examples/whatsapp-pilot/.env.example b/examples/whatsapp-pilot/.env.example new file mode 100644 index 00000000..6c34bc86 --- /dev/null +++ b/examples/whatsapp-pilot/.env.example @@ -0,0 +1,34 @@ +# Copy to .env in this directory. .env is gitignored; never commit real values. + +# Model provider for agents.yml. +OPENAI_API_KEY=replace-with-your-model-key + +# Storage: docker-compose.yml sets EXTRA_DB_URL to a file on a durable volume. +# For PostgreSQL instead: +# EXTRA_DB_BACKEND=postgres +# EXTRA_DB_URL=postgresql://user:password@host/database + +# Stage 1 — webhook intake. Messages are stored; nothing is answered yet. +EXTRA_WHATSAPP_ENABLED=false +EXTRA_WHATSAPP_CONNECTION_ID=whatsapp +EXTRA_WHATSAPP_ACCOUNT_ID=replace-with-business-account-id +EXTRA_WHATSAPP_PHONE_NUMBER_ID=replace-with-phone-number-id +EXTRA_WHATSAPP_APP_SECRET=replace-with-meta-app-secret +EXTRA_WHATSAPP_VERIFY_TOKEN=replace-with-a-random-verify-token + +# Stage 2 — worker. Requires stage 1. Answers are saved but not sent. +EXTRA_WHATSAPP_WORKER_ENABLED=false +EXTRA_WHATSAPP_ORGANIZATION_ID=replace-with-organization-id +EXTRA_WHATSAPP_IDENTITY_RESOLVER=whatsapp_identity:AllowlistIdentityResolver +# Path inside the container; docker-compose.yml sets this to /workspace/senders.json. +# WHATSAPP_PILOT_SENDERS_FILE=./senders.json + +# Stage 3 — sending. Requires stage 2. +EXTRA_WHATSAPP_SENDER_ENABLED=false +# Pin the Graph API version your account supports (format vNN.0). +EXTRA_WHATSAPP_API_VERSION=replace-with-pinned-version +EXTRA_WHATSAPP_ACCESS_TOKEN=replace-with-system-user-access-token +EXTRA_WHATSAPP_SUPPORT_CONTACT=replace-with-support-contact + +# Optional: seconds between `channel status:` log lines (default 60). +# EXTRA_WHATSAPP_STATUS_LOG_SECONDS=60 diff --git a/examples/whatsapp-pilot/.gitignore b/examples/whatsapp-pilot/.gitignore new file mode 100644 index 00000000..6f731aa2 --- /dev/null +++ b/examples/whatsapp-pilot/.gitignore @@ -0,0 +1 @@ +senders.json diff --git a/examples/whatsapp-pilot/README.md b/examples/whatsapp-pilot/README.md new file mode 100644 index 00000000..101993a0 --- /dev/null +++ b/examples/whatsapp-pilot/README.md @@ -0,0 +1,42 @@ +# WhatsApp pilot + +A deployable starting point for the WhatsApp channel: one tool-free assistant, +a sender allowlist, and the conversation manager on durable storage. Read the +[runbook](../../docs/WHATSAPP_RUNBOOK.md) before enabling anything. + +| File | Purpose | +| --- | --- | +| `agents.yml`, `prompts/assistant.md` | The assistant. It has no tools, so it never waits for an approval that WhatsApp cannot answer. | +| `whatsapp_identity.py` | `AllowlistIdentityResolver`: links listed senders to pilot users. For a closed pilot only. | +| `senders.example.json` | Allowlist format, keyed by the sender ID Meta sends in `from`. | +| `.env.example` | Every setting, with all three stages off. | +| `docker-compose.yml` | Runs `agent-manager` with the database on a durable volume. | + +## Start + +```bash +cp .env.example .env +``` + +```bash +cp senders.example.json senders.json +``` + +Fill in `.env` and `senders.json`, build the image from the repository root +with `make build`, then: + +```bash +docker compose up -d +``` + +Turn on the stages one at a time, in this order, restarting after each: +intake (`EXTRA_WHATSAPP_ENABLED`), worker (`EXTRA_WHATSAPP_WORKER_ENABLED`), +sending (`EXTRA_WHATSAPP_SENDER_ENABLED`). + +The manager listens on `127.0.0.1:8100` only. It also serves the web +conversation API, which accepts anonymous visitors by default, so do not publish +the port. Configure your HTTPS reverse proxy to forward only +`https:///channels/whatsapp/webhook` to it. + +`senders.json` and `.env` contain personal data and secrets. Keep them out of +version control. diff --git a/examples/whatsapp-pilot/agents.yml b/examples/whatsapp-pilot/agents.yml new file mode 100644 index 00000000..11f17307 --- /dev/null +++ b/examples/whatsapp-pilot/agents.yml @@ -0,0 +1,23 @@ +system: + name: "WhatsApp Pilot Assistant" + +defaults: + model: + provider: openai + name: gpt-4o-mini + temperature: 0.2 + +# No tools: a pending approval cannot be answered over WhatsApp, so the pilot +# agent must never need one. +execution: + max_iterations: 4 + +agents: + pilot_assistant: + name: "Pilot Assistant" + description: "Answers pilot users' questions in short plain-text messages." + prompts: + system: prompts/assistant.md + +graph: + pilot_assistant: diff --git a/examples/whatsapp-pilot/docker-compose.yml b/examples/whatsapp-pilot/docker-compose.yml new file mode 100644 index 00000000..959cd5d1 --- /dev/null +++ b/examples/whatsapp-pilot/docker-compose.yml @@ -0,0 +1,28 @@ +# The image's default `agentctl serve` does not host conversations or the +# WhatsApp channel, so this service runs `agent-manager` instead. +# Build the image first from the repository root: `make build`. +services: + manager: + image: ghcr.io/extra-org/extra:local + command: ["agent-manager", "--config", "/workspace/agents.yml", "--host", "0.0.0.0", "--port", "8100"] + env_file: + - .env + environment: + # A value in .env (for example PostgreSQL) takes precedence over this file. + EXTRA_DB_URL: ${EXTRA_DB_URL:-sqlite+aiosqlite:////data/pilot.db} + WHATSAPP_PILOT_SENDERS_FILE: /workspace/senders.json + volumes: + - .:/workspace + # The queue and outbox live here; keep this volume across restarts. + - pilot-data:/data + ports: + # Loopback only. The manager also serves the web conversation API, which + # is anonymous by default; expose just the webhook path through a proxy. + - "127.0.0.1:8100:8100" + restart: unless-stopped + # Longer than the worker drain (EXTRA_WHATSAPP_WORKER_DRAIN_SECONDS, 15 s + # by default), so a stop lets active turns finish before the process exits. + stop_grace_period: 30s + +volumes: + pilot-data: diff --git a/examples/whatsapp-pilot/prompts/assistant.md b/examples/whatsapp-pilot/prompts/assistant.md new file mode 100644 index 00000000..26d7ca89 --- /dev/null +++ b/examples/whatsapp-pilot/prompts/assistant.md @@ -0,0 +1,8 @@ +You are a pilot assistant that people reach over WhatsApp. + +- Reply in the language the person writes in. Hebrew and English are both common. +- Keep replies short: a few sentences of plain text. WhatsApp does not render + Markdown tables or headings. +- If you do not know something, say so. Do not guess facts about the organization. +- Never ask for passwords, one-time codes, or payment details. +- When the person needs a human, tell them to send /help for the support contact. diff --git a/examples/whatsapp-pilot/senders.example.json b/examples/whatsapp-pilot/senders.example.json new file mode 100644 index 00000000..cb89c3cb --- /dev/null +++ b/examples/whatsapp-pilot/senders.example.json @@ -0,0 +1,7 @@ +{ + "15550001111": { + "user_id": "pilot-user-1", + "display_name": "Pilot User", + "roles": ["employee"] + } +} diff --git a/examples/whatsapp-pilot/whatsapp_identity.py b/examples/whatsapp-pilot/whatsapp_identity.py new file mode 100644 index 00000000..c698fd4e --- /dev/null +++ b/examples/whatsapp-pilot/whatsapp_identity.py @@ -0,0 +1,52 @@ +"""Pilot identity: an administrator-managed allowlist of WhatsApp senders. + +Possessing the phone number is the only proof. Use this for a closed pilot and +replace it with the organization's account-linking resolver before internal +information is exposed. +""" + +import asyncio +import json +import os +from pathlib import Path + +from agent_manager.domain.channels.identity import ( + ChannelIdentityResolver, + ChannelIdentityUnavailable, +) +from agent_manager.domain.channels.models import ChannelAddress +from agent_manager.domain.identity import Principal + + +class AllowlistIdentityResolver(ChannelIdentityResolver): + def __init__(self) -> None: + self._path = Path(os.environ["WHATSAPP_PILOT_SENDERS_FILE"]) + self._organization_id = os.environ["EXTRA_WHATSAPP_ORGANIZATION_ID"] + + async def resolve(self, address: ChannelAddress) -> Principal | None: + # Read on every lookup, so removing a sender revokes access immediately. + try: + text = await asyncio.to_thread(self._path.read_text, encoding="utf-8") + senders = json.loads(text) + except (OSError, ValueError) as exc: + raise ChannelIdentityUnavailable("sender allowlist is unreadable") from exc + entry = senders.get(address.sender_id) if isinstance(senders, dict) else None + if not isinstance(entry, dict): + return None + user_id, roles = entry.get("user_id"), entry.get("roles", []) + display_name = entry.get("display_name") + # A malformed entry denies access rather than granting partial roles. + if ( + not isinstance(user_id, str) + or not user_id.strip() + or not isinstance(roles, list) + or not all(isinstance(role, str) for role in roles) + or not (display_name is None or isinstance(display_name, str)) + ): + return None + return Principal.external( + user_id, + display_name=display_name, + roles=tuple(roles), + organization_id=self._organization_id, + ) diff --git a/tasks/0021-whatsapp-pilot-packaging.md b/tasks/0021-whatsapp-pilot-packaging.md new file mode 100644 index 00000000..83f55879 --- /dev/null +++ b/tasks/0021-whatsapp-pilot-packaging.md @@ -0,0 +1,105 @@ +# Task 0021 — WhatsApp pilot packaging and operation (WA-06) + +## Goal / authorization + +Implement the final planned stage requested by the user after the WA-05 review: +an administrator runbook, a deployable pilot example, operational monitoring, +and fake-provider evidence for the complete round trip and durable restart. +The user chose structured log lines for monitoring instead of a new HTTP or CLI +interface. The real-number smoke test remains a separately authorized +deployment activity. + +## Read first + +AGENTS.md; project-architecture, senior-python-engineering, testing, +documentation, code-review; the WhatsApp plan §6–§7, ADRs 0006–0010, task +0019/0020, and the existing channel implementation and tests. + +## Scope / files allowed to change + +- This task, tasks/README.md, docs/README.md, docs/WHATSAPP_INTEGRATION_PLAN.md, + and new docs/WHATSAPP_RUNBOOK.md. +- New examples/whatsapp-pilot/: README, agent config and prompt, an allowlist + identity resolver plugin, `.env.example`, `senders.example.json`, and a + `docker-compose.yml` that starts `agent-manager` on durable storage. +- Under src/agent_manager/: new domain/channels/status.py and + infrastructure/persistence/channels/status.py; existing + application/channels/{worker,worker_options}.py, api/routes/whatsapp.py, + composition.py, config.py. +- New tests/agent_manager/test_channel_status.py and + test_whatsapp_pilot_rehearsal.py; existing test_whatsapp_webhook.py for + rejection-log assertions. + +## Acceptance criteria + +1. The worker logs one bounded `channel status:` line per configured interval + with inbox counts by attention state, the oldest active inbox age, pending + delivery parts and their oldest age, and unknown, failed, and blocked + delivery counts. No message text, sender identifier, token, or provider + message ID appears in it. A failed status read is logged and never stops + the worker. +2. Every rejected webhook request logs `whatsapp webhook rejected:` with a + fixed reason (signature, subscription, encoding, content type, length, + size) and no request content. +3. The interval is configurable and validated; the default needs no setting. +4. A fake-provider rehearsal proves: intake while the worker is disabled keeps + the message queued; a restart with the worker and sender enabled on the same + database answers and sends it exactly once; a signed `delivered` status + callback is projected; a further restart sends nothing again. +5. The runbook documents preflight, configuration, the actual enablement order + and pause states, deployment command and migrations, health, monitoring, + restart, rollback, retention, and the manual smoke-test checklist, matching + the implemented behavior. +6. The example configuration validates, contains no secrets, and keeps the + agent free of approval-dependent tools. +7. `make check` passes. + +## Out of scope + +Real Meta credentials or traffic, the smoke test itself, Google/JWT account +linking, a metrics endpoint or dependency, templates, media, and any schema +migration. + +## Commands / report + +`make check`; `agentctl validate` for the example; report per AGENTS.md §9. + +## Implementation result + +- `SqlChannelStatusReader` reads grouped counts and oldest ages without + selecting text or identifiers. The worker bounds each read to five seconds + and logs the result each `EXTRA_WHATSAPP_STATUS_LOG_SECONDS` (default 60); a + failed read logs its type. +- The webhook route logs a fixed reason for every rejection. +- `examples/whatsapp-pilot/` validates offline. Its allowlist resolver denies + malformed entries, re-reads the file per lookup, and fails closed when the + file is unreadable. +- The rehearsal test proves paused intake, a single answer and send after + restart, a projected `delivered` callback, and no resend on a later restart. +- The runbook's rollback command was rehearsed on a temporary database: + downgrading to 0006 removes the channel tables and keeps conversations. + +### Self-review / corrections + +- The example published port 8100 on all interfaces. The manager's web + conversation API is anonymous by default, so that bypassed the WhatsApp + identity check. The port is now bound to loopback, and the runbook requires a + proxy that forwards only the webhook path. +- Compose `environment` overrode `EXTRA_DB_URL` from `.env`, pairing a + PostgreSQL backend with a SQLite URL. The value now defaults only when unset. +- The rollback commands did not work inside the image, whose entrypoint passes + other commands to `agentctl`. The runbook now opens a shell with + `--entrypoint sh`. +- The status read had no bound inside the scheduling loop. A stalled database + could delay scheduling and turn a graceful drain into cancellation. It now + times out after five seconds. +- Delivery outcomes counted parts while blocked replies counted replies. All + outcome fields now count replies; `outbox_pending` still counts parts. + +### Limits / next task + +- No automatic retention for channel data exists. Proposed next task: a + retention job with a configurable period for channel rows. +- The status query scans delivery history without retention; acceptable at + pilot volume. +- The real-number smoke test and organization account linking remain pending. diff --git a/tasks/README.md b/tasks/README.md index 11827899..816351bf 100644 --- a/tasks/README.md +++ b/tasks/README.md @@ -30,5 +30,6 @@ finishing, Expected final report.** | 0018 | [SQL test resource cleanup](0018-sql-test-resource-cleanup.md) | 0017 review verification | | 0019 | [WhatsApp outbound delivery and controls](0019-whatsapp-delivery-and-controls.md) | 0013–0018; WhatsApp WA-05 | | 0020 | [WhatsApp delivery review](0020-whatsapp-delivery-review.md) | 0019 | +| 0021 | [WhatsApp pilot packaging](0021-whatsapp-pilot-packaging.md) | 0020; WhatsApp WA-06 | Before starting any task, read `AGENTS.md` and the skill(s) the task lists. diff --git a/tests/agent_manager/test_whatsapp_pilot_rehearsal.py b/tests/agent_manager/test_whatsapp_pilot_rehearsal.py new file mode 100644 index 00000000..6937f264 --- /dev/null +++ b/tests/agent_manager/test_whatsapp_pilot_rehearsal.py @@ -0,0 +1,192 @@ +"""Pilot rehearsal: paused intake survives restarts and is answered and sent exactly once.""" + +import asyncio +import logging +import time +from collections.abc import Callable, Iterator +from contextlib import contextmanager +from datetime import UTC, datetime +from pathlib import Path + +import pytest +from fastapi.testclient import TestClient +from sqlmodel import SQLModel + +from agent_engine.core.validator import SystemSpecValidator +from agent_engine.parsers.yaml.parser import YAMLParser +from agent_manager.api.app import create_app +from agent_manager.config import Settings +from agent_manager.domain.channels.delivery import SendOutcome +from agent_manager.domain.channels.identity import ChannelIdentityUnavailable +from agent_manager.domain.channels.models import ChannelAddress +from agent_manager.infrastructure.channels.identity_loader import load_channel_identity_resolver +from agent_manager.infrastructure.persistence.channels.tables import ( + ChannelDeliveryPartRow, + ChannelInboxRow, +) +from agent_manager.infrastructure.persistence.database import create_db_engine +from tests.agent_manager.test_whatsapp_delivery_lifecycle import outbound_values +from tests.agent_manager.test_whatsapp_webhook import ( + ENDPOINT, + StartupEngine, + change, + encoded, + message, + payload, + saved_rows, + settings_values, + signed, +) + +CONFIG = Path(__file__).parents[2] / "examples/enterprise-knowledge-assistant/agents.yaml" +SENDER = "opaque-sender" + + +class Provider: + def __init__(self) -> None: + self.calls: list[tuple[str, str]] = [] + self.closed = False + + async def send(self, recipient: str, text: str) -> SendOutcome: + self.calls.append((recipient, text)) + return SendOutcome("accepted", f"wamid.reply-{len(self.calls)}") + + async def close(self) -> None: + self.closed = True + + +def wait_until(condition: Callable[[], bool], seconds: float = 3) -> None: + deadline = time.monotonic() + seconds + while not condition(): + assert time.monotonic() < deadline, "rehearsal step did not complete in time" + time.sleep(0.01) + + +def post(client: TestClient, value: dict) -> None: + body = encoded(value) + assert client.post(ENDPOINT, content=body, headers=signed(body)).status_code == 200 + + +@pytest.fixture +def url(tmp_path: Path) -> str: + url = f"sqlite+aiosqlite:///{tmp_path / 'pilot.db'}" + + async def schema() -> None: + database = create_db_engine(url) + try: + async with database.begin() as connection: + await connection.run_sync(SQLModel.metadata.create_all) + finally: + await database.dispose() + + asyncio.run(schema()) + return url + + +def test_paused_intake_is_answered_once_after_restart_and_delivery_is_projected( + url: str, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + engine = StartupEngine() + providers: list[Provider] = [] + + def provider_factory(*args: object, **kwargs: object) -> Provider: + providers.append(Provider()) + return providers[-1] + + monkeypatch.setattr("agent_manager.api.app.LangGraphEngine", lambda *a, **kw: engine) + monkeypatch.setattr("agent_manager.composition.WhatsAppSender", provider_factory) + # Startup replaces logging handlers; keep pytest's capture in place. + monkeypatch.setattr("agent_manager.api.app.configure_logging", lambda: None) + caplog.set_level(logging.INFO, logger="agent_manager.application.channels.worker") + + @contextmanager + def deployment(values: dict) -> Iterator[TestClient]: + app = create_app(str(CONFIG), Settings.from_values(database_url=url, **values)) + with TestClient(app) as client: + assert client.get("/health").status_code == 200 + yield client + + # Intake only: the worker is paused, so the accepted message waits durably. + with deployment(settings_values()) as client: + post(client, payload(change(message("question", text={"body": "hello"})))) + assert [row.state for row in saved_rows(url, ChannelInboxRow)] == ["queued"] + assert engine.prompts == [] and providers == [] + + # Restart with the worker and sender: answered once, then a signed callback lands. + full = outbound_values(extra_whatsapp_status_log_seconds=0.05) + with deployment(full) as client: + wait_until(lambda: providers[0].calls == [(SENDER, "answer:hello")]) + delivered = { + "id": "wamid.reply-1", + "recipient_id": SENDER, + "status": "delivered", + "timestamp": str(int(datetime.now(UTC).timestamp())), + } + post(client, payload(change(statuses=[delivered]))) + wait_until( + lambda: ( + [part.state for part in saved_rows(url, ChannelDeliveryPartRow)] == ["delivered"] + ) + ) + assert providers[0].closed + + # A further restart finds nothing to redo. + with deployment(full): + time.sleep(0.3) + assert providers[1].calls == [] + assert engine.prompts == ["hello"] + assert [part.state for part in saved_rows(url, ChannelDeliveryPartRow)] == ["delivered"] + + status_lines = [ + record.getMessage() + for record in caplog.records + if record.getMessage().startswith("channel status:") + ] + assert status_lines + for private in (SENDER, "hello", "answer:hello", "wamid."): + assert not any(private in line for line in status_lines) + + +EXAMPLE = Path(__file__).parents[2] / "examples/whatsapp-pilot" + + +def test_pilot_example_config_passes_startup_validation() -> None: + spec = YAMLParser().parse(str(EXAMPLE / "agents.yml")) + assert SystemSpecValidator().validate(spec, EXAMPLE) == [] + + +async def test_pilot_allowlist_resolver_links_revokes_and_fails_closed( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + senders = tmp_path / "senders.json" + senders.write_text((EXAMPLE / "senders.example.json").read_text(), encoding="utf-8") + monkeypatch.setenv("WHATSAPP_PILOT_SENDERS_FILE", str(senders)) + monkeypatch.setenv("EXTRA_WHATSAPP_ORGANIZATION_ID", "org") + # The manager adds the config directory to sys.path at startup. + monkeypatch.syspath_prepend(str(EXAMPLE)) + resolver = load_channel_identity_resolver("whatsapp_identity:AllowlistIdentityResolver") + listed = ChannelAddress("whatsapp", "system", "15550001111") + + principal = await resolver.resolve(listed) + assert principal is not None + assert (principal.external_id, principal.organization_id, principal.roles) == ( + "pilot-user-1", + "org", + ("employee",), + ) + assert await resolver.resolve(ChannelAddress("whatsapp", "system", "15559999999")) is None + + for malformed in ( + '{"user_id": "u", "roles": "employee"}', + '{"user_id": " "}', + '{"user_id": 7}', + ): + senders.write_text(f'{{"15550001111": {malformed}}}', encoding="utf-8") + assert await resolver.resolve(listed) is None + + senders.write_text("{}", encoding="utf-8") + assert await resolver.resolve(listed) is None + + senders.write_text("not json", encoding="utf-8") + with pytest.raises(ChannelIdentityUnavailable): + await resolver.resolve(listed)