Catch-up 4/4: email replies, docs, fixture cleanup

README.md

@@ -10,20 +10,23 @@ One Cloudflare Worker serves the SPA, the REST API, an MCP server, and WebSocket
 
 Task managers are slow, expensive per seat, and hostile to agents. Flow inverts all three:
 
-- **Agent-first.** A built-in MCP server at `/mcp` exposes 15 tools with proper annotations and structured output. Every tool calls the same Durable Object RPC method the UI and REST routes call, so a task created by Claude fires exactly the automations a task created by a human does. API keys impersonate real users, and the audit trail records which key did what — you always know whether a change came from a person or their agent.
+- **Agent-first.** A built-in MCP server at `/mcp` exposes 18 tools with proper annotations and structured output. Every tool calls the same Durable Object RPC method the UI and REST routes call, so a task created by Claude fires exactly the automations a task created by a human does. API keys impersonate real users, and the audit trail records which key did what — you always know whether a change came from a person or their agent.
 - **Fast in a way SaaS can't be.** The whole workspace lives in one Durable Object's SQLite, so reads are single-digit-millisecond, in-process queries — no network hop to a database. The client gets one snapshot, then applies deltas over a WebSocket forever; there is no "refetch the board" path anywhere in the codebase. Mutations are optimistic in the UI, and a localStorage boot cache paints the last known board before the bundle has even finished loading.
 - **Yours.** Your tasks live in your Cloudflare account, in SQLite you can query, with an audit log of every mutation. No seat pricing, no data egress negotiation, no vendor deciding which API endpoints agents are allowed to use.
 
 ## Features
 
 - **Boards** with per-list status pipelines (exactly one `open` status first, one `closed` last, custom statuses between), drag-and-drop with fractional positions, filters, and a command palette
 - **Tasks** with markdown descriptions, single assignee, priority, due/start dates, tags, snooze ("waiting on…" with a wake time), and comments
-- **Asana-style subtasks** — lightweight done/not-done checklist steps with optional assignee and due date, deliberately not miniature tasks with their own status pipeline
+- **Asana-style subtasks** — lightweight done/not-done checklist steps with optional assignee, due date and a plain-text note, deliberately not miniature tasks with their own status pipeline
+- **My Work** — everything assigned to you, subtasks included, bucketed by due date, plus a rail of work you've assigned out
 - **Real-time sync** — delta protocol with a monotonic `seq`, replay on reconnect, hibernating WebSockets so idle connections cost nothing
-- **MCP server** — streamable HTTP at `/mcp`, 15 tools, structured output, per-tool annotations, error messages written so agents can self-correct
+- **MCP server** — streamable HTTP at `/mcp`, 18 tools, structured output, per-tool annotations, error messages written so agents can self-correct
 - **REST API** — the full surface, curl-friendly, with per-item results on bulk operations (up to 200 at a time)
 - **Automations** — trigger → conditions → actions rules evaluated inline in the mutation turn (depth-capped at 5), with a per-run log; outbound webhooks (HMAC-signed) and email go through a queue
 - **Email notifications** — assigned-to-me, comment-on-my-task, status changes, per-user preferences, sent via Cloudflare Email Sending with a dry-run gate that defaults to on
+- **Agent handoffs** — one person's agent files a request for a teammate's agent as a task with a state (requested → working / needs input → done / declined), over MCP or REST, with the trust model spelled out in the tool descriptions
+- **Email intake** — each list can take tickets at `<alias>@your-domain` from allowlisted senders, with threading (replies become comments), Cloudflare-verified sender authentication, optional acknowledgements, and comments you can email back to the requester
 - **Inbound webhooks** — per-list `inb_` tokens so a bug tracker or form tool can create tasks in exactly one list, with idempotent delivery
 - **Per-space visibility** — spaces are workspace-visible or private to a member list; snapshots, deltas, search results, and writes all enforce it
 - **Audit log** — every mutation records who, via what (`ui`/`api`/`mcp`/`webhook`/`automation`/`import`), which API key, which automation rule, and the diff
@@ -118,7 +121,7 @@ That's the whole integration. The agent gets workspace orientation, search, task
 |---|---|
 | [docs/ARCHITECTURE.md](docs/ARCHITECTURE.md) | The single-DO decision and its limits, the delta protocol, snapshot slimming, the automations engine, auth model, performance techniques |
 | [docs/SELF_HOSTING.md](docs/SELF_HOSTING.md) | End-to-end deployment runbook: Cloudflare setup, Access, email, R2, Queues, bootstrap |
