Developers · September 26, 2026

Resume an Agent Workflow When a Reply Arrives

How to resume an agent workflow when an email reply arrives: key on thread_id, dedupe message.received by event ID, and catch replies that break the thread.

Your agent emails a supplier asking for a revised quote, and the workflow has nothing to do until they answer. The answer might take four minutes or four days. Holding a process open for that is fragile, and polling the inbox in a loop is wasteful. So the real question is how to resume an agent workflow when an email reply arrives, on a different worker, days later, with none of the original process left.

This piece settles the mechanics: which event fires, which identifier survives the round trip, how to record that identifier without a crash window, what the resume handler looks like, and what to do when a reply arrives that the key cannot place. Header mechanics are covered in depth in keeping an agent’s replies in the right thread, and deciding how long to wait before nudging is left to what to do when the reply never comes.

How to resume an agent workflow when an email reply arrives

Treat the reply as an event that rehydrates stored state, not as the return value of a blocking call. The workflow sends, writes down where it is, and stops. Nothing is running while it waits. When the reply lands, an event handler finds the stored state, marks the workflow runnable, and a worker picks it up from the step it recorded.

That shape has four parts, and each one has a specific failure it exists to prevent:

The rest of this article builds those four parts against the developer API, in the order a message travels.

Which event fires: message.received, not email.received

The event you subscribe to is message.received. It is in the declared catalogue in packages/contracts/src/events.ts, and it is actually emitted: once for live inbound mail, after the message, its recipients and its content are committed, and once for each simulated message injected into a test inbox.

email.received is the name of the upstream mail provider’s callback to this system, which is where the confusion comes from. It is not in the customer event catalogue and is never delivered to your webhook, so a handler registered for it never runs.

The message.received payload is deliberately small:

{
  "id": "evt_...",
  "type": "message.received",
  "api_version": "2026-09-19",
  "data": {
    "message_id": "msg_...",
    "thread_id": "thd_...",
    "inbox_id": "ibx_...",
    "environment": "live",
    "simulated": false
  }
}

That is the webhook envelope: id, type, api_version and data, with the event ID repeated in the webhook-id header. thread_id is in the payload, which is what makes a one-lookup resume possible. What is not in the payload is just as important: no sender, no subject, no body, and nothing that says which of your messages this one answered. Anything the resumed step needs to reason about, it reads from the API after it has been woken. Test-mode events carry "simulated": true and "environment": "test", so the same handler runs in both.

The polled cursor at GET /v1/projects/{project_id}/events returns the same events with sequence, occurred_at and environment added. Use the webhook to wake the workflow and the cursor to catch up after downtime; the events and webhooks reference has the delivery rules and signature format.

Why thread_id is the key, and not the subject or your Message-ID

Key the workflow on thread_id, because it is the only identifier that is assigned by the inbox, stable for the life of the conversation, and present on both the send response and the inbound event.

The alternatives each fail in a specific way.

The subject belongs to the other person. They rewrite it, their client localises the Re: prefix, and a generic subject like “Quote” collides across unrelated conversations. The inbox never consults it when threading inbound mail, so neither should you.

The RFC Message-ID you sent is per message, not per conversation. The workflow’s second outbound email has a different one, and a reply to that email does not carry the first. It is also not something you can hold: it is assigned by the provider at submission, and the REST message listings do not return rfc_message_id, in_reply_to or references at all.

thread_id comes back in the send response, arrives in every message.received, and is what the inbox itself resolves inbound replies to. The resolution is covered header by header in the threading piece linked above; the part that matters for correlation is short. The inbox tries the In-Reply-To value first, then the References values newest to oldest, against Message-IDs in the same inbox. A hit joins that thread. A miss opens a new one. There is no subject fallback, no sender fallback and no cross-inbox match.

Two consequences follow. A client that drops References but keeps In-Reply-To still threads, because In-Reply-To alone is a candidate. And a recipient who changes the subject line on their reply stays in the thread, because the subject was never an input.

Recording the key without a crash window

Write the workflow row before you send, and derive the send’s Idempotency-Key from the workflow and step, so that a crash after the send can be repaired by repeating it.

The naive order is: send, read thread_id from the 202, store it. If the process dies between the 202 and the commit, the email went out, the recipient will reply, and you have no row that maps their thread to anything. The fix uses a property of the send endpoint: reusing an Idempotency-Key with an identical payload replays the stored response instead of sending again. The replayed body is the original one, so it contains the original thread_id.

