Task Creation from Email or Chat Messages in Macro: The Complete Workflow

Macro converts plain emails and chat messages into structured task objects through a multi-stage pipeline involving email parsing, scheduled action persistence, background dispatch, and final document storage.

The macro-inc/macro repository implements an end-to-end workflow for task creation from email or chat messages. Incoming communications are parsed, transformed into JSON payloads, persisted to PostgreSQL, and ultimately stored as documents that downstream services can process. This article breaks down each stage with source code references from the actual implementation.

---

How Email Messages Trigger Task Creation

The email flow begins when an inbound message arrives at Macro's email service. The system parses MIME payloads and extracts structured task data.

Email Reception and Parsing

In email_service/src/util/sync_contacts.rs, the email service handles inbound webhooks. The parser scans email bodies for task indicators—typically a Task: block or specially formatted commands—and constructs a JSON task object containing the title, description, due date, and owner information.

This extracted payload is then forwarded to the scheduled action system for persistence and execution.

Scheduled Action API Entry Point

The parsed task JSON arrives at the HTTP endpoint defined in scheduled_action/src/inbound/axum_router.rs. The POST /scheduled_action/create handler receives a request body with:

• owner: The task assignee (e.g., macro|user@user.com)
• name: Display name for the scheduled action
• schedule: ISO 8601 timestamp for execution
• kind: Action type identifier (set to "task" for task creation)
• timezone: Scheduling timezone
• task: The nested task definition object

POST /scheduled_action/create
Content-Type: application/json

{
  "owner": "macro|user@user.com",
  "name": "Follow‑up on Q3 report",
  "schedule": "2024-09-01T09:00:00Z",
  "kind": "task",
  "timezone": "America/Los_Angeles",
  "task": {
    "title": "Review Q3 financials",
    "description": "Ask finance team for the latest numbers",
    "due_at": "2024-09-05T17:00:00Z"
  }
}

---

Database Persistence and Dispatch

Once the API receives the payload, the system persists it for reliable, at-least-once execution.

Scheduled Action Repository

The scheduled_action/src/outbound/pg_scheduled_action_repo.rs module handles database operations. It executes an INSERT into the scheduled_action table, storing:

• Raw JSON payload
• Owner identity
• Schedule metadata
• The serialized task column

This persistence decouples receipt from execution, ensuring tasks survive service restarts.

Background Polling Dispatcher

The scheduled_action/src/outbound/pg_polling_dispatcher.rs runs continuously, querying for pending scheduled_action rows. For each due record, it spawns a Tokio task via TaskTracker to execute the payload asynchronously.

The dispatcher determines execution path based on the kind field:

• Worker tasks: Trigger external ECS containers
• Document tasks: Call internal document service APIs

---

Task Execution Paths

Macro supports two execution modes depending on task requirements.

External Worker Execution (ECS)

For tasks requiring isolated compute resources, the dispatcher invokes worker_trigger/src/service/ecs/run_task.rs. This module calls AWS ECS RunTask with a task_definition ARN, launching a containerized worker that processes the payload.

This path suits resource-intensive or security-sensitive operations.

Document Service Task Creation

Most email-derived tasks follow the internal path to documents/src/inbound/axum_router/create_task.rs. This handler:

1. Validates the CreateTaskRequest
2. Generates a task ID
3. Inserts a row into the documents table via documents/src/outbound/pg_document_repo.rs
4. Returns the created task identifier

POST /documents/create_task
Content-Type: application/json
Authorization: Bearer <jwt>

{
  "title": "Update product roadmap",
  "description": "Add Q4 milestones",
  "owner": "macro|user@user.com",
  "metadata": { "priority": "high" }
}

The documents table serves as the canonical store for all tasks, enabling downstream services like search indexing and notification systems to react to new entries.

---

Chat Message Task Creation

Chat platforms like Slack follow a parallel but unified path through the notification service.

Notification Service Event Handling

When a chat message contains a task assignment, notification_service/src/api/user_notification.rs receives a task_assigned event. The handler:

• Persists the event to the notifications table
• Spawns an in-process executor task

In-Process Agent Task Execution

The scheduled_action/src/outbound/inprocess_executor/agent_task.rs module extracts task descriptions from chat payloads and constructs API calls to the Documents service. This Rust code illustrates the pattern:

let resp = client
    .post("/documents/create_task")
    .json(&CreateTaskRequest {
        title: "Follow‑up on demo".into(),
        description: Some("Schedule a follow‑up call".into()),
        owner: "macro|user@user.com".into(),
        metadata: None,
    })
    .send()
    .await?;

This design reuses the same create_task endpoint regardless of origin—email or chat—ensuring consistent validation, persistence, and downstream processing.

---

Architecture Comparison: Email vs. Chat Flows

Aspect	Email Flow	Chat Flow
Entry point	email_service webhook	notification_service event
Parsing location	sync_contacts.rs	user_notification.rs
Initial storage	scheduled_action table	notifications table
Dispatcher	pg_polling_dispatcher.rs	In-process executor
Task creation	Documents API (async)	Documents API (sync/async hybrid)
Latency	Minutes (scheduled)	Seconds (real-time)

Both paths converge at the Documents service, maintaining a single source of truth for task state.

---

Summary

• Email parsing happens in email_service/src/util/sync_contacts.rs, extracting task JSON from MIME bodies.
• Scheduled actions provide durable persistence via pg_scheduled_action_repo.rs and reliable execution through pg_polling_dispatcher.rs.
• Worker tasks launch ECS containers through worker_trigger/src/service/ecs/run_task.rs.
• Document tasks persist to the documents table via documents/src/inbound/axum_router/create_task.rs.
• Chat messages route through notification_service and execute via inprocess_executor/agent_task.rs, reusing the same Documents API.

---

Frequently Asked Questions

How does Macro identify tasks inside email bodies?

The email service in sync_contacts.rs scans for structured markers like Task: prefixes or command syntax. It parses the surrounding text into JSON fields including title, description, and due dates. The parser handles plain text and HTML MIME parts.

What happens if the scheduled action dispatcher crashes?

The pg_polling_dispatcher.rs uses PostgreSQL row-level locking with SKIP LOCKED to ensure exactly-once processing. Unacknowledged tasks remain in the table and are retried on restart. The scheduled_action table persistence guarantees no task loss.

Why are chat tasks handled differently from email tasks?

Chat requires lower latency. The notification service uses an in-process executor in agent_task.rs rather than polling, enabling sub-second task creation. Email tolerates minutes of delay, making the polling dispatcher more efficient for batch processing.

Can external systems trigger this same workflow?

Yes. Any source can POST to /scheduled_action/create with a valid task payload. The API in axum_router.rs accepts authenticated requests from integrations, not just internal services.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}