Slack delivery: fast group-DM addressing without directory clobbering, and no duplicate replies

src/api/app-messaging.ts

@@ -302,9 +302,9 @@ export function createMessagingMethods(
       return found;
     },
 
-    async upsertDirectory(members) {
+    async upsertDirectory(members, syncedAt) {
       const previous = await deps.directory.list();
-      await deps.directory.replace(members);
+      if (!(await deps.directory.replace(members, syncedAt))) return;
       const present = members.filter((m) => m.type === "internal").map((m) => m.principalId);
       const presentSet = new Set(present);
       const removed = previous.map((m) => m.principalId).filter((id) => !presentSet.has(id));
@@ -329,11 +329,11 @@ export function createMessagingMethods(
         });
       }
     },
-    async upsertChannels(channels, channelMembers) {
-      await deps.directory.replaceChannels(channels, channelMembers);
+    async upsertChannels(channels, channelMembers, syncedAt) {
+      await deps.directory.replaceChannels(channels, channelMembers, syncedAt);
     },
-    async upsertGroups(groupMembers) {
-      await deps.directory.replaceGroups(groupMembers);
+    async upsertGroups(groupMembers, syncedAt) {
+      await deps.directory.replaceGroups(groupMembers, syncedAt);
     },
     async setDirectoryWorkspaceUrl(url) {
       await deps.directory.setWorkspaceUrl(url);

src/api/app-types.ts

@@ -349,9 +349,9 @@ export interface App {
   ackDelivery(id: string, slackApiMs?: number): Promise<void>;
   ackDeliveryByKey(idempotencyKey: string): Promise<void>;
   setRunDeliveryState(runId: string, state: RunDeliveryState): Promise<boolean>;
-  upsertDirectory(members: DirectoryMember[]): Promise<void>;
-  upsertChannels(channels: DirectoryChannel[], channelMembers?: ChannelMembership[]): Promise<void>;
-  upsertGroups(groupMembers: GroupMembership[]): Promise<void>;
+  upsertDirectory(members: DirectoryMember[], syncedAt?: number): Promise<void>;
+  upsertChannels(channels: DirectoryChannel[], channelMembers?: ChannelMembership[], syncedAt?: number): Promise<void>;
+  upsertGroups(groupMembers: GroupMembership[], syncedAt?: number): Promise<void>;
   setDirectoryWorkspaceUrl(url: string): Promise<void>;
   directoryMeta(): Promise<DirectoryMeta>;
   resolveRecipient(query: string): Promise<RecipientResolution>;

src/api/routes/directory.ts

@@ -4,6 +4,8 @@ import { sendJson } from "../http.ts";
 import { audit, isObj, orgScope } from "./shared.ts";
 import { type ApiCtx, type Route } from "./route.ts";
 
+const numOrUndef = (v: unknown): number | undefined => (typeof v === "number" && Number.isFinite(v) ? v : undefined);
+
 async function deactivatePrincipal(ctx: ApiCtx): Promise<void> {
   const { res, deps } = ctx;
   if (!deps.identity) return sendJson(res, 404, { error: "not_found" });
@@ -32,6 +34,9 @@ async function pushDirectory(ctx: ApiCtx): Promise<void> {
     channelMembers?: unknown;
     groupMembers?: unknown;
     workspaceUrl?: unknown;
+    membersSyncedAt?: unknown;
+    channelsSyncedAt?: unknown;
+    groupsSyncedAt?: unknown;
   };
   if (!Array.isArray(b.members) && !Array.isArray(b.channels) && !Array.isArray(b.groupMembers)) {
     return sendJson(res, 400, {
@@ -58,7 +63,7 @@ async function pushDirectory(ctx: ApiCtx): Promise<void> {
         type: m.type,
         ...(typeof m.slackId === "string" && m.slackId ? { slackId: m.slackId } : {}),
       }));
-    await app.upsertDirectory(members);
+    await app.upsertDirectory(members, numOrUndef(b.membersSyncedAt));
     memberCount = members.length;
   }
   let channelCount: number | undefined;
@@ -73,7 +78,7 @@ async function pushDirectory(ctx: ApiCtx): Promise<void> {
             isObj(m) && typeof m.channelId === "string" && typeof m.principalId === "string",
         )
       : undefined;
-    await app.upsertChannels(channels, channelMembers);
+    await app.upsertChannels(channels, channelMembers, numOrUndef(b.channelsSyncedAt));
     channelCount = channels.length;
   }
   let groupMemberCount: number | undefined;
@@ -82,7 +87,7 @@ async function pushDirectory(ctx: ApiCtx): Promise<void> {
       (m): m is { groupId: string; principalId: string } =>
         isObj(m) && typeof m.groupId === "string" && typeof m.principalId === "string",
     );
-    await app.upsertGroups(groupMembers);
+    await app.upsertGroups(groupMembers, numOrUndef(b.groupsSyncedAt));
     groupMemberCount = groupMembers.length;
   }
   return sendJson(res, 200, {

src/api/slack-core-client.ts

@@ -48,6 +48,9 @@ interface DirectoryPush {
   channelMembers?: Array<{ channelId: string; principalId: string }>;
   groupMembers?: Array<{ groupId: string; principalId: string }>;
   workspaceUrl?: string;
+  membersSyncedAt?: number;
+  channelsSyncedAt?: number;
+  groupsSyncedAt?: number;
 }
 
 export interface SlackCoreClient {
@@ -275,9 +278,9 @@ export function createSlackCoreClient(deps: SlackCoreClientDeps): SlackCoreClien
 
     async pushDirectory(body) {
       if (body.workspaceUrl) await deps.app.setDirectoryWorkspaceUrl(body.workspaceUrl);
-      if (body.members) await deps.app.upsertDirectory(body.members);
-      if (body.channels) await deps.app.upsertChannels(body.channels, body.channelMembers);
-      if (body.groupMembers) await deps.app.upsertGroups(body.groupMembers);
+      if (body.members) await deps.app.upsertDirectory(body.members, body.membersSyncedAt);
+      if (body.channels) await deps.app.upsertChannels(body.channels, body.channelMembers, body.channelsSyncedAt);
+      if (body.groupMembers) await deps.app.upsertGroups(body.groupMembers, body.groupsSyncedAt);
     },
 
     claimDeliveries(type, claimMs) {

src/core/orchestrator.ts

@@ -2300,79 +2300,101 @@ export function createOrchestrator(deps: OrchestratorDeps): Orchestrator {
           !result.silent
         ) {
           const firstTapeWriteFailed = !!result.tapeWriteFailed;
-          const nudgeHistory = filterHistory(
-            forModelContext(await deps.sessions.getEntries(session.id), { includeSecurityTainted: false }),
-          );
-          const nudgeTape = tapeRows
-            ? await deps.sessions
-                .getTape(session.id)
-                .then(async (allRows) => {
-                  const rows = filterTapeForAudience(allRows, conversation.audience, scopeId, resolution.orgScopeId);
-                  const sameHarness = rows.every(
-                    (row) => row.kind !== "message" || row.harness === undefined || row.harness === "pi",
-                  );
-                  const eventsEntitled = tapeEventsEntitled(
-                    rows,
-                    conversation.audience,
-                    scopeId,
-                    resolution.orgScopeId,
-                  );
-                  const primarySubturnComplete =
-                    primarySubturnEndSeq !== undefined &&
-                    rows.some(
-                      (row) =>
-                        row.kind === "annotation" &&
-                        row.entrySeq === primarySubturnEndSeq &&
-                        (row.payload as { subturnEnd?: unknown } | null)?.subturnEnd === true,
+          // The model already wrote a reply as plain assistant text — deliver that text
+          // directly instead of nudging it to re-post (a nudge here re-sends near-identical
+          // text, which surfaces that render assistant entries show twice).
+          const primaryReply = stripAckPrefix(result.reply ?? "", spineAckText).trim();
+          if (primaryReply && defaultDestination && deps.deliveries) {
+            try {
+              const directKey = `post:${session.id}:${randomUUID()}`;
+              await reachEnqueue({
+                deliveries: deps.deliveries,
+                destination: defaultDestination,
+                text: primaryReply,
+                idempotencyKey: directKey,
+                provenance: postProvenance(directKey),
+              });
+              spine.surfaceOutboundCount += 1;
+              if (input.runId) deps.turnStream?.markSurfacePosted(input.runId);
+            } catch (e) {
+              console.error(`[orchestrator] direct reply delivery failed session=${session.id}:`, errMessage(e));
+            }
+          }
+          if (spine.surfaceOutboundCount === 0) {
+            const nudgeHistory = filterHistory(
+              forModelContext(await deps.sessions.getEntries(session.id), { includeSecurityTainted: false }),
+            );
+            const nudgeTape = tapeRows
+              ? await deps.sessions
+                  .getTape(session.id)
+                  .then(async (allRows) => {
+                    const rows = filterTapeForAudience(allRows, conversation.audience, scopeId, resolution.orgScopeId);
+                    const sameHarness = rows.every(
+                      (row) => row.kind !== "message" || row.harness === undefined || row.harness === "pi",
                     );
-                  if (
-                    primaryServedTape &&
-                    !firstTapeWriteFailed &&
-                    sameHarness &&
-                    eventsEntitled &&
-                    primarySubturnComplete
-                  ) {
-                    const fold = await rehydrateTape(foldTape(rows));
-                    if (fold.length && lintFold(fold).ok) return { rows, mode: "serve" as const, fold };
-                  }
-                  return { rows, mode: "shadow" as const };
-                })
-                .catch((e) => {
-                  swallow("tape: nudge read", e);
-                  return undefined;
-                })
-            : undefined;
-          result = await runHarnessTurn(
-            "[system] You were addressed directly. Reply with the `slack` tool's `post` action, or decline explicitly with stay_silent — ending the turn without either is not allowed here.",
-            {
-              ...(turnEnvironment ? { environment: turnEnvironment } : {}),
-              ...(nudgeTape?.mode !== "serve" && inbound.images.length ? { images: inbound.images } : {}),
-            },
-            { history: nudgeHistory, ...(nudgeTape ? { tape: nudgeTape } : {}) },
-          );
-          if (firstTapeWriteFailed || result.tapeWriteFailed) result = { ...result, tapeWriteFailed: true };
-          if (spine.surfaceOutboundCount === 0 && spine.staySilentReason === undefined && !result.silent) {
-            const fallback = stripAckPrefix(result.reply ?? "", spineAckText).trim();
-            if (fallback && defaultDestination && deps.deliveries) {
-              try {
-                const fallbackKey = `post:${session.id}:${randomUUID()}`;
-                await reachEnqueue({
-                  deliveries: deps.deliveries,
-                  destination: defaultDestination,
-                  text: fallback,
-                  idempotencyKey: fallbackKey,
-                  provenance: postProvenance(fallbackKey),
-                });
-                spine.surfaceOutboundCount += 1;
-                if (input.runId) deps.turnStream?.markSurfacePosted(input.runId);
-              } catch (e) {
-                console.error(
-                  `[orchestrator] shed-reply fallback delivery failed session=${session.id}:`,
-                  errMessage(e),
-                );
+                    const eventsEntitled = tapeEventsEntitled(
+                      rows,
+                      conversation.audience,
+                      scopeId,
+                      resolution.orgScopeId,
+                    );
+                    const primarySubturnComplete =
+                      primarySubturnEndSeq !== undefined &&
+                      rows.some(
+                        (row) =>
+                          row.kind === "annotation" &&
+                          row.entrySeq === primarySubturnEndSeq &&
+                          (row.payload as { subturnEnd?: unknown } | null)?.subturnEnd === true,
+                      );
+                    if (
+                      primaryServedTape &&
+                      !firstTapeWriteFailed &&
+                      sameHarness &&
+                      eventsEntitled &&
+                      primarySubturnComplete
+                    ) {
+                      const fold = await rehydrateTape(foldTape(rows));
+                      if (fold.length && lintFold(fold).ok) return { rows, mode: "serve" as const, fold };
+                    }
+                    return { rows, mode: "shadow" as const };
+                  })
+                  .catch((e) => {
+                    swallow("tape: nudge read", e);
+                    return undefined;
+                  })
+              : undefined;
+            result = await runHarnessTurn(
+              "[system] You were addressed directly. Reply with the `slack` tool's `post` action, or decline explicitly with stay_silent — ending the turn without either is not allowed here.",
+              {
+                ...(turnEnvironment ? { environment: turnEnvironment } : {}),
+                ...(nudgeTape?.mode !== "serve" && inbound.images.length ? { images: inbound.images } : {}),
+              },
+              { history: nudgeHistory, ...(nudgeTape ? { tape: nudgeTape } : {}) },
+            );
+            if (firstTapeWriteFailed || result.tapeWriteFailed) result = { ...result, tapeWriteFailed: true };
+            if (spine.surfaceOutboundCount === 0 && spine.staySilentReason === undefined && !result.silent) {
+              const fallback = stripAckPrefix(result.reply ?? "", spineAckText).trim();
+              if (fallback && defaultDestination && deps.deliveries) {
+                try {
+                  const fallbackKey = `post:${session.id}:${randomUUID()}`;
+                  await reachEnqueue({
+                    deliveries: deps.deliveries,
+                    destination: defaultDestination,
+                    text: fallback,
+                    idempotencyKey: fallbackKey,
+                    provenance: postProvenance(fallbackKey),
+                  });
+                  spine.surfaceOutboundCount += 1;
+                  if (input.runId) deps.turnStream?.markSurfacePosted(input.runId);
+                } catch (e) {
+                  console.error(
+                    `[orchestrator] shed-reply fallback delivery failed session=${session.id}:`,
+                    errMessage(e),
+                  );
+                }
+              } else {
+                console.error(`[orchestrator] addressed turn ended silent after nudge session=${session.id}`);
               }
-            } else {
-              console.error(`[orchestrator] addressed turn ended silent after nudge session=${session.id}`);
             }
           }
         }

src/directory/directory-store.ts

@@ -41,8 +41,12 @@ export type ChannelResolution =
   { kind: "one"; channel: DirectoryChannel } | { kind: "ambiguous"; candidates: DirectoryChannel[] } | { kind: "none" };
 
 export interface DirectoryStore {
-  replace(members: DirectoryMember[]): Promise<void>;
-  replaceChannels(channels: DirectoryChannel[], channelMembers?: ChannelMembership[]): Promise<void>;
+  replace(members: DirectoryMember[], syncedAt?: number): Promise<boolean>;
+  replaceChannels(
+    channels: DirectoryChannel[],
+    channelMembers?: ChannelMembership[],
+    syncedAt?: number,
+  ): Promise<boolean>;
   list(): Promise<DirectoryMember[]>;
   listChannels(): Promise<DirectoryChannel[]>;
   get(principalId: string): Promise<DirectoryMember | null>;
@@ -51,7 +55,7 @@ export interface DirectoryStore {
   channelMember(channelId: string, principalId: string): Promise<boolean>;
   channelMembership(channelId: string, principalId: string): Promise<boolean | undefined>;
   channelPrivacy(channelId: string): Promise<boolean | undefined>;
-  replaceGroups(groupMembers: GroupMembership[]): Promise<void>;
+  replaceGroups(groupMembers: GroupMembership[], syncedAt?: number): Promise<boolean>;
   upsertGroup(groupId: string, principalIds: readonly string[]): Promise<void>;
   resolveGroupByParticipants(participants: readonly string[]): Promise<GroupResolution>;
   groupMember(groupId: string, principalId: string): Promise<boolean>;
@@ -92,6 +96,18 @@ export function createDirectoryStore(): DirectoryStore {
   let groupMembers: Map<string, Set<string>> | undefined;
   let groupsSynced = false;
   let workspaceUrl: string | undefined;
+  const syncedAts = new Map<string, number>();
+
+  function acceptSync(section: string, syncedAt: number | undefined): boolean {
+    if (syncedAt === undefined) return true;
+    const stored = syncedAts.get(section);
+    if (stored !== undefined && stored > syncedAt) {
+      console.warn(`[directory] refused stale ${section} swap: stamped ${stored - syncedAt}ms behind`);
+      return false;
+    }
+    syncedAts.set(section, syncedAt);
+    return true;
+  }
 
   return {
     async setWorkspaceUrl(url) {
@@ -100,10 +116,13 @@ export function createDirectoryStore(): DirectoryStore {
     async meta() {
       return { workspaceUrl: workspaceUrl ?? null };
     },
-    async replace(next) {
+    async replace(next, syncedAt) {
+      if (!acceptSync("members", syncedAt)) return false;
       members = next.filter((m) => m.principalId && m.type === "internal");
+      return true;
     },
-    async replaceChannels(nextChannels, nextChannelMembers) {
+    async replaceChannels(nextChannels, nextChannelMembers, syncedAt) {
+      if (!acceptSync("channels", syncedAt)) return false;
       channels = nextChannels.filter((c) => c.channelId && c.name);
       if (nextChannelMembers !== undefined) {
         const byChannel = new Map<string, Set<string>>();
@@ -113,6 +132,7 @@ export function createDirectoryStore(): DirectoryStore {
         }
         channelMembers = byChannel;
       }
+      return true;
     },
     async channelMember(channelId, principalId) {
       return channelMembers?.get(channelId)?.has(principalId) ?? false;
@@ -128,21 +148,25 @@ export function createDirectoryStore(): DirectoryStore {
       const channel = channels.find((candidate) => candidate.channelId === channelId);
       return channel ? channel.isPrivate === true : undefined;
     },
-    async replaceGroups(nextGroupMembers) {
+    async replaceGroups(nextGroupMembers, syncedAt) {
+      if (!acceptSync("groups", syncedAt)) return false;
       const byGroup = new Map<string, Set<string>>();
       for (const m of nextGroupMembers) {
         if (!m.groupId || !m.principalId) continue;
         (byGroup.get(m.groupId) ?? byGroup.set(m.groupId, new Set()).get(m.groupId)!).add(m.principalId);
       }
       groupMembers = byGroup;
       groupsSynced = true;
+      return true;
     },
     async upsertGroup(groupId, principalIds) {
       const ids = principalIds.filter(Boolean);
       if (!groupId || !ids.length) return;
       const byGroup = groupMembers ?? new Map<string, Set<string>>();
       byGroup.set(groupId, new Set(ids));
       groupMembers = byGroup;
+      const stored = syncedAts.get("groups");
+      syncedAts.set("groups", Math.max(stored ?? 0, Date.now()));
     },
     async resolveGroupByParticipants(participants) {
       const key = groupParticipantsKey(participants);