CREATE TABLE workflows (
  id            text PRIMARY KEY,
  inbox_id      text NOT NULL,
  counterparty  text NOT NULL,          -- address you expect the answer from
  step          integer NOT NULL,
  status        text NOT NULL,          -- sending | waiting | runnable | running
                                        -- awaiting_approval | needs_review | done
  send_key      text,
  send_payload  jsonb,                  -- exact body, so a replay hashes identically
  lease_until   timestamptz
);

CREATE TABLE workflow_threads (
  thread_id    text PRIMARY KEY,
  workflow_id  text NOT NULL REFERENCES workflows(id),
  matched_by   text NOT NULL            -- send | sender_fallback | human
);

CREATE TABLE resume_inputs (
  event_id     text PRIMARY KEY,
  workflow_id  text NOT NULL REFERENCES workflows(id),
  message_id   text NOT NULL,
  consumed     boolean NOT NULL DEFAULT false
);

CREATE TABLE processed_events (event_id text PRIMARY KEY);

CREATE TABLE unmatched_replies (
  event_id text PRIMARY KEY, inbox_id text NOT NULL,
  thread_id text NOT NULL, message_id text NOT NULL
);

The send side, in TypeScript on Node 20 with pg:

import pg from "pg";

const API = "https://api.emailforagents.ai/v1";
const PROJECT = process.env.EFA_PROJECT_ID!;
const KEY = process.env.EFA_API_KEY!;
export const db = new pg.Pool();

export async function efa(path: string, init: RequestInit = {}) {
  const res = await fetch(`${API}/projects/${PROJECT}${path}`, {
    ...init,
    headers: {
      Authorization: `Bearer ${KEY}`,
      "Content-Type": "application/json",
      ...init.headers,
    },
  });
  const body = await res.json();
  if (!res.ok) throw Object.assign(new Error(body.error?.code), { status: res.status, body });
  return body;
}

export async function tx<T>(fn: (c: pg.PoolClient) => Promise<T>): Promise<T> {
  const c = await db.connect();
  try {
    await c.query("BEGIN");
    const out = await fn(c);
    await c.query("COMMIT");
    return out;
  } catch (err) {
    await c.query("ROLLBACK");
    throw err;
  } finally {
    c.release();
  }
}

// Step 1 of 2: persist intent. Commit this before any network call.
export async function planSend(wfId: string, step: number, payload: object) {
  await db.query(
    `UPDATE workflows SET status='sending', step=$2, send_key=$3, send_payload=$4 WHERE id=$1`,
    [wfId, step, `wf-${wfId}-step-${step}`, payload],
  );
}

// Step 2 of 2: send, or replay. Safe to call any number of times.
export async function executeSend(wfId: string) {
  const {
    rows: [wf],
  } = await db.query(`SELECT * FROM workflows WHERE id=$1`, [wfId]);
  if (wf.status !== "sending") return;
  const r = await efa(`/inboxes/${wf.inbox_id}/messages`, {
    method: "POST",
    headers: { "Idempotency-Key": wf.send_key },
    body: JSON.stringify(wf.send_payload),
  });
  if (r.result === "queued") {
    await tx(async (c) => {
      await c.query(
        `INSERT INTO workflow_threads (thread_id, workflow_id, matched_by)
         VALUES ($1, $2, 'send') ON CONFLICT (thread_id) DO NOTHING`,
        [r.thread_id, wfId],
      );
      await c.query(`UPDATE workflows SET status='waiting' WHERE id=$1 AND status='sending'`, [
        wfId,
      ]);
    });
  } else if (r.result === "approval_required") {
    await db.query(`UPDATE workflows SET status='awaiting_approval' WHERE id=$1`, [wfId]);
  }
}

A sweep that calls executeSend on every row stuck in sending for more than a minute closes the crash window. Resend the stored payload unchanged, because a changed body returns 409 idempotency_conflict. Treat 503 temporarily_unavailable (“A matching send is already in progress”) as “try again shortly”. Finish well inside the 30-day replay window, after which the key returns idempotency_expired. The full set of outcomes is in idempotency keys for agent email.

If the step continues an existing conversation, send with reply_to_message_id set to the inbound message you are answering. The reply joins the parent’s thread, so the thread_id in the response is the one already in workflow_threads, and the insert is a no-op.

