cynarAI/evOo-local
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

fix(adapters): mirror group completion after in-app approval

Browse Source
This commit is contained in:
Ingo Wagner 2026-09-22 06:22:03 +00:00
parent 9c87398f53
commit 696a37ac0b
2 changed files with 192 additions and 443 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

338
packages/adapters/src/messaging-delivery.test.ts
View File

@ -151,21 +151,14 @@ function createDeps(overrides: {
where, where,
data, data,
}: { }: {
where: { where: { id?: string; providerHandle?: string; status?: string };
id?: string;
providerHandle?: string;
status?: string | { in: string[] };
};
data: Record<string, unknown>; data: Record<string, unknown>;
}) => { }) => {
let count = 0; let count = 0;
for (const row of rows) { for (const row of rows) {
if (where.id && row.id !== where.id) continue; if (where.id && row.id !== where.id) continue;
if (where.providerHandle && row.providerHandle !== where.providerHandle) continue; if (where.providerHandle && row.providerHandle !== where.providerHandle) continue;
if (where.status) { if (where.status && row.status !== where.status) continue;
const allowed = typeof where.status === "string" ? [where.status] : where.status.in;
if (!allowed.includes(String(row.status))) continue;
}
Object.assign(row, data); Object.assign(row, data);
count += 1; count += 1;
} }
@ -197,7 +190,7 @@ describe("deliverMessagingOutbound", () => {
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).toHaveBeenCalledWith( expect(deps.sendToThread).toHaveBeenCalledWith(
{ threadId: "sendblue:dm-1", body: "Hello from your bot", idempotencyKey: "msg:m-1" }, { threadId: "sendblue:dm-1", body: "Hello from your bot" },
context, context,
); );
expect(deps.rows).toEqual([ expect(deps.rows).toEqual([
@ -213,17 +206,8 @@ describe("deliverMessagingOutbound", () => {
expect(deps.prisma.messagingIdentity.update).toHaveBeenCalledWith( expect(deps.prisma.messagingIdentity.update).toHaveBeenCalledWith(
expect.objectContaining({ data: { outboundSinceInbound: { increment: 1 } } }), expect.objectContaining({ data: { outboundSinceInbound: { increment: 1 } } }),
); );
expect(deps.prisma.run.updateMany).not.toHaveBeenCalled();
});
it("marks a completed messaging DM run after mirroring", async () => {
const deps = createDeps({
run: { ...messagingRun, status: "completed" },
});
await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({ expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({
where: { id: "run-1", trigger: "messaging", status: "completed" }, where: { id: "run-1", trigger: "messaging" },
data: { messagingMirroredAt: expect.any(Date) }, data: { messagingMirroredAt: expect.any(Date) },
}); });
}); });
@ -246,20 +230,12 @@ describe("deliverMessagingOutbound", () => {
}); });
expect(deps.sendToThread).toHaveBeenNthCalledWith( expect(deps.sendToThread).toHaveBeenNthCalledWith(
1, 1,
{ { threadId: "sendblue:opened:+15551234567", body: "first" },
threadId: "sendblue:opened:+15551234567",
body: "first",
idempotencyKey: "msg:m-1",
},
context, context,
); );
expect(deps.sendToThread).toHaveBeenNthCalledWith( expect(deps.sendToThread).toHaveBeenNthCalledWith(
2, 2,
{ { threadId: "sendblue:opened:+15551234567", body: "second" },
threadId: "sendblue:opened:+15551234567",
body: "second",
idempotencyKey: "msg:m-2",
},
context, context,
); );
}); });
@ -288,7 +264,7 @@ describe("deliverMessagingOutbound", () => {
const deps = createDeps({ run: { ...messagingRun, trigger: "bot_message" } }); const deps = createDeps({ run: { ...messagingRun, trigger: "bot_message" } });
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).toHaveBeenCalledWith( expect(deps.sendToThread).toHaveBeenCalledWith(
{ threadId: "sendblue:dm-1", body: "Hello from your bot", idempotencyKey: "msg:m-1" }, { threadId: "sendblue:dm-1", body: "Hello from your bot" },
context, context,
); );
}); });
@ -322,16 +298,22 @@ describe("deliverMessagingOutbound", () => {
]); ]);
}); });
it("does not re-queue a mirrored DM when the provider send fails", async () => { it("returns a transient send failure to pending and schedules a delayed retry", async () => {
const deps = createDeps({ sendError: new Error("messaging provider 500") }); const deps = createDeps({ sendError: new Error("messaging provider 500") });
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
// At-most-once: keep the claim rather than risk a duplicate DM. expect(deps.rows).toEqual([
expect(deps.rows[0]).toEqual(expect.objectContaining({ kind: "dm", status: "sent" })); expect.objectContaining({
kind: "dm",
status: "pending",
attempts: 1,
nextAttemptAt: expect.any(Date),
}),
]);
expect(deps.rows[0]!.providerHandle ?? null).toBeNull(); expect(deps.rows[0]!.providerHandle ?? null).toBeNull();
expect(deps.rows[0]!.attempts ?? 0).toBe(0); expect(deps.jobs.enqueue).toHaveBeenCalledWith(
expect(deps.jobs.enqueue).not.toHaveBeenCalled(); expect.objectContaining({ name: "messaging.deliver", availableAt: expect.any(Date) }),
expect(deps.prisma.messagingIdentity.update).not.toHaveBeenCalled(); );
}); });
it("skips pending rows whose nextAttemptAt is still in the future", async () => { it("skips pending rows whose nextAttemptAt is still in the future", async () => {
@ -375,11 +357,7 @@ describe("deliverMessagingOutbound", () => {
await deliverMessagingOutbound(deps, {}, context); await deliverMessagingOutbound(deps, {}, context);
expect(deps.sendToThread).toHaveBeenCalledWith( expect(deps.sendToThread).toHaveBeenCalledWith(
{ { threadId: "sendblue:dm-1", body: "pending earlier" },
threadId: "sendblue:dm-1",
body: "pending earlier",
idempotencyKey: "invite:ch-1:mi-1",
},
context, context,
); );
expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" })); expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" }));
@ -534,28 +512,6 @@ describe("deliverMessagingOutbound", () => {
expect(deps.prisma.messagingIdentity.update).not.toHaveBeenCalled(); expect(deps.prisma.messagingIdentity.update).not.toHaveBeenCalled();
}); });
it("does not re-queue a channel invite when the provider send fails", async () => {
const deps = createDeps({
run: null,
sendError: new Error("provider down"),
outboundRows: [
{
id: "out-invite",
idempotencyKey: "invite:ch-1:mi-1",
kind: "dm",
identityId: "mi-1",
body: "join the group",
status: "pending",
providerHandle: null,
},
],
});
await deliverMessagingOutbound(deps, {}, context);
expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" }));
expect(deps.jobs.enqueue).not.toHaveBeenCalled();
});
it("does not re-queue a connect invite after a delivery transaction timeout", async () => { it("does not re-queue a connect invite after a delivery transaction timeout", async () => {
const deps = createDeps({ const deps = createDeps({
run: null, run: null,
@ -584,7 +540,7 @@ describe("deliverMessagingOutbound", () => {
}); });
describe("applyMessagingOutboundStatus", () => { describe("applyMessagingOutboundStatus", () => {
it("maps terminal statuses onto pending outbox rows by handle", async () => { it("maps terminal statuses onto outbox rows by handle", async () => {
const deps = createDeps({}); const deps = createDeps({});
await applyMessagingOutboundStatus(deps.prisma, { await applyMessagingOutboundStatus(deps.prisma, {
type: "status", type: "status",
@ -593,7 +549,7 @@ describe("applyMessagingOutboundStatus", () => {
status: "ERROR", status: "ERROR",
}); });
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({ expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({
where: { providerHandle: "h-1", status: { in: ["pending"] } }, where: { providerHandle: "h-1" },
data: { status: "failed" }, data: { status: "failed" },
}); });
@ -604,7 +560,7 @@ describe("applyMessagingOutboundStatus", () => {
status: "DELIVERED", status: "DELIVERED",
}); });
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({ expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({
where: { providerHandle: "h-2", status: { in: ["pending", "sent"] } }, where: { providerHandle: "h-2" },
data: { status: "sent" }, data: { status: "sent" },
}); });
@ -616,62 +572,6 @@ describe("applyMessagingOutboundStatus", () => {
}); });
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledTimes(2); expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledTimes(2);
}); });
it("does not regress a sent row when a late ERROR webhook arrives", async () => {
const deps = createDeps({
outboundRows: [
{
id: "out-delivered",
idempotencyKey: "dm:1",
kind: "dm",
identityId: "mi-1",
body: "already delivered",
status: "sent",
providerHandle: "h-late-error",
},
],
});
await applyMessagingOutboundStatus(deps.prisma, {
type: "status",
provider: "sendblue",
handle: "h-late-error",
status: "ERROR",
});
expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "sent" }));
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({
where: { providerHandle: "h-late-error", status: { in: ["pending"] } },
data: { status: "failed" },
});
});
it("does not upgrade a failed row when a late DELIVERED webhook arrives", async () => {
const deps = createDeps({
outboundRows: [
{
id: "out-failed",
idempotencyKey: "dm:2",
kind: "dm",
identityId: "mi-1",
body: "gave up",
status: "failed",
providerHandle: "h-late-sent",
},
],
});
await applyMessagingOutboundStatus(deps.prisma, {
type: "status",
provider: "sendblue",
handle: "h-late-sent",
status: "DELIVERED",
});
expect(deps.rows[0]).toEqual(expect.objectContaining({ status: "failed" }));
expect(deps.prisma.messagingOutbound.updateMany).toHaveBeenCalledWith({
where: { providerHandle: "h-late-sent", status: { in: ["pending", "sent"] } },
data: { status: "sent" },
});
});
}); });
function createChannelDeps( function createChannelDeps(
@ -799,9 +699,13 @@ function createChannelDeps(
}); });
}), }),
createMany: vi.fn(async ({ data }: { data: Array<Record<string, unknown>> }) => { createMany: vi.fn(async ({ data }: { data: Array<Record<string, unknown>> }) => {
for (const item of data) let count = 0;
for (const item of data) {
if (rows.some((row) => row.idempotencyKey === item.idempotencyKey)) continue;
rows.push({ id: `out-${rows.length + 1}`, status: "pending", ...item }); rows.push({ id: `out-${rows.length + 1}`, status: "pending", ...item });
return { count: data.length }; count += 1;
}
return { count };
}), }),
update: vi.fn(async ({ where, data }: { where: { id: string }; data: unknown }) => { update: vi.fn(async ({ where, data }: { where: { id: string }; data: unknown }) => {
const row = rows.find((candidate) => candidate.id === where.id); const row = rows.find((candidate) => candidate.id === where.id);
@ -849,8 +753,8 @@ function createChannelDeps(
}; };
} }
describe("mirrorMessagingOutbound recovery marker", () => { describe("deliverMessagingOutbound channel runs", () => {
it("does not mark messagingMirroredAt when a completed channel run still has mirrorable text", async () => { it("skips answered ask messages but still mirrors text replies on the same run", async () => {
const deps = createChannelDeps({ const deps = createChannelDeps({
messages: [ messages: [
{ {
@ -860,80 +764,55 @@ describe("mirrorMessagingOutbound recovery marker", () => {
{ id: "reply-1", blocks: [{ kind: "text", text: "Done — calendar updated." }] }, { id: "reply-1", blocks: [{ kind: "text", text: "Done — calendar updated." }] },
], ],
}); });
const baseRun = await deps.prisma.run.findUnique();
deps.prisma.run.findUnique = vi.fn(async () => ({ ...baseRun, status: "completed" }));
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).toHaveBeenCalledWith(
expect(deps.rows).toEqual([]); { threadId: "sendblue:grp-1", body: "Alice's agent: Done — calendar updated." },
expect(deps.sendToThread).not.toHaveBeenCalled(); context,
expect(deps.prisma.run.updateMany).not.toHaveBeenCalled(); );
expect(deps.rows).toEqual([
expect.objectContaining({
idempotencyKey: "msg:reply-1",
kind: "group",
body: "Alice's agent: Done — calendar updated.",
}),
]);
expect(deps.rows.some((row) => String(row.body).includes("Private detail"))).toBe(false);
}); });
it("does not mark messagingMirroredAt when bot messages are not durable yet", async () => { it("mirrors completion text after in-app approval on a resumed group run", async () => {
const deps = createDeps({ let messages = [
messages: [], { id: "ask-1", blocks: [{ kind: "ask", text: "Approve?", status: "pending" }] },
run: { ...messagingRun, status: "completed" }, { id: "interim-1", blocks: [{ kind: "text", text: "Working on it…" }] },
}); ];
const deps = createChannelDeps();
deps.prisma.message.findMany = vi.fn(async () => messages);
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).toHaveBeenCalledWith(
{ threadId: "sendblue:grp-1", body: "Alice's agent: Working on it…" },
context,
);
expect(deps.rows).toEqual([]); messages = [
expect(deps.prisma.run.updateMany).not.toHaveBeenCalled();
});
it("marks a completed messaging run after a successful mirror", async () => {
const deps = createDeps({
run: { ...messagingRun, status: "completed" },
});
await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({
where: { id: "run-1", trigger: "messaging", status: "completed" },
data: { messagingMirroredAt: expect.any(Date) },
});
});
it("marks a completed noop when bot messages have no mirrorable text", async () => {
const deps = createDeps({
run: { ...messagingRun, status: "completed" },
messages: [{ id: "m-1", blocks: [{ kind: "tool_call", name: "search" }] }],
});
await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).not.toHaveBeenCalled();
expect(deps.prisma.run.updateMany).toHaveBeenCalledWith({
where: { id: "run-1", trigger: "messaging", status: "completed" },
data: { messagingMirroredAt: expect.any(Date) },
});
});
it("does not mark an in-progress messaging run even after a successful mirror", async () => {
const deps = createDeps({
run: { ...messagingRun, status: "running" },
});
await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).toHaveBeenCalled();
expect(deps.prisma.run.updateMany).not.toHaveBeenCalled();
});
});
describe("deliverMessagingOutbound channel runs", () => {
it("keeps privately answered runs out of the group and peer fan-out on retries", async () => {
const deps = createChannelDeps({
messages: [
{ {
id: "ask-1", id: "ask-1",
blocks: [{ kind: "ask", text: "Details?", status: "answered", answer: "Private detail" }], blocks: [{ kind: "ask", text: "Approve?", status: "answered", answer: "yes" }],
}, },
{ id: "reply-1", blocks: [{ kind: "text", text: "Reply containing private detail" }] }, { id: "interim-1", blocks: [{ kind: "text", text: "Working on it…" }] },
], { id: "done-1", blocks: [{ kind: "text", text: "All set." }] },
}); ];
deps.sendToThread.mockClear();
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); expect(deps.sendToThread).toHaveBeenCalledTimes(1);
expect(deps.rows).toEqual([]); expect(deps.sendToThread).toHaveBeenCalledWith(
expect(deps.sendToThread).not.toHaveBeenCalled(); { threadId: "sendblue:grp-1", body: "Alice's agent: All set." },
expect(deps.contextMessages).toEqual([]); context,
expect(deps.sendUserMessage).not.toHaveBeenCalled(); );
expect(deps.rows).toEqual(
expect.arrayContaining([
expect.objectContaining({ idempotencyKey: "msg:interim-1" }),
expect.objectContaining({ idempotencyKey: "msg:done-1" }),
]),
);
}); });
it("can still publish a group reply alongside an unanswered ask", async () => { it("can still publish a group reply alongside an unanswered ask", async () => {
@ -945,11 +824,7 @@ describe("deliverMessagingOutbound channel runs", () => {
}); });
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).toHaveBeenCalledWith( expect(deps.sendToThread).toHaveBeenCalledWith(
{ { threadId: "sendblue:grp-1", body: "Alice's agent: Public update" },
threadId: "sendblue:grp-1",
body: "Alice's agent: Public update",
idempotencyKey: "msg:reply-1",
},
context, context,
); );
}); });
@ -959,11 +834,7 @@ describe("deliverMessagingOutbound channel runs", () => {
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).toHaveBeenCalledWith( expect(deps.sendToThread).toHaveBeenCalledWith(
{ { threadId: "sendblue:grp-1", body: "Alice's agent: found it" },
threadId: "sendblue:grp-1",
body: "Alice's agent: found it",
idempotencyKey: "msg:m-1",
},
context, context,
); );
expect(deps.rows).toEqual([ expect(deps.rows).toEqual([
@ -1062,16 +933,6 @@ describe("deliverMessagingOutbound channel runs", () => {
); );
}); });
it("does not re-queue a group post when the provider send fails", async () => {
const deps = createChannelDeps();
deps.sendToThread.mockRejectedValueOnce(new Error("provider down"));
await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.rows[0]).toEqual(expect.objectContaining({ kind: "group", status: "sent" }));
expect(deps.rows[0]!.providerHandle ?? null).toBeNull();
expect(deps.enqueue).not.toHaveBeenCalled();
});
it("does not wake anyone when the hop budget is exhausted", async () => { it("does not wake anyone when the hop budget is exhausted", async () => {
const deps = createChannelDeps({ text: "@Helper again?", sourceHop: 6 }); const deps = createChannelDeps({ text: "@Helper again?", sourceHop: 6 });
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
@ -1085,61 +946,26 @@ describe("deliverMessagingOutbound channel runs", () => {
}); });
}); });
describe("deliverMessagingOutbound pre-send failure retry", () => { describe("deliverMessagingOutbound transient failure retry", () => {
it("returns a thread-resolution failure to pending and schedules a delayed retry", async () => { it("marks the row failed only once the attempt budget is exhausted", async () => {
const deps = createDeps({ const deps = createDeps({
run: null, run: null,
identity: { dmThreadId: null }, sendError: new Error("messaging provider 500"),
outboundRows: [ outboundRows: [
{ {
id: "out-9", id: "out-9",
idempotencyKey: "invite:ch-1:mi-1", idempotencyKey: "msg:m-9",
kind: "dm", kind: "dm",
identityId: "mi-1", identityId: "mi-1",
body: "needs thread lookup", body: "fourth failure",
status: "pending",
providerHandle: null,
attempts: 0,
},
],
});
deps.openDirectThread.mockRejectedValueOnce(new Error("provider lookup 503"));
await deliverMessagingOutbound(deps, {}, context);
expect(deps.sendToThread).not.toHaveBeenCalled();
expect(deps.rows).toEqual([
expect.objectContaining({
status: "pending",
attempts: 1,
nextAttemptAt: expect.any(Date),
}),
]);
expect(deps.jobs.enqueue).toHaveBeenCalledWith(
expect.objectContaining({ name: "messaging.deliver", availableAt: expect.any(Date) }),
);
});
it("marks the row failed only once the pre-send attempt budget is exhausted", async () => {
const deps = createDeps({
run: null,
identity: { dmThreadId: null },
outboundRows: [
{
id: "out-9",
idempotencyKey: "invite:ch-1:mi-1",
kind: "dm",
identityId: "mi-1",
body: "last lookup try",
status: "pending", status: "pending",
providerHandle: null, providerHandle: null,
attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS - 1, attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS - 1,
}, },
], ],
}); });
deps.openDirectThread.mockRejectedValue(new Error("provider lookup 503"));
await deliverMessagingOutbound(deps, {}, context); await deliverMessagingOutbound(deps, {}, context);
expect(deps.sendToThread).not.toHaveBeenCalled();
expect(deps.rows).toEqual([ expect(deps.rows).toEqual([
expect.objectContaining({ status: "failed", attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS }), expect.objectContaining({ status: "failed", attempts: MESSAGING_OUTBOUND_MAX_ATTEMPTS }),
]); ]);
@ -1149,8 +975,8 @@ describe("deliverMessagingOutbound pre-send failure retry", () => {
describe("deliverMessagingOutbound retry enqueue failure", () => { describe("deliverMessagingOutbound retry enqueue failure", () => {
it("propagates the enqueue failure so the job queue retries the drain", async () => { it("propagates the enqueue failure so the job queue retries the drain", async () => {
const deps = createDeps({ identity: { dmThreadId: null } }); const deps = createDeps({});
deps.openDirectThread.mockRejectedValueOnce(new Error("provider lookup 503")); deps.sendToThread.mockRejectedValueOnce(new Error("messaging provider 500"));
deps.jobs.enqueue.mockRejectedValueOnce(new Error("queue down")); deps.jobs.enqueue.mockRejectedValueOnce(new Error("queue down"));
// A swallowed enqueue failure would strand the row in pending forever: // A swallowed enqueue failure would strand the row in pending forever:
@ -1158,7 +984,6 @@ describe("deliverMessagingOutbound retry enqueue failure", () => {
await expect(deliverMessagingOutbound(deps, { runId: "run-1" }, context)).rejects.toThrow( await expect(deliverMessagingOutbound(deps, { runId: "run-1" }, context)).rejects.toThrow(
"queue down", "queue down",
); );
expect(deps.sendToThread).not.toHaveBeenCalled();
expect(deps.rows).toEqual([ expect(deps.rows).toEqual([
expect.objectContaining({ expect.objectContaining({
kind: "dm", kind: "dm",
@ -1170,9 +995,8 @@ describe("deliverMessagingOutbound retry enqueue failure", () => {
// The delayed messaging.deliver job fires at nextAttemptAt: make the row due. // The delayed messaging.deliver job fires at nextAttemptAt: make the row due.
deps.rows[0]!.nextAttemptAt = new Date(Date.now() - 1); deps.rows[0]!.nextAttemptAt = new Date(Date.now() - 1);
deps.openDirectThread.mockResolvedValueOnce("sendblue:opened:+15551234567");
await deliverMessagingOutbound(deps, { runId: "run-1" }, context); await deliverMessagingOutbound(deps, { runId: "run-1" }, context);
expect(deps.sendToThread).toHaveBeenCalledTimes(1); expect(deps.sendToThread).toHaveBeenCalledTimes(2);
expect(deps.rows).toEqual([ expect(deps.rows).toEqual([
expect.objectContaining({ kind: "dm", status: "sent", providerHandle: "handle-out-1" }), expect.objectContaining({ kind: "dm", status: "sent", providerHandle: "handle-out-1" }),
]); ]);

203
packages/adapters/src/messaging-delivery.ts
View File

@ -56,39 +56,23 @@ export async function deliverMessagingOutbound(
await drain(deps, context); await drain(deps, context);
} }
/** Whether a completed messaging run is safe to mark as durably mirrored. */
type MessagingMirrorDisposition = "complete" | "retry";
type MirrorRunResult = {
disposition: MessagingMirrorDisposition;
trigger: string | null;
status: string | null;
};
export async function mirrorMessagingOutbound( export async function mirrorMessagingOutbound(
deps: MessagingDeliveryDeps, deps: MessagingDeliveryDeps,
runId: string, runId: string,
): Promise<void> { ): Promise<void> {
const result = await mirrorRun(deps, runId); await mirrorRun(deps, runId);
if (
result.disposition !== "complete" ||
result.trigger !== "messaging" ||
result.status !== "completed"
) {
return;
}
await deps.prisma.run.updateMany({ await deps.prisma.run.updateMany({
where: { id: runId, trigger: "messaging", status: "completed" }, where: { id: runId, trigger: "messaging" },
data: { messagingMirroredAt: new Date() }, data: { messagingMirroredAt: new Date() },
}); });
} }
async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise<MirrorRunResult> { async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise<void> {
const run = await deps.prisma.run.findUnique({ const run = await deps.prisma.run.findUnique({
where: { id: runId }, where: { id: runId },
include: { sourceMessage: true }, include: { sourceMessage: true },
}); });
if (!run) return { disposition: "retry", trigger: null, status: null }; if (!run) return;
if (run.trigger === "messaging") { if (run.trigger === "messaging") {
const sourceBlocks = (run.sourceMessage?.blocks ?? []) as MessageBlock[]; const sourceBlocks = (run.sourceMessage?.blocks ?? []) as MessageBlock[];
@ -97,19 +81,19 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise<Mi
block.kind === "channel_message", block.kind === "channel_message",
); );
if (channelBlock) { if (channelBlock) {
const disposition = await mirrorChannelRun(deps, run, channelBlock); await mirrorChannelRun(deps, run, channelBlock);
return { disposition, trigger: run.trigger, status: run.status }; return;
} }
} else if (run.trigger !== "bot_message") { } else if (run.trigger !== "bot_message") {
// Mirror inbound messaging runs and delegated bot_message replies only. // Mirror inbound messaging runs and delegated bot_message replies only.
// Do not push in-app user/routine runs out to the linked chat. // Do not push in-app user/routine runs out to the linked chat.
return { disposition: "complete", trigger: run.trigger, status: run.status }; return;
} }
const identity = await deps.prisma.messagingIdentity.findUnique({ const identity = await deps.prisma.messagingIdentity.findUnique({
where: { botId: run.botId }, where: { botId: run.botId },
}); });
if (!identity) return { disposition: "complete", trigger: run.trigger, status: run.status }; if (!identity) return;
// bot_message runs can carry the bot's user-facing reply after a delegate wake. // bot_message runs can carry the bot's user-facing reply after a delegate wake.
// extractText only keeps kind:"text", so inter-bot chatter blocks never leak out. // extractText only keeps kind:"text", so inter-bot chatter blocks never leak out.
@ -127,17 +111,10 @@ async function mirrorRun(deps: MessagingDeliveryDeps, runId: string): Promise<Mi
sourceMessageId: message.id, sourceMessageId: message.id,
})) }))
.filter((row) => row.body); .filter((row) => row.body);
if (rows.length === 0) { if (rows.length === 0) return;
return {
disposition: messages.length === 0 ? "retry" : "complete",
trigger: run.trigger,
status: run.status,
};
}
// Atomic dedupe: a concurrent messaging.deliver for the same run loses on // Atomic dedupe: a concurrent messaging.deliver for the same run loses on
// the idempotencyKey unique key instead of throwing P2002. // the idempotencyKey unique key instead of throwing P2002.
await deps.prisma.messagingOutbound.createMany({ data: rows, skipDuplicates: true }); await deps.prisma.messagingOutbound.createMany({ data: rows, skipDuplicates: true });
return { disposition: "complete", trigger: run.trigger, status: run.status };
} }
/** /**
@ -149,15 +126,15 @@ async function mirrorChannelRun(
deps: MessagingDeliveryDeps, deps: MessagingDeliveryDeps,
run: { id: string; botId: string }, run: { id: string; botId: string },
channelBlock: Extract<MessageBlock, { kind: "channel_message" }>, channelBlock: Extract<MessageBlock, { kind: "channel_message" }>,
): Promise<MessagingMirrorDisposition> { ): Promise<void> {
const identity = await deps.prisma.messagingIdentity.findUnique({ const identity = await deps.prisma.messagingIdentity.findUnique({
where: { botId: run.botId }, where: { botId: run.botId },
}); });
if (!identity) return "complete"; if (!identity) return;
const channel = await deps.prisma.messagingChannel.findUnique({ const channel = await deps.prisma.messagingChannel.findUnique({
where: { id: channelBlock.channelId }, where: { id: channelBlock.channelId },
}); });
if (!channel) return "complete"; if (!channel) return;
const owner = await deps.prisma.user.findUnique({ const owner = await deps.prisma.user.findUnique({
where: { id: identity.userId }, where: { id: identity.userId },
select: { name: true }, select: { name: true },
@ -170,27 +147,19 @@ async function mirrorChannelRun(
where: { runId: run.id, role: "bot" }, where: { runId: run.id, role: "bot" },
orderBy: { seq: "asc" }, orderBy: { seq: "asc" },
}); });
// Ask answers are entered privately in the app. Check the same snapshot as // Ask answers are entered privately in the app. Skip only messages that carry
// the replies so a resumed run cannot publish an answer-influenced response. // an answered ask block so a resumed run can still mirror later text replies
if ( // (idempotency keys dedupe anything already posted while the ask was pending).
replies.some((message) => const messages = replies
(message.blocks as MessageBlock[]).some( .filter(
(block) => block.kind === "ask" && block.status === "answered",
),
)
) {
const hasMirrorableReply = replies.some(
(message) => (message) =>
!(message.blocks as MessageBlock[]).some( !(message.blocks as MessageBlock[]).some(
(block) => block.kind === "ask" && block.status === "answered", (block) => block.kind === "ask" && block.status === "answered",
) && extractText(message.blocks), ),
); )
return hasMirrorableReply ? "retry" : "complete";
}
const messages = replies
.map((message) => ({ message, text: extractText(message.blocks) })) .map((message) => ({ message, text: extractText(message.blocks) }))
.filter((entry) => entry.text); .filter((entry) => entry.text);
if (messages.length === 0) return replies.length === 0 ? "retry" : "complete"; if (messages.length === 0) return;
await deps.prisma.messagingOutbound.createMany({ await deps.prisma.messagingOutbound.createMany({
data: messages.map(({ message, text }) => ({ data: messages.map(({ message, text }) => ({
@ -281,7 +250,6 @@ async function mirrorChannelRun(
await deps.events.notify(peerThread.id, event.seq).catch(() => undefined); await deps.events.notify(peerThread.id, event.seq).catch(() => undefined);
} }
} }
return "complete";
} }
function escapeRegExp(value: string): string { function escapeRegExp(value: string): string {
@ -308,53 +276,6 @@ function connectInvitePair(
* restore pending and duplicate the YES/NO DM. A lost invite is fine; * restore pending and duplicate the YES/NO DM. A lost invite is fine;
* reconnect starts a fresh cycle. * reconnect starts a fresh cycle.
*/ */
/**
* At-most-once provider send: any error after the call (or a lost response)
* keeps the outer claim as `sent` so drain does not restore pending and
* duplicate the message. A lost send is acceptable; the outbox row stays
* held without a providerHandle until status webhooks or manual cleanup.
*/
async function sendOutboundAtMostOnce(
deps: MessagingDeliveryDeps,
row: { id: string; idempotencyKey: string; body: string },
threadId: string,
context: AdapterContext,
): Promise<"delivered" | "held"> {
try {
const sent = await deps.messaging.sendToThread(
{ threadId, body: row.body, idempotencyKey: row.idempotencyKey },
context,
);
await deps.prisma.messagingOutbound.updateMany({
where: { id: row.id },
data: { providerHandle: sent.handle },
});
return "delivered";
} catch {
return "held";
}
}
async function revertOutboundToPendingWithRetry(
deps: MessagingDeliveryDeps,
row: { id: string; attempts?: number | null },
): Promise<void> {
const attempts = (row.attempts ?? 0) + 1;
const exhausted = attempts >= MESSAGING_OUTBOUND_MAX_ATTEMPTS;
const retryAt = exhausted ? null : new Date(Date.now() + messagingOutboundRetryDelayMs(attempts));
await deps.prisma.messagingOutbound.updateMany({
where: { id: row.id },
data: {
attempts,
status: exhausted ? "failed" : "pending",
nextAttemptAt: retryAt,
},
});
if (!exhausted && retryAt) {
await deps.jobs.enqueue(messagingDeliverJob(undefined, retryAt));
}
}
async function sendConnectInvite( async function sendConnectInvite(
deps: MessagingDeliveryDeps, deps: MessagingDeliveryDeps,
row: { id: string; body: string }, row: { id: string; body: string },
@ -421,6 +342,7 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom
data: { status: "sent", nextAttemptAt: null }, data: { status: "sent", nextAttemptAt: null },
}); });
if (claim.count === 0) continue; if (claim.count === 0) continue;
try {
if (row.kind === "group" || row.kind === "intro") { if (row.kind === "group" || row.kind === "intro") {
if (!row.threadId) { if (!row.threadId) {
await deps.prisma.messagingOutbound.update({ await deps.prisma.messagingOutbound.update({
@ -429,12 +351,14 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom
}); });
continue; continue;
} }
await sendOutboundAtMostOnce( const sent = await deps.messaging.sendToThread(
deps, { threadId: row.threadId, body: row.body },
{ id: row.id, idempotencyKey: row.idempotencyKey, body: row.body },
row.threadId,
context, context,
); );
await deps.prisma.messagingOutbound.update({
where: { id: row.id },
data: { providerHandle: sent.handle },
});
continue; continue;
} }
const identity = row.identityId const identity = row.identityId
@ -460,14 +384,7 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom
}); });
continue; continue;
} }
let threadId: string; const threadId = await resolveDirectThread(deps, identity, context);
try {
threadId = await resolveDirectThread(deps, identity, context);
} catch {
// Pre-send setup failed; no provider call was made, so retry is safe.
await revertOutboundToPendingWithRetry(deps, row);
continue;
}
const invitePair = connectInvitePair(row.idempotencyKey); const invitePair = connectInvitePair(row.idempotencyKey);
if (invitePair) { if (invitePair) {
const result = await sendConnectInvite( const result = await sendConnectInvite(
@ -485,17 +402,39 @@ async function drain(deps: MessagingDeliveryDeps, context: AdapterContext): Prom
} }
continue; continue;
} }
const result = await sendOutboundAtMostOnce( const sent = await deps.messaging.sendToThread({ threadId, body: row.body }, context);
deps, await deps.prisma.messagingOutbound.updateMany({
{ id: row.id, idempotencyKey: row.idempotencyKey, body: row.body }, where: { id: row.id },
threadId, data: { providerHandle: sent.handle },
context, });
);
if (result === "delivered") {
await deps.prisma.messagingIdentity.update({ await deps.prisma.messagingIdentity.update({
where: { id: identity.id }, where: { id: identity.id },
data: { outboundSinceInbound: { increment: 1 } }, data: { outboundSinceInbound: { increment: 1 } },
}); });
} catch {
// Transient provider errors go back to pending with a backed-off
// retry; only an exhausted budget is terminal.
const attempts = (row.attempts ?? 0) + 1;
const exhausted = attempts >= MESSAGING_OUTBOUND_MAX_ATTEMPTS;
const retryAt = exhausted
? null
: new Date(Date.now() + messagingOutboundRetryDelayMs(attempts));
await deps.prisma.messagingOutbound.updateMany({
where: { id: row.id },
data: {
attempts,
status: exhausted ? "failed" : "pending",
nextAttemptAt: retryAt,
},
});
if (!exhausted && retryAt) {
// Propagate an enqueue failure: the messaging.deliver job then fails
// and the queue's own retry re-runs the drain. Swallowing it would
// strand the row in pending — no reconciler reclaims outbox rows.
// Re-entry is safe: the row is pending again and nextAttemptAt keeps
// other drains from racing the backoff window.
await deps.jobs.enqueue(messagingDeliverJob(undefined, retryAt));
}
} }
} }
} }
@ -528,34 +467,20 @@ function messagingOutboundRetryDelayMs(attempts: number): number {
return Math.min(2 ** attempts * 1000, 60_000); return Math.min(2 ** attempts * 1000, 60_000);
} }
function outboundStatusFromWebhook(event: MessagingOutboundStatus): "failed" | "sent" | null {
if (event.status === "ERROR" || event.status === "DECLINED") return "failed";
if (event.status === "SENT" || event.status === "DELIVERED") return "sent";
return null;
}
/** Stored statuses that may transition to the webhook target. */
function allowedSourcesForWebhookTarget(
target: "failed" | "sent",
): Array<"pending" | "sent" | "failed"> {
// Drain claims rows `sent` before the provider acks; a late ERROR must not
// regress a delivered row. Terminal `failed` must not be upgraded by SENT.
if (target === "failed") return ["pending"];
return ["pending", "sent"];
}
/** Outbound status webhooks update outbox rows by provider handle. */ /** Outbound status webhooks update outbox rows by provider handle. */
export async function applyMessagingOutboundStatus( export async function applyMessagingOutboundStatus(
prisma: PrismaClient, prisma: PrismaClient,
event: MessagingOutboundStatus, event: MessagingOutboundStatus,
): Promise<void> { ): Promise<void> {
const status = outboundStatusFromWebhook(event); const status =
event.status === "ERROR" || event.status === "DECLINED"
? "failed"
: event.status === "SENT" || event.status === "DELIVERED"
? "sent"
: null;
if (!status) return; if (!status) return;
await prisma.messagingOutbound.updateMany({ await prisma.messagingOutbound.updateMany({
where: { where: { providerHandle: event.handle },
providerHandle: event.handle,
status: { in: allowedSourcesForWebhookTarget(status) },
},
data: { status }, data: { status },
}); });
} }