fix(discord-bot): keep prior final when a follow-up is queued

apps/discord-bot/src/features/DiscordDelivery.test.ts

@@ -9,7 +9,9 @@ import {
   excludeFinalizedAssistants,
   initialDeliveryEpochState,
   isGrownFinalizedText,
+  retainUnfinalizedStreamText,
   shouldRecreateTip,
+  unfinalizedPriorAssistantForCatchUp,
   type DeliveryEpochState,
 } from "./DiscordDelivery.ts";
 
@@ -210,6 +212,100 @@ describe("assistantMessagesForDelivery", () => {
   });
 });
 
+describe("unfinalizedPriorAssistantForCatchUp", () => {
+  it("recovers the prior turn answer when queue drain advanced latestTurn", () => {
+    // Empasa: parked follow-up started turn 2 before Discord finalized turn 1.
+    const messages = [
+      msg("u1", "user", "why did the shard manager not recover", "t1"),
+      msg(
+        "a1",
+        "assistant",
+        "Shard-lock recovery is still a bug in this Effect version; later ones auto-recover.",
+        "t1",
+      ),
+      msg("u2", "user", "so everything has recovered now, labels are printing?", "t2"),
+    ];
+    const prior = unfinalizedPriorAssistantForCatchUp({
+      messages,
+      currentTurnId: "t2",
+      lastFinalizedAssistantId: null,
+    });
+    expect(prior?.id).toBe("a1");
+    expect(prior?.turnId).toBe("t1");
+    expect(
+      assistantMessagesForDelivery({
+        messages,
+        turnId: "t2",
+        turnInProgress: true,
+        hasLatestTurn: true,
+      }),
+    ).toEqual([]);
+  });
+
+  it("does not replay a prior answer Discord already finalized", () => {
+    const messages = [
+      msg("u1", "user", "first", "t1"),
+      msg("a1", "assistant", "first answer already posted to Discord", "t1"),
+      msg("u2", "user", "second", "t2"),
+    ];
+    expect(
+      unfinalizedPriorAssistantForCatchUp({
+        messages,
+        currentTurnId: "t2",
+        lastFinalizedAssistantId: "a1",
+        lastFinalizedText: "first answer already posted to Discord",
+      }),
+    ).toBeNull();
+  });
+
+  it("ignores the in-progress turn's own assistants", () => {
+    const messages = [
+      msg("u1", "user", "first", "t1"),
+      msg("a1", "assistant", "I'll trace the shard-lock unhealthy path.", "t1"),
+    ];
+    expect(
+      unfinalizedPriorAssistantForCatchUp({
+        messages,
+        currentTurnId: "t1",
+        lastFinalizedAssistantId: null,
+      }),
+    ).toBeNull();
+  });
+});
+
+describe("retainUnfinalizedStreamText", () => {
+  it("keeps the prior stream body when a new epoch enters awaiting", () => {
+    expect(
+      retainUnfinalizedStreamText({
+        phase: "awaiting",
+        streamText: "",
+        priorLastAssistantText:
+          "Shard-lock recovery is still a bug in this Effect version; later ones auto-recover.",
+      }),
+    ).toBe("Shard-lock recovery is still a bug in this Effect version; later ones auto-recover.");
+  });
+
+  it("does not keep a Working placeholder", () => {
+    expect(
+      retainUnfinalizedStreamText({
+        phase: "awaiting",
+        streamText: "",
+        priorLastAssistantText: "_Working.._",
+      }),
+    ).toBe("");
+  });
+
+  it("uses live stream text while streaming", () => {
+    expect(
+      retainUnfinalizedStreamText({
+        phase: "streaming",
+        streamText: "Yes. Printing is live.",
+        priorLastAssistantText: "I'll trace the shard-lock unhealthy path.",
+      }),
+    ).toBe("Yes. Printing is live.");
+  });
+});
+
 describe("excludeFinalizedAssistants", () => {
   it("keeps only assistants after the finalized bubble in message order", () => {
     const messages = [

apps/discord-bot/src/features/DiscordDelivery.ts

@@ -283,6 +283,78 @@ export function assistantMessagesForDelivery(input: {
   });
 }
 
+/**
+ * Prior-turn assistant Discord still owes a final for.
+ *
+ * Queue drain can advance `latestTurn` to the parked follow-up before the
+ * previous answer is Discord-finalized. {@link assistantMessagesForDelivery}
+ * returns [] in that window on purpose (must not stream the prior body under a
+ * new Working tip). Callers catch-up-finalize this bubble as its own final.
+ */
+export function unfinalizedPriorAssistantForCatchUp(input: {
+  readonly messages: ReadonlyArray<{
+    readonly id: string;
+    readonly role: string;
+    readonly turnId: string | null;
+    readonly text: string;
+  }>;
+  readonly currentTurnId: string | null;
+  readonly lastFinalizedAssistantId: string | null;
+  readonly lastFinalizedText?: string | null;
+}): {
+  readonly id: string;
+  readonly text: string;
+  readonly turnId: string | null;
+} | null {
+  const { messages, currentTurnId } = input;
+  const lastFinalizedText = input.lastFinalizedText ?? null;
+  let prior: {
+    readonly id: string;
+    readonly text: string;
+    readonly turnId: string | null;
+  } | null = null;
+  for (const message of messages) {
+    if (message.role !== "assistant") continue;
+    if (currentTurnId !== null && message.turnId === currentTurnId) continue;
+    const text = message.text.trimEnd();
+    if (text === "" || (/^_Working/i.test(text) && text.length < 40)) continue;
+    prior = { id: message.id, text: message.text, turnId: message.turnId };
+  }
+  if (prior === null) return null;
+
+  const remaining = excludeFinalizedAssistants({
+    messages,
+    assistants: [prior],
+    lastFinalizedAssistantId: input.lastFinalizedAssistantId,
+    lastFinalizedText,
+  });
+  return remaining[0] ?? null;
+}
+
+/**
+ * Keep a substantial unfinalized stream body when the epoch enters awaiting
+ * (fresh Working / next turn with no assistants yet). Wiping it here used to
+ * drop the prior answer on queue drain before the next turn streamed.
+ */
+export function retainUnfinalizedStreamText(input: {
+  readonly phase: DeliveryPhase;
+  readonly streamText: string;
+  readonly priorLastAssistantText: string;
+}): string {
+  if (input.phase === "streaming" && input.streamText.trim() !== "") {
+    return input.streamText;
+  }
+  const prior = input.priorLastAssistantText.trim();
+  const awaitingOrEmptyStream =
+    input.phase === "awaiting" || (input.phase === "streaming" && input.streamText.trim() === "");
+  if (awaitingOrEmptyStream && prior !== "" && !(/^_Working/i.test(prior) && prior.length < 40)) {
+    return input.priorLastAssistantText;
+  }
+  if (input.phase === "awaiting") return "";
+  if (input.phase === "streaming") return input.streamText;
+  return input.priorLastAssistantText;
+}
+
 export function deliveryTextFromAssistants(
   assistants: ReadonlyArray<{ readonly text: string }>,
   mode: "progress" | "answer",

apps/discord-bot/src/features/MentionRouter.ts

@@ -109,6 +109,7 @@ import {
   projectTopicFromParentLookup,
   normalizeWorkspacePath,
   resolveDiscordFollowUpDelivery,
+  shouldPostWorkingAckForContinue,
   type ProjectTopicLookup,
 } from "../presentation/mentions.ts";
 import {
@@ -1203,6 +1204,7 @@ const make = (botConfig: DiscordBotConfig) =>
             });
           }
           const turnAlreadyRunning = hasInterruptibleTurn(currentThread);
+          const followUpDelivery = resolveDiscordFollowUpDelivery(input.flags);
           const liveBridge = yield* getLiveDiscordBridge(
             input.discordThreadId,
             existing.t3ThreadId,
@@ -1236,35 +1238,37 @@ const make = (botConfig: DiscordBotConfig) =>
             reuseLiveBridge,
           });
 
-          // Always post a fresh Working.. tip for interactive continues — including mid-turn
-          // steers while a live bridge is already streaming. Skipping the ack (old
-          // reuseLiveBridge path) left Discord editing the *previous* Working+Stop above
-          // the human message with no visible response to the new mention.
-          // Live bridge adopts this id: freezes/clears old tip, streams under the new one.
-          const workingAckMessageId =
-            input.presentationMode === "final-only"
-              ? null
-              : yield* rest
-                  .createMessage(input.discordThreadId, {
-                    ...workingMessageFields("_Working.._", existing.t3ThreadId),
-                  })
-                  .pipe(
-                    Effect.map((msg) => msg.id as string),
-                    Effect.tap((messageId) =>
-                      Effect.logInfo("Posted Working.. ack", {
-                        messageId,
-                        midTurnSteer: reuseLiveBridge,
-                      }),
-                    ),
-                    Effect.result,
-                    Effect.flatMap((result) => {
-                      if (Result.isSuccess(result)) return Effect.succeed(result.success);
-                      return Effect.logError("Failed to post Working.. ack").pipe(
-                        Effect.andThen(Effect.logError(result.failure)),
-                        Effect.as(null),
-                      );
+          // Interactive continues post a fresh Working.. tip, including mid-turn
+          // steers (live bridge adopts it: freezes the old tip, streams under the new
+          // one). Parked follow-ups must not — adopting would freeze the in-flight
+          // answer and the queued turn's final would replace it.
+          const workingAckMessageId = shouldPostWorkingAckForContinue({
+            presentationMode: input.presentationMode,
+            turnAlreadyRunning,
+            followUpDelivery,
+          })
+            ? yield* rest
+                .createMessage(input.discordThreadId, {
+                  ...workingMessageFields("_Working.._", existing.t3ThreadId),
+                })
+                .pipe(
+                  Effect.map((msg) => msg.id as string),
+                  Effect.tap((messageId) =>
+                    Effect.logInfo("Posted Working.. ack", {
+                      messageId,
+                      midTurnSteer: reuseLiveBridge,
                     }),
-                  );
+                  ),
+                  Effect.result,
+                  Effect.flatMap((result) => {
+                    if (Result.isSuccess(result)) return Effect.succeed(result.success);
+                    return Effect.logError("Failed to post Working.. ack").pipe(
+                      Effect.andThen(Effect.logError(result.failure)),
+                      Effect.as(null),
+                    );
+                  }),
+                )
+            : null;
           const pendingDiscordUserMessageId = newMessageId();
 
           // Ensure bridge (reuses live fiber for same thread; only restarts when needed).
@@ -1350,7 +1354,6 @@ const make = (botConfig: DiscordBotConfig) =>
           // Server parks busy-thread follow-ups. Default Discord policy is **queue**
           // (badge with 📥; delete user message to remove; /omegent steernow to flush).
           // `--steer` / `/omegent steer` inject immediately after startTurn.
-          const followUpDelivery = resolveDiscordFollowUpDelivery(input.flags);
           const t3MessageId = MessageId.make(startedTurn.messageId);
           if (followUpDelivery === "steer" && turnAlreadyRunning) {
             const steered = yield* t3

apps/discord-bot/src/features/ResponseBridge.ts

@@ -100,7 +100,9 @@ import {
   decideAssistantDelivery,
   decideHeartbeat,
   initialDeliveryEpochState,
+  retainUnfinalizedStreamText,
   shouldRecreateTip,
+  unfinalizedPriorAssistantForCatchUp,
   type DeliveryEpochState,
 } from "./DiscordDelivery.ts";
 import { upsertThreadInfoPin } from "./ThreadInfoPin.ts";
@@ -5075,6 +5077,54 @@ export const runBridge = (
             lastFinalizedAssistantId: lastFinalizedForDelivery,
             lastFinalizedText: lastFinalizedTextForDelivery,
           });
+          // Queue drain can make latestTurn the parked follow-up before Discord
+          // finalized the prior answer. Do not stream that body under a new
+          // Working tip — post it as its own final first.
+          if (deliveryAssistants.length === 0 && prior.finalizedTurnId === null) {
+            const priorCatchUp = unfinalizedPriorAssistantForCatchUp({
+              messages: threadMessagesForDelivery,
+              currentTurnId: activeTurnId,
+              lastFinalizedAssistantId: lastFinalizedForDelivery,
+              lastFinalizedText: lastFinalizedTextForDelivery,
+            });
+            const hasUnfinalizedTip =
+              allStreamIds(prior).length > 0 || prior.lastAssistantText.trim() !== "";
+            if (priorCatchUp !== null && hasUnfinalizedTip) {
+              yield* Effect.logInfo(
+                "Catch-up finalizing prior turn before queued follow-up delivery",
+                {
+                  t3ThreadId: input.t3ThreadId,
+                  priorTurnId: priorCatchUp.turnId,
+                  priorAssistantId: priorCatchUp.id,
+                  nextTurnId: activeTurnId,
+                  textLen: priorCatchUp.text.length,
+                },
+              );
+              yield* postOrEditAssistant({
+                turnId: priorCatchUp.turnId,
+                t3MessageId: priorCatchUp.id,
+                text: priorCatchUp.text,
+                streaming: false,
+                images: [],
+                worktreePath: thread.worktreePath,
+              }).pipe(
+                Effect.catchCause((cause) =>
+                  Effect.logError("Failed to catch-up finalize prior turn", {
+                    t3ThreadId: input.t3ThreadId,
+                    priorTurnId: priorCatchUp.turnId,
+                    priorAssistantId: priorCatchUp.id,
+                    cause: formatAlertCause(cause, 300),
+                  }),
+                ),
+                Effect.asVoid,
+              );
+            }
+          }
+          const afterCatchUp = yield* Ref.get(stateRef);
+          const lastFinalizedAfterCatchUp =
+            afterCatchUp.delivery.lastFinalizedAssistantId ?? lastFinalizedForDelivery;
+          const lastFinalizedTextAfterCatchUp =
+            afterCatchUp.delivery.lastFinalizedText ?? lastFinalizedTextForDelivery;
           const deliveryAssistantIds = new Set(deliveryAssistants.map((entry) => entry.id));
           const turnAssistants = thread.messages.filter(
             (message) => message.role === "assistant" && deliveryAssistantIds.has(message.id),
@@ -5090,10 +5140,10 @@ export const runBridge = (
 
           const decision = decideAssistantDelivery({
             state: {
-              ...prior.delivery,
+              ...afterCatchUp.delivery,
               // Keep durable finalize memory from the link store in sync.
-              lastFinalizedAssistantId: lastFinalizedForDelivery,
-              lastFinalizedText: lastFinalizedTextForDelivery,
+              lastFinalizedAssistantId: lastFinalizedAfterCatchUp,
+              lastFinalizedText: lastFinalizedTextAfterCatchUp,
             },
             turnId: activeTurnId,
             turnInProgress: turnInProgressNow,
@@ -5141,12 +5191,11 @@ export const runBridge = (
               delivery: decision.state,
               adoptedInitialSnapshot: true,
               seededWorkingAckPending: decision.state.phase === "awaiting",
-              lastAssistantText:
-                decision.state.phase === "streaming"
-                  ? decision.state.streamText
-                  : decision.state.phase === "awaiting"
-                    ? ""
-                    : current.lastAssistantText,
+              lastAssistantText: retainUnfinalizedStreamText({
+                phase: decision.state.phase,
+                streamText: decision.state.streamText,
+                priorLastAssistantText: current.lastAssistantText,
+              }),
               finalizedTurnId:
                 decision.state.phase === "awaiting" || reopenedAfterPrematureFinal
                   ? null