The resume handler: dedupe, look up, mark runnable

The webhook handler does three things in one transaction and nothing else: record the event ID, record the input against the workflow, and flip waiting to runnable. The agent’s actual work happens in a worker, outside the callback.

import http from "node:http";
import { createHmac, timingSafeEqual } from "node:crypto";
import { tx } from "./send";

const SECRET = process.env.EFA_WEBHOOK_SECRET!;

function verified(h: http.IncomingHttpHeaders, raw: string) {
  const id = String(h["webhook-id"]),
    ts = String(h["webhook-timestamp"]);
  if (!Number.isFinite(Number(ts)) || Math.abs(Date.now() / 1000 - Number(ts)) > 300) return false;
  const want = Buffer.from(
    "v1," + createHmac("sha256", SECRET).update(`${id}.${ts}.${raw}`).digest("base64"),
  );
  const got = Buffer.from(String(h["webhook-signature"]));
  return want.length === got.length && timingSafeEqual(want, got);
}

function onEvent(evt: { id: string; type: string; data: any }) {
  return tx(async (c) => {
    const fresh = await c.query(
      `INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING RETURNING event_id`,
      [evt.id],
    );
    if (fresh.rowCount === 0 || evt.type !== "message.received") return;
    const { thread_id, message_id, inbox_id } = evt.data;
    const hit = await c.query(
      `SELECT w.id, w.status FROM workflow_threads t JOIN workflows w ON w.id = t.workflow_id
       WHERE t.thread_id = $1 FOR UPDATE OF w`,
      [thread_id],
    );
    if (hit.rowCount === 0) {
      await c.query(`INSERT INTO unmatched_replies VALUES ($1, $2, $3, $4)`, [
        evt.id,
        inbox_id,
        thread_id,
        message_id,
      ]);
    } else {
      const wf = hit.rows[0];
      await c.query(
        `INSERT INTO resume_inputs (event_id, workflow_id, message_id) VALUES ($1, $2, $3)`,
        [evt.id, wf.id, message_id],
      );
      if (wf.status === "waiting") {
        await c.query(`UPDATE workflows SET status='runnable' WHERE id=$1`, [wf.id]);
      }
    }
  });
}

http
  .createServer((req, res) => {
    const chunks: Buffer[] = [];
    req.on("data", (b) => chunks.push(b));
    req.on("end", async () => {
      const raw = Buffer.concat(chunks).toString("utf8");
      if (!verified(req.headers, raw)) return res.writeHead(400).end();
      try {
        await onEvent(JSON.parse(raw));
        res.writeHead(204).end();
      } catch {
        res.writeHead(500).end(); // retried by the dispatcher; dedupe makes that safe
      }
    });
  })
  .listen(8080);

The dedupe insert and the state change share a transaction on purpose. If you commit the event ID and crash before flipping the workflow, the redelivery is deduplicated away and the workflow waits forever. If you flip the workflow first and crash before recording the ID, the redelivery resumes it a second time. One commit makes both impossible. Why the key is the event ID and not a payload hash, and why every duplicate must still get a 2xx, is set out in handling webhook duplicates.

Notice that the handler never drops an input because the workflow is busy. A person who sends a reply and then a correction thirty seconds later produces two events on one thread. The second finds the workflow running, records its input, and leaves the status alone. The worker handles that at the end of its run:

import { db, efa, tx, executeSend } from "./send";

export const OWN_ADDRESS = process.env.EFA_INBOX_ADDRESS!.toLowerCase();

