Author actual launch email and authentic receipt duplicate journey

tests/runner-e2e/chat-catalog-domains/runtime-email-duplicate.test.ts

@@ -0,0 +1,25 @@
+import { expect, it } from "vitest";
+import { randomUUID, createHash } from "node:crypto";
+import { mkdtemp, realpath, rm } from "node:fs/promises";
+import { tmpdir } from "node:os";
+import path from "node:path";
+import { createRequire } from "node:module";
+import { EventEmitter } from "node:events";
+import { emailDuplicateRequestMatches, matchesDuplicateSend, installOwnedEmailReceiptDuplicate, armOwnedEmailReceiptDuplicate, readOwnedEmailDuplicate, type EmailDuplicateRequest, type EmailDuplicateReceipt } from "./runtime-email-duplicate.js";
+function fixture() { const nonce = randomUUID(), request: EmailDuplicateRequest = { executionId: "chat-features.runner-codex-subscription.local.chat-channels.outbound-email", attemptId: "a".repeat(32), inboxId: "qa@example.test", marker: `Launch draft ${nonce}`, nonce, expiresAt: Date.now() + 60000 }, publication = randomUUID(), url = new URL(`https://api.agentmail.to/v0/inboxes/${encodeURIComponent(request.inboxId)}/messages/send`), init: RequestInit = { method: "POST", headers: { "Idempotency-Key": publication }, body: JSON.stringify({ to: [request.inboxId], text: request.marker, headers: { "X-Paperclip-Publication-Id": publication } }) }; return { request, publication, url, init }; }
+it("accepts only exact selected, bounded, attempt-owned receipt duplication", () => { const f = fixture(); expect(emailDuplicateRequestMatches(f.request, f.request)).toBe(true); for (const change of [{ executionId: "goals.list" }, { attemptId: "b".repeat(32) }, { expiresAt: Date.now() - 1 }, { marker: "unrelated draft" }, { extra: true }]) expect(emailDuplicateRequestMatches({ ...f.request, ...change }, f.request)).toBe(false); });
+it("observes only the authorized original publication and selected real provider/inbox", () => { const f = fixture(); expect(matchesDuplicateSend(f.request, f.url, f.init)).toBe(f.publication); expect(matchesDuplicateSend(f.request, new URL(f.url.href.replace("api.agentmail.to", "evil.example.test")), f.init)).toBeUndefined(); for (const change of [{ to: ["other@example.test"] }, { text: "wrong draft" }, { headers: { "X-Paperclip-Publication-Id": randomUUID() } }]) expect(matchesDuplicateSend(f.request, f.url, { ...f.init, body: JSON.stringify({ ...JSON.parse(String(f.init.body)), ...change }) })).toBeUndefined(); });
+it("installs nothing for unrelated executions", async () => { const original = globalThis.fetch; expect(await installOwnedEmailReceiptDuplicate({ executionIds: ["goals.list"], attemptId: "", temporaryRoot: "/not-read" })).toBeUndefined(); expect(globalThis.fetch).toBe(original); });
+it("duplicates one identical authentic receipt only; unrelated and later events remain single, sends remain one", async () => {
+ const f = fixture(), root = await realpath(await mkdtemp(path.join(tmpdir(), "email-duplicate-control-"))), prior = globalThis.fetch; let sends = 0;
+ globalThis.fetch = (async () => { sends++; return new Response(JSON.stringify({ message_id: "actual-message", thread_id: "actual-thread" }), { status: 202 }); }) as typeof fetch;
+ const WS = createRequire(new URL("../../../server/package.json", import.meta.url))("ws") as { prototype: EventEmitter }, originalEmit = WS.prototype.emit;
+ const owned = await installOwnedEmailReceiptDuplicate({ executionIds: [f.request.executionId], attemptId: f.request.attemptId, temporaryRoot: root });
+ try { await armOwnedEmailReceiptDuplicate(root, f.request); const response = await fetch(f.url, f.init); expect(response.status).toBe(202); expect((await response.json()).message_id).toBe("actual-message");
+  const socket = Object.setPrototypeOf(new EventEmitter(), WS.prototype) as EventEmitter & { _url: string }; socket._url = "wss://ws.agentmail.to/v0"; const observed: Buffer[] = []; socket.on("message", bytes => observed.push(bytes));
+  const other = Buffer.from(JSON.stringify({ event_type: "message.sent", send: { inbox_id: f.request.inboxId, message_id: "other" } })); socket.emit("message", other); expect(observed).toHaveLength(1);
+  const real = Buffer.from(JSON.stringify({ event_type: "message.sent", send: { inbox_id: f.request.inboxId, message_id: "actual-message" } })); socket.emit("message", real); expect(observed).toHaveLength(3); expect(observed[1]).toBe(real); expect(observed[2]).toBe(real);
+  socket.emit("message", real); expect(observed).toHaveLength(4); expect(sends).toBe(1); await owned!.close();
+  const receipt = readOwnedEmailDuplicate<EmailDuplicateReceipt>(root, "receipt.json"); expect(receipt.publicationId).toBe(f.publication); expect(receipt.authenticPayloadSha256).toBe(createHash("sha256").update(real).digest("hex")); expect(receipt.injectedDuplicateCount).toBe(1); expect(receipt.released).toBe(true);
+ } finally { await owned?.close(); globalThis.fetch = prior; await rm(root, { recursive: true, force: true }); } expect(WS.prototype.emit).toBe(originalEmit);
+});

