From b21a3763d51e7ae758607c3f0a09f3ade1556387 Mon Sep 17 00:00:00 2001 From: "omegent-app[bot]" <306514130+omegent-app[bot]@users.noreply.github.com> Date: Mon, 31 Aug 2026 08:05:37 +0000 Subject: [PATCH] fix(discord-bot): keep prior final when a follow-up is queued Parked mid-turn mentions were still posting a fresh Working tip, which froze the in-flight stream and let queue-drain start the next turn before Discord posted the previous answer. Skip Working/adopt for queued follow-ups, catch-up-finalize the prior turn when latestTurn already advanced, and do not wipe lastAssistantText on awaiting. Co-authored-by: Patrick Roza <42661+patroza@users.noreply.github.com> --- .../src/features/DiscordDelivery.test.ts | 96 +++++++++++++++++++ .../src/features/DiscordDelivery.ts | 72 ++++++++++++++ .../discord-bot/src/features/MentionRouter.ts | 61 ++++++------ .../src/features/ResponseBridge.ts | 67 +++++++++++-- .../src/presentation/mentions.test.ts | 37 +++++++ apps/discord-bot/src/presentation/mentions.ts | 18 ++++ 6 files changed, 313 insertions(+), 38 deletions(-) diff --git a/apps/discord-bot/src/features/DiscordDelivery.test.ts b/apps/discord-bot/src/features/DiscordDelivery.test.ts index c9f6c593812c..fa2e22712202 100644 --- a/apps/discord-bot/src/features/DiscordDelivery.test.ts +++ b/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 = [ diff --git a/apps/discord-bot/src/features/DiscordDelivery.ts b/apps/discord-bot/src/features/DiscordDelivery.ts index 3429cb5fcc37..cd20144b972e 100644 --- a/apps/discord-bot/src/features/DiscordDelivery.ts +++ b/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", diff --git a/apps/discord-bot/src/features/MentionRouter.ts b/apps/discord-bot/src/features/MentionRouter.ts index 1dcaadb5e26f..f0d3ed004fdd 100644 --- a/apps/discord-bot/src/features/MentionRouter.ts +++ b/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 diff --git a/apps/discord-bot/src/features/ResponseBridge.ts b/apps/discord-bot/src/features/ResponseBridge.ts index 66b695f9dc78..56b78298c8e4 100644 --- a/apps/discord-bot/src/features/ResponseBridge.ts +++ b/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 diff --git a/apps/discord-bot/src/presentation/mentions.test.ts b/apps/discord-bot/src/presentation/mentions.test.ts index 0110c8a3a56b..0753884523cd 100644 --- a/apps/discord-bot/src/presentation/mentions.test.ts +++ b/apps/discord-bot/src/presentation/mentions.test.ts @@ -10,6 +10,7 @@ import { projectTopicFromParentLookup, readChannelTopic, resolveDiscordFollowUpDelivery, + shouldPostWorkingAckForContinue, } from "./mentions.ts"; import { chunkDiscordContent, @@ -140,6 +141,42 @@ describe("resolveDiscordFollowUpDelivery", () => { }); }); +describe("shouldPostWorkingAckForContinue", () => { + it("does not post Working for a parked mid-turn follow-up", () => { + expect( + shouldPostWorkingAckForContinue({ + turnAlreadyRunning: true, + followUpDelivery: "queue", + }), + ).toBe(false); + }); + + it("posts Working for a mid-turn steer and for idle continues", () => { + expect( + shouldPostWorkingAckForContinue({ + turnAlreadyRunning: true, + followUpDelivery: "steer", + }), + ).toBe(true); + expect( + shouldPostWorkingAckForContinue({ + turnAlreadyRunning: false, + followUpDelivery: "queue", + }), + ).toBe(true); + }); + + it("never posts Working in final-only presentation", () => { + expect( + shouldPostWorkingAckForContinue({ + presentationMode: "final-only", + turnAlreadyRunning: false, + followUpDelivery: "steer", + }), + ).toBe(false); + }); +}); + describe("parseMentionIntent", () => { it("recognizes stop words as interrupt commands", () => { expect(parseMentionIntent("stop")).toEqual({ kind: "interrupt" }); diff --git a/apps/discord-bot/src/presentation/mentions.ts b/apps/discord-bot/src/presentation/mentions.ts index f3c4ea16ddcd..5a15069a7df3 100644 --- a/apps/discord-bot/src/presentation/mentions.ts +++ b/apps/discord-bot/src/presentation/mentions.ts @@ -31,6 +31,24 @@ export function resolveDiscordFollowUpDelivery( return flags.followUpDelivery ?? "queue"; } +/** + * Whether a continue-mention should post/adopt a fresh `_Working.._` tip. + * + * Parked mid-turn follow-ups must not. Adopting a new Working freezes the live + * tip and bumps the delivery epoch, so the in-flight answer is never posted + * (queued follow-up final lands; prior final is stuck as frozen progress). + * `--steer` still posts a new tip under the new mention. + */ +export function shouldPostWorkingAckForContinue(input: { + readonly presentationMode?: "full" | "final-only" | undefined; + readonly turnAlreadyRunning: boolean; + readonly followUpDelivery: DiscordFollowUpDelivery; +}): boolean { + if (input.presentationMode === "final-only") return false; + if (input.turnAlreadyRunning && input.followUpDelivery === "queue") return false; + return true; +} + export type ParsedMentionIntent = | { readonly kind: "interrupt" } | { readonly kind: "help" }