export async function runOne() {
  const {
    rows: [wf],
  } = await db.query(
    `UPDATE workflows SET status='running', lease_until=now() + interval '10 minutes'
     WHERE id = (SELECT id FROM workflows
                 WHERE status='runnable' OR (status='running' AND lease_until < now())
                 ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED)
     RETURNING *`,
  );
  if (!wf) return;
  const inputs = (
    await db.query(
      `SELECT event_id, message_id FROM resume_inputs WHERE workflow_id=$1 AND NOT consumed`,
      [wf.id],
    )
  ).rows;
  const threadIds = (
    await db.query(`SELECT thread_id FROM workflow_threads WHERE workflow_id=$1`, [wf.id])
  ).rows.map((r) => r.thread_id);

  const history = [];
  for (const t of threadIds) {
    let cursor = "";
    do {
      const page = await efa(`/inboxes/${wf.inbox_id}/threads/${t}/messages?limit=100${cursor}`);
      history.push(...page.data);
      cursor = page.has_more ? `&cursor=${encodeURIComponent(page.next_cursor)}` : "";
    } while (cursor);
  }
  const wanted = new Set(inputs.map((i) => i.message_id));
  const replies = history.filter(
    (m) =>
      wanted.has(m.id) && m.direction === "inbound" && m.from.email.toLowerCase() !== OWN_ADDRESS,
  );

  const next = await agentStep(wf, history, replies); // your model call; returns next payload or null

  await tx(async (c) => {
    await c.query(`UPDATE resume_inputs SET consumed=true WHERE event_id = ANY($1)`, [
      inputs.map((i) => i.event_id),
    ]);
    const pending = await c.query(
      `SELECT 1 FROM resume_inputs WHERE workflow_id=$1 AND NOT consumed LIMIT 1`,
      [wf.id],
    );
    if (next) {
      await c.query(
        `UPDATE workflows SET status='sending', step=step+1, send_key=$2, send_payload=$3 WHERE id=$1`,
        [wf.id, `wf-${wf.id}-step-${wf.step + 1}`, next],
      );
    } else {
      await c.query(`UPDATE workflows SET status=$2 WHERE id=$1`, [
        wf.id,
        pending.rowCount ? "runnable" : "waiting",
      ]);
    }
  });
  if (next) await executeSend(wf.id);
}

declare function agentStep(wf: any, history: any[], replies: any[]): Promise<object | null>;

The worker reads the conversation from the thread messages endpoint, which returns messages in sequence order, and fetches bodies from .../messages/{message_id}/body. It needs messages:read, and messages:send for the next send. Planning that send in the same commit that consumes the inputs is what keeps a crash from either losing the reply or answering it twice. The request shapes behind efa() are in the agent email API loop.

What breaks the thread key, and the fallback

The key breaks whenever the arriving message carries no In-Reply-To or References value that matches a message in your inbox. The inbox then opens a new thread, the event arrives with a thread_id your table has never seen, and the handler files it under unmatched_replies. Here is when that happens:

Two rows need explaining. The fast reply: the provider assigns your Message-ID at submission and a separate job records it afterwards, so a reply can reference it before the inbox knows it. When a reply carrying references arrives while a recent send in that inbox still has no recorded Message-ID, ingestion first asks the provider for it directly. Only if that lookup fails, or several unconfirmed sends are candidates, does it defer the message and retry, while the message is under ten minutes old by the provider’s received time. After that it stores the message regardless, possibly as a new thread. Preserving mail wins over grouping it. For your handler, the effect is that message.received for a very fast reply can arrive minutes late, because it is emitted only once the message is stored.

The forward: the parent lookup walks every References value, so if the forwarding client copied your chain into the forward, the colleague’s reply can land in your thread. The key matched, and the person who answered is not the one you asked.

The fallback is a sender match inside the same inbox, and it resumes only when it finds exactly one candidate. For each unmatched reply, read the thread’s messages, take the new message’s from.email, and look for workflows in that inbox, in waiting or sending, whose counterparty is that address. Including sending covers a reply that beat your sweep after a crash.

export async function resolveUnmatched(u: {
  event_id: string;
  inbox_id: string;
  thread_id: string;
  message_id: string;
}) {
  const page = await efa(`/inboxes/${u.inbox_id}/threads/${u.thread_id}/messages?limit=100`);
  const msg = page.data.find((m: any) => m.id === u.message_id);
  const sender = msg?.from.email.toLowerCase();
  if (!sender || sender === OWN_ADDRESS)
    return db.query(`DELETE FROM unmatched_replies WHERE event_id=$1`, [u.event_id]);

  const { rows } = await db.query(
    `SELECT id FROM workflows WHERE inbox_id=$1 AND lower(counterparty)=$2 AND status IN ('waiting','sending')`,
    [u.inbox_id, sender],
  );
  if (rows.length !== 1) return; // zero or several: stays queued for a human
  await tx(async (c) => {
    await c.query(
      `INSERT INTO workflow_threads VALUES ($1, $2, 'sender_fallback') ON CONFLICT DO NOTHING`,
      [u.thread_id, rows[0].id],
    );
    await c.query(
      `INSERT INTO resume_inputs (event_id, workflow_id, message_id) VALUES ($1, $2, $3)`,
      [u.event_id, rows[0].id, u.message_id],
    );
    await c.query(`UPDATE workflows SET status='runnable' WHERE id=$1 AND status='waiting'`, [
      rows[0].id,
    ]);
    await c.query(`DELETE FROM unmatched_replies WHERE event_id=$1`, [u.event_id]);
  });
}

