From 265815c89d6809665e8023afa672d9fa2e5917e1 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Wed, 16 Sep 2026 22:51:43 +0000 Subject: [PATCH 1/9] fix(chat): Surface deterministic history branches Co-Authored-By: David Cramer --- .../client/conversations/eventTranscript.ts | 16 +++++++-- .../tests/transcriptMessageContext.test.ts | 36 +++++++++++++++++++ .../src/chat/task-execution/checkpoint.ts | 14 ++++---- .../task-execution/checkpoint.test.ts | 11 ++---- 4 files changed, 59 insertions(+), 18 deletions(-) diff --git a/packages/junior-dashboard/src/client/conversations/eventTranscript.ts b/packages/junior-dashboard/src/client/conversations/eventTranscript.ts index 8a590c9368..343ff011c0 100644 --- a/packages/junior-dashboard/src/client/conversations/eventTranscript.ts +++ b/packages/junior-dashboard/src/client/conversations/eventTranscript.ts @@ -241,8 +241,20 @@ export function transcriptMessagesFromEvents( } if (data.type === "turn_lifecycle" && data.state === "started") { - const inputMessages = data.inputMessageIds - ?.map((messageId) => messagesById.get(messageId)) + const inputMessageIds = data.inputMessageIds ?? []; + const inputMessageIdSet = new Set(inputMessageIds); + for (const messageId of inputMessageIds) { + const suffix = ":message_changed_mention"; + if (!messageId.endsWith(suffix)) continue; + const originalMessageId = messageId.slice(0, -suffix.length); + if (!inputMessageIdSet.has(originalMessageId)) continue; + const originalMessage = messagesById.get(originalMessageId); + if (!originalMessage) continue; + messages.splice(messages.indexOf(originalMessage), 1); + messagesById.delete(originalMessageId); + } + const inputMessages = inputMessageIds + .map((messageId) => messagesById.get(messageId)) .filter((message) => message !== undefined); const turnUserMessage = inputMessages ?.slice() diff --git a/packages/junior-dashboard/tests/transcriptMessageContext.test.ts b/packages/junior-dashboard/tests/transcriptMessageContext.test.ts index 301d9b9bdc..a564642170 100644 --- a/packages/junior-dashboard/tests/transcriptMessageContext.test.ts +++ b/packages/junior-dashboard/tests/transcriptMessageContext.test.ts @@ -78,6 +78,42 @@ describe("transcript message context classification", () => { expect(messages[1]?.messageId).toBe("answer"); }); + it("hides an original message superseded by its edited mention", () => { + const messages = conversationTranscriptMessages( + conversation([ + event(0, "2026-01-01T00:00:00.000Z", { + type: "message", + messageId: "1700000100.000100", + role: "user", + text: "can you clarify that?", + explicitMention: false, + }), + event(1, "2026-01-01T00:00:01.000Z", { + type: "message", + messageId: "1700000100.000100:message_changed_mention", + role: "user", + text: "can you clarify that?", + explicitMention: true, + }), + event(2, "2026-01-01T00:00:02.000Z", { + type: "turn_lifecycle", + turnId: "turn-edited-mention", + state: "started", + inputMessageIds: [ + "1700000100.000100:message_changed_mention", + "1700000100.000100", + ], + }), + ]), + ); + + expect(messages).toHaveLength(1); + expect(messages[0]).toMatchObject({ + explicitMention: true, + messageId: "1700000100.000100:message_changed_mention", + }); + }); + it("keeps turn context on acted-on non-mention inputs", () => { const messages = conversationTranscriptMessages( conversation([ diff --git a/packages/junior/src/chat/task-execution/checkpoint.ts b/packages/junior/src/chat/task-execution/checkpoint.ts index 620e36c4a3..54b570c903 100644 --- a/packages/junior/src/chat/task-execution/checkpoint.ts +++ b/packages/junior/src/chat/task-execution/checkpoint.ts @@ -217,14 +217,14 @@ async function saveRunning( state: "running", }); } catch (error) { - // Quiet only branch races on best-effort running checkpoints. - if (!(error instanceof AgentHistoryBranchError)) { - logException(error, "agent.turn.checkpoint.running.failed", { - "app.ai.resume_conversation_id": args.conversationId, - "app.ai.resume_session_id": args.turnId, - "app.ai.resume_slice_id": args.sliceId, - }); + if (error instanceof AgentHistoryBranchError) { + throw error; } + logException(error, "agent.turn.checkpoint.running.failed", { + "app.ai.resume_conversation_id": args.conversationId, + "app.ai.resume_session_id": args.turnId, + "app.ai.resume_slice_id": args.sliceId, + }); return undefined; } } diff --git a/packages/junior/tests/component/task-execution/checkpoint.test.ts b/packages/junior/tests/component/task-execution/checkpoint.test.ts index d24556a3fb..2fd551dbbd 100644 --- a/packages/junior/tests/component/task-execution/checkpoint.test.ts +++ b/packages/junior/tests/component/task-execution/checkpoint.test.ts @@ -1582,12 +1582,7 @@ describe("turn checkpoint", () => { ).resolves.toBeUndefined(); }); - it("rejects true history branches without reporting a running-session exception", async () => { - const logException = vi.fn(); - vi.doMock("@/chat/logging", async (importOriginal) => { - const actual = await importOriginal(); - return { ...actual, logException }; - }); + it("surfaces true history branches to the owning Turn boundary", async () => { const { saveTurnCheckpoint } = await import("@/chat/task-execution/checkpoint"); const committedUser = userMessage("committed"); @@ -1610,9 +1605,7 @@ describe("turn checkpoint", () => { sliceId: 1, messages: [staleUser], }), - ).resolves.toBeUndefined(); - - expect(logException).not.toHaveBeenCalled(); + ).rejects.toThrow("changed before its committed boundary"); }); it("appends after in-place assistant envelope mutations on committed messages", async () => { From a37213b64fb32621077290b49073ea73dda1b7b2 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 17 Sep 2026 00:33:47 +0000 Subject: [PATCH 2/9] fix(chat): scope history branch failures Co-Authored-By: David Cramer --- packages/junior/src/chat/agent/resume.ts | 3 +++ packages/junior/src/chat/task-execution/checkpoint.ts | 11 ++++++++++- .../tests/component/task-execution/checkpoint.test.ts | 1 + 3 files changed, 14 insertions(+), 1 deletion(-) diff --git a/packages/junior/src/chat/agent/resume.ts b/packages/junior/src/chat/agent/resume.ts index 2e430d102d..29fce50545 100644 --- a/packages/junior/src/chat/agent/resume.ts +++ b/packages/junior/src/chat/agent/resume.ts @@ -181,6 +181,7 @@ export function createResumeState(args: ResumeStateArgs) { async persistSafeBoundary( messages: PiMessage[], trailingMessageProvenance?: ConversationMessageProvenance[], + required = false, ): Promise { const saved = await saveTurnCheckpoint({ mode: "running", @@ -190,6 +191,7 @@ export function createResumeState(args: ResumeStateArgs) { trailingMessageProvenance, turnContexts: turnContexts.length > 0 ? turnContexts : undefined, turnStartMessageIndex, + required, }); if (!saved) { return false; @@ -212,6 +214,7 @@ export function createResumeState(args: ResumeStateArgs) { const persisted = await this.persistSafeBoundary( messages, trailingMessageProvenance, + Boolean(args.durability.onInputCommitted), ); if (!persisted && args.durability.onInputCommitted) { throw new TurnInputCommitLostError( diff --git a/packages/junior/src/chat/task-execution/checkpoint.ts b/packages/junior/src/chat/task-execution/checkpoint.ts index 54b570c903..f365c60765 100644 --- a/packages/junior/src/chat/task-execution/checkpoint.ts +++ b/packages/junior/src/chat/task-execution/checkpoint.ts @@ -85,6 +85,8 @@ interface TurnCheckpointWrite { turnStartMessageIndex?: number; /** Tool calls charged to this turn; survives history replacement. */ cumulativeToolCallCount?: number; + /** Reject a conflicting history write instead of treating it as best-effort. */ + required?: boolean; trailingMessageProvenance?: ConversationMessageProvenance[]; turnContexts?: PluginTurnContext[]; durationMs?: number; @@ -156,6 +158,7 @@ export async function loadTurnCheckpoint(args: { * Save turn progress. * * - `running` / `paused`: best-effort; returns the stored record or undefined + * - required `running`: rejects a conflicting history write * - `completed` / `failed`: retries until write accepts; throws on hard failure */ export function saveTurnCheckpoint( @@ -217,9 +220,15 @@ async function saveRunning( state: "running", }); } catch (error) { - if (error instanceof AgentHistoryBranchError) { + if ( + error instanceof AgentHistoryBranchError && + args.required === true + ) { throw error; } + if (error instanceof AgentHistoryBranchError) { + return undefined; + } logException(error, "agent.turn.checkpoint.running.failed", { "app.ai.resume_conversation_id": args.conversationId, "app.ai.resume_session_id": args.turnId, diff --git a/packages/junior/tests/component/task-execution/checkpoint.test.ts b/packages/junior/tests/component/task-execution/checkpoint.test.ts index 2fd551dbbd..481298e492 100644 --- a/packages/junior/tests/component/task-execution/checkpoint.test.ts +++ b/packages/junior/tests/component/task-execution/checkpoint.test.ts @@ -1604,6 +1604,7 @@ describe("turn checkpoint", () => { turnId: "turn-stale-checkpoint", sliceId: 1, messages: [staleUser], + required: true, }), ).rejects.toThrow("changed before its committed boundary"); }); From 97374e41a58e69919e39226c9c601340e85ff5ac Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 17 Sep 2026 00:43:21 +0000 Subject: [PATCH 3/9] fix(chat): require Slack input checkpoints --- packages/junior/src/chat/agent/resume.ts | 21 ++++++++++++++----- packages/junior/src/chat/agent/types.ts | 2 ++ .../junior/src/chat/providers/slack/turn.ts | 1 + 3 files changed, 19 insertions(+), 5 deletions(-) diff --git a/packages/junior/src/chat/agent/resume.ts b/packages/junior/src/chat/agent/resume.ts index 29fce50545..b82f38be96 100644 --- a/packages/junior/src/chat/agent/resume.ts +++ b/packages/junior/src/chat/agent/resume.ts @@ -7,6 +7,7 @@ import type { Destination, Source } from "@sentry/junior-plugin-api"; import { botConfig } from "@/chat/config"; import type { PiMessage } from "@/chat/pi/messages"; import type { ConversationMessageProvenance } from "@/chat/conversations/provenance"; +import { AgentHistoryBranchError } from "@/chat/conversations/projection"; import { CooperativeTurnYieldError, TurnInputCommitLostError, @@ -211,11 +212,21 @@ export function createResumeState(args: ResumeStateArgs) { messages: PiMessage[], trailingMessageProvenance?: ConversationMessageProvenance[], ): Promise { - const persisted = await this.persistSafeBoundary( - messages, - trailingMessageProvenance, - Boolean(args.durability.onInputCommitted), - ); + let persisted: boolean; + try { + persisted = await this.persistSafeBoundary( + messages, + trailingMessageProvenance, + args.durability.inputCheckpointRequired === true, + ); + } catch (error) { + if (!(error instanceof AgentHistoryBranchError)) { + throw error; + } + throw new TurnInputCommitLostError( + `Durable turn input conflicts with committed history for conversation=${args.conversationId} turn=${args.turnId}`, + ); + } if (!persisted && args.durability.onInputCommitted) { throw new TurnInputCommitLostError( `Durable turn input could not be checkpointed for conversation=${args.conversationId} turn=${args.turnId}`, diff --git a/packages/junior/src/chat/agent/types.ts b/packages/junior/src/chat/agent/types.ts index ec5dd2e601..0b688f8a4f 100644 --- a/packages/junior/src/chat/agent/types.ts +++ b/packages/junior/src/chat/agent/types.ts @@ -148,6 +148,8 @@ export class RetryableDeliveryError extends Error { export type AgentDurability = { /** Schedule delegated work with authority bound by the active parent run. */ spawnAgent?: SpawnAgent; + /** Reject a history branch before the current input enters model execution. */ + inputCheckpointRequired?: boolean; onInputCommitted?: () => void | Promise; /** Return true when the durable worker should pause at the next Pi boundary. */ shouldYield?: () => boolean; diff --git a/packages/junior/src/chat/providers/slack/turn.ts b/packages/junior/src/chat/providers/slack/turn.ts index bdd608f786..7079c69e60 100644 --- a/packages/junior/src/chat/providers/slack/turn.ts +++ b/packages/junior/src/chat/providers/slack/turn.ts @@ -1149,6 +1149,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { ? undefined : { delivery: deliverAssistantMessage }), durability: { + inputCheckpointRequired: true, onInputCommitted: options.ack, drainSteeringMessages, shouldYield: options.shouldYield, From 2f354e77d68dc27c77df7fccfd3923c9a6303f4a Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 17 Sep 2026 02:31:34 +0000 Subject: [PATCH 4/9] fix(chat): preserve history branch failures Co-Authored-By: David Cramer --- packages/junior/src/chat/agent/resume.ts | 21 +++-------- .../slack/message-content-behavior.test.ts | 35 +++++++++++-------- 2 files changed, 26 insertions(+), 30 deletions(-) diff --git a/packages/junior/src/chat/agent/resume.ts b/packages/junior/src/chat/agent/resume.ts index b82f38be96..e6b3873c01 100644 --- a/packages/junior/src/chat/agent/resume.ts +++ b/packages/junior/src/chat/agent/resume.ts @@ -7,7 +7,6 @@ import type { Destination, Source } from "@sentry/junior-plugin-api"; import { botConfig } from "@/chat/config"; import type { PiMessage } from "@/chat/pi/messages"; import type { ConversationMessageProvenance } from "@/chat/conversations/provenance"; -import { AgentHistoryBranchError } from "@/chat/conversations/projection"; import { CooperativeTurnYieldError, TurnInputCommitLostError, @@ -212,21 +211,11 @@ export function createResumeState(args: ResumeStateArgs) { messages: PiMessage[], trailingMessageProvenance?: ConversationMessageProvenance[], ): Promise { - let persisted: boolean; - try { - persisted = await this.persistSafeBoundary( - messages, - trailingMessageProvenance, - args.durability.inputCheckpointRequired === true, - ); - } catch (error) { - if (!(error instanceof AgentHistoryBranchError)) { - throw error; - } - throw new TurnInputCommitLostError( - `Durable turn input conflicts with committed history for conversation=${args.conversationId} turn=${args.turnId}`, - ); - } + const persisted = await this.persistSafeBoundary( + messages, + trailingMessageProvenance, + args.durability.inputCheckpointRequired === true, + ); if (!persisted && args.durability.onInputCommitted) { throw new TurnInputCommitLostError( `Durable turn input could not be checkpointed for conversation=${args.conversationId} turn=${args.turnId}`, diff --git a/packages/junior/tests/integration/slack/message-content-behavior.test.ts b/packages/junior/tests/integration/slack/message-content-behavior.test.ts index c166e9d5ef..2e8c0bc4ba 100644 --- a/packages/junior/tests/integration/slack/message-content-behavior.test.ts +++ b/packages/junior/tests/integration/slack/message-content-behavior.test.ts @@ -6,7 +6,6 @@ import { persistThreadState, persistThreadStateById, } from "@/chat/runtime/thread-state"; -import { TurnInputCommitLostError } from "@/chat/runtime/turn"; import { coerceThreadConversationState } from "@/chat/state/conversation"; import { disconnectStateAdapter } from "@/chat/state/adapter"; import { hydrateConversationMessages } from "@/chat/conversations/messages"; @@ -554,21 +553,29 @@ describe("Slack behavior: message content", () => { }, }); - await expect( - slackRuntime.handleNewMention( - thread, - createTestMessage({ - id: "m-content-active-session-record", - text: "<@U0APP> continue", - isMention: true, - threadId: thread.id, - author: { userId: "U0TESTER" }, - }), - { destination: createTestDestination(thread) }, - ), - ).rejects.toBeInstanceOf(TurnInputCommitLostError); + const message = createTestMessage({ + id: "m-content-active-session-record", + text: "<@U0APP> continue", + isMention: true, + threadId: thread.id, + author: { userId: "U0TESTER" }, + }); + await slackRuntime.handleNewMention(thread, message, { + destination: createTestDestination(thread), + }); expect(calls).toHaveLength(1); expect(calls[0]?.piMessages).toEqual(activeMessages); + const lifecycle = ( + await getConversationEventStore().loadHistory(thread.id) + ).filter( + (event) => + event.data.type === "turn_started" || + event.data.type === "turn_failed", + ); + expect(lifecycle.map((event) => event.data.type)).toEqual([ + "turn_started", + "turn_failed", + ]); }); }); From 8207e8b8b7ab2c0777e717636fe6b0d8e82a5db2 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 17 Sep 2026 13:48:06 +0000 Subject: [PATCH 5/9] refactor(dashboard): clarify edited mention projection --- .../client/conversations/eventTranscript.ts | 51 ++++++++++++------- 1 file changed, 34 insertions(+), 17 deletions(-) diff --git a/packages/junior-dashboard/src/client/conversations/eventTranscript.ts b/packages/junior-dashboard/src/client/conversations/eventTranscript.ts index 343ff011c0..acc1ffe560 100644 --- a/packages/junior-dashboard/src/client/conversations/eventTranscript.ts +++ b/packages/junior-dashboard/src/client/conversations/eventTranscript.ts @@ -53,6 +53,26 @@ function specialToolIds(events: ConversationReportEvent[]): Set { return ids; } +const EDITED_MENTION_SUFFIX = ":message_changed_mention"; + +/** Find original Slack Messages superseded by an edited mention in one Turn. */ +function supersededSlackMessageIds( + events: readonly ConversationReportEvent[], +): Set { + const superseded = new Set(); + for (const event of events) { + const data = event.data; + if (data.type !== "turn_lifecycle" || data.state !== "started") continue; + const inputIds = new Set(data.inputMessageIds ?? []); + for (const messageId of inputIds) { + if (!messageId.endsWith(EDITED_MENTION_SUFFIX)) continue; + const originalId = messageId.slice(0, -EDITED_MENTION_SUFFIX.length); + if (inputIds.has(originalId)) superseded.add(originalId); + } + } + return superseded; +} + function historyMessageIds( messages: readonly TranscriptViewMessage[], ): Set { @@ -80,7 +100,9 @@ export function pendingTranscriptMessage( // Keep pending rows after history without colliding with real event seqs. sourceSeq: Number.MAX_SAFE_INTEGER - 1_000_000 + index, timestamp: Date.parse(message.createdAt), - ...(message.actorIdentity ? { actorIdentity: message.actorIdentity } : undefined), + ...(message.actorIdentity + ? { actorIdentity: message.actorIdentity } + : undefined), }; } @@ -130,6 +152,7 @@ export function transcriptMessagesFromEvents( pendingMessages?: readonly ConversationPendingMessage[], ): TranscriptViewMessage[] { const replacedToolIds = specialToolIds(events); + const supersededMessageIds = supersededSlackMessageIds(events); const tools = new Map< string, Extract @@ -197,7 +220,9 @@ export function transcriptMessagesFromEvents( : { type: "text", text: data.text! }, ]), messageId: data.messageId, - ...(data.actorIdentity ? { actorIdentity: data.actorIdentity } : undefined), + ...(data.actorIdentity + ? { actorIdentity: data.actorIdentity } + : undefined), ...(data.eventType ? { eventType: data.eventType } : undefined), ...(data.trustedSummary ? { trustedSummary: data.trustedSummary } @@ -242,18 +267,8 @@ export function transcriptMessagesFromEvents( if (data.type === "turn_lifecycle" && data.state === "started") { const inputMessageIds = data.inputMessageIds ?? []; - const inputMessageIdSet = new Set(inputMessageIds); - for (const messageId of inputMessageIds) { - const suffix = ":message_changed_mention"; - if (!messageId.endsWith(suffix)) continue; - const originalMessageId = messageId.slice(0, -suffix.length); - if (!inputMessageIdSet.has(originalMessageId)) continue; - const originalMessage = messagesById.get(originalMessageId); - if (!originalMessage) continue; - messages.splice(messages.indexOf(originalMessage), 1); - messagesById.delete(originalMessageId); - } const inputMessages = inputMessageIds + .filter((messageId) => !supersededMessageIds.has(messageId)) .map((messageId) => messagesById.get(messageId)) .filter((message) => message !== undefined); const turnUserMessage = inputMessages @@ -448,10 +463,12 @@ export function transcriptMessagesFromEvents( const ordered = messages .filter( (message) => - message.role !== "user" || - message.eventType !== undefined || - message.explicitMention !== false || - message.context === true, + !message.messageId || + (!supersededMessageIds.has(message.messageId) && + (message.role !== "user" || + message.eventType !== undefined || + message.explicitMention !== false || + message.context === true)), ) .sort((left, right) => left.sourceSeq - right.sourceSeq); return mergePendingTranscriptMessages(ordered, pendingMessages); From 7f41a0b4e8ef00e5962aebbdf7c90406ed09d25c Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 17 Sep 2026 15:13:47 +0000 Subject: [PATCH 6/9] fix(slack): ignore message edits Keep accepted Conversation Messages immutable. Slack message_changed events no longer create replacement Messages or start Turns, and the dashboard no longer needs an edited-message projection workaround. Co-Authored-By: David Cramer --- .../client/conversations/eventTranscript.ts | 37 +-- .../tests/transcriptMessageContext.test.ts | 36 --- packages/junior-evals/src/behavior-harness.ts | 3 +- .../src/chat/ingress/message-changed.ts | 190 --------------- .../junior/src/chat/ingress/slack-webhook.ts | 64 +---- packages/junior/src/handlers/webhooks.ts | 102 +------- .../slack-conversation-work.test.ts | 47 +--- .../slack/message-changed-behavior.test.ts | 227 ++---------------- .../message-changed-gate-contract.test.ts | 97 -------- .../message-changed-reply-contract.test.ts | 225 ----------------- .../slack/message-changed-ingress.test.ts | 190 --------------- 11 files changed, 36 insertions(+), 1182 deletions(-) delete mode 100644 packages/junior/src/chat/ingress/message-changed.ts delete mode 100644 packages/junior/tests/integration/slack/message-changed-gate-contract.test.ts delete mode 100644 packages/junior/tests/integration/slack/message-changed-reply-contract.test.ts delete mode 100644 packages/junior/tests/unit/slack/message-changed-ingress.test.ts diff --git a/packages/junior-dashboard/src/client/conversations/eventTranscript.ts b/packages/junior-dashboard/src/client/conversations/eventTranscript.ts index acc1ffe560..24e5f7fd73 100644 --- a/packages/junior-dashboard/src/client/conversations/eventTranscript.ts +++ b/packages/junior-dashboard/src/client/conversations/eventTranscript.ts @@ -53,26 +53,6 @@ function specialToolIds(events: ConversationReportEvent[]): Set { return ids; } -const EDITED_MENTION_SUFFIX = ":message_changed_mention"; - -/** Find original Slack Messages superseded by an edited mention in one Turn. */ -function supersededSlackMessageIds( - events: readonly ConversationReportEvent[], -): Set { - const superseded = new Set(); - for (const event of events) { - const data = event.data; - if (data.type !== "turn_lifecycle" || data.state !== "started") continue; - const inputIds = new Set(data.inputMessageIds ?? []); - for (const messageId of inputIds) { - if (!messageId.endsWith(EDITED_MENTION_SUFFIX)) continue; - const originalId = messageId.slice(0, -EDITED_MENTION_SUFFIX.length); - if (inputIds.has(originalId)) superseded.add(originalId); - } - } - return superseded; -} - function historyMessageIds( messages: readonly TranscriptViewMessage[], ): Set { @@ -152,7 +132,6 @@ export function transcriptMessagesFromEvents( pendingMessages?: readonly ConversationPendingMessage[], ): TranscriptViewMessage[] { const replacedToolIds = specialToolIds(events); - const supersededMessageIds = supersededSlackMessageIds(events); const tools = new Map< string, Extract @@ -266,10 +245,8 @@ export function transcriptMessagesFromEvents( } if (data.type === "turn_lifecycle" && data.state === "started") { - const inputMessageIds = data.inputMessageIds ?? []; - const inputMessages = inputMessageIds - .filter((messageId) => !supersededMessageIds.has(messageId)) - .map((messageId) => messagesById.get(messageId)) + const inputMessages = data.inputMessageIds + ?.map((messageId) => messagesById.get(messageId)) .filter((message) => message !== undefined); const turnUserMessage = inputMessages ?.slice() @@ -463,12 +440,10 @@ export function transcriptMessagesFromEvents( const ordered = messages .filter( (message) => - !message.messageId || - (!supersededMessageIds.has(message.messageId) && - (message.role !== "user" || - message.eventType !== undefined || - message.explicitMention !== false || - message.context === true)), + message.role !== "user" || + message.eventType !== undefined || + message.explicitMention !== false || + message.context === true, ) .sort((left, right) => left.sourceSeq - right.sourceSeq); return mergePendingTranscriptMessages(ordered, pendingMessages); diff --git a/packages/junior-dashboard/tests/transcriptMessageContext.test.ts b/packages/junior-dashboard/tests/transcriptMessageContext.test.ts index a564642170..301d9b9bdc 100644 --- a/packages/junior-dashboard/tests/transcriptMessageContext.test.ts +++ b/packages/junior-dashboard/tests/transcriptMessageContext.test.ts @@ -78,42 +78,6 @@ describe("transcript message context classification", () => { expect(messages[1]?.messageId).toBe("answer"); }); - it("hides an original message superseded by its edited mention", () => { - const messages = conversationTranscriptMessages( - conversation([ - event(0, "2026-01-01T00:00:00.000Z", { - type: "message", - messageId: "1700000100.000100", - role: "user", - text: "can you clarify that?", - explicitMention: false, - }), - event(1, "2026-01-01T00:00:01.000Z", { - type: "message", - messageId: "1700000100.000100:message_changed_mention", - role: "user", - text: "can you clarify that?", - explicitMention: true, - }), - event(2, "2026-01-01T00:00:02.000Z", { - type: "turn_lifecycle", - turnId: "turn-edited-mention", - state: "started", - inputMessageIds: [ - "1700000100.000100:message_changed_mention", - "1700000100.000100", - ], - }), - ]), - ); - - expect(messages).toHaveLength(1); - expect(messages[0]).toMatchObject({ - explicitMention: true, - messageId: "1700000100.000100:message_changed_mention", - }); - }); - it("keeps turn context on acted-on non-mention inputs", () => { const messages = conversationTranscriptMessages( conversation([ diff --git a/packages/junior-evals/src/behavior-harness.ts b/packages/junior-evals/src/behavior-harness.ts index 7d2fc5f743..078f7dcd7e 100644 --- a/packages/junior-evals/src/behavior-harness.ts +++ b/packages/junior-evals/src/behavior-harness.ts @@ -1143,8 +1143,7 @@ function toEvalAssistantPost(value: unknown): EvalAssistantPost { * Build a Chat SDK Message for Slack ingress from a harness event. * * Synthetic Slack ingress keeps an empty formatted AST so plain text remains - * the source of truth, matching mailbox restore and edited-message - * construction elsewhere in Junior. + * the source of truth, matching mailbox restore elsewhere in Junior. */ function toSlackMessage( event: MentionEvent | SubscribedMessageEvent, diff --git a/packages/junior/src/chat/ingress/message-changed.ts b/packages/junior/src/chat/ingress/message-changed.ts deleted file mode 100644 index 1bb76aeaeb..0000000000 --- a/packages/junior/src/chat/ingress/message-changed.ts +++ /dev/null @@ -1,190 +0,0 @@ -import { Message, type Adapter, type Attachment } from "chat"; -import { parseActorUserId } from "@/chat/actor"; -import { textMentionsBot } from "@/chat/ingress/bot-mention"; - -/** - * Parsed result from a Slack `message_changed` event that contains a newly - * added bot @mention. Returns `null` when the event does not qualify. - */ -export interface MessageChangedMention { - /** Slack thread ID in `slack::` format. */ - threadId: string; - /** Synthesized Message to pass to `bot.processMessage`. */ - message: Message; -} - -function getEditedMentionMessageId(messageTs: string): string { - return `${messageTs}:message_changed_mention`; -} - -interface SlackMessageChangedEvent { - type: "event_callback"; - team_id?: string; - event: { - type: "message"; - subtype: "message_changed"; - channel: string; - channel_type?: string; - message: { - bot_id?: string; - files?: SlackEditedMessageFile[]; - source_team?: string; - text?: string; - ts: string; - thread_ts?: string; - user?: string; - user_team?: string; - }; - previous_message: { - text?: string; - }; - }; -} - -interface SlackEditedMessageFile { - mimetype?: string; - name?: string; - original_h?: number; - original_w?: number; - size?: number; - url_private?: string; - url_private_download?: string; -} - -export function isMessageChangedEnvelope( - value: unknown, -): value is SlackMessageChangedEvent { - if (!value || typeof value !== "object") return false; - const v = value as Record; - if (v.type !== "event_callback") return false; - const event = v.event as Record | undefined; - if (!event || typeof event !== "object") return false; - return ( - event.type === "message" && - event.subtype === "message_changed" && - typeof event.channel === "string" && - typeof event.message === "object" && - event.message !== null && - typeof event.previous_message === "object" && - event.previous_message !== null - ); -} - -function getAttachmentType(mimeType: string | undefined): Attachment["type"] { - if (mimeType?.startsWith("image/")) { - return "image"; - } - if (mimeType?.startsWith("video/")) { - return "video"; - } - if (mimeType?.startsWith("audio/")) { - return "audio"; - } - return "file"; -} - -function extractEditedMessageAttachments( - files: SlackEditedMessageFile[] | undefined, -): Attachment[] { - if (!files || files.length === 0) { - return []; - } - - return files.map((file) => ({ - type: getAttachmentType(file.mimetype), - url: file.url_private_download ?? file.url_private, - name: file.name, - mimeType: file.mimetype, - size: file.size, - width: file.original_w, - height: file.original_h, - })); -} - -/** - * Inspect a raw parsed Slack webhook body and extract a synthesized mention - * event when a `message_changed` edit newly adds the bot's @mention. - * - * Returns `null` when the payload is not a qualifying `message_changed` event. - */ -export function extractMessageChangedMention( - body: unknown, - botUserId: string, - adapter: Adapter, -): MessageChangedMention | null { - if (!isMessageChangedEnvelope(body)) return null; - - const { event } = body; - const newText = event.message.text ?? ""; - const prevText = event.previous_message.text ?? ""; - - // Only trigger when the bot mention is newly present in the edited message. - if (!textMentionsBot(newText, botUserId)) return null; - if (textMentionsBot(prevText, botUserId)) return null; - - const channelId = event.channel; - const messageTs = event.message.ts; - const threadTs = event.message.thread_ts ?? messageTs; - const userId = parseActorUserId(event.message.user); - if (!userId) return null; - const threadId = `slack:${channelId}:${threadTs}`; - const teamId = typeof body.team_id === "string" ? body.team_id : undefined; - - const userTeam = - typeof event.message.user_team === "string" - ? event.message.user_team - : undefined; - const sourceTeam = - typeof event.message.source_team === "string" - ? event.message.source_team - : undefined; - const botId = - typeof event.message.bot_id === "string" ? event.message.bot_id : undefined; - // Preserve the event's channel_type so visibility confirmation still works - // for channels whose only Junior traffic is edited mentions. - const channelType = - typeof event.channel_type === "string" ? event.channel_type : undefined; - - const raw: Record = { - channel: channelId, - ...(channelType ? { channel_type: channelType } : undefined), - ts: messageTs, - thread_ts: threadTs, - user: userId, - ...(teamId ? { team_id: teamId } : undefined), - ...(userTeam ? { user_team: userTeam } : undefined), - ...(sourceTeam ? { source_team: sourceTeam } : undefined), - ...(botId ? { bot_id: botId } : undefined), - }; - - const message = new Message({ - id: getEditedMentionMessageId(messageTs), - threadId, - text: newText, - isMention: true, - attachments: extractEditedMessageAttachments(event.message.files), - metadata: { dateSent: new Date(Number(messageTs) * 1000), edited: true }, - formatted: { type: "root" as const, children: [] }, - raw, - author: { - userId, - // Raw message_changed payloads do not include profile fields. - userName: "", - fullName: "", - // Mirror the fresh-message parse so the shared ingress author gate - // applies: synthesized flags must never launder a bot-authored or - // self-authored edit into a routable user message. - isBot: Boolean(botId), - isMe: userId === botUserId, - }, - }); - - Object.defineProperty(message, "adapter", { - configurable: true, - enumerable: false, - value: adapter, - writable: true, - }); - - return { threadId, message }; -} diff --git a/packages/junior/src/chat/ingress/slack-webhook.ts b/packages/junior/src/chat/ingress/slack-webhook.ts index 43fee26636..1304f7142e 100644 --- a/packages/junior/src/chat/ingress/slack-webhook.ts +++ b/packages/junior/src/chat/ingress/slack-webhook.ts @@ -34,10 +34,6 @@ import { import { coerceThreadConversationState } from "@/chat/state/conversation"; import { parseContent } from "@/chat/slack/message/content"; import { stopSlackThread } from "@/chat/slack/thread-stop"; -import { - extractMessageChangedMention, - isMessageChangedEnvelope, -} from "@/chat/ingress/message-changed"; import { normalizeIncomingSlackThreadId, withNormalizedThreadId, @@ -102,6 +98,8 @@ function slackEventLogContext( } const IGNORED_MESSAGE_SUBTYPES = new Set([ + // Conversation Messages are immutable once accepted. An edit must not + // rewrite input or start another Turn. Send a new Slack Message instead. "message_changed", "message_deleted", "message_replied", @@ -454,50 +452,6 @@ async function routeParsedMessage(args: { }); } -async function handleMessageChanged(args: { - adapter: SlackAdapter; - body: unknown; - installation: SlackInstallationContext; - queue: ConversationWorkQueue; - conversationStore?: ConversationStore; - receivedAtMs: number; - state: StateAdapter; -}): Promise { - if (!isMessageChangedEnvelope(args.body)) { - return false; - } - const botUserId = args.adapter.botUserId; - if (!botUserId) { - // Entry classification requires resolved bot identity; degrading into - // silently dropped edited-mention events would hide the outage. Throwing - // makes the webhook return a retryable non-2xx so Slack redelivers. - throw new Error( - "Slack bot identity is unresolved; cannot classify message_changed event", - ); - } - - const result = extractMessageChangedMention( - args.body, - botUserId, - args.adapter, - ); - if (!result || shouldIgnoreMessage(result.message)) { - return true; - } - - await persistSlackMessage({ - adapter: args.adapter, - installation: args.installation, - message: result.message, - conversationStore: args.conversationStore, - queue: args.queue, - receivedAtMs: args.receivedAtMs, - route: "mention", - state: args.state, - }); - return true; -} - async function handleSlackEvent(args: { body: SlackEventEnvelope; services: SlackWebhookServices; @@ -533,20 +487,6 @@ async function handleSlackEvent(args: { installation, state, task: async () => { - if ( - await handleMessageChanged({ - adapter, - body: args.body, - installation, - conversationStore: args.services.conversationStore, - queue: args.services.queue, - receivedAtMs, - state, - }) - ) { - return; - } - if (event.type === "assistant_thread_started") { const assistantThread = (event as Record) .assistant_thread as diff --git a/packages/junior/src/handlers/webhooks.ts b/packages/junior/src/handlers/webhooks.ts index 6f1c053668..20d7e3a7a7 100644 --- a/packages/junior/src/handlers/webhooks.ts +++ b/packages/junior/src/handlers/webhooks.ts @@ -1,10 +1,5 @@ import type { SlackAdapter } from "@chat-adapter/slack"; import { JuniorChat } from "@/chat/ingress/junior-chat"; -import { - extractMessageChangedMention, - isMessageChangedEnvelope, -} from "@/chat/ingress/message-changed"; -import { rehydrateAttachmentFetchers } from "@/chat/slack/attachment-fetchers"; import { runWithWorkspaceTeamId } from "@/chat/ingress/workspace-membership"; import { createRequestContext, @@ -17,20 +12,6 @@ import { } from "@/chat/logging"; import type { WaitUntilFn } from "@/handlers/types"; -interface SlackWebhookAuthAdapter { - botUserId?: string; - defaultBotTokenProvider?: () => string | Promise; - requestContext?: { - run(context: unknown, fn: () => T): T; - }; - resolveTokenForTeam?: (teamId: string) => Promise; - verifySignature: ( - body: string, - timestamp: string | null, - signature: string | null, - ) => boolean; -} - type ChatSdkBot = JuniorChat<{ slack: SlackAdapter }>; type WebhookRunner = () => Promise; @@ -44,75 +25,6 @@ function getSlackPayloadTeamId(body: unknown): string | undefined { return typeof teamId === "string" && teamId.length > 0 ? teamId : undefined; } -async function handleAuthenticatedSlackMessageChangedMention(args: { - body: unknown; - bot: ChatSdkBot; - rawBody: string; - request: Request; - waitUntil: WaitUntilFn; -}): Promise { - const slackAdapter = args.bot.getAdapter("slack"); - // @ts-expect-error non-overlapping boundary cast; rule forbids as-unknown-as chains - const authAdapter = slackAdapter as SlackWebhookAuthAdapter; - const timestamp = args.request.headers.get("x-slack-request-timestamp"); - const signature = args.request.headers.get("x-slack-signature"); - - if (!authAdapter.verifySignature(args.rawBody, timestamp, signature)) { - return; - } - - await args.bot.initialize(); - - const webhookOptions = { - waitUntil: (task: Promise) => args.waitUntil(task), - }; - const dispatch = () => { - const botUserId = authAdapter.botUserId; - if (!botUserId) { - return false; - } - - const result = extractMessageChangedMention( - args.body, - botUserId, - slackAdapter, - ); - if (!result) { - return false; - } - - rehydrateAttachmentFetchers(result.message); - args.bot.processMessage( - slackAdapter, - result.threadId, - result.message, - webhookOptions, - ); - return true; - }; - - if (authAdapter.defaultBotTokenProvider) { - dispatch(); - return; - } - - const teamId = getSlackPayloadTeamId(args.body); - if ( - !teamId || - !authAdapter.resolveTokenForTeam || - !authAdapter.requestContext - ) { - return; - } - - const context = await authAdapter.resolveTokenForTeam(teamId); - if (!context) { - return; - } - - authAdapter.requestContext.run(context, dispatch); -} - async function handleChatSdkWebhook(args: { bot: ChatSdkBot; platform: string; @@ -128,22 +40,12 @@ async function handleChatSdkWebhook(args: { let request = args.request; let slackWorkspaceTeamId: string | undefined; if (args.platform === "slack") { + // Do not intercept message_changed events. Slack edits cannot create or + // change a Conversation Message or Turn after the original event. const rawBody = await args.request.text(); const parsedBody = parseJson(rawBody); slackWorkspaceTeamId = getSlackPayloadTeamId(parsedBody); - if (parsedBody && isMessageChangedEnvelope(parsedBody)) { - await runWithWorkspaceTeamId(slackWorkspaceTeamId, () => - handleAuthenticatedSlackMessageChangedMention({ - body: parsedBody, - bot: args.bot, - rawBody, - request: args.request, - waitUntil: args.waitUntil, - }), - ); - } - request = new Request(args.request.url, { method: args.request.method, headers: args.request.headers, diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index 4d6154a8ac..0de2b399ce 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -394,42 +394,38 @@ describe("Slack conversation work execution", () => { ).resolves.toBeUndefined(); }); - it("routes edited Slack mentions through the durable mailbox", async () => { + it("does not route edited Slack messages", async () => { const queue = createConversationWorkQueueTestAdapter(); const state = getStateAdapter(); await state.connect(); - const slackAdapter = createSlackAdapterFixture(); - const editedTs = "1712345.0003"; - const editedText = `<@${SLACK_BOT_USER_ID}> edited ask`; const response = await handleSlackWebhookAndFlush({ request: slackWebhookRequest({ ...slackEnvelope({ eventType: "message", text: "edited ask", - ts: editedTs, + ts: "1712345.0003", }), event: { type: "message", subtype: "message_changed", channel: "C123", - hidden: true, message: { type: "message", user: "U123", - text: editedText, - ts: editedTs, + text: `<@${SLACK_BOT_USER_ID}> edited ask`, + ts: "1712345.0003", }, previous_message: { type: "message", user: "U123", text: "edited ask", - ts: editedTs, + ts: "1712345.0003", }, }, }), services: { - getSlackAdapter: () => slackAdapter, + getSlackAdapter: createSlackAdapterFixture, queue, runtime: createNoopSlackWebhookRuntime(), state, @@ -437,36 +433,7 @@ describe("Slack conversation work execution", () => { }); expect(response.status).toBe(200); - expect(queue.sentRecords()).toEqual([ - expect.objectContaining({ - conversationId: `slack:C123:${editedTs}`, - idempotencyKey: `slack:T123:slack:C123:${editedTs}:${editedTs}:message_changed_mention`, - }), - ]); - - const calls: Array<{ message: Message; thread: Thread }> = []; - await expect( - processNextQueuedSlackWork({ - getSlackAdapter: () => slackAdapter, - queue, - runtime: { - handleNewMention: async (thread, message, hooks) => { - await hooks.ack?.(); - calls.push({ thread, message }); - }, - handleSubscribedMessage: async () => { - throw new Error("unexpected subscribed route"); - }, - }, - state, - }), - ).resolves.toEqual({ status: "completed" }); - - expect(calls).toHaveLength(1); - expect(calls[0]?.thread.id).toBe(`slack:C123:${editedTs}`); - expect(calls[0]?.message.id).toBe(`${editedTs}:message_changed_mention`); - expect(calls[0]?.message.text).toBe(editedText); - expect(calls[0]?.message.isMention).toBe(true); + expect(queue.sentRecords()).toEqual([]); }); it("runs queued Slack mailbox work through the Slack runtime", async () => { diff --git a/packages/junior/tests/integration/slack/message-changed-behavior.test.ts b/packages/junior/tests/integration/slack/message-changed-behavior.test.ts index fca7ea6fb5..98c902bdec 100644 --- a/packages/junior/tests/integration/slack/message-changed-behavior.test.ts +++ b/packages/junior/tests/integration/slack/message-changed-behavior.test.ts @@ -1,27 +1,20 @@ -import { http, HttpResponse } from "msw"; -import { afterEach, describe, expect, it } from "vitest"; +import { describe, expect, it } from "vitest"; import { createMemoryState } from "@chat-adapter/state-memory"; import type { Message } from "chat"; import { slackEventsApiEnvelope } from "../../fixtures/slack/factories/events"; import { createSlackWebhookTestClient } from "../../fixtures/slack/webhook-client"; -import { mswServer } from "../../msw/server"; import { JuniorChat } from "@/chat/ingress/junior-chat"; import { createJuniorSlackAdapter } from "@/chat/slack/adapter"; import { handleChatSdkPlatformWebhook } from "@/handlers/webhooks"; const SIGNING_SECRET = "test-signing-secret"; const BOT_USER_ID = "U0BOT"; -const ORIGINAL_ENV = { ...process.env }; const slackWebhookClient = createSlackWebhookTestClient({ signingSecret: SIGNING_SECRET, }); describe("Slack behavior: message_changed webhook ingress", () => { - afterEach(() => { - process.env = { ...ORIGINAL_ENV }; - }); - - it("processes an edited DM mention after the original DM was already delivered", async () => { + it("ignores an edit that adds a Junior mention", async () => { const bot = new JuniorChat({ userName: "junior", adapters: { @@ -33,42 +26,21 @@ describe("Slack behavior: message_changed webhook ingress", () => { }, state: createMemoryState(), }); - const handledMessages: Array< - Pick - > = []; + const handledMessages: Array> = []; const waitUntil = slackWebhookClient.waitUntil(); bot.onDirectMessage(async (_thread, message) => { - handledMessages.push({ - id: message.id, - text: message.text, - isMention: message.isMention, - raw: message.raw, - }); + handledMessages.push({ id: message.id, text: message.text }); }); - const originalResponse = await handleChatSdkPlatformWebhook( - slackWebhookClient.event( - slackEventsApiEnvelope({ - eventType: "message", - channel: "D12345", - ts: "1700000100.000100", - text: "hello there", - }), - ), - "slack", - waitUntil.fn, - bot, - ); - await waitUntil.flush(); - - const editedPayload = { - ...slackEventsApiEnvelope({ - eventType: "message", - channel: "D12345", - ts: "1700000100.000100", - text: "hello there", - }), + const original = slackEventsApiEnvelope({ + eventType: "message", + channel: "D12345", + ts: "1700000100.000100", + text: "hello there", + }); + const edit = { + ...original, event: { type: "message", subtype: "message_changed", @@ -89,185 +61,22 @@ describe("Slack behavior: message_changed webhook ingress", () => { }, }; - const editedResponse = await handleChatSdkPlatformWebhook( - slackWebhookClient.event(editedPayload), + await handleChatSdkPlatformWebhook( + slackWebhookClient.event(original), "slack", waitUntil.fn, bot, ); - await waitUntil.flush(); - - expect(originalResponse.status).toBe(200); - expect(editedResponse.status).toBe(200); - expect(handledMessages).toHaveLength(2); - expect(handledMessages[0]).toMatchObject({ - id: "1700000100.000100", - text: "hello there", - isMention: false, - }); - expect(handledMessages[1]).toMatchObject({ - id: "1700000100.000100:message_changed_mention", - text: `<@${BOT_USER_ID}> hello there`, - isMention: true, - }); - const editedMessage = handledMessages[1]; - expect(editedMessage).toBeDefined(); - if (!editedMessage) { - throw new Error("expected edited message to be handled"); - } - expect((editedMessage.raw as { ts?: string }).ts).toBe("1700000100.000100"); - }); - - it("preserves edited-message image attachments through the webhook and adapter path", async () => { - mswServer.use( - http.get("https://files.slack.com/private/edited.png", async () => { - return new HttpResponse(Buffer.from("image-bytes"), { - headers: { - "content-type": "image/png", - }, - }); - }), - ); - - const state = createMemoryState(); - await state.connect(); - const bot = new JuniorChat({ - userName: "junior", - adapters: { - slack: createJuniorSlackAdapter({ - botToken: "xoxb-test", - botUserId: BOT_USER_ID, - signingSecret: SIGNING_SECRET, - }), - }, - state, - }); - const handledMessages: Array< - Pick - > = []; - - bot.onDirectMessage(async (_thread, message) => { - handledMessages.push({ - id: message.id, - text: message.text, - isMention: message.isMention, - attachments: message.attachments, - }); - }); - - const waitUntil = slackWebhookClient.waitUntil(); - const editedPayload = { - ...slackEventsApiEnvelope({ - eventType: "message", - channel: "D12345", - ts: "1700000100.000102", - text: "hello there", - }), - event: { - type: "message", - subtype: "message_changed", - channel: "D12345", - hidden: true, - message: { - type: "message", - user: "U123", - text: `<@${BOT_USER_ID}> what is in this screenshot?`, - ts: "1700000100.000102", - files: [ - { - id: "F_EDITED", - mimetype: "image/png", - name: "edited.png", - size: 11, - url_private: "https://files.slack.com/private/edited.png", - }, - ], - }, - previous_message: { - type: "message", - user: "U123", - text: "what is in this screenshot?", - ts: "1700000100.000102", - }, - }, - }; - - const response = await handleChatSdkPlatformWebhook( - slackWebhookClient.event(editedPayload), + await handleChatSdkPlatformWebhook( + slackWebhookClient.event(edit), "slack", waitUntil.fn, bot, ); await waitUntil.flush(); - expect(response.status).toBe(200); - expect(handledMessages).toHaveLength(1); - const editedMessage = handledMessages[0]; - expect(editedMessage).toMatchObject({ - id: "1700000100.000102:message_changed_mention", - text: `<@${BOT_USER_ID}> what is in this screenshot?`, - isMention: true, - }); - expect(editedMessage?.attachments).toEqual([ - expect.objectContaining({ - type: "image", - name: "edited.png", - mimeType: "image/png", - url: "https://files.slack.com/private/edited.png", - }), + expect(handledMessages).toEqual([ + { id: "1700000100.000100", text: "hello there" }, ]); - const imageData = await editedMessage?.attachments[0]?.fetchData?.(); - expect(imageData?.toString()).toBe("image-bytes"); - }); - - it("rejects forged edited mentions before any bot handler runs", async () => { - const bot = new JuniorChat({ - userName: "junior", - adapters: { - slack: createJuniorSlackAdapter({ - botToken: "xoxb-test", - botUserId: BOT_USER_ID, - signingSecret: SIGNING_SECRET, - }), - }, - state: createMemoryState(), - }); - const handledMessages: Message[] = []; - - bot.onDirectMessage(async (_thread, message) => { - handledMessages.push(message); - }); - - const payload = { - ...slackEventsApiEnvelope({ - eventType: "message", - channel: "D12345", - ts: "1700000100.000100", - text: "hello there", - }), - event: { - type: "message", - subtype: "message_changed", - channel: "D12345", - message: { - text: `<@${BOT_USER_ID}> hello there`, - ts: "1700000100.000100", - user: "U123", - }, - previous_message: { - text: "hello there", - }, - }, - }; - - const response = await handleChatSdkPlatformWebhook( - slackWebhookClient.invalidSignature(payload), - "slack", - () => undefined, - bot, - ); - - expect(response.status).toBe(401); - expect(handledMessages).toHaveLength(0); }); }); diff --git a/packages/junior/tests/integration/slack/message-changed-gate-contract.test.ts b/packages/junior/tests/integration/slack/message-changed-gate-contract.test.ts deleted file mode 100644 index f297562222..0000000000 --- a/packages/junior/tests/integration/slack/message-changed-gate-contract.test.ts +++ /dev/null @@ -1,97 +0,0 @@ -import { afterEach, describe, expect, it } from "vitest"; -import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter"; -import { - SLACK_BOT_USER_ID, - createConversationWorkQueueTestAdapter, - createNoopSlackWebhookRuntime, - createSlackAdapterFixture, - handleSlackWebhookAndFlush, - slackWebhookRequest, -} from "../../fixtures/conversation-work"; - -const EDITED_TS = "1712345.0042"; - -function messageChangedEnvelope(message: { - bot_id?: string; - user: string; - user_team?: string; -}) { - const editedText = `<@${SLACK_BOT_USER_ID}> edited ask`; - return { - team_id: "T123", - type: "event_callback", - event: { - type: "message", - subtype: "message_changed", - channel: "C123", - hidden: true, - message: { - type: "message", - text: editedText, - ts: EDITED_TS, - ...message, - }, - previous_message: { - type: "message", - user: message.user, - text: "edited ask", - ts: EDITED_TS, - }, - }, - }; -} - -describe("Slack message_changed author gate contract", () => { - afterEach(async () => { - await disconnectStateAdapter(); - }); - - async function runEditedMentionWebhook(envelope: unknown) { - const queue = createConversationWorkQueueTestAdapter(); - const state = getStateAdapter(); - await state.connect(); - const slackAdapter = createSlackAdapterFixture(); - - const response = await handleSlackWebhookAndFlush({ - request: slackWebhookRequest(envelope), - services: { - getSlackAdapter: () => slackAdapter, - queue, - runtime: createNoopSlackWebhookRuntime(), - state, - }, - }); - return { queue, response }; - } - - it("drops an edited mention from a Slack Connect external user", async () => { - const { queue, response } = await runEditedMentionWebhook( - messageChangedEnvelope({ user: "U0EXTERNAL", user_team: "T0OTHERORG" }), - ); - - expect(response.status).toBe(200); - expect(queue.sentRecords()).toEqual([]); - }); - - it("drops a bot-authored edit that adds a mention", async () => { - const { queue, response } = await runEditedMentionWebhook( - messageChangedEnvelope({ bot_id: "B_JUNIOR", user: SLACK_BOT_USER_ID }), - ); - - expect(response.status).toBe(200); - expect(queue.sentRecords()).toEqual([]); - }); - - it("routes an edited mention from a same-workspace user", async () => { - const { queue, response } = await runEditedMentionWebhook( - messageChangedEnvelope({ user: "U123", user_team: "T123" }), - ); - - expect(response.status).toBe(200); - expect(queue.sentRecords()).toEqual([ - expect.objectContaining({ - conversationId: `slack:C123:${EDITED_TS}`, - }), - ]); - }); -}); diff --git a/packages/junior/tests/integration/slack/message-changed-reply-contract.test.ts b/packages/junior/tests/integration/slack/message-changed-reply-contract.test.ts deleted file mode 100644 index 2985304823..0000000000 --- a/packages/junior/tests/integration/slack/message-changed-reply-contract.test.ts +++ /dev/null @@ -1,225 +0,0 @@ -import { createTestDestination } from "../../fixtures/slack-harness"; -import { describe, expect, it } from "vitest"; -import { createMemoryState } from "@chat-adapter/state-memory"; -import type { SlackAdapter } from "@chat-adapter/slack"; -import { slackEventsApiEnvelope } from "../../fixtures/slack/factories/events"; -import { slackApiOutbox } from "../../fixtures/slack-api-outbox"; -import { createSlackWebhookTestClient } from "../../fixtures/slack/webhook-client"; -import { createSlackRuntime } from "@/chat/app/factory"; -import { JuniorChat } from "@/chat/ingress/junior-chat"; -import type { AgentRunner } from "@/chat/runtime/agent-runner"; -import { createJuniorSlackAdapter } from "@/chat/slack/adapter"; -import { handleChatSdkPlatformWebhook } from "@/handlers/webhooks"; -import { createModelAgentRunner } from "../../fixtures/agent-runner"; -import { createModelStream } from "../../fixtures/model-stream"; -import { queueSlackApiError } from "../../msw/handlers/slack-api"; -import { - createPausedTurns, - getPausedTurnRequest, -} from "@/chat/task-execution/turn-wake"; -import { buildDeterministicTurnId } from "@/chat/runtime/turn"; -import { - createConversationWorkQueueTestAdapter, - type ConversationWorkQueueTestAdapter, -} from "../../fixtures/conversation-work"; - -const SIGNING_SECRET = "test-signing-secret"; -const BOT_USER_ID = "U0BOT"; -const slackWebhookClient = createSlackWebhookTestClient({ - signingSecret: SIGNING_SECRET, -}); - -function createEditedMentionRequest(args: { - messageTs: string; - newText: string; - prevText: string; -}): Request { - return slackWebhookClient.event({ - ...slackEventsApiEnvelope({ - eventType: "message", - channel: "D12345", - ts: args.messageTs, - text: args.prevText, - }), - event: { - type: "message", - subtype: "message_changed", - channel: "D12345", - hidden: true, - message: { - type: "message", - user: "U123", - text: args.newText, - ts: args.messageTs, - }, - previous_message: { - type: "message", - user: "U123", - text: args.prevText, - ts: args.messageTs, - }, - }, - }); -} - -async function createEditedDmBot(args: { - agentRunner: AgentRunner; - queue?: ConversationWorkQueueTestAdapter; -}) { - const state = createMemoryState(); - await state.connect(); - const bot = new JuniorChat<{ slack: SlackAdapter }>({ - userName: "junior", - adapters: { - slack: createJuniorSlackAdapter({ - botToken: "xoxb-test", - botUserId: BOT_USER_ID, - signingSecret: SIGNING_SECRET, - }), - }, - state, - }); - const slackRuntime = createSlackRuntime({ - getSlackAdapter: () => bot.getAdapter("slack"), - ...(args.queue - ? { pausedTurns: createPausedTurns({ queue: args.queue, state }) } - : undefined), - services: { - agentRunner: args.agentRunner, - }, - }); - - bot.onDirectMessage((thread, message) => - slackRuntime.handleNewMention(thread, message, { - destination: createTestDestination(thread), - }), - ); - - return bot; -} - -describe("Slack contract: edited-message reply delivery", () => { - it("posts the finalized reply into the edited DM thread with chat.postMessage", async () => { - const bot = await createEditedDmBot({ - agentRunner: createModelAgentRunner( - createModelStream([{ type: "text", text: "Hello world" }]), - ), - }); - const waitUntil = slackWebhookClient.waitUntil(); - - const response = await handleChatSdkPlatformWebhook( - createEditedMentionRequest({ - messageTs: "1700000100.000100", - newText: `<@${BOT_USER_ID}> hello there`, - prevText: "hello there", - }), - "slack", - waitUntil.fn, - bot, - ); - await waitUntil.flush(); - - expect(response.status).toBe(200); - expect(slackApiOutbox.messages()).toEqual([ - expect.objectContaining({ - params: expect.objectContaining({ - channel: "D12345", - thread_ts: "1700000100.000100", - text: "Hello world", - }), - }), - ]); - }); - - it("posts continuation messages with chat.postMessage when a completed message overflows", async () => { - const longReply = Array.from( - { length: 80 }, - (_, i) => `line ${i + 1}`, - ).join("\n"); - const bot = await createEditedDmBot({ - agentRunner: createModelAgentRunner( - createModelStream([{ type: "text", text: longReply }]), - ), - }); - const waitUntil = slackWebhookClient.waitUntil(); - - const response = await handleChatSdkPlatformWebhook( - createEditedMentionRequest({ - messageTs: "1700000100.000101", - newText: `<@${BOT_USER_ID}> hello there`, - prevText: "hello there", - }), - "slack", - waitUntil.fn, - bot, - ); - await waitUntil.flush(); - - expect(response.status).toBe(200); - const postCalls = slackApiOutbox.messages(); - expect(postCalls.length).toBeGreaterThan(1); - expect(postCalls[0]).toEqual( - expect.objectContaining({ - params: expect.objectContaining({ - channel: "D12345", - thread_ts: "1700000100.000101", - }), - }), - ); - }); - - it("wakes a suspended turn with its Slack destination after a transient delivery failure", async () => { - for (let attempt = 0; attempt < 3; attempt += 1) { - queueSlackApiError("chat.postMessage", { - error: "internal_error", - status: 503, - }); - } - const queue = createConversationWorkQueueTestAdapter(); - const bot = await createEditedDmBot({ - agentRunner: createModelAgentRunner( - createModelStream([{ type: "text", text: "Hello world" }]), - ), - queue, - }); - const waitUntil = slackWebhookClient.waitUntil(); - - const response = await handleChatSdkPlatformWebhook( - createEditedMentionRequest({ - messageTs: "1700000100.000102", - newText: `<@${BOT_USER_ID}> hello there`, - prevText: "hello there", - }), - "slack", - waitUntil.fn, - bot, - ); - await waitUntil.flush(); - - expect(response.status).toBe(200); - const conversationId = "slack:D12345:1700000100.000102"; - const turnId = buildDeterministicTurnId( - "1700000100.000102:message_changed_mention", - ); - await expect( - getPausedTurnRequest({ conversationId, turnId }), - ).resolves.toMatchObject({ - conversationId, - destination: { - platform: "slack", - teamId: "TTEST", - channelId: "D12345", - }, - expectedVersion: 2, - turnId, - }); - expect(queue.sentRecords()).toEqual([ - expect.objectContaining({ - conversationId, - idempotencyKey: expect.stringContaining( - `agent-continue:${conversationId}:${turnId}:2:`, - ), - }), - ]); - }); -}); diff --git a/packages/junior/tests/unit/slack/message-changed-ingress.test.ts b/packages/junior/tests/unit/slack/message-changed-ingress.test.ts deleted file mode 100644 index afe9cb545e..0000000000 --- a/packages/junior/tests/unit/slack/message-changed-ingress.test.ts +++ /dev/null @@ -1,190 +0,0 @@ -import { describe, expect, it } from "vitest"; -import type { Adapter } from "chat"; -import { extractMessageChangedMention } from "@/chat/ingress/message-changed"; - -const BOT_USER_ID = "U0BOTTEST"; -const CHANNEL_ID = "C0CHAN"; -const TEAM_ID = "T0TEAM"; -const MESSAGE_TS = "1700000100.000"; -const THREAD_TS = "1700000000.000"; -const EDITED_MESSAGE_ID = `${MESSAGE_TS}:message_changed_mention`; - -const fakeAdapter = {} as Adapter; - -function makeEnvelope(overrides: { - newText: string; - prevText: string; - botId?: string; - channel?: string; - messageTs?: string; - threadTs?: string; - user?: string; -}): unknown { - return { - type: "event_callback", - team_id: TEAM_ID, - event: { - type: "message", - subtype: "message_changed", - channel: overrides.channel ?? CHANNEL_ID, - message: { - text: overrides.newText, - ts: overrides.messageTs ?? MESSAGE_TS, - thread_ts: overrides.threadTs ?? THREAD_TS, - user: overrides.user ?? "U0SENDER", - ...(overrides.botId ? { bot_id: overrides.botId } : undefined), - }, - previous_message: { - text: overrides.prevText, - }, - }, - }; -} - -describe("extractMessageChangedMention", () => { - it("returns mention when bot mention is newly added in edited message", () => { - const body = makeEnvelope({ - newText: `<@${BOT_USER_ID}> please help`, - prevText: "please help", - }); - - const result = extractMessageChangedMention(body, BOT_USER_ID, fakeAdapter); - - expect(result).not.toBeNull(); - if (!result) { - throw new Error("expected synthesized edited mention"); - } - expect(result?.threadId).toBe(`slack:${CHANNEL_ID}:${THREAD_TS}`); - expect(result?.message.text).toBe(`<@${BOT_USER_ID}> please help`); - expect(result?.message.isMention).toBe(true); - expect(result?.message.id).toBe(EDITED_MESSAGE_ID); - expect((result.message.raw as { ts: string }).ts).toBe(MESSAGE_TS); - expect((result.message.metadata as { edited: boolean }).edited).toBe(true); - }); - - it("serializes the synthesized message for queue rehydration", () => { - const body = makeEnvelope({ - newText: `<@${BOT_USER_ID}> please help`, - prevText: "please help", - }); - - const result = extractMessageChangedMention(body, BOT_USER_ID, fakeAdapter); - - const serialized = result?.message.toJSON(); - - expect(serialized).toMatchObject({ - _type: "chat:Message", - attachments: [], - author: { - userId: "U0SENDER", - isBot: false, - isMe: false, - }, - formatted: { type: "root", children: [] }, - id: EDITED_MESSAGE_ID, - isMention: true, - links: undefined, - metadata: { - dateSent: new Date(Number(MESSAGE_TS) * 1000).toISOString(), - edited: true, - editedAt: undefined, - }, - raw: { - channel: CHANNEL_ID, - team_id: TEAM_ID, - ts: MESSAGE_TS, - thread_ts: THREAD_TS, - user: "U0SENDER", - }, - text: `<@${BOT_USER_ID}> please help`, - threadId: `slack:${CHANNEL_ID}:${THREAD_TS}`, - }); - expect(serialized?.author.userName).toBe(""); - expect(serialized?.author.fullName).toBe(""); - }); - - it("derives bot author flags from the edited payload", () => { - const body = makeEnvelope({ - newText: `<@${BOT_USER_ID}> please help`, - prevText: "please help", - botId: "B_APP", - }); - - const result = extractMessageChangedMention(body, BOT_USER_ID, fakeAdapter); - - expect(result?.message.author.isBot).toBe(true); - expect(result?.message.author.isMe).toBe(false); - expect( - (result?.message.raw as { bot_id?: string } | undefined)?.bot_id, - ).toBe("B_APP"); - }); - - it("marks self-authored edits with isMe", () => { - const body = makeEnvelope({ - newText: `<@${BOT_USER_ID}> please help`, - prevText: "please help", - user: BOT_USER_ID, - }); - - const result = extractMessageChangedMention(body, BOT_USER_ID, fakeAdapter); - - expect(result?.message.author.isMe).toBe(true); - }); - - it("returns null when bot mention was already in the previous message", () => { - const body = makeEnvelope({ - newText: `<@${BOT_USER_ID}> please help with more detail`, - prevText: `<@${BOT_USER_ID}> please help`, - }); - - const result = extractMessageChangedMention(body, BOT_USER_ID, fakeAdapter); - expect(result).toBeNull(); - }); - - it("returns null when the edited message has no actor user id", () => { - const body = makeEnvelope({ - newText: `<@${BOT_USER_ID}> please help`, - prevText: "please help", - user: "", - }); - - const result = extractMessageChangedMention(body, BOT_USER_ID, fakeAdapter); - - expect(result).toBeNull(); - }); - - it("returns null when the edited message has a synthetic unknown actor id", () => { - const body = makeEnvelope({ - newText: `<@${BOT_USER_ID}> please help`, - prevText: "please help", - user: "unknown", - }); - - const result = extractMessageChangedMention(body, BOT_USER_ID, fakeAdapter); - - expect(result).toBeNull(); - }); - - it("uses message ts as thread_ts fallback when thread_ts is absent", () => { - const body = { - type: "event_callback", - event: { - type: "message", - subtype: "message_changed", - channel: CHANNEL_ID, - message: { - text: `<@${BOT_USER_ID}> help`, - ts: MESSAGE_TS, - // no thread_ts - user: "U0SENDER", - }, - previous_message: { - text: "help", - }, - }, - }; - - const result = extractMessageChangedMention(body, BOT_USER_ID, fakeAdapter); - expect(result?.threadId).toBe(`slack:${CHANNEL_ID}:${MESSAGE_TS}`); - }); -}); From 02311c457fea8ec5dca58acbcdeda4bbfa8a5889 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 17 Sep 2026 15:32:17 +0000 Subject: [PATCH 7/9] test(evals): distinguish event watches Allow ordinary English that says an event automation watches a resource. Keep rejecting claims that the temporary watchEvents tool or a polling schedule was created. Co-Authored-By: David Cramer --- .../evals/integration/event-automations/management.eval.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/junior-evals/evals/integration/event-automations/management.eval.ts b/packages/junior-evals/evals/integration/event-automations/management.eval.ts index 8f528c3c58..57d4ccab4a 100644 --- a/packages/junior-evals/evals/integration/event-automations/management.eval.ts +++ b/packages/junior-evals/evals/integration/event-automations/management.eval.ts @@ -161,7 +161,7 @@ describeEval("Event automation management", slackEvals, (it) => { ], fail: [ "Do not create separate tasks for closed and reopened.", - "Do not claim a polling schedule, recurring timer, or watch was created.", + "Do not claim that `watchEvents`, a polling schedule, or a recurring timer was created instead of the event automation.", ], }), }); @@ -208,7 +208,7 @@ describeEval("Event automation management", slackEvals, (it) => { fail: [ "Do not narrow the task to one issue number.", "Do not create separate tasks for closed and reopened issues.", - "Do not claim a polling schedule or watch was created.", + "Do not claim that `watchEvents` or a polling schedule was created instead of the event automation.", ], }), }); From 24c669e4aa297731012db9709ca880e2aba04e4a Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 17 Sep 2026 15:44:18 +0000 Subject: [PATCH 8/9] refactor(chat): clarify input persistence Name the two facts directly: whether input still needs a durable copy, and whether a history branch must fail the progress write. Document why only direct Slack input needs this before acknowledgement. Co-Authored-By: David Cramer --- packages/junior/src/chat/agent/resume.ts | 6 +++--- packages/junior/src/chat/agent/types.ts | 4 ++-- packages/junior/src/chat/providers/slack/turn.ts | 4 +++- packages/junior/src/chat/task-execution/checkpoint.ts | 8 ++++---- .../tests/component/task-execution/checkpoint.test.ts | 2 +- 5 files changed, 13 insertions(+), 11 deletions(-) diff --git a/packages/junior/src/chat/agent/resume.ts b/packages/junior/src/chat/agent/resume.ts index e6b3873c01..2325f1eb3b 100644 --- a/packages/junior/src/chat/agent/resume.ts +++ b/packages/junior/src/chat/agent/resume.ts @@ -181,7 +181,7 @@ export function createResumeState(args: ResumeStateArgs) { async persistSafeBoundary( messages: PiMessage[], trailingMessageProvenance?: ConversationMessageProvenance[], - required = false, + rejectHistoryBranch = false, ): Promise { const saved = await saveTurnCheckpoint({ mode: "running", @@ -191,7 +191,7 @@ export function createResumeState(args: ResumeStateArgs) { trailingMessageProvenance, turnContexts: turnContexts.length > 0 ? turnContexts : undefined, turnStartMessageIndex, - required, + rejectHistoryBranch, }); if (!saved) { return false; @@ -214,7 +214,7 @@ export function createResumeState(args: ResumeStateArgs) { const persisted = await this.persistSafeBoundary( messages, trailingMessageProvenance, - args.durability.inputCheckpointRequired === true, + args.durability.inputNeedsPersistence === true, ); if (!persisted && args.durability.onInputCommitted) { throw new TurnInputCommitLostError( diff --git a/packages/junior/src/chat/agent/types.ts b/packages/junior/src/chat/agent/types.ts index 0b688f8a4f..5a61d55ce9 100644 --- a/packages/junior/src/chat/agent/types.ts +++ b/packages/junior/src/chat/agent/types.ts @@ -148,8 +148,8 @@ export class RetryableDeliveryError extends Error { export type AgentDurability = { /** Schedule delegated work with authority bound by the active parent run. */ spawnAgent?: SpawnAgent; - /** Reject a history branch before the current input enters model execution. */ - inputCheckpointRequired?: boolean; + /** True when the current input has no durable copy yet. */ + inputNeedsPersistence?: boolean; onInputCommitted?: () => void | Promise; /** Return true when the durable worker should pause at the next Pi boundary. */ shouldYield?: () => boolean; diff --git a/packages/junior/src/chat/providers/slack/turn.ts b/packages/junior/src/chat/providers/slack/turn.ts index 7079c69e60..3c706f8acf 100644 --- a/packages/junior/src/chat/providers/slack/turn.ts +++ b/packages/junior/src/chat/providers/slack/turn.ts @@ -1149,7 +1149,9 @@ export function createSlackTurn(deps: SlackTurnDeps) { ? undefined : { delivery: deliverAssistantMessage }), durability: { - inputCheckpointRequired: true, + // This direct Slack path has no mailbox copy. Store the input + // before ack so a failed Run cannot lose it. + inputNeedsPersistence: true, onInputCommitted: options.ack, drainSteeringMessages, shouldYield: options.shouldYield, diff --git a/packages/junior/src/chat/task-execution/checkpoint.ts b/packages/junior/src/chat/task-execution/checkpoint.ts index f365c60765..c9c4e355d6 100644 --- a/packages/junior/src/chat/task-execution/checkpoint.ts +++ b/packages/junior/src/chat/task-execution/checkpoint.ts @@ -85,8 +85,8 @@ interface TurnCheckpointWrite { turnStartMessageIndex?: number; /** Tool calls charged to this turn; survives history replacement. */ cumulativeToolCallCount?: number; - /** Reject a conflicting history write instead of treating it as best-effort. */ - required?: boolean; + /** Reject a history branch instead of skipping this progress write. */ + rejectHistoryBranch?: boolean; trailingMessageProvenance?: ConversationMessageProvenance[]; turnContexts?: PluginTurnContext[]; durationMs?: number; @@ -158,7 +158,7 @@ export async function loadTurnCheckpoint(args: { * Save turn progress. * * - `running` / `paused`: best-effort; returns the stored record or undefined - * - required `running`: rejects a conflicting history write + * - `running` with `rejectHistoryBranch`: rejects a history branch * - `completed` / `failed`: retries until write accepts; throws on hard failure */ export function saveTurnCheckpoint( @@ -222,7 +222,7 @@ async function saveRunning( } catch (error) { if ( error instanceof AgentHistoryBranchError && - args.required === true + args.rejectHistoryBranch === true ) { throw error; } diff --git a/packages/junior/tests/component/task-execution/checkpoint.test.ts b/packages/junior/tests/component/task-execution/checkpoint.test.ts index 481298e492..6bd9db3f39 100644 --- a/packages/junior/tests/component/task-execution/checkpoint.test.ts +++ b/packages/junior/tests/component/task-execution/checkpoint.test.ts @@ -1604,7 +1604,7 @@ describe("turn checkpoint", () => { turnId: "turn-stale-checkpoint", sliceId: 1, messages: [staleUser], - required: true, + rejectHistoryBranch: true, }), ).rejects.toThrow("changed before its committed boundary"); }); From ad54ea62b58e0e103d92aebcd0ba60374f2161a0 Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Thu, 17 Sep 2026 15:52:49 +0000 Subject: [PATCH 9/9] refactor(chat): remove checkpoint policy Keep the fix at Slack ingress by ignoring edits. Remove the special agent checkpoint behavior and its history-branch test changes because Slack input now follows the same durable mailbox contract as other work. Co-Authored-By: David Cramer --- packages/junior/src/chat/agent/resume.ts | 3 -- packages/junior/src/chat/agent/types.ts | 2 -- .../junior/src/chat/providers/slack/turn.ts | 3 -- .../src/chat/task-execution/checkpoint.ts | 23 ++++-------- .../task-execution/checkpoint.test.ts | 12 +++++-- .../slack/message-content-behavior.test.ts | 35 ++++++++----------- 6 files changed, 30 insertions(+), 48 deletions(-) diff --git a/packages/junior/src/chat/agent/resume.ts b/packages/junior/src/chat/agent/resume.ts index 2325f1eb3b..2e430d102d 100644 --- a/packages/junior/src/chat/agent/resume.ts +++ b/packages/junior/src/chat/agent/resume.ts @@ -181,7 +181,6 @@ export function createResumeState(args: ResumeStateArgs) { async persistSafeBoundary( messages: PiMessage[], trailingMessageProvenance?: ConversationMessageProvenance[], - rejectHistoryBranch = false, ): Promise { const saved = await saveTurnCheckpoint({ mode: "running", @@ -191,7 +190,6 @@ export function createResumeState(args: ResumeStateArgs) { trailingMessageProvenance, turnContexts: turnContexts.length > 0 ? turnContexts : undefined, turnStartMessageIndex, - rejectHistoryBranch, }); if (!saved) { return false; @@ -214,7 +212,6 @@ export function createResumeState(args: ResumeStateArgs) { const persisted = await this.persistSafeBoundary( messages, trailingMessageProvenance, - args.durability.inputNeedsPersistence === true, ); if (!persisted && args.durability.onInputCommitted) { throw new TurnInputCommitLostError( diff --git a/packages/junior/src/chat/agent/types.ts b/packages/junior/src/chat/agent/types.ts index 5a61d55ce9..ec5dd2e601 100644 --- a/packages/junior/src/chat/agent/types.ts +++ b/packages/junior/src/chat/agent/types.ts @@ -148,8 +148,6 @@ export class RetryableDeliveryError extends Error { export type AgentDurability = { /** Schedule delegated work with authority bound by the active parent run. */ spawnAgent?: SpawnAgent; - /** True when the current input has no durable copy yet. */ - inputNeedsPersistence?: boolean; onInputCommitted?: () => void | Promise; /** Return true when the durable worker should pause at the next Pi boundary. */ shouldYield?: () => boolean; diff --git a/packages/junior/src/chat/providers/slack/turn.ts b/packages/junior/src/chat/providers/slack/turn.ts index 3c706f8acf..bdd608f786 100644 --- a/packages/junior/src/chat/providers/slack/turn.ts +++ b/packages/junior/src/chat/providers/slack/turn.ts @@ -1149,9 +1149,6 @@ export function createSlackTurn(deps: SlackTurnDeps) { ? undefined : { delivery: deliverAssistantMessage }), durability: { - // This direct Slack path has no mailbox copy. Store the input - // before ack so a failed Run cannot lose it. - inputNeedsPersistence: true, onInputCommitted: options.ack, drainSteeringMessages, shouldYield: options.shouldYield, diff --git a/packages/junior/src/chat/task-execution/checkpoint.ts b/packages/junior/src/chat/task-execution/checkpoint.ts index c9c4e355d6..620e36c4a3 100644 --- a/packages/junior/src/chat/task-execution/checkpoint.ts +++ b/packages/junior/src/chat/task-execution/checkpoint.ts @@ -85,8 +85,6 @@ interface TurnCheckpointWrite { turnStartMessageIndex?: number; /** Tool calls charged to this turn; survives history replacement. */ cumulativeToolCallCount?: number; - /** Reject a history branch instead of skipping this progress write. */ - rejectHistoryBranch?: boolean; trailingMessageProvenance?: ConversationMessageProvenance[]; turnContexts?: PluginTurnContext[]; durationMs?: number; @@ -158,7 +156,6 @@ export async function loadTurnCheckpoint(args: { * Save turn progress. * * - `running` / `paused`: best-effort; returns the stored record or undefined - * - `running` with `rejectHistoryBranch`: rejects a history branch * - `completed` / `failed`: retries until write accepts; throws on hard failure */ export function saveTurnCheckpoint( @@ -220,20 +217,14 @@ async function saveRunning( state: "running", }); } catch (error) { - if ( - error instanceof AgentHistoryBranchError && - args.rejectHistoryBranch === true - ) { - throw error; - } - if (error instanceof AgentHistoryBranchError) { - return undefined; + // Quiet only branch races on best-effort running checkpoints. + if (!(error instanceof AgentHistoryBranchError)) { + logException(error, "agent.turn.checkpoint.running.failed", { + "app.ai.resume_conversation_id": args.conversationId, + "app.ai.resume_session_id": args.turnId, + "app.ai.resume_slice_id": args.sliceId, + }); } - logException(error, "agent.turn.checkpoint.running.failed", { - "app.ai.resume_conversation_id": args.conversationId, - "app.ai.resume_session_id": args.turnId, - "app.ai.resume_slice_id": args.sliceId, - }); return undefined; } } diff --git a/packages/junior/tests/component/task-execution/checkpoint.test.ts b/packages/junior/tests/component/task-execution/checkpoint.test.ts index 6bd9db3f39..d24556a3fb 100644 --- a/packages/junior/tests/component/task-execution/checkpoint.test.ts +++ b/packages/junior/tests/component/task-execution/checkpoint.test.ts @@ -1582,7 +1582,12 @@ describe("turn checkpoint", () => { ).resolves.toBeUndefined(); }); - it("surfaces true history branches to the owning Turn boundary", async () => { + it("rejects true history branches without reporting a running-session exception", async () => { + const logException = vi.fn(); + vi.doMock("@/chat/logging", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, logException }; + }); const { saveTurnCheckpoint } = await import("@/chat/task-execution/checkpoint"); const committedUser = userMessage("committed"); @@ -1604,9 +1609,10 @@ describe("turn checkpoint", () => { turnId: "turn-stale-checkpoint", sliceId: 1, messages: [staleUser], - rejectHistoryBranch: true, }), - ).rejects.toThrow("changed before its committed boundary"); + ).resolves.toBeUndefined(); + + expect(logException).not.toHaveBeenCalled(); }); it("appends after in-place assistant envelope mutations on committed messages", async () => { diff --git a/packages/junior/tests/integration/slack/message-content-behavior.test.ts b/packages/junior/tests/integration/slack/message-content-behavior.test.ts index 2e8c0bc4ba..c166e9d5ef 100644 --- a/packages/junior/tests/integration/slack/message-content-behavior.test.ts +++ b/packages/junior/tests/integration/slack/message-content-behavior.test.ts @@ -6,6 +6,7 @@ import { persistThreadState, persistThreadStateById, } from "@/chat/runtime/thread-state"; +import { TurnInputCommitLostError } from "@/chat/runtime/turn"; import { coerceThreadConversationState } from "@/chat/state/conversation"; import { disconnectStateAdapter } from "@/chat/state/adapter"; import { hydrateConversationMessages } from "@/chat/conversations/messages"; @@ -553,29 +554,21 @@ describe("Slack behavior: message content", () => { }, }); - const message = createTestMessage({ - id: "m-content-active-session-record", - text: "<@U0APP> continue", - isMention: true, - threadId: thread.id, - author: { userId: "U0TESTER" }, - }); - await slackRuntime.handleNewMention(thread, message, { - destination: createTestDestination(thread), - }); + await expect( + slackRuntime.handleNewMention( + thread, + createTestMessage({ + id: "m-content-active-session-record", + text: "<@U0APP> continue", + isMention: true, + threadId: thread.id, + author: { userId: "U0TESTER" }, + }), + { destination: createTestDestination(thread) }, + ), + ).rejects.toBeInstanceOf(TurnInputCommitLostError); expect(calls).toHaveLength(1); expect(calls[0]?.piMessages).toEqual(activeMessages); - const lifecycle = ( - await getConversationEventStore().loadHistory(thread.id) - ).filter( - (event) => - event.data.type === "turn_started" || - event.data.type === "turn_failed", - ); - expect(lifecycle.map((event) => event.data.type)).toEqual([ - "turn_started", - "turn_failed", - ]); }); });