elie222 · wflanagan · Sep 3, 2026
diff --git a/.env.example b/.env.example
@@ -89,6 +89,7 @@ TELEGRAM_WEBHOOK_SECRET_TOKEN=
 # Bot events: app_mention, message.im, message.channels, message.groups, message.mpim.
 SLACK_APP_TOKEN=
 SLACK_BOT_TOKEN=
+SLACK_WORKSPACE_ID=
 SLACK_RAKAZO_BOT_ID=
 # Optional lower-cost model for ambient Slack judgments. Configure both or neither.
 TEAM_CHAT_JUDGE_PROVIDER=

diff --git a/apps/api/src/team-chat-bridge.test.ts b/apps/api/src/team-chat-bridge.test.ts
@@ -103,9 +103,9 @@ describe("team chat bridge", () => {
           records.push(record);
           return record;
         }),
-        findMany: vi.fn(async ({ where }: { where: { status: string } }) =>
+        findMany: vi.fn(async ({ where }: { where: { status: string | { in: string[] } } }) =>
           records
-            .filter((record) => record.status === where.status)
+            .filter((record) => matchesStatus(record.status, where.status))
             .map((record) => ({
               ...record,
               externalConversation: conversation,
@@ -120,7 +120,29 @@ describe("team chat bridge", () => {
             return record;
           },
         ),
-        updateMany: vi.fn(async () => ({ count: 0 })),
+        updateMany: vi.fn(
+          async ({
+            where,
+            data,
+          }: {
+            where: { id?: string | { in?: string[] }; status?: string | { in: string[] } };
+            data: Record<string, unknown>;
+          }) => {
+            const ids =
+              typeof where.id === "string" ? [where.id] : (where.id as { in?: string[] })?.in;
+            let count = 0;
+            for (const record of records) {
+              if (
+                (!ids || ids.includes(String(record.id))) &&
+                matchesStatus(record.status, where.status)
+              ) {
+                Object.assign(record, data);
+                count += 1;
+              }
+            }
+            return { count };
+          },
+        ),
       },
       run: { findMany: vi.fn(async () => []) },
       message: {
@@ -198,6 +220,7 @@ describe("team chat bridge", () => {
         conversationId: "C-1",
         replyThreadId: "100.1",
         content: "The launch plan is ready.",
+        idempotencyKey: "external-message:external-1",
       },
     ]);
     expect(records[0]).toMatchObject({
@@ -355,6 +378,7 @@ describe("team chat bridge", () => {
         conversationId: "C-1",
         replyThreadId: "100.1",
         content: "Research found that the launch should move.",
+        idempotencyKey: "team-chat-run:run-arthur-result",
       },
     ]);
     expect(prisma.run.updateMany).toHaveBeenCalledWith({
@@ -478,6 +502,118 @@ describe("team chat bridge", () => {
     expect(enqueue).toHaveBeenCalledTimes(1);
   });
 
+  it("does not discard an elected ambient trigger before promoting it", async () => {
+    const records: Array<Record<string, unknown>> = [];
+    const sendUserMessage = vi.fn(async (input: { createRun?: boolean }) =>
+      input.createRun === false
+        ? { messageId: "message-visible", taskId: null, runId: null }
+        : { messageId: "message-prompt", taskId: "task-ambient", runId: "run-ambient" },
+    );
+    const judge = {
+      decide: vi.fn(async () => ({
+        act: true,
+        reason: "A committed launch date changed.",
+        askedByEventId: "Ev-ambient",
+      })),
+    };
+    const conversation = {
+      id: "conversation-ambient",
+      provider: "slack",
+      workspaceId: "T-1",
+      externalKey: "channel:C-1",
+      conversationId: "C-1",
+      displayName: "launch",
+      spaceId: "space-1",
+      botId: "bot-1",
+      userId: "owner-1",
+      thread: { id: "thread-ambient" },
+    };
+    const prisma = ambientPrisma(records, conversation, true) as unknown as PrismaClient;
+    const bridge = new TeamChatBridge({
+      prisma,
+      events: { sendUserMessage },
+      jobs: { enqueue: vi.fn(async () => undefined) },
+      provider: new FakeTeamChatProvider(),
+      judge,
+      botId: "bot-1",
+      ambientDebounceMs: 0,
+      reconcileIntervalMs: 60_000,
+    });
+
+    await bridge.start();
+    await bridge.receive(ambientMessage());
+    await bridge.stop();
+
+    expect(prisma.externalMessage.updateMany).toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: { id: "external-1", status: "observed" },
+        data: expect.objectContaining({ status: "received" }),
+      }),
+    );
+    expect(prisma.externalMessage.updateMany).not.toHaveBeenCalledWith(
+      expect.objectContaining({
+        where: { id: { in: ["external-1"] }, status: "observed" },
+        data: expect.objectContaining({ status: "ignored" }),
+      }),
+    );
+  });
+
+  it("does not resend a completed run when delivery was already reserved", async () => {
+    const provider = new FakeTeamChatProvider();
+    const prisma = {
+      bot: {
+        findFirst: vi.fn(async () => ({
+          id: "bot-1",
+          spaceId: "space-1",
+          userId: "owner-1",
+          name: "Arthur",
+        })),
+      },
+      externalMessage: {
+        findMany: vi.fn(async ({ where }: { where: { status?: { in: string[] } | string } }) =>
+          where.status && matchesStatus("delivering", where.status)
+            ? [
+                {
+                  id: "external-1",
+                  status: "delivering",
+                  attempts: 0,
+                  kind: "mention",
+                  runId: "run-1",
+                  replyThreadId: "100.1",
+                  run: { id: "run-1", status: "completed" },
+                  externalConversation: {
+                    provider: "slack",
+                    botId: "bot-1",
+                    conversationId: "C-1",
+                  },
+                },
+              ]
+            : [],
+        ),
+        updateMany: vi.fn(async () => ({ count: 0 })),
+      },
+      run: { findMany: vi.fn(async () => []) },
+      message: {
+        findFirst: vi.fn(async () => ({
+          blocks: [{ kind: "text", text: "Already being sent." }],
+        })),
+      },
+    } as unknown as PrismaClient;
+    const bridge = new TeamChatBridge({
+      prisma,
+      events: { sendUserMessage: vi.fn() },
+      jobs: { enqueue: vi.fn(async () => undefined) },
+      provider,
+      botId: "bot-1",
+      reconcileIntervalMs: 60_000,
+    });
+
+    await bridge.start();
+    await bridge.stop();
+
+    expect(provider.sent).toEqual([]);
+  });
+
   it("uses room listening and guidance instead of Arthur's defaults", async () => {
     const records: Array<Record<string, unknown>> = [];
     const sendUserMessage = vi.fn(async (input: { createRun?: boolean }) =>
@@ -746,9 +882,9 @@ function ambientPrisma(
         records.push(record);
         return record;
       }),
-      findMany: vi.fn(async ({ where }: { where: { status: string } }) =>
+      findMany: vi.fn(async ({ where }: { where: { status: string | { in: string[] } } }) =>
         records
-          .filter((record) => record.status === where.status)
+          .filter((record) => matchesStatus(record.status, where.status))
           .map((record) => ({
             ...record,
             externalConversation: conversation,
@@ -764,15 +900,17 @@ function ambientPrisma(
           data: Record<string, unknown>;
         }) => {
           const ids = (where.id as { in?: string[] } | undefined)?.in;
+          let count = 0;
           for (const record of records) {
             if (
               (!ids || ids.includes(String(record.id))) &&
-              record.status === (where.status ?? record.status)
+              matchesStatus(record.status, where.status ?? String(record.status))
             ) {
               Object.assign(record, data);
+              count += 1;
             }
           }
-          return { count: records.length };
+          return { count };
         },
       ),
       update: vi.fn(
@@ -789,6 +927,12 @@ function ambientPrisma(
   };
 }
 
+function matchesStatus(value: unknown, status: string | { in: string[] } | undefined): boolean {
+  if (!status) return true;
+  const current = String(value);
+  return typeof status === "string" ? current === status : status.in.includes(current);
+}
+
 class FakeTeamChatProvider implements TeamChatProvider {
   readonly id = "slack";
   readonly sent: TeamChatSendRequest[] = [];

diff --git a/apps/api/src/team-chat-bridge.ts b/apps/api/src/team-chat-bridge.ts
@@ -11,6 +11,7 @@ const BATCH_SIZE = 20;
 const AMBIENT_BATCH_SIZE = 100;
 const AMBIENT_CONTEXT_MESSAGES = 20;
 const AMBIENT_CONTEXT_MESSAGE_CHARS = 2_000;
+const DELIVERY_RESERVATION_MS = 2 * 60_000;
 
 interface TeamChatBridgeDeps {
   prisma: PrismaClient;
@@ -269,6 +270,7 @@ export class TeamChatBridge {
     for (const message of received)
       await this.queue(message).catch((error) => this.retry(message, error));
 
+    await this.recoverStaleDeliveries(now);
     const running = await this.deps.prisma.externalMessage.findMany({
       where: {
         status: "running",
@@ -338,6 +340,7 @@ export class TeamChatBridge {
           conversationId: origin.externalConversation.conversationId,
           replyThreadId: origin.replyThreadId,
           content,
+          idempotencyKey: `team-chat-run:${run.id}`,
         });
       }
       await this.markTeamChatMirrored(run.id);
@@ -478,31 +481,45 @@ export class TeamChatBridge {
       const trigger =
         judgedMessages.find((message) => message.providerEventId === decision.askedByEventId) ??
         latest;
-      await this.markAmbientIgnored(evaluated, now);
-      if (!decision.act) continue;
-      await this.deps.prisma.externalMessage.update({
-        where: { id: trigger.id },
-        data: {
-          status: "received",
-          judgedAt: now,
-          engagementReason: decision.reason ?? null,
-          batchContext: teamChatAmbientPrompt({
-            provider: this.deps.provider.id,
-            channelId: latest.externalConversation.conversationId,
-            channelName: latest.externalConversation.displayName,
-            rules,
-            reason: decision.reason,
-            messages: judgedMessages.map((message) => ({
-              senderId: message.senderId,
-              senderName: message.senderName,
-              content: message.content,
-            })),
-          }),
-        },
+      if (!decision.act) {
+        await this.markAmbientIgnored(evaluated, now);
+        continue;
+      }
+      const promoted = await this.promoteAmbientTrigger(trigger.id, {
+        judgedAt: now,
+        engagementReason: decision.reason ?? null,
+        batchContext: teamChatAmbientPrompt({
+          provider: this.deps.provider.id,
+          channelId: latest.externalConversation.conversationId,
+          channelName: latest.externalConversation.displayName,
+          rules,
+          reason: decision.reason,
+          messages: judgedMessages.map((message) => ({
+            senderId: message.senderId,
+            senderName: message.senderName,
+            content: message.content,
+          })),
+        }),
       });
+      if (promoted)
+        await this.markAmbientIgnored(
+          evaluated.filter(({ id }) => id !== trigger.id),
+          now,
+        );
     }
   }
 
+  private async promoteAmbientTrigger(
+    id: string,
+    data: { judgedAt: Date; engagementReason: string | null; batchContext: string },
+  ): Promise<boolean> {
+    const result = await this.deps.prisma.externalMessage.updateMany({
+      where: { id, status: "observed" },
+      data: { status: "received", ...data },
+    });
+    return result.count === 1;
+  }
+
   private async markAmbientIgnored(messages: Array<{ id: string }>, judgedAt: Date): Promise<void> {
     if (messages.length === 0) return;
     await this.deps.prisma.externalMessage.updateMany({
@@ -610,10 +627,13 @@ export class TeamChatBridge {
       await this.markDelivered(message.id, "silent");
       return;
     }
+    const reserved = await this.reserveDelivery(message.id);
+    if (!reserved) return;
     const sent = await this.deps.provider.send({
       conversationId: message.externalConversation.conversationId,
       replyThreadId: message.replyThreadId,
       content,
+      idempotencyKey: `external-message:${message.id}`,
     });
     await this.markDelivered(message.id, sent.handle);
   }
@@ -628,14 +648,35 @@ export class TeamChatBridge {
       await this.markDelivered(message.id, "silent-failure");
       return;
     }
+    const reserved = await this.reserveDelivery(message.id);
+    if (!reserved) return;
     const sent = await this.deps.provider.send({
       conversationId: message.externalConversation.conversationId,
       replyThreadId: message.replyThreadId,
       content: `${this.target?.name ?? "The agent"} could not complete that request. Open Rakazo for details.`,
+      idempotencyKey: `external-message:${message.id}:failure`,
     });
     await this.markDelivered(message.id, sent.handle);
   }
 
+  private async reserveDelivery(id: string): Promise<boolean> {
+    const result = await this.deps.prisma.externalMessage.updateMany({
+      where: { id, status: "running" },
+      data: {
+        status: "delivering",
+        nextAttemptAt: new Date(Date.now() + DELIVERY_RESERVATION_MS),
+      },
+    });
+    return result.count === 1;
+  }
+
+  private async recoverStaleDeliveries(now: Date): Promise<void> {
+    await this.deps.prisma.externalMessage.updateMany({
+      where: { status: "delivering", nextAttemptAt: { lte: now } },
+      data: { status: "running", nextAttemptAt: null },
+    });
+  }
+
   private async markDelivered(id: string, handle: string): Promise<void> {
     await this.deps.prisma.externalMessage.update({
       where: { id },

diff --git a/packages/adapter-kit/src/types.ts b/packages/adapter-kit/src/types.ts
@@ -540,6 +540,8 @@ export interface TeamChatSendRequest {
   conversationId: string;
   replyThreadId: string | null;
   content: string;
+  /** Stable retry key the provider can use to dedupe externally accepted sends. */
+  idempotencyKey?: string;
 }
 
 export interface TeamChatSendResult {