Writing the new thread into workflow_threads means the next message on it takes the primary path, and the worker reads both threads as one history. Recording matched_by lets the agent’s next outbound say so honestly and lets you audit how often the fallback fires.

What the fallback deliberately does not do: match on subject, match across inboxes, or pick the most recent of several candidates. A supplier with three open quote requests who writes “yes, go ahead” in a fresh email is ambiguous, and guessing which order they approved is worse than asking. Those rows stay in unmatched_replies for a person. A reply from someone other than the counterparty, a forward recipient or an assistant, finds no candidate and also waits for a person, because the right answer depends on whether that person has authority to answer for them.

Three things the key cannot tell you

A thread match tells you which conversation a reply belongs to. It does not tell you three other things, and a resume handler that assumes them will misbehave.

Which of your messages it answers. The event carries no parent reference and the message listings expose no in_reply_to, so a late reply to step 1 that arrives after the workflow has sent step 2 looks the same as an answer to step 2. Validate the content against what the current step asked for, and let the agent treat a mismatch as information rather than an answer. If you need a hard boundary, give each question its own step number in your own records and have the step decide whether the reply addresses it.

That a human wrote it. A copy of your own outbound message that comes back in, from a list reflector or the agent’s own address on the Cc line, attaches to your thread and emits message.received like any other inbound message. That is why the worker filters on from.email against the inbox’s own address. Without the filter, the workflow resumes on its own echo.

That there is a thread yet. A send that returns approval_required returns a draft_id and an approval_id, not a thread_id, because no message exists yet. When a human approves, the send is submitted and message.queued fires, but its payload is message_id, thread_id and inbox_id with no approval or draft reference, so it cannot be joined to the approval_id you stored. approval.rejected is emitted; approval.approved is declared and not emitted by this build. Replaying your original Idempotency-Key does not help either: it returns the stored approval_required body, because the approved send is recorded under its own key.

That is a gap in the product, not a design to copy, and the workaround below is a stopgap. When a message.queued event arrives whose thread_id is not in workflow_threads, read that thread’s messages, take the outbound one, and compare its to addresses and subject with the payload stored on each workflow in awaiting_approval in that inbox. Bind on exactly one match; leave zero or several for a person. This does not contradict “the subject is not a key”. The subject fails as a key for inbound mail because the other person controls it. Here you match your own outbound message, and approval is bound to a hash of the draft you submitted, so the approved subject and recipients are the ones you wrote, not ones a person typed. Make the match unambiguous by putting your own reference in the subject of any first send that may be held, such as “Quote request Q-1042”. If the step is a reply, none of this is needed: reply_to_message_id fixes the thread before approval. How the approval gate holds and releases a message is covered in human approval before an agent sends.

Testing the resume path before anything is routable

Every branch above can be exercised in a test project except live header matching. POST /v1/projects/{project_id}/test/inbound injects a message into a test inbox and emits message.received with "simulated": true, but it accepts no In-Reply-To or References and always opens a new thread. That makes it the right tool for the fallback and the wrong one for the primary path.

1. Create a workflow, send step 1 from a test inbox, and confirm a workflow_threads row with matched_by = 'send'.
2. Kill the process after the POST returns and before the commit. Run the sweep and confirm the replayed response writes the same thread_id.
3. Simulate an inbound from the counterparty’s address. It arrives on a new thread, lands in unmatched_replies, and resolveUnmatched binds it with matched_by = 'sender_fallback'.
4. Replay the same webhook body with the same webhook-id. Assert 204 and no second resume_inputs row.
5. Simulate two messages from the counterparty a second apart and assert one run that sees both, or two runs in sequence, never two in parallel.
6. Simulate one from an unknown sender and assert it stays unmatched.

The primary path, a real reply joining the thread through its headers, needs one live send to a mailbox you control, answered from the client your counterparties actually use. Do that once per client family you care about, because the header behaviour in the table above is theirs, not ours.

api reliability webhooks events

Keep reading

All guides · Documentation · Developers