tests/runner-e2e/chat-catalog-domains/runtime-email-duplicate.ts

@@ -0,0 +1,58 @@
+/** Exact selected network duplicate of one authentic AgentMail sent receipt.
+ * The real send and first provider notification are forwarded unchanged. */
+import { createHash, randomUUID } from "node:crypto";
+import { createRequire } from "node:module";
+import { EventEmitter } from "node:events";
+import { constants, openSync, readFileSync, closeSync, fstatSync, lstatSync, realpathSync } from "node:fs";
+import { mkdir, writeFile, rename } from "node:fs/promises";
+import path from "node:path";
+import { faultNotification } from "./runtime-email-fault.js";
+export const duplicateEmailScenario = "chat-channels.outbound-email";
+export interface EmailDuplicateRequest { executionId: string; attemptId: string; inboxId: string; marker: string; nonce: string; expiresAt: number }
+export interface EmailDuplicateReceipt { executionId: string; attemptId: string; nonce: string; publicationId: string; providerMessageId: string; authenticPayloadSha256: string; injectedDuplicateCount: number; released: boolean; expired: boolean }
+const sha = (value: string | Buffer) => createHash("sha256").update(value).digest("hex");
+const uuid = (v: unknown) => typeof v === "string" && /^[a-f0-9]{8}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{4}-[a-f0-9]{12}$/i.test(v);
+function directory(root: string) { return path.join(root, "email-receipt-duplicate"); }
+async function atomic(root: string, name: string, value: unknown) { const target = path.join(directory(root), name), temporary = `${target}.${randomUUID()}.tmp`; await writeFile(temporary, JSON.stringify(value), { mode: 0o600, flag: "wx" }); await rename(temporary, target); }
+function read<T>(root: string, name: string): T {
+ const dir = directory(root), d = lstatSync(dir), target = path.join(dir, name), s = lstatSync(target);
+ if (!d.isDirectory() || d.mode & 0o077 || d.uid !== process.getuid?.() || realpathSync(dir) !== dir || !s.isFile() || s.isSymbolicLink() || s.nlink !== 1 || s.mode & 0o077 || s.uid !== process.getuid?.() || s.size > 32768) throw new Error("Owned email duplicate control ownership changed");
+ const fd = openSync(target, constants.O_RDONLY | constants.O_NOFOLLOW);
+ try { const opened = fstatSync(fd); if (opened.ino !== s.ino || opened.dev !== s.dev) throw new Error("Email duplicate control changed before read"); const bytes = readFileSync(fd), after = fstatSync(fd), named = lstatSync(target); if (after.size !== opened.size || after.mtimeMs !== opened.mtimeMs || named.ino !== opened.ino || named.dev !== opened.dev) throw new Error("Email duplicate control changed during read"); return JSON.parse(bytes.toString()) as T; } finally { closeSync(fd); }
+}
+export function emailDuplicateRequestMatches(r: EmailDuplicateRequest, expected: { executionId: string; attemptId: string }, now = Date.now()) {
+ return !!r && Object.keys(r).sort().join() === ["executionId", "attemptId", "inboxId", "marker", "nonce", "expiresAt"].sort().join() && r.executionId === expected.executionId && r.executionId.endsWith(`.local.${duplicateEmailScenario}`) && r.attemptId === expected.attemptId && /^[a-f0-9]{32}$/.test(r.attemptId) && uuid(r.nonce) && /^[^\s@]+@[^\s@]+\.[^\s@]+$/.test(r.inboxId) && r.marker.includes(r.nonce) && r.marker.length <= 200 && now < r.expiresAt && r.expiresAt <= now + 480000;
+}
+export function matchesDuplicateSend(r: EmailDuplicateRequest, url: URL, init?: RequestInit) {
+ if (url.origin !== "https://api.agentmail.to" || url.pathname !== `/v0/inboxes/${encodeURIComponent(r.inboxId)}/messages/send` || url.search || init?.method !== "POST" || typeof init.body !== "string" || Buffer.byteLength(init.body) > 131072) return undefined;
+ try { const body = JSON.parse(init.body), pub = new Headers(init.headers).get("Idempotency-Key"); return uuid(pub) && body.headers?.["X-Paperclip-Publication-Id"] === pub && JSON.stringify(body.to) === JSON.stringify([r.inboxId]) && !body.cc?.length && !body.bcc?.length && typeof body.text === "string" && body.text.includes(r.marker) ? pub! : undefined; } catch { return undefined; }
+}
+export async function installOwnedEmailReceiptDuplicate(input: { executionIds: string[]; attemptId: string; temporaryRoot: string }) {
+ if (!input.executionIds.some(id => id.endsWith(`.local.${duplicateEmailScenario}`))) return undefined;
+ if (input.executionIds.length !== 1 || !/^[a-f0-9]{32}$/.test(input.attemptId) || !path.isAbsolute(input.temporaryRoot) || realpathSync(input.temporaryRoot) !== input.temporaryRoot) throw new Error("Receipt duplicate requires its exact isolated execution");
+ await mkdir(directory(input.temporaryRoot), { mode: 0o700 });
+ const WebSocket = createRequire(new URL("../../../server/package.json", import.meta.url))("ws") as { prototype: EventEmitter }, originalEmit = WebSocket.prototype.emit, originalFetch = globalThis.fetch;
+ let request: EmailDuplicateRequest | undefined, publicationId: string | undefined, providerMessageId: string | undefined, pending: Promise<void> | undefined, settle: (() => void) | undefined, receipt: EmailDuplicateReceipt | undefined, closed = false, buffered = 0, writes = Promise.resolve();
+ const save = () => { if (receipt) { const snapshot = { ...receipt }; writes = writes.then(() => atomic(input.temporaryRoot, "receipt.json", snapshot)); } };
+ const load = () => { if (request || closed) return request; try { const r = read<EmailDuplicateRequest>(input.temporaryRoot, "request.json"); if (!emailDuplicateRequestMatches(r, { executionId: input.executionIds[0]!, attemptId: input.attemptId })) throw new Error("Stale or undeclared email duplicate request"); return request = r; } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; throw error; } };
+ const release = async (expired = false) => { if (closed) return; closed = true; clearInterval(timer); if (globalThis.fetch === wrappedFetch) globalThis.fetch = originalFetch; if (WebSocket.prototype.emit === wrappedEmit) WebSocket.prototype.emit = originalEmit; if (receipt) { receipt.released = true; receipt.expired = expired; save(); } await writes; await atomic(input.temporaryRoot, "closed.json", { executionId: input.executionIds[0], attemptId: input.attemptId, nonce: request?.nonce ?? null, released: true, expired }); };
+ const wrappedFetch: typeof fetch = async (target, init) => {
+  const r = load(); if (!r || closed) return originalFetch(target, init); if (Date.now() >= r.expiresAt) { await release(true); return originalFetch(target, init); }
+  const url = new URL(typeof target === "string" ? target : target instanceof URL ? target.href : target.url), candidate = matchesDuplicateSend(r, url, init);
+  if (!candidate || publicationId) return originalFetch(target, init); publicationId = candidate; pending = new Promise(resolve => { settle = resolve; });
+  try { const response = await originalFetch(target, init); if (response.ok) { const clone = response.clone(), reader = clone.body?.getReader(); if (!reader) throw new Error("Actual send has no provider receipt"); const parts: Uint8Array[] = []; let size = 0; try { while (true) { const item = await reader.read(); if (item.done) break; size += item.value.byteLength; if (size > 32768) throw new Error("Actual provider receipt exceeds finite bounds"); parts.push(item.value); } } finally { await reader.cancel(); } const body = JSON.parse(Buffer.concat(parts).toString()) as { message_id?: unknown }; if (typeof body.message_id !== "string" || !body.message_id || body.message_id.length > 998) throw new Error("Actual provider send lacks message identity"); providerMessageId = body.message_id; } return response; } finally { settle?.(); }
+ };
+ const wrappedEmit: EventEmitter["emit"] = function (this: EventEmitter, event, ...args) {
+  const r = request, data = args[0]; if (event !== "message" || !r || closed || String((this as EventEmitter & { url?: string }).url) !== "wss://ws.agentmail.to/v0" || !Buffer.isBuffer(data) && typeof data !== "string" || Buffer.byteLength(data as string | Buffer) > 131072) return originalEmit.call(this, event, ...args);
+  let parsed: { event_type?: string; type?: string }; try { parsed = JSON.parse(String(data)); } catch { return originalEmit.call(this, event, ...args); } const notification = faultNotification(parsed);
+  if ((parsed.event_type ?? parsed.type?.replace(/^message_/, "message.")) !== "message.sent" || notification?.inboxId !== r.inboxId || !publicationId || receipt) return originalEmit.call(this, event, ...args);
+  const deliver = () => { const first = originalEmit.call(this, event, ...args); if (!closed && notification.messageId === providerMessageId && !receipt) { receipt = { executionId: r.executionId, attemptId: r.attemptId, nonce: r.nonce, publicationId: publicationId!, providerMessageId, authenticPayloadSha256: sha(data as string | Buffer), injectedDuplicateCount: 1, released: false, expired: false }; originalEmit.call(this, event, ...args); save(); } return first; };
+  if (!providerMessageId && pending) { if (++buffered > 128) { void release(true); return originalEmit.call(this, event, ...args); } void pending.then(() => { buffered--; deliver(); }); return true; } return deliver();
+ };
+ globalThis.fetch = wrappedFetch; WebSocket.prototype.emit = wrappedEmit;
+ const timer = setInterval(() => { void (async () => { const r = load(); if (!r || closed) return; let releaseRequest: { nonce?: string; attemptId?: string; executionId?: string; release?: boolean } | undefined; try { releaseRequest = read(input.temporaryRoot, "release.json"); } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; } if (releaseRequest) { if (Object.keys(releaseRequest).sort().join() !== ["nonce", "attemptId", "executionId", "release"].sort().join() || releaseRequest.nonce !== r.nonce || releaseRequest.attemptId !== r.attemptId || releaseRequest.executionId !== r.executionId || releaseRequest.release !== true) throw new Error("Wrong email duplicate release"); await release(); } else if (Date.now() >= r.expiresAt) await release(true); })().catch(() => { void release(true); }); }, 200); timer.unref();
+ await atomic(input.temporaryRoot, "installed.json", { executionId: input.executionIds[0], attemptId: input.attemptId, pid: process.pid }); return { close: release };
+}
+export async function armOwnedEmailReceiptDuplicate(root: string, request: EmailDuplicateRequest) { if (!emailDuplicateRequestMatches(request, request)) throw new Error("Cannot arm an unsafe email receipt duplicate"); await atomic(root, "request.json", request); }
+export function readOwnedEmailDuplicate<T>(root: string, name: "receipt.json" | "installed.json" | "closed.json") { return read<T>(root, name); }
+export async function releaseOwnedEmailReceiptDuplicate(root: string, request: EmailDuplicateRequest) { await atomic(root, "release.json", { executionId: request.executionId, attemptId: request.attemptId, nonce: request.nonce, release: true }); const deadline = Date.now() + 10000; while (Date.now() < deadline) { try { const receipt = read<{ executionId: string; attemptId: string; nonce: string; released: boolean; expired: boolean }>(root, "closed.json"); if (receipt.executionId !== request.executionId || receipt.attemptId !== request.attemptId || receipt.nonce !== request.nonce || !receipt.released || receipt.expired) throw new Error("Exact email duplicate teardown failed"); return; } catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; } await new Promise(resolve => setTimeout(resolve, 100)); } throw new Error("Email duplicate teardown was not confirmed"); }

