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/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/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 15c506b9..a81143a2 100644 --- a/tasks/README.md +++ b/tasks/README.md @@ -30,6 +30,7 @@ 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 | | PR 148 | [Atomic run leg accounting contract](pr-0148-run-leg-accounting-contract.md) | issue 140 review follow-up | Before starting any task, read `AGENTS.md` and the skill(s) the task lists. 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_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) 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