-| [docs/AGENTS_AND_MCP.md](docs/AGENTS_AND_MCP.md) | MCP server, the 15 tools, API keys and impersonation, webhooks in and out |
+| [docs/AGENTS_AND_MCP.md](docs/AGENTS_AND_MCP.md) | MCP server, the 18 tools, agent handoffs, API keys and impersonation, webhooks in and out |
 | [docs/API.md](docs/API.md) | REST reference, every route with method, auth, and a curl example |
 | [docs/AUTOMATIONS.md](docs/AUTOMATIONS.md) | The rule vocabulary: triggers, conditions, actions, templates, the run log |
 | [docs/IMPORTING.md](docs/IMPORTING.md) | Migrating from ClickUp; where Asana/Trello extractors would go |

apps/api/README.md

@@ -405,7 +405,7 @@ flowcurl "$FLOW/api/handoffs?direction=outgoing&states=done,declined&limit=20"
 # Reply: a message (posted as a comment), a new state, or both.
 flowcurl -X POST $FLOW/api/tasks/tk_hnd123/handoff \
   -H 'Content-Type: application/json' \
-  -d '{"message":"| page | sessions | vs Feb |\n|---|---|---|\n| /implants | 1,204 | +8% |","state":"done"}'
+  -d '{"message":"| page | sessions | vs Feb |\n|---|---|---|\n| /pricing | 1,204 | +8% |","state":"done"}'
 # {"task":{...},"comment":{...}}     comment is null when only the state moved
 ```
 
@@ -674,6 +674,26 @@ flowcurl -X PUT $FLOW/api/lists/ls_def456/email-intake \
 flowcurl -X DELETE $FLOW/api/lists/ls_def456/email-intake
 ```
 
+**Replying by email.** With `EMAIL_INTAKE_REPLY="true"`, a comment posted with
+`"emailToRequester": true` also goes to the sender whose email opened the
+ticket, verified or not (the address is always on the allowlist, so a forger
+never receives the reply): from `EMAIL_FROM`, as `Re: <title>`, threaded on
+their messages, with Reply-To on the intake address. The queue consumer sends
+it, then marks the comment `sent`; a dead-lettered send marks it `failed`, and
+a failed report never retries into a second email. A person in the app can
+send one; an api key needs `canEmailRequesters`, which an owner or admin sets
+with `PATCH /api/api-keys/:id`; MCP never can.
+
+```bash
+flowcurl -X PATCH $FLOW/api/api-keys/ak_abc123 \
+  -H 'Content-Type: application/json' -d '{"canEmailRequesters":true}'
+
+flowcurl -X POST $FLOW/api/tasks/tk_ghi789/comments \
+  -H 'Content-Type: application/json' \
+  -d '{"body":"Done, here it is: https://example.com/page","emailToRequester":true}'
+# {"id":"cm_...","emailTo":"front.desk@example.org","emailStatus":"sending",...}
+```
+
 ### Import
 
 Admin-only, and used by `apps/importer` during the ClickUp migration.
@@ -1024,7 +1044,7 @@ src/
   mcp/context.ts      per-request identity + memoised id -> name index
   mcp/views.ts        result shaping — ids to names                   (pure)
   mcp/work.ts         due-date bucketing for flow_list_my_work        (pure)