tests/runner-e2e/chat-catalog-domains/runtime-email-outbound.test.ts

@@ -0,0 +1,8 @@
+import { expect, it } from "vitest";
+import { gradeOutboundEmail, truthfulEmailOutcome } from "./runtime-email-outbound.js";
+function valid() { return { companyId: "company", leadId: "ceo", scenarioId: "chat-channels.outbound-email", origin: "http://127.0.0.1:42123", reply: "", agents: [], tasks: [], activity: [], links: [], humanQueries: [], runs: [1,2,3,4,5].map(id => ({ id: String(id), agentId: "ceo", status: "succeeded" })), state: { actualHumanDraftBound: true, actualDraftTransmitted: true, exactAuthorizedRecipient: true, feedbackRequested: true, normalTaskBound: true, actualNativeSendBound: true, nativeLedgerCausal: true, independentProviderCount: 1, originalProviderReceiptMatched: true, authenticDuplicateBound: true, duplicateReleased: true, finalIndependentProviderCount: 1, oneSavedPublication: true, actualThreadLinked: true, factualReplyBound: true, factualReply: "The provider accepted the original launch email; its conversation and message receipt are saved with the task. Recipient delivery remains unconfirmed.", actualOutcome: "sent" } }; }
+it("accepts actual authorized draft, delegated native send, authentic duplicate and one physical provider message", () => expect(gradeOutboundEmail(valid()).passed).toBe(true));
+it("accepts substantive natural outcome paraphrases without requiring a fixed URL or wording", () => { const o = valid(); o.state.factualReply = "Your launch draft was sent for review and the original provider message is recorded. I cannot confirm whether it was delivered."; expect(gradeOutboundEmail(o).passed).toBe(true); });
+it("rejects inaccessible/wrong draft, wrong recipient and an acknowledgement without task-native send", () => { for (const field of ["actualDraftTransmitted", "exactAuthorizedRecipient", "actualNativeSendBound", "nativeLedgerCausal"]) { const o = valid(); (o.state as Record<string, unknown>)[field] = false; expect(gradeOutboundEmail(o).passed).toBe(false); } });
+it("rejects forged/unobserved duplicate proof or an extra physical provider message", () => { const o = valid(); o.state.authenticDuplicateBound = false; expect(gradeOutboundEmail(o).passed).toBe(false); const p = valid(); p.state.finalIndependentProviderCount = 2; expect(gradeOutboundEmail(p).passed).toBe(false); });
+it("distinguishes actual recipient delivery from accepted send and requires the correct factual reply phase", () => { expect(truthfulEmailOutcome("It was sent and was delivered.", "sent")).toBe(false); expect(truthfulEmailOutcome("It was sent and was delivered.", "delivered")).toBe(true); const o = valid(); o.state.factualReplyBound = false; expect(gradeOutboundEmail(o).passed).toBe(false); });

