Event-driven mail and agent work

The Cloudflare consumer keeps idle mailbox checks out of the model context. afmail cloudflare watch is an ordinary program: it polls Cloudflare Queues, imports mail into the local workspace, and invokes a configured command only when durable new-message work exists. No model, model credential, or agent SDK is needed to wait for mail.

Email Worker -> immutable raw EML in R2
                         |
                  R2 create notification
                         |
                  Cloudflare Queue
                         |
                  VPS: afmail cloudflare watch
                         |
           local import + durable notification task
                         |
           configured command -> agent scheduler

Setup

First complete the Cloudflare mailbox setup. Keep the existing spool control endpoint configured: Queue acknowledgement does not replace R2 cleanup or the Durable Object quota ledger.

Create a queue and enable its HTTP pull consumer. In the R2 bucket settings, add an object-create notification for the pending prefix (inbox/ by default) with this queue as its destination. Exclude the imported/archive prefix: copying an imported object must not create new mail work. See Cloudflare’s R2 event notifications and HTTP pull consumers. The queue token needs Queues read and write permissions; acknowledgement changes queue state. It is separate from the R2 S3 credentials and spool control token.

Configure the workspace using secret sources rather than live tokens in argv:

afmail config set cloudflare.account_id YOUR_ACCOUNT_ID
afmail config set cloudflare.queue.queue_id YOUR_QUEUE_ID
afmail config set cloudflare.queue.api_token_secret 'file:/etc/afmail/secrets.json#queue_token_secret'
afmail config set cloudflare.consumer.command '["/usr/local/bin/enqueue-mail-agent"]'
afmail cloudflare consume --reconcile
afmail cloudflare watch

The first command runs one finite tick for inspection. The second stays in the foreground; run it under a service manager on the VPS. An example systemd unit:

[Unit]
Description=Receive mail and enqueue agent work
Wants=network-online.target
After=network-online.target

[Service]
Type=simple
User=mailagent
WorkingDirectory=/srv/private-mail
ExecStart=/usr/local/bin/afmail cloudflare watch
Restart=on-failure
RestartSec=10
KillMode=control-group

[Install]
WantedBy=multi-user.target

The service account needs access to the private workspace, credentials, and configured handler. Installing a unit and provisioning Cloudflare resources are operator deployment steps; running the local tests does neither.

Configuration

HTTP pull uses short polling, not a held-open connection. Latency depends on the poll interval and ongoing work. A handler runs synchronously; use a quick durable scheduler handoff to keep receiving mail while the agent works. No subsecond or exactly-once delivery guarantee is made.

watch reloads configuration between ticks, emits cloudflare_consumer_ready once and cloudflare_consumer_tick for work or warnings, and stays quiet on idle ticks. These are structured progress events on its ordered output stream. Stop it with the service manager or Ctrl-C. The finite consume command accepts --reconcile to scan immediately; without it, it consumes events and retries local work without listing the R2 bucket.

Handler contract

The command is executed directly, without a shell, in the workspace root. No subject, sender, body, recipient, object key, or attachment name is inserted into argv. Stdin and child stdout/stderr are disconnected from the afmail protocol stream; the handler should manage its own logs without logging secrets. Two environment variables supply trusted task locators:

• AFMAIL_TASK_PATH: absolute path to a persisted JSON batch.
• AFMAIL_TASK_ID: stable idempotency key for that batch, unchanged on retries.

A task has this shape (the example ID and message ID are illustrative):

{
  "schema_name": "mail_received_task",
  "schema_version": 1,
  "task_id": "mail_0123456789abcdef",
  "messages": [
    {
      "message_id": "message_20261024_0123456789abcdef",
      "raw_path": ".afmail/messages/message_20261024_0123456789abcdef.eml"
    }
  ]
}

The agent can read the referenced raw mail or use afmail message show MESSAGE_ID. The workspace lock is released while the handler runs so ordinary afmail commands remain available. Several new messages received in one tick can share a task. Repeated deliveries with identical raw bytes merge into the existing message, including envelope recipients, and do not create another task.

Exit zero only after the scheduler durably accepts the task (or after inline processing completes). The scheduler must deduplicate by AFMAIL_TASK_ID: a crash after successful handoff but before recording completion can invoke the handler again. The trigger is at-least-once. A nonzero exit, start failure, or timeout leaves the same active batch pending. Fix the handler and resume the consumer; the retry uses the same task ID and inputs. An active failed batch is retried before creating another batch. Remaining new mail stays pending locally. Scheduling acceptance and agent processing are separate: after durable scheduler acceptance, the scheduler owns agent execution retries.

Treat all mail content as untrusted data. A trigger authorizes the configured handler, not commands embedded in a message. Existing push/send approval requirements remain in force. This consumer does not automatically push mail.

Persistence and recovery

The order is:

1. Persist a preparing notification intent for previously unseen raw bytes.
2. Import raw mail and its canonical delivery metadata; make the task pending and persist the imported-object receipt.
3. Delete/archive the R2 spool object through the existing control interface when managed; persist cleanup completion.
4. Acknowledge the Queue event. If acknowledgement fails, a repeated event can be settled from its receipt without scheduling the agent again.
5. Persist a stable batch and run the handler. Record task completion only after the handler exits zero.

A crash between import and task readiness leaves the intent in place; replay merges the imported bytes and makes that intent ready. Failed remote cleanup keeps its event retryable while the local task can already be dispatched. Receipts repair cleanup after an ambiguous response. A missing R2 object with no local receipt is an error and is never acknowledged as successful import.

State lives in .afmail/cloudflare-consumer/journal.json, with task input files in .afmail/cloudflare-consumer/batches/. Preserve these as private canonical workspace state, including completed tasks and receipts; they are not disposable render caches. Restore the journal and mail evidence together. Do not delete or hand-edit the journal to retry work. A corrupt/unsupported journal is reported, not silently reset. The process lock is machine-local and is not a distributed claim across cloned workspaces.

R2 remains the recovery source until cleanup. watch periodically scans pending objects even when the Queue API is temporarily unavailable. This recovers mail whose event expired, reached a dead-letter queue, or was never configured. Scanning retains the ordinary pull count/byte limits. Inspect warnings and pending_message_count in finite results and watch events; a successful tick with warnings does not mean every event or handler succeeded. Configure and monitor a dead-letter queue for repeatedly failing events, and retain pending R2 objects long enough for recovery.

Use the generated Worker or another spool writer with unique, immutable object keys. Reusing an already consumed key violates the receipt identity contract. One consumer tick per workspace is protected through dispatch by a process lock; the normal workspace lock protects import and journal mutations. Run one watch service for a workspace. Multiple hosts pulling one Queue compete for messages; they do not each receive a complete mailbox copy. This feature does not add cross-host workspace coordination or a shared helpdesk.

Validation

Run the repository entrypoints:

bash scripts/test.sh consumer
bash scripts/test.sh pull
bash scripts/test.sh static

The consumer gate uses local HTTP Queue/R2 fixtures and a real subprocess handler. It checks idle behavior, direct object import, batched duplicate suppression, handler access to afmail while dispatch runs, stable retries, remote cleanup and Queue acknowledgement failures, malformed/missing events, R2 recovery during a Queue outage, crash recovery, and credential redaction. No Cloudflare account or model is contacted. Provisioned R2 notifications and live Queue delivery still need deployment verification against the operator’s account.