-  side-effects/       owned by the automations agent
+  side-effects/       queue consumer: webhook + email delivery, retries, DLQ
 ```
 
 `src/do.ts` is worth reading first if you are touching the DO: it is the full

apps/api/src/auth.test.ts

@@ -48,6 +48,7 @@ const key: ApiKey = {
   createdAt: 1_700_000_000_000,
   lastUsedAt: null,
   revokedAt: null,
+  canEmailRequesters: false,
 };
 
 const env = {

apps/api/src/do.ts

@@ -111,6 +111,8 @@ export interface WorkspaceApi extends WorkspaceRpc {
     actor: string | Actor
   ): Promise<ApiKey>;
   revokeApiKey(keyId: string, actor: string | Actor): Promise<Ok>;
+  /** Owner/admin, from the web app only: allow or stop a key emailing ticket requesters. */
+  setApiKeyEmailRequesters(keyId: string, allowed: boolean, actor: string | Actor): Promise<ApiKey>;
 
   // --- tasks ----------------------------------------------------------------
   createTask(input: CreateTaskInput, actor: string | Actor): Promise<Task>;
@@ -287,6 +289,12 @@ export interface WorkspaceApi extends WorkspaceRpc {
     kind: "webhook" | "email";
     detail: string;
   }): Promise<Ok>;
+  /**
+   * The side-effects consumer's report on an emailed comment: "sent" once the
+   * mail service accepted it, "failed" once delivery gave up. Moves only a
+   * comment still "sending"; `ok` is false when there was nothing to move.
+   */
+  markEmailReply(commentId: string, status: "sent" | "failed"): Promise<{ ok: boolean }>;
 
   // --- audit ----------------------------------------------------------------
   /**

apps/api/src/email-intake.test.ts

@@ -320,8 +320,8 @@ describe("authenticationLog", () => {
 describe("titleFromSubject", () => {
   it("collapses whitespace and strips reply/forward prefixes", () => {
     expect(titleFromSubject("Re: FW:  Printer \n  broken")).toBe("Printer broken");
-    expect(titleFromSubject("RE: Fwd: Re[2]: x-ray sensor")).toBe("x-ray sensor");
-    expect(titleFromSubject("Recall notice")).toBe("Recall notice");
+    expect(titleFromSubject("RE: Fwd: Re[2]: printer jam")).toBe("printer jam");
+    expect(titleFromSubject("Renewal notice")).toBe("Renewal notice");
   });
 
   it("falls back for a blank or prefix-only subject", () => {
@@ -1302,13 +1302,13 @@ describe("deleteUploads", () => {
     remove.mockResolvedValueOnce(undefined).mockRejectedValueOnce(new Error("R2 500")).mockResolvedValueOnce(undefined);
     const uploads = Array.from({ length: 2500 }, (_, i) => ({
       id: `at_${i}`,
-      r2Key: `email/abc/at_${i}/patient-${i}.pdf`,
+      r2Key: `email/abc/at_${i}/private-name-${i}.pdf`,
     }));
     await deleteUploads(makeEnv(), uploads);
     expect(remove.mock.calls.map((call) => (call[0] as string[]).length)).toEqual([1000, 1000, 500]);
     const logged = errors.join("\n");
     expect(logged).toContain("at_1000");
-    expect(logged).not.toContain("patient-");
+    expect(logged).not.toContain("private-name-");
     expect(logged).not.toContain("email/");
   });
 });

apps/api/src/email-intake.ts

@@ -10,12 +10,14 @@ import {
   type IngestEmailResult,
 } from "@flow/core";
 import { id, ID_PREFIX } from "@flow/core/id";
+import { SYNTHETIC_ID_PREFIX } from "@flow/core/email-intake";
 import { LIMITS, type Actor } from "@flow/shared";
 import { findOwnerActor } from "./auth.js";
 import { workspace } from "./do.js";
 import type { Env } from "./env.js";
 import { contentDisposition, mimeEssence, sanitizeFilename } from "./routes/attachments.js";
 import { emailBrandName, emailFrom, renderEmailHtml } from "./side-effects/index.js";
+import { HEADER_VALUE_MAX_BYTES, errorCode, sendWithHeaders, threadingHeaders } from "./email-thread.js";
 
 /**
  * Email intake: the Worker half.
@@ -55,14 +57,13 @@ export const PERMANENT_REJECT = "This message could not be filed. Please contact
 const UNREADABLE_REJECT = "This message could not be read. Please send it again as a regular email.";
 
 /** Prefix of the dedupe key minted for a message that arrives without a Message-ID. */
-export const SYNTHETIC_ID_PREFIX = "sha256:";
+export { SYNTHETIC_ID_PREFIX };
 
 
 /** Fixed, so nothing the sender wrote is ever reflected back out in our name. */
 export const ACK_SUBJECT = "We received your request";
 
-/** Email Service caps each custom header value at 2,048 bytes. */
-export const HEADER_VALUE_MAX_BYTES = 2048;
+export { HEADER_VALUE_MAX_BYTES, threadingHeaders };
 
 /** Text and HTML parts are cut to this many characters before any regex or normalization runs. */
 export const BODY_SOURCE_MAX = 2 * LIMITS.descriptionMax;
@@ -833,7 +834,7 @@ export function planAttachments(parsed: Pick<Email, "attachments" | "html">, mes
 /**
  * Best effort, in batches R2 accepts: an object left behind is unreferenced
  * garbage, never a broken link. Logs ids and counts only, since keys and
- * filenames can name patients.
+ * filenames can name people.
  */
 export async function deleteUploads(
   env: Env,
@@ -899,28 +900,6 @@ export function acknowledgementMarkdown(): string {
   ].join("\n");
 }
 
-function utf8Length(text: string): number {
-  return new TextEncoder().encode(text).byteLength;
-}
-
-/**
- * In-Reply-To and References for the acknowledgement, so the client's reply
- * carries the original Message-ID in its References and threads back onto the
- * ticket. When the chain outgrows the header value cap, ids are dropped from
- * the middle: RFC 5322 §3.6.4 keeps the thread's first id and its newest ones.
- * Null when the message had no real Message-ID to point at.
- */
-export function threadingHeaders(messageId: string, parentIds: readonly string[]): Record<string, string> | null {
-  if (messageId.startsWith(SYNTHETIC_ID_PREFIX) || /[\r\n]/.test(messageId)) return null;
-  const inReplyTo = `<${messageId}>`;
-  if (utf8Length(inReplyTo) > HEADER_VALUE_MAX_BYTES) return null;
-  const ids = [...new Set([...parentIds, messageId])];
-  const format = () => ids.map((value) => `<${value}>`).join(" ");
-  while (ids.length > 2 && utf8Length(format()) > HEADER_VALUE_MAX_BYTES) ids.splice(1, 1);
-  const references = utf8Length(format()) > HEADER_VALUE_MAX_BYTES ? inReplyTo : format();
-  return { "In-Reply-To": inReplyTo, References: references };
-}
-
 export interface Acknowledgement {
   /** The send without custom headers; the fallback when they are refused. */
   base: EmailMessageBuilder;
@@ -964,12 +943,6 @@ export function buildAcknowledgement(input: {
   };
 }
 
-function errorCode(err: unknown): string {
-  const code = (err as { code?: unknown } | null)?.code;
-  if (typeof code === "string") return code;
-  return /\bE_HEADER\w*/.exec(errorMessage(err))?.[0] ?? "";
-}
-
 /** "Name <address>" for the dry-run log, whichever form the builder holds. */
 function senderLabel(from: EmailMessageBuilder["from"]): string {
   return typeof from === "string" ? from : `${from.name ?? ""} <${from.email}>`.trim();
@@ -995,20 +968,16 @@ export async function sendAcknowledgement(env: Env, ack: Acknowledgement, taskId
   if (!binding) {
     throw new Error('EMAIL_DRY_RUN is off but the SEND_EMAIL binding is missing from wrangler.jsonc');
   }
-  try {
-    await binding.send({ ...ack.base, headers: ack.headers });
-  } catch (err) {
-    if (!errorCode(err).startsWith("E_HEADER")) throw err;
+  await sendWithHeaders(binding, ack.base, ack.headers, (code) => {
     console.warn(
       JSON.stringify({
         level: "warn",
         msg: "email intake ack: threading headers were refused; resending without them",
         taskId,
-        code: errorCode(err),
+        code,
       })
     );
-    await binding.send(ack.base);
-  }
+  });
 }
 
 // --- the handler --------------------------------------------------------------

apps/api/src/email-thread.ts

@@ -0,0 +1,69 @@
+/// <reference types="@cloudflare/workers-types" />
+// Threading and sending for mail Flow writes back to the person who emailed a
+// ticket in: the intake acknowledgement (./email-intake.ts) and the comments a
+// person chooses to email back (./side-effects). Kept apart from both so
+// neither has to import the other.
+
+import { SYNTHETIC_ID_PREFIX } from "@flow/core/email-intake";
+
+/** Email Service caps each custom header value at 2,048 bytes. */
+export const HEADER_VALUE_MAX_BYTES = 2048;
+
+function utf8Length(text: string): number {
+  return new TextEncoder().encode(text).byteLength;
+}
+
+/** A message id that can go inside angle brackets in a header: no whitespace, brackets or synthetic key. */
+function headerSafeId(value: string): boolean {
+  return value !== "" && !value.startsWith(SYNTHETIC_ID_PREFIX) && !/[\s<>]/.test(value);
+}
+
+/**
+ * In-Reply-To and References, so the requester's reply carries the ticket's
+ * Message-IDs in its References and threads back onto it. When the chain
+ * outgrows the header value cap, ids are dropped from the middle: RFC 5322
+ * §3.6.4 keeps the thread's first id and its newest ones. Null when there is
+ * no real Message-ID to point at.
+ */
+export function threadingHeaders(messageId: string, parentIds: readonly string[]): Record<string, string> | null {
+  if (!headerSafeId(messageId)) return null;
+  const inReplyTo = `<${messageId}>`;
+  if (utf8Length(inReplyTo) > HEADER_VALUE_MAX_BYTES) return null;
+  const ids = [...new Set([...parentIds.filter(headerSafeId), messageId])];
+  const format = () => ids.map((value) => `<${value}>`).join(" ");
+  while (ids.length > 2 && utf8Length(format()) > HEADER_VALUE_MAX_BYTES) ids.splice(1, 1);
+  const references = utf8Length(format()) > HEADER_VALUE_MAX_BYTES ? inReplyTo : format();
+  return { "In-Reply-To": inReplyTo, References: references };
+}
+
+function errorMessage(err: unknown): string {
+  return err instanceof Error ? err.message : String(err);
+}
+
+/** The Email Service error code (E_HEADER_…, E_RATE_LIMIT…), or "" when there isn't one. */
+export function errorCode(err: unknown): string {
+  const code = (err as { code?: unknown } | null)?.code;
+  if (typeof code === "string") return code;
+  return /\bE_HEADER\w*/.exec(errorMessage(err))?.[0] ?? "";
+}
+
+/**
+ * Send with custom headers, and once more without them if the binding refuses
+ * a header: mail that lands outside the thread beats mail that never lands.
+ * `onHeadersRefused` gets the refusal's code so the caller can log it.
+ */
+export async function sendWithHeaders(
+  binding: SendEmail,
+  base: EmailMessageBuilder,
+  headers: Record<string, string>,
+  onHeadersRefused: (code: string) => void
+): Promise<void> {
+  try {
+    await binding.send({ ...base, headers });
+  } catch (err) {
+    const code = errorCode(err);
+    if (!code.startsWith("E_HEADER")) throw err;
+    onHeadersRefused(code);
+    await binding.send(base);
+  }
+}

apps/api/src/env.ts

@@ -40,7 +40,14 @@ export type Env = CoreEnv & {
    * writes, because an ack only goes to senders that header vouches for.
    */
   EMAIL_INTAKE_ACK?: string;
-  /** Email Service binding (`send_email` in wrangler.jsonc): notifications, automations, intake acks. */
+  /**
+   * "true" lets a person in the web app (or an api key allowed to) email a
+   * comment back to the requester of a ticket that came in by email. Read by
+   * the DO, which refuses replies while it's off; the panel then hides the
+   * option.
+   */
+  EMAIL_INTAKE_REPLY?: string;
+  /** Email Service binding (`send_email` in wrangler.jsonc): notifications, automations, intake acks and replies. */
   SEND_EMAIL?: SendEmail;
   /**
    * Local-dev escape hatch. Auth resolves to the owner user when this is

apps/api/src/mcp/schemas-out.ts

@@ -106,6 +106,9 @@ export const CommentView = z.object({
   createdAt: z.number(),
   /** True when the author's agent wrote it (posted with a key); absent otherwise. */
   agent: z.boolean().optional(),
+  /** Set when this comment was also emailed to the ticket's requester: who, and where delivery stands. */
+  emailedTo: z.string().optional(),
+  emailStatus: z.enum(["sending", "sent", "failed"]).optional(),
 });
 export type CommentView = z.infer<typeof CommentView>;
 

apps/api/src/mcp/schemas.ts

@@ -46,7 +46,8 @@ export const UpdateTaskArgs = UpdateTaskInput;
 export const MoveTaskArgs = MoveTaskInput;
 export const BulkUpdateTasksArgs = BulkUpdateInput;
 export const ToggleSubtaskArgs = ToggleSubtaskInput;
-export const CommentOnTaskArgs = CreateCommentInput;
+/** No emailToRequester: emailing a client is for a person in the app, never an agent. */
+export const CommentOnTaskArgs = CreateCommentInput.omit({ emailToRequester: true });
 export const UpsertAutomationArgs = UpsertAutomationInput;
 export const RequestHandoffArgs = RequestHandoffInput;
 export const ListHandoffsArgs = ListHandoffsInput.extend({ format: Format });

apps/api/src/mcp/server.test.ts

@@ -122,6 +122,7 @@ const auth: AuthContext = {
     createdAt: 0,
     lastUsedAt: null,
     revokedAt: null,
+    canEmailRequesters: false,
   },
   actor: { userId: "us_alice", via: "api", apiKeyId: "ak_1", automationRuleId: null },
 };