tests/runner-e2e/chat-catalog-domains/runtime-email-outbound.ts

@@ -0,0 +1,75 @@
+import { createHash, randomUUID } from "node:crypto";
+import { expect } from "@playwright/test";
+import type { EmailPublicationSummary, EmailThreadSummary } from "../../../packages/shared/src/types/email.js";
+import type { ChatCatalogContext } from "../chat-catalog-contract.js";
+import { sendChatMessage } from "../chat-flow.js";
+import { pollUntil } from "../api.js";
+import { collectRunEvents } from "../run-observations.js";
+import { checks, type Entity } from "./internal-support.js";
+import { openStandingChat, waitForReply } from "./identity-support.js";
+import { runtimeDefinition, runtimeObservation, observeRuntime } from "./runtime-support.js";
+import { externalRequirements } from "./external-fixtures.js";
+import { actualOriginalEmailSend, assertsEmailDelivery, prepareOwnedEmailConnection, inspectOwnedEmailPublication, cleanupOwnedEmailConnection } from "./runtime-email.js";
+import { armOwnedEmailReceiptDuplicate, readOwnedEmailDuplicate, releaseOwnedEmailReceiptDuplicate, type EmailDuplicateRequest, type EmailDuplicateReceipt } from "./runtime-email-duplicate.js";
+const hash = (v: string) => createHash("sha256").update(v).digest("hex"), normalized = (v: string) => v.replace(/\s+/g, " ").trim();
+const duplicateRequests = new WeakMap<ChatCatalogContext, EmailDuplicateRequest>();
+export function truthfulEmailOutcome(reply: string, outcome: string) { return /sent|accepted|provider.*(?:receipt|message)|queued.*(?:accepted|sent)/i.test(reply) && (outcome === "delivered" || !assertsEmailDelivery(reply)) && !/\b(?:did not|never|not) (?:actually )?send\b/i.test(reply); }
+export function gradeOutboundEmail(value: unknown) {
+ const c = checks(), o = runtimeObservation(value); c.add("actual-authorized-task-bound-email", !!o, "A real authorized draft, normal task execution and independently accepted provider message are required."); if (!o) return c.result(); const s = o.state;
+ c.add("real-human-draft-and-exact-recipient", s.actualHumanDraftBound === true && s.actualDraftTransmitted === true && s.exactAuthorizedRecipient === true && s.feedbackRequested === true, "Actual task-bound bytes contain the authorized human launch draft and request feedback from exactly the designated synthetic recipient.");
+ c.add("ordinary-chat-delegation-and-native-send", s.normalTaskBound === true && s.actualNativeSendBound === true && s.nativeLedgerCausal === true && o.runs.length === 5 && o.runs.every(row => row.agentId === o.leadId && row.status === "succeeded"), "Canonical Agent Chat delegates one real same-agent normal task; its own admitted native execution alone performs the email send, with separately attributed completion and factual reply.");
+ c.add("one-real-provider-message-despite-authentic-network-duplicate", s.independentProviderCount === 1 && s.originalProviderReceiptMatched === true && s.authenticDuplicateBound === true && s.duplicateReleased === true && s.finalIndependentProviderCount === 1 && s.oneSavedPublication === true, "One actual official sent receipt is replayed byte-for-byte once at the selected network boundary; actual provider and persisted publication checks still prove one email.");
+ c.add("linked-task-thread-and-truthful-reported-outcome", s.actualThreadLinked === true && s.factualReplyBound === true && truthfulEmailOutcome(String(s.factualReply), String(s.actualOutcome)), "The saved returned conversation remains linked to the delegated task; a later run-attributed chat answer reports real acceptance and distinguishes unproven delivery.");
+ return c.result();
+}
+const requirement = externalRequirements["chat-channels.outbound-email"]!;
+export const runtimeOutboundEmailCases = [runtimeDefinition("chat-channels.outbound-email", async ctx => {
+ try {
+  const root = process.env.PAPERCLIP_RUNNER_E2E_TEMP_ROOT, attemptId = process.env.PAPERCLIP_E2E_ACCOUNTING_ATTEMPT_ID; if (!root || !attemptId) throw new Error("Exact owned receipt duplicate server hook is required before provider setup");
+  const installed = readOwnedEmailDuplicate<{ executionId: string; attemptId: string; pid: number }>(root, "installed.json"); if (installed.executionId !== ctx.input.execution.id || installed.attemptId !== attemptId || !Number.isSafeInteger(installed.pid)) throw new Error("Selected outbound receipt duplicate hook does not bind this exact attempt");
+  const source = await openStandingChat(ctx), connection = await prepareOwnedEmailConnection(ctx), nonce = randomUUID(), marker = `Launch draft ${nonce}`, draft = `${marker}\nAcorn helps independent teams launch accessible product pages. The launch page introduces the product, explains the lightweight workflow, and invites readers to join the early-access list. The launch date is November 12 and the public contact is launch@example.test.`;
+  await ctx.input.page.goto(`/${ctx.input.fixtures.company.issuePrefix}/issues/${source.id}`, { waitUntil: "domcontentloaded" });
+  for (const scope of [{ scopeType: "company", scopeId: source.companyId }, { scopeType: "agent", scopeId: ctx.input.fixtures.agent.id }]) await ctx.input.api.post(`/api/companies/${source.companyId}/budgets/policies`, { ...scope, amount: 500, hardStopEnabled: true, isActive: true, unpricedUsagePolicy: "allow" });
+  const setupQuery = `This is my authorized launch draft for my next request. Retain the complete draft in this conversation, including its reference. Do not email it yet.\n\n${draft}`;
+  await sendChatMessage(ctx.input.page, setupQuery); await waitForReply(ctx);
+  const comments = await ctx.input.api.get<Entity[]>(`/api/issues/${source.id}/comments?order=asc`), human = comments.filter(row => !row.authorAgentId), runs = await ctx.allRuns();
+  const actualHumanDraftBound = human.length === 1 && human[0]?.body === setupQuery && human[0]?.authorUserId === connection.personId && runs.length === 1 && runs[0]?.status === "succeeded" && runs[0]?.contextSnapshot?.issueId === source.id;
+  await ctx.input.api.post(`/api/companies/${source.companyId}/tools/policies`, { name: `Exact self-addressed launch review ${nonce}`, policyType: "allow", priority: 10, selectors: { connectionId: connection.endpoint.connectionId, toolNames: ["email.send"] }, conditions: { arguments: { fieldEquals: { to: [connection.endpoint.address] } } }, config: {} });
+  const request: EmailDuplicateRequest = { executionId: ctx.input.execution.id, attemptId, inboxId: connection.endpoint.address!, marker, nonce, expiresAt: Date.now() + 480000 }; await armOwnedEmailReceiptDuplicate(root, request); duplicateRequests.set(ctx, request);
+  return { ...connection, sourceId: source.id, draft, marker, setupQuery, setupRunId: runs[0]?.id, actualHumanDraftBound, emailRecipient: { id: connection.endpoint.id, companyId: source.companyId, connectionId: connection.endpoint.connectionId, email: connection.endpoint.address }, startedAt: new Date().toISOString() };
+ } catch (error) { await cleanupOwnedEmailConnection(ctx); throw error; }
+}, gradeOutboundEmail, {
+ authentication: "human-session", setupHumanMessageCount: 1, expectedHumanMessageCount: 3, expectedRunCount: 5, turnTimeoutMs: 120000, attemptTimeoutMs: 1020000,
+ externalInput: { provider: requirement.provider, fields: requirement.fields }, queryBindings: [{ catalogToken: "reviewer@example.test", stateKey: "emailRecipient", resourceKind: "agentmail-recipient", valueField: "email" }],
+ sourceFiles: ["chat-catalog-domains/runtime-email-outbound.ts", "chat-catalog-domains/runtime-email.ts", "chat-catalog-domains/runtime-email-private.ts", "chat-catalog-domains/runtime-email-private-worker.ts", "chat-catalog-domains/runtime-email-duplicate.ts", "chat-catalog-domains/runtime-email-fault.ts", "server-entry.ts", "chat-query-bindings.ts", "../../server/src/services/email-channels.ts", "../../server/src/services/agentmail-api.ts", "../../server/src/services/connectors/agentmail.ts", "chat-scenarios/chat-channels.yaml"],
+ async act(ctx, s) {
+  const api = ctx.input.api, companyId = ctx.input.fixtures.company.id, sourceId = String(s.sourceId), endpointId = (s.endpoint as { id: string }).id;
+  const sent = await pollUntil({ label: "Actual delegated task sends the authorized launch draft and returns ordinary completion", deadlineAt: Date.now() + 360000, load: async () => {
+   await ctx.refreshIssue(); const runs = await ctx.allRuns(), tasks = await api.get<Entity[]>(`/api/companies/${companyId}/issues`), work = tasks.find(row => row.parentId === sourceId && row.assigneeAgentId === ctx.input.fixtures.agent.id), emails = work ? tasks.filter(row => row.parentId === work.id) : [], threads = await Promise.all(emails.map(row => api.get<EmailThreadSummary | null>(`/api/companies/${companyId}/email/tasks/${row.id}`))); return { runs, tasks, work, thread: threads.filter((row): row is EmailThreadSummary => !!row)[0] };
+  }, accept: v => ctx.issue()?.conversationState === "waiting" && v.runs.length === 4 && v.runs.every(row => row.status === "succeeded") && v.work?.status === "done" && v.thread?.publications.length === 1 && ["sent", "delivered"].includes(v.thread.publications[0]!.outcome), reject: v => v.runs.length > 4 ? "Email delegation exceeded its declared causal run bound" : undefined });
+  const thread = sent.thread!, publication = await api.get<EmailPublicationSummary>(`/api/companies/${companyId}/email/deliveries/${thread.publications[0]!.id}`), work = sent.work!, run = sent.runs.find(row => row.contextSnapshot?.issueId === work.id), request = publication.request!;
+  s.emailTaskId = thread.issueId; s.workTaskId = work.id; s.publicationId = publication.id; s.actualOutcome = publication.outcome;
+  s.actualDraftTransmitted = normalized(request.text).includes(normalized(String(s.draft))); s.feedbackRequested = /feedback|review|thoughts|comments|suggestions|what (?:do you|you) think/i.test(request.text);
+  s.exactAuthorizedRecipient = JSON.stringify(request.to) === JSON.stringify([(s.emailRecipient as { email: string }).email]) && !request.cc?.length && !request.bcc?.length;
+  s.normalTaskBound = work.companyId === companyId && (work as Entity & { workMode?: string }).workMode === "standard" && work.parentId === sourceId && work.assigneeAgentId === ctx.input.fixtures.agent.id && !!run && run.agentId === work.assigneeAgentId && run.runtimeMode === "native";
+  const events = run ? await collectRunEvents<{ seq?: number; eventType?: string; payload?: unknown }>((afterSeq, limit) => api.get(`/api/heartbeat-runs/${run.id}/events?afterSeq=${afterSeq}&limit=${limit}`)) : []; s.actualNativeSendBound = actualOriginalEmailSend(events, { companyId, taskId: work.id, endpointId, publicationId: publication.id });
+  const direct = await inspectOwnedEmailPublication(ctx, publication, String(s.startedAt)); s.independentProviderCount = direct.matchedSent?.length; s.originalProviderReceiptMatched = direct.originalMessageMatched;
+  const duplicateRequest = duplicateRequests.get(ctx)!, root = process.env.PAPERCLIP_RUNNER_E2E_TEMP_ROOT!;
+  const duplicate = await pollUntil({ label: "Actual authentic sent-receipt network duplicate observed", deadlineAt: Date.now() + 30000, load: async () => { try { return readOwnedEmailDuplicate<EmailDuplicateReceipt>(root, "receipt.json"); } catch (error) { if ((error as NodeJS.ErrnoException).code === "ENOENT") return undefined; throw error; } }, accept: row => !!row });
+  s.authenticDuplicateBound = duplicate?.executionId === duplicateRequest.executionId && duplicate.attemptId === duplicateRequest.attemptId && duplicate.nonce === duplicateRequest.nonce && duplicate.publicationId === publication.id && duplicate.providerMessageId === publication.providerMessageId && /^[a-f0-9]{64}$/.test(duplicate.authenticPayloadSha256) && duplicate.injectedDuplicateCount === 1 && !duplicate.expired;
+  await ctx.input.evidence("outbound-authentic-network-duplicate.json", { ...duplicate, injection: "one identical actual official message.sent frame replayed once; no fabricated provider response" });
+  await releaseOwnedEmailReceiptDuplicate(root, duplicateRequest); s.duplicateReleased = true; duplicateRequests.delete(ctx);
+  const before = sent.runs.map(row => row.id), followup = "Please inspect the saved email conversation and its actual provider receipt, then report the real outcome and what is known about recipient delivery. Do not send another email.";
+  await sendChatMessage(ctx.input.page, followup);
+  const final = await pollUntil({ label: "Actual launch email outcome reported in Agent Chat", deadlineAt: Date.now() + 180000, load: async () => { await ctx.refreshIssue(); return ctx.allRuns(); }, accept: rows => ctx.issue()?.conversationState === "waiting" && rows.length === 5 && rows.every(row => row.status === "succeeded") });
+  const factual = final.find(row => !before.includes(row.id))!, comments = await api.get<Entity[]>(`/api/issues/${sourceId}/comments?order=asc`), human = comments.filter(row => !row.authorAgentId), canonical = sent.runs.find(row => Array.isArray(row.contextSnapshot?.wakeCommentIds) && row.contextSnapshot.wakeCommentIds.includes(human[1]?.id)), completion = sent.runs.filter(row => Array.isArray(row.contextSnapshot?.chatCompletionUpdates) && row.contextSnapshot.chatCompletionUpdates.some((update: Entity) => update.id === work.id && update.status === "done"));
+  s.factualReplyBound = human.length === 3 && human[2]?.body === followup && factual.contextSnapshot?.issueId === sourceId && Array.isArray(factual.contextSnapshot.wakeCommentIds) && factual.contextSnapshot.wakeCommentIds.length === 1 && factual.contextSnapshot.wakeCommentIds[0] === human[2]?.id;
+  s.factualReply = comments.filter(row => row.createdByRunId === factual.id && row.authorAgentId === ctx.input.fixtures.agent.id).map(row => row.body).join("\n");
+  s.nativeLedgerCausal = !!canonical && !!run && completion.length === 1 && new Set([s.setupRunId, canonical.id, run.id, completion[0]!.id, factual.id]).size === 5 && final.every(row => [s.setupRunId, canonical.id, run.id, completion[0]!.id, factual.id].includes(row.id));
+  const saved = await api.get<EmailThreadSummary>(`/api/companies/${companyId}/email/tasks/${thread.issueId}`), emailTask = await api.get<Entity>(`/api/issues/${thread.issueId}`); const ordinary = (await api.get<Array<Entity & { workMode?: string }>>(`/api/companies/${companyId}/issues`)).filter(row => row.workMode !== "chat"); s.oneSavedPublication = saved.publications.length === 1 && saved.publications[0]?.id === publication.id && ordinary.length === 2 && ordinary.every(row => [work.id, thread.issueId].includes(row.id)); s.actualThreadLinked = emailTask.companyId === companyId && emailTask.parentId === work.id && saved.conversationId === publication.conversationId && saved.endpoint.id === endpointId;
+  const last = await inspectOwnedEmailPublication(ctx, publication, String(s.startedAt)); s.finalIndependentProviderCount = last.matchedSent?.length;
+  (s.providerProbes as unknown[]).push(...[direct, last].map(row => ({ phase: "actual-outbound-provider-receipt", directProviderProbeCount: row.directProviderProbeCount, status: "verified", usage: "unavailable", cost: "unavailable" })));
+ },
+ async verifyResult(ctx, s) { const source = ctx.input.page.url(); await ctx.input.page.goto(`/${ctx.input.fixtures.company.issuePrefix}/issues/${s.emailTaskId}`, { waitUntil: "domcontentloaded" }); const task = await ctx.input.api.get<Entity>(`/api/issues/${s.emailTaskId}`); await expect(ctx.input.page.getByTestId("issue-detail-header").getByRole("heading", { name: task.title!, exact: true })).toBeVisible(); await ctx.input.capture("actual-outbound-email-thread", "The task-bound launch email and actual provider result", "outbound-email-thread.png"); await ctx.input.page.goto(source, { waitUntil: "domcontentloaded" }); },
+ async cleanup(ctx) { try { const r = duplicateRequests.get(ctx); if (r) await releaseOwnedEmailReceiptDuplicate(process.env.PAPERCLIP_RUNNER_E2E_TEMP_ROOT!, r); } finally { duplicateRequests.delete(ctx); await cleanupOwnedEmailConnection(ctx); } },
+})];

tests/runner-e2e/server-entry.ts

@@ -5,9 +5,11 @@ import { PaperclipRunnerToolAuthority } from "../../server/src/services/native-r
 import { canonicalDocumentIssueId, contextCommentGateSelected, holdCommittedDocumentResponse, isArmed } from "./context-comment-gate.js";
 import { holdInteractionResponse } from "./interaction-response-gate.js";
 import { installOwnedEmailResponseLoss } from "./chat-catalog-domains/runtime-email-fault.js";
+import { installOwnedEmailReceiptDuplicate } from "./chat-catalog-domains/runtime-email-duplicate.js";
 
 const ids: string[] = JSON.parse(process.env.PAPERCLIP_RUNNER_E2E_EXECUTION_IDS ?? "[]");
 await installOwnedEmailResponseLoss({ executionIds: ids, attemptId: process.env.PAPERCLIP_E2E_ACCOUNTING_ATTEMPT_ID ?? "", temporaryRoot: process.env.PAPERCLIP_RUNNER_E2E_TEMP_ROOT ?? "" });
+await installOwnedEmailReceiptDuplicate({ executionIds: ids, attemptId: process.env.PAPERCLIP_E2E_ACCOUNTING_ATTEMPT_ID ?? "", temporaryRoot: process.env.PAPERCLIP_RUNNER_E2E_TEMP_ROOT ?? "" });
 const contextCommentGate = contextCommentGateSelected(ids);
 const completionBusyGate = ids.some(id => id.startsWith("completion-updates.") && id.endsWith(".handoff-completion-busy"));
 if (ids.some((id) => id.endsWith(".accept-while-running"))) {