diff --git a/packages/junior-dashboard/e2e/conversation-send.spec.ts b/packages/junior-dashboard/e2e/conversation-send.spec.ts index 32574e4910..856739c0d6 100644 --- a/packages/junior-dashboard/e2e/conversation-send.spec.ts +++ b/packages/junior-dashboard/e2e/conversation-send.spec.ts @@ -453,7 +453,7 @@ test("hands web and external messages from queue to history without gaps", async historyHeld = undefined; queued = [web]; await expect( - pending.getByRole("button", { name: "Remove queued message" }), + pending.getByRole("button", { name: "Delete queued message" }), ).toBeVisible(); await expect(thinking).toBeHidden(); @@ -518,7 +518,7 @@ test("hands web and external messages from queue to history without gaps", async await composer.fill(second.text); await page.getByRole("button", { name: "Send", exact: true }).click(); await expect( - pending.getByRole("button", { name: "Remove queued message" }), + pending.getByRole("button", { name: "Delete queued message" }), ).toBeVisible(); await expect(page.getByText(second.text, { exact: true })).toHaveCount(1); committed = [web, second]; @@ -554,11 +554,11 @@ test("hands web and external messages from queue to history without gaps", async timeout: 15_000, }); await expect(thinking).toBeHidden(); - await pending.getByRole("button", { name: "Remove queued message" }).click(); - await expect(pending.getByText("Could not remove. Try again.")).toBeVisible(); + await pending.getByRole("button", { name: "Delete queued message" }).click(); + await expect(pending.getByText("Could not delete. Try again.")).toBeVisible(); failCancel = false; await pending - .getByRole("button", { name: "Could not remove. Try again." }) + .getByRole("button", { name: "Could not delete. Try again." }) .click(); await expect(pending).toBeHidden(); }); @@ -698,8 +698,9 @@ test("edits a queued message back into the composer", async ({ source: "web", text: "Queued follow-up", }; - // Junior reads this one before the edit lands, so the cancel finds nothing. - const readMessage = { + // The reader steers this one. Junior reads it before the edit lands, so the + // cancel finds nothing. + const readMessage: ConversationPendingMessage & { text: string } = { ...queuedMessage, attachments: undefined, inboundMessageId: "queued-read", @@ -707,6 +708,19 @@ test("edits a queued message back into the composer", async ({ text: "Already read", }; let queued: ConversationPendingMessage[] = [queuedMessage, readMessage]; + const steerRequests: string[][] = []; + await page.route(`${path}/pending-messages/steer`, async (route) => { + const ids = route.request().postDataJSON().inboundMessageIds as string[]; + steerRequests.push(ids); + queued = queued.map((message) => + ids.includes(message.inboundMessageId) + ? { ...message, delivery: "interrupt" } + : message, + ); + await route.fulfill({ + json: { conversationId, steeredInboundMessageIds: ids }, + }); + }); await page.route(`${path}/pending-messages`, async (route) => { if (route.request().method() !== "DELETE") { await route.fulfill({ @@ -747,6 +761,18 @@ test("edits a queued message back into the composer", async ({ ).toBeVisible(); await screenshot(page, "conversation-queue-compact"); + const readRow = pending.getByRole("article").filter({ + hasText: readMessage.text, + }); + await readRow.getByRole("button", { name: "Steer queued message" }).click(); + await expect(readRow.getByLabel("Steered", { exact: true })).toBeVisible(); + await expect( + readRow.getByRole("button", { name: "Steer queued message" }), + ).toBeHidden(); + expect(steerRequests).toEqual([[readMessage.inboundMessageId]]); + // Close the delivery tooltip that opened under the pointer. + await page.mouse.move(0, 0); + // A draft with no room for the queued image keeps the message queued. await composer.evaluate((element, data) => { const dataTransfer = new DataTransfer(); diff --git a/packages/junior-dashboard/src/client/conversations/ComposerImages.tsx b/packages/junior-dashboard/src/client/conversations/ComposerImages.tsx index e9c25f6c58..5685d08678 100644 --- a/packages/junior-dashboard/src/client/conversations/ComposerImages.tsx +++ b/packages/junior-dashboard/src/client/conversations/ComposerImages.tsx @@ -6,6 +6,12 @@ import { type InputImage, type MessageAttachment, } from "@sentry/junior/api/schema"; +import { Button } from "../components/ui/button"; +import { + Tooltip, + TooltipContent, + TooltipTrigger, +} from "../components/ui/tooltip"; import { conversationAttachmentUrl, inputImageSrc, @@ -24,13 +30,6 @@ export function composerImagesFit(images: readonly InputImage[]): boolean { ); } -import { Button } from "../components/ui/button"; -import { - Tooltip, - TooltipContent, - TooltipTrigger, -} from "../components/ui/tooltip"; - /** Read a bounded image selection for the message request and local previews. */ export async function readComposerImages( files: File[], diff --git a/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx b/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx index 05f7d51e39..32e3cd8670 100644 --- a/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx +++ b/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx @@ -19,6 +19,7 @@ import { useAppendConversationMessage, useArchiveConversation, useCancelConversationPendingMessages, + useSteerConversationPendingMessages, useConversationData, useStopConversationTurn, type PendingArchiveConversationUpdate, @@ -566,7 +567,7 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { ); const [editTargetInboundMessageId, setEditTargetInboundMessageId] = useState(); - const [editError, setEditError] = useState(); + const [queueActionError, setQueueActionError] = useState(); // Keep submit identity stable across mutation status flips so the memoized // composer does not re-render while the reader is still typing. const appendMessageRef = useRef(appendMessage); @@ -617,7 +618,7 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { cancelPendingMessagesRef.current.reset(); } if (!stopTurnRef.current.isPending) stopTurnRef.current.reset(); - setEditError(undefined); + setQueueActionError(undefined); onPinRequestRef.current(); }, []); const cancellableMessageIds = props.pendingMessages @@ -659,7 +660,7 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { editingRef.current = true; const inboundMessageId = message.inboundMessageId; setEditTargetInboundMessageId(inboundMessageId); - setEditError(undefined); + setQueueActionError(undefined); try { // Read images first. A failed read must leave the message queued. const images = await readStoredComposerImages( @@ -670,7 +671,7 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { if (!composerRef.current) return; // Check before the cancel. After it, images that do not fit are lost. if (!composerRef.current.canRestoreImages(images)) { - setEditError("Remove some draft images to edit that message."); + setQueueActionError("Remove some draft images to edit that message."); return; } const result = await editCancelRef.current.mutateAsync({ @@ -680,7 +681,7 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { // Junior can read the message before the cancel lands. Then it is // already in the Conversation and there is nothing to edit. if (!result.cancelledInboundMessageIds.includes(inboundMessageId)) { - setEditError("Junior already read that message."); + setQueueActionError("Junior already read that message."); return; } const restored = { images, text: message.text ?? "" }; @@ -692,7 +693,7 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { restoreStoredDraftText(props.conversationId, restored.text); } } catch { - setEditError("Could not edit the queued message. Try again."); + setQueueActionError("Could not edit the queued message. Try again."); } finally { editingRef.current = false; setEditTargetInboundMessageId(undefined); @@ -701,6 +702,28 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { [props.conversationId, props.pendingGeneratedAtRef], ); + const steerPendingMessages = useSteerConversationPendingMessages( + props.conversationId, + ); + const steerPendingMessagesRef = useRef(steerPendingMessages); + steerPendingMessagesRef.current = steerPendingMessages; + const onSteerMessage = useCallback((message: ConversationMailboxMessage) => { + if (steerPendingMessagesRef.current.isPending) return; + setQueueActionError(undefined); + steerPendingMessagesRef.current.mutate( + { inboundMessageIds: [message.inboundMessageId] }, + { + onError: () => + setQueueActionError( + "Could not send the queued message now. Try again.", + ), + }, + ); + }, []); + const steerTargetInboundMessageId = steerPendingMessages.isPending + ? steerPendingMessages.variables?.inboundMessageIds[0] + : undefined; + const onStop = useCallback(async () => { const result = await stopTurnRef.current.mutateAsync(); const restored = result.stoppedMessages.map((message) => message.text); @@ -737,6 +760,8 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { onEditMessage={onEditMessage} onLayoutChange={onMailboxLayoutChange} onRetry={onRetry} + onSteerMessage={props.stoppable ? onSteerMessage : undefined} + steerTargetInboundMessageId={steerTargetInboundMessageId} viewerEmail={viewerEmail} /> @@ -745,7 +770,7 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { , -): { icon: LucideIcon | typeof Spinner; label: string } { +): { icon: LucideIcon | typeof Spinner; label: string; tooltip: string } { if (message.clientStatus === "failed") { - return { icon: AlertCircle, label: "Failed to send" }; + return { + icon: AlertCircle, + label: "Failed to send", + tooltip: "Failed to send", + }; } if (message.clientStatus === "sending") { - return { icon: Spinner, label: "Sending" }; + return { icon: Spinner, label: "Sending", tooltip: "Sending" }; } if (message.delivery === "interrupt") { - return { icon: SkipForward, label: "Interrupt" }; + return { + icon: STEER_ICON, + label: "Steered", + tooltip: "Steered: Junior reads it at its next step", + }; } - return { icon: Clock3, label: "Queued" }; + return { + icon: Clock3, + label: "Queued", + tooltip: "Queued: Junior reads it after the current reply", + }; } -function PendingMetaIcon(props: { children: ReactElement; label: string }) { +function PendingMetaIcon(props: { + children: ReactElement; + label: string; + tooltip?: string; +}) { return ( } /> - {props.label} + + {props.tooltip ?? props.label} + ); } +/** One icon-only row action. The tooltip says what the action does. */ function PendingActionButton(props: { disabled: boolean; icon: LucideIcon; @@ -81,28 +105,33 @@ function PendingActionButton(props: { onClick(): void; pending: boolean; tooltip: string; + variant?: "danger"; }) { const Icon = props.icon; return ( {props.pending ? ( ); } @@ -115,13 +144,15 @@ function PendingMetaIcons(props: { message: ConversationMailboxMessage; onCancel?: () => void; onEdit?: () => void; + onSteer?: () => void; showSlack: boolean; + steerPending: boolean; }) { const delivery = pendingDeliveryMeta(props.message); const DeliveryIcon = delivery.icon; return ( - + {props.showSlack ? ( } /> - Slack + From Slack ) : null} - + + {props.onSteer ? ( + + ) : null} {props.onEdit ? ( ) : null} {props.onCancel ? ( ) : null} @@ -236,12 +282,15 @@ function PendingRow(props: { cancelError: boolean; cancelPending: boolean; canEdit: boolean; + canSteer: boolean; conversationId?: string; editPending: boolean; message: ConversationMailboxMessage; onCancel?(message: ConversationMailboxMessage): void; onEdit?(message: ConversationMailboxMessage): void; onRetry?(message: ConversationMailboxMessage): void; + onSteer?(message: ConversationMailboxMessage): void; + steerPending: boolean; }) { const text = props.message.text ?? ""; const redacted = Boolean(props.message.redacted); @@ -295,7 +344,13 @@ function PendingRow(props: { ? () => props.onEdit?.(props.message) : undefined } + onSteer={ + props.canSteer && props.onSteer + ? () => props.onSteer?.(props.message) + : undefined + } showSlack={showSlack} + steerPending={props.steerPending} /> } /> @@ -315,7 +370,7 @@ function PendingRow(props: { ) : null} {props.cancelError ? (

- Could not remove. Try again. + Could not delete. Try again.

) : null} @@ -383,6 +438,10 @@ export const PendingMailboxStack = memo(function PendingMailboxStack(props: { /** Fires after expand/collapse changes the stack height above the composer. */ onLayoutChange?: () => void; onRetry?(message: ConversationMailboxMessage): void; + /** Ask Junior to read the viewer's own queued web message now. */ + onSteerMessage?: (message: ConversationMailboxMessage) => void; + /** Inbound id of the row that a steer is sending now. */ + steerTargetInboundMessageId?: string; /** Email of the signed-in reader. Only their own web rows can be edited. */ viewerEmail?: string; }): ReactNode { @@ -415,23 +474,26 @@ export const PendingMailboxStack = memo(function PendingMailboxStack(props: { const hiddenCount = Math.max(0, rows.length - COLLAPSED_PENDING_ROW_COUNT); const toggleExpanded = () => setExpanded((value) => !value); const actionsDisabled = - Boolean(props.cancelPending) || Boolean(props.editTargetInboundMessageId); + Boolean(props.cancelPending) || + Boolean(props.editTargetInboundMessageId) || + Boolean(props.steerTargetInboundMessageId); const renderRow = (message: ConversationMailboxMessage) => { const isTarget = props.cancelTargetInboundMessageId === message.inboundMessageId; // Only accepted server rows can leave the queue. Local rows are still sending. const accepted = message.clientStatus === undefined; + const ownWebRow = + accepted && + message.source === "web" && + !message.redacted && + sameEmail(message.actorIdentity?.email, props.viewerEmail); return ( ); }; diff --git a/packages/junior-dashboard/src/client/conversations/queries.ts b/packages/junior-dashboard/src/client/conversations/queries.ts index 01a283cb41..80d151e9e2 100644 --- a/packages/junior-dashboard/src/client/conversations/queries.ts +++ b/packages/junior-dashboard/src/client/conversations/queries.ts @@ -26,6 +26,7 @@ import { conversationEventPageSchema, conversationPendingMessagesReportSchema, forkConversationResponseSchema, + steerConversationPendingMessagesResponseSchema, stopConversationTurnResponseSchema, } from "@sentry/junior/api/schema"; @@ -396,6 +397,52 @@ export function useCancelConversationPendingMessages(conversationId: string) { }); } +/** Ask Junior to read the viewer's own queued web messages now. */ +export function useSteerConversationPendingMessages(conversationId: string) { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: (args: { inboundMessageIds: string[] }) => + post( + steerConversationPendingMessagesResponseSchema, + `/api/conversations/${encodeURIComponent(conversationId)}/pending-messages/steer`, + args, + ), + onSuccess: async (result) => { + const steered = new Set(result.steeredInboundMessageIds); + if (steered.size === 0) return; + // A live poll that read the mailbox before the steer would restore + // `defer`. Cancel it; onSettled refetches after the write. + await queryClient.cancelQueries({ + exact: true, + queryKey: conversationDetailQueryKey(conversationId), + }); + queryClient.setQueryData( + conversationDetailQueryKey(conversationId), + (current) => + current?.mailbox + ? { + ...current, + mailbox: { + ...current.mailbox, + messages: current.mailbox.messages.map((message) => + steered.has(message.inboundMessageId) + ? { ...message, delivery: "interrupt" as const } + : message, + ), + }, + } + : current, + ); + }, + onSettled: async () => { + await queryClient.invalidateQueries({ + exact: true, + queryKey: conversationDetailQueryKey(conversationId), + }); + }, + }); +} + /** * Local cache of queued rows that a stop removed. The worker drops them only * when it observes the stop, so the mailbox can still list them for a while. diff --git a/packages/junior-dashboard/tests/pending-mailbox-stack.test.tsx b/packages/junior-dashboard/tests/pending-mailbox-stack.test.tsx index 5ce45ee151..1b2f0d77ed 100644 --- a/packages/junior-dashboard/tests/pending-mailbox-stack.test.tsx +++ b/packages/junior-dashboard/tests/pending-mailbox-stack.test.tsx @@ -41,8 +41,8 @@ describe("PendingMailboxStack remove control", () => { />, ); - expect(accepted).toContain("Remove queued message"); - expect(localOnly).not.toContain("Remove queued message"); + expect(accepted).toContain("Delete queued message"); + expect(localOnly).not.toContain("Delete queued message"); }); it("keeps remove available for accepted rows while a local send is pending", () => { @@ -60,7 +60,7 @@ describe("PendingMailboxStack remove control", () => { />, ); - expect(html.match(/aria-label="Remove queued message"/g)).toHaveLength(1); + expect(html.match(/aria-label="Delete queued message"/g)).toHaveLength(1); }); it("does not show remove for failed local outbox rows", () => { @@ -79,8 +79,8 @@ describe("PendingMailboxStack remove control", () => { />, ); - expect(html).not.toContain("Remove queued message"); - expect(html).not.toContain("Could not remove. Try again."); + expect(html).not.toContain("Delete queued message"); + expect(html).not.toContain("Could not delete. Try again."); }); it("shows an inline remove error on the target accepted row", () => { @@ -93,8 +93,49 @@ describe("PendingMailboxStack remove control", () => { />, ); - expect(html).toContain("Could not remove. Try again."); - expect(html).toContain('aria-label="Could not remove. Try again."'); + expect(html).toContain("Could not delete. Try again."); + expect(html).toContain('aria-label="Could not delete. Try again."'); + }); + + it("shows edit and steer only on the viewer's own accepted web rows", () => { + const own = { email: "Reader@example.com" }; + const html = renderToStaticMarkup( + undefined} + onSteerMessage={() => undefined} + viewerEmail="reader@example.com" + />, + ); + + expect(html.match(/aria-label="Edit queued message"/g)).toHaveLength(2); + // A steered row is already an interrupt, so it cannot be steered again. + expect(html.match(/aria-label="Steer queued message"/g)).toHaveLength(1); }); it("shows edit only on the viewer's own accepted web rows", () => { diff --git a/packages/junior-evals/evals/integration/conversation/steering.eval.ts b/packages/junior-evals/evals/integration/conversation/steering.eval.ts index e5df8a3d71..5ca559edcb 100644 --- a/packages/junior-evals/evals/integration/conversation/steering.eval.ts +++ b/packages/junior-evals/evals/integration/conversation/steering.eval.ts @@ -3,6 +3,7 @@ import { slackMention, person, slackThreadMessage, + webMessage, } from "@junior-evals/fixture/inputs"; import { rubric, RubricJudge } from "@junior-evals/fixture/judge"; import { sendDuringFirstModelRequest } from "@junior-evals/fixture/progress"; @@ -148,3 +149,36 @@ describe("Slack Turn Steering", () => { expect(next.turns).toEqual([]); }); }); + +describe("Dashboard Turn Steering", () => { + test("when the author steers a queued dashboard message, the running turn takes it", async ({ + run, + }) => { + const conversation = await run(webMessage(INCIDENT_REQUEST), { + onProgress: sendDuringFirstModelRequest([ + webMessage("include the rollback owner: Dana", { steer: true }), + ]), + }); + await expect(conversation).toSatisfyJudge( + RubricJudge, + rubric({ + pass: ["The summary names Dana as the rollback owner."], + }), + ); + + // Without the steer, the queued message would wait for its own turn. + expect(conversation.turns.map((turn) => turn.status)).toEqual([ + "succeeded", + ]); + expect( + conversation.session.events.filter( + (event) => event.type === "message" && event.role === "user", + ), + ).toEqual([ + expect.objectContaining({ content: INCIDENT_REQUEST }), + expect.objectContaining({ + content: "include the rollback owner: Dana", + }), + ]); + }); +}); diff --git a/packages/junior-evals/src/fixture/agent.ts b/packages/junior-evals/src/fixture/agent.ts index 3f8a594522..9b6e7dcf60 100644 --- a/packages/junior-evals/src/fixture/agent.ts +++ b/packages/junior-evals/src/fixture/agent.ts @@ -15,7 +15,6 @@ import { type HarnessRun, type TranscriptEvent } from "vitest-evals/harness"; import { createApp, type JuniorAppOptions } from "@/app"; import { createJuniorApi } from "@/api"; import type { JuniorApiEnv } from "@/api/route"; -import { acceptedConversationMessageSchema } from "@/api/schema"; import { forkConversationResponseSchema } from "@/api/schema"; import { createConversationId, @@ -27,11 +26,16 @@ import { runEvalWork } from "../eval-work"; import { completeAuthorization, forgetAuthorizations } from "./auth"; import { installBlobMock } from "./blob"; import { installGatewayObserver, type GatewayProgress } from "./gateway"; +import { + expectAccepted, + jsonRequest, + steerQueuedWebMessage, + webImage, +} from "./dashboard-requests"; import { fixtureId } from "./ids"; import { installWebReplay } from "./web"; import type { AutomationInput, - FileInput, HistoryItem, HistoryReply, Input, @@ -392,7 +396,7 @@ export async function createFixtureAgent( const path = started ? `/api/conversations/${encodeURIComponent(record.conversationId)}/messages` : "/api/conversations"; - await expectAccepted( + const accepted = await expectAccepted( await api.request( path, jsonRequest(record.viewerEmail, { @@ -406,6 +410,13 @@ export async function createFixtureAgent( }), ), ); + if (input.steer) { + await steerQueuedWebMessage(api, { + conversationId: record.conversationId, + inboundMessageId: accepted.messageId, + viewerEmail: record.viewerEmail, + }); + } return; } if (record.surface !== "slack") { @@ -963,34 +974,3 @@ function heartbeatSecret(): string { } return secret; } - -/** An image as the dashboard sends it with a message. */ -function webImage(image: FileInput) { - if (!image.content) throw new Error(`Image ${image.name} needs content`); - return { - contentType: image.mimeType, - data: image.content.toString("base64"), - filename: image.name, - }; -} - -/** A JSON POST signed in as `viewerEmail`. */ -function jsonRequest(viewerEmail: string, body: unknown): RequestInit { - return { - method: "POST", - headers: { - "content-type": "application/json", - [VIEWER_HEADER]: viewerEmail, - }, - body: JSON.stringify(body), - }; -} - -async function expectAccepted(response: Response) { - if (response.status !== 200) { - throw new Error( - `Web message returned ${response.status}: ${await response.text()}`, - ); - } - return acceptedConversationMessageSchema.parse(await response.json()); -} diff --git a/packages/junior-evals/src/fixture/dashboard-requests.ts b/packages/junior-evals/src/fixture/dashboard-requests.ts new file mode 100644 index 0000000000..16c210712c --- /dev/null +++ b/packages/junior-evals/src/fixture/dashboard-requests.ts @@ -0,0 +1,75 @@ +/** + * Requests that the fixture sends to the dashboard API as a signed-in person. + */ +import type { Hono } from "hono"; +import type { JuniorApiEnv } from "@/api/route"; +import { + acceptedConversationMessageSchema, + steerConversationPendingMessagesResponseSchema, +} from "@/api/schema"; +import type { FileInput } from "./inputs"; +import { VIEWER_HEADER } from "./results"; + +/** An image as the dashboard sends it with a message. */ +export function webImage(image: FileInput) { + if (!image.content) throw new Error(`Image ${image.name} needs content`); + return { + contentType: image.mimeType, + data: image.content.toString("base64"), + filename: image.name, + }; +} + +/** A JSON POST signed in as `viewerEmail`. */ +export function jsonRequest(viewerEmail: string, body: unknown): RequestInit { + return { + method: "POST", + headers: { + "content-type": "application/json", + [VIEWER_HEADER]: viewerEmail, + }, + body: JSON.stringify(body), + }; +} + +/** Read the accepted message of a web message post, or fail the test. */ +export async function expectAccepted(response: Response) { + if (response.status !== 200) { + throw new Error( + `Web message returned ${response.status}: ${await response.text()}`, + ); + } + return acceptedConversationMessageSchema.parse(await response.json()); +} + +/** + * Steer one queued web message into the running turn, as the steer action in + * the dashboard queue does. + */ +export async function steerQueuedWebMessage( + api: Hono, + args: { + conversationId: string; + inboundMessageId: string; + viewerEmail: string; + }, +): Promise { + const response = await api.request( + `/api/conversations/${encodeURIComponent(args.conversationId)}/pending-messages/steer`, + jsonRequest(args.viewerEmail, { + inboundMessageIds: [args.inboundMessageId], + }), + ); + if (response.status !== 200) { + throw new Error( + `Steer returned ${response.status}: ${await response.text()}`, + ); + } + const result = steerConversationPendingMessagesResponseSchema.parse( + await response.json(), + ); + // An empty result means Junior read the message before the steer. + if (!result.steeredInboundMessageIds.includes(args.inboundMessageId)) { + throw new Error("The message left the queue before the steer"); + } +} diff --git a/packages/junior-evals/src/fixture/inputs.ts b/packages/junior-evals/src/fixture/inputs.ts index 845f08db69..440b5d9640 100644 --- a/packages/junior-evals/src/fixture/inputs.ts +++ b/packages/junior-evals/src/fixture/inputs.ts @@ -80,6 +80,8 @@ export interface WebMessageInput { kind: "web_message"; /** Images that the person added to the message. */ images?: FileInput[]; + /** After the post, the person steers the queued message into the running turn. */ + steer?: boolean; text: string; } @@ -195,7 +197,7 @@ export function slackCommand( /** Send a message from the dashboard. */ export function webMessage( text: string, - options: { images?: FileInput[] } = {}, + options: { images?: FileInput[]; steer?: boolean } = {}, ): WebMessageInput { return { kind: "web_message", text, ...options }; } diff --git a/packages/junior/src/api/conversations/routes.ts b/packages/junior/src/api/conversations/routes.ts index a76b3772e0..175dfb5544 100644 --- a/packages/junior/src/api/conversations/routes.ts +++ b/packages/junior/src/api/conversations/routes.ts @@ -10,6 +10,8 @@ import { archiveConversationResponseSchema, cancelConversationPendingMessagesBodySchema, cancelConversationPendingMessagesResponseSchema, + steerConversationPendingMessagesBodySchema, + steerConversationPendingMessagesResponseSchema, conversationAttachmentParamsSchema, conversationDetailQuerySchema, conversationDetailReportSchema, @@ -44,6 +46,7 @@ import { forkConversationForViewer } from "./fork"; import { readConversationEvents } from "./event-list"; import { readConversationFeed } from "./list"; import { cancelConversationPendingMessagesForViewer } from "./cancel-pending-messages"; +import { steerConversationPendingMessagesForViewer } from "./steer-pending-messages"; import { requireConversationPendingMessages } from "./pending-messages"; import { readConversationStats } from "./stats"; import { stopConversationTurnForViewer } from "./stop"; @@ -283,6 +286,34 @@ export function createConversationRoutes( }, ); + app.post( + "/:conversationId/pending-messages/steer", + requireViewer, + validateRequest( + "param", + conversationParamsSchema, + "Invalid route parameters.", + ), + validateRequest( + "json", + steerConversationPendingMessagesBodySchema, + "Invalid request body.", + ), + async (context) => { + const viewer = context.get("viewer"); + const { conversationId } = context.req.valid("param"); + const body = context.req.valid("json"); + return jsonResponse( + steerConversationPendingMessagesResponseSchema, + await steerConversationPendingMessagesForViewer( + viewer, + conversationId, + body, + ), + ); + }, + ); + app.delete( "/:conversationId/pending-messages", requireViewer, diff --git a/packages/junior/src/api/conversations/steer-pending-messages.ts b/packages/junior/src/api/conversations/steer-pending-messages.ts new file mode 100644 index 0000000000..8a623f50de --- /dev/null +++ b/packages/junior/src/api/conversations/steer-pending-messages.ts @@ -0,0 +1,49 @@ +import type { User } from "@sentry/junior-plugin-api"; +import { webActorFromEmail } from "@/chat/conversations/web-input"; +import { getConversationStore, getDb } from "@/chat/db"; +import { steerWebPendingMessages } from "@/chat/task-execution/store"; +import { throwApiError } from "../http"; +import type { + SteerConversationPendingMessagesBody, + SteerConversationPendingMessagesResponse, +} from "../schema/conversation"; +import { readConversationAccessFromSql } from "./access"; + +/** + * Steer the viewer's own queued web messages into the conversation now. + * + * Only the author can steer a message. Ids that are not the viewer's queued + * web messages, or that Junior already read, are left out of the result. + */ +export async function steerConversationPendingMessagesForViewer( + viewer: User, + conversationId: string, + body: SteerConversationPendingMessagesBody, +): Promise { + const conversation = await getConversationStore().get({ conversationId }); + if (!conversation) { + throwApiError(404, "Conversation not found."); + } + + const access = await readConversationAccessFromSql( + getDb(), + [conversationId], + viewer, + ); + if (!access.get(conversationId)?.isParticipant) { + throwApiError( + 403, + "Only conversation participants can steer queued messages.", + ); + } + + const result = await steerWebPendingMessages({ + authorId: webActorFromEmail(viewer.email).userId, + conversationId, + inboundMessageIds: body.inboundMessageIds, + }); + return { + conversationId, + steeredInboundMessageIds: result.steeredInboundMessageIds, + }; +} diff --git a/packages/junior/src/api/schema.ts b/packages/junior/src/api/schema.ts index b3340c0481..ebaa54e6f4 100644 --- a/packages/junior/src/api/schema.ts +++ b/packages/junior/src/api/schema.ts @@ -42,6 +42,8 @@ export { archiveConversationResponseSchema, cancelConversationPendingMessagesBodySchema, cancelConversationPendingMessagesResponseSchema, + steerConversationPendingMessagesBodySchema, + steerConversationPendingMessagesResponseSchema, conversationAuxiliaryCostsSchema, conversationDetailQuerySchema, conversationDetailReportSchema, @@ -80,6 +82,8 @@ export type { ArchiveConversationResponse, CancelConversationPendingMessagesBody, CancelConversationPendingMessagesResponse, + SteerConversationPendingMessagesBody, + SteerConversationPendingMessagesResponse, ActorIdentity, ConversationActivityPreview, ConversationAgentTreeNode, diff --git a/packages/junior/src/api/schema/conversation.ts b/packages/junior/src/api/schema/conversation.ts index 2c78096b6d..b944a3f840 100644 --- a/packages/junior/src/api/schema/conversation.ts +++ b/packages/junior/src/api/schema/conversation.ts @@ -203,6 +203,19 @@ export const cancelConversationPendingMessagesResponseSchema = z }) .strict(); +/** Queued web messages that the viewer wants Junior to read now. */ +export const steerConversationPendingMessagesBodySchema = z + .object({ inboundMessageIds: z.array(z.string().min(1)).min(1) }) + .strict(); + +/** Result of steering queued web messages into the conversation. */ +export const steerConversationPendingMessagesResponseSchema = z + .object({ + conversationId: z.string().min(1), + steeredInboundMessageIds: z.array(z.string().min(1)), + }) + .strict(); + export const conversationAuxiliaryCostsSchema = z .object({ costUsd: z.number().finite().nonnegative(), @@ -934,3 +947,9 @@ export type CancelConversationPendingMessagesBody = z.infer< export type CancelConversationPendingMessagesResponse = z.infer< typeof cancelConversationPendingMessagesResponseSchema >; +export type SteerConversationPendingMessagesBody = z.infer< + typeof steerConversationPendingMessagesBodySchema +>; +export type SteerConversationPendingMessagesResponse = z.infer< + typeof steerConversationPendingMessagesResponseSchema +>; diff --git a/packages/junior/src/chat/agent/types.ts b/packages/junior/src/chat/agent/types.ts index 2c313cbc9b..fbe4834b87 100644 --- a/packages/junior/src/chat/agent/types.ts +++ b/packages/junior/src/chat/agent/types.ts @@ -64,6 +64,8 @@ export type AgentSteeringMessage = { text: string; timestampMs?: number; attachments?: AgentAttachment[]; + /** Stored images that the agent runner loads into `attachments`. */ + storedAttachments?: readonly MessageAttachment[]; }; /** Model-safe input for one asynchronously delegated child-agent task. */ diff --git a/packages/junior/src/chat/runtime/agent-runner.ts b/packages/junior/src/chat/runtime/agent-runner.ts index 976a38ea4d..366043c0d2 100644 --- a/packages/junior/src/chat/runtime/agent-runner.ts +++ b/packages/junior/src/chat/runtime/agent-runner.ts @@ -1,6 +1,8 @@ import type { StreamFn } from "@earendil-works/pi-agent-core"; +import type { MessageAttachment } from "@/chat/attachments/input"; import { isAgentRunFeatureDisabled, + type AgentAttachment, type AgentRun, type MessageAgent, type SpawnAgent, @@ -64,19 +66,49 @@ export function createAgentRunner( } : undefined), }; - if (run.instruction.storedAttachments?.length) { - // Image storage loads runtime config. Keep it behind execution so the - // local CLI can set its defaults before config is first read. + // Image storage loads runtime config. Keep it behind execution so the + // local CLI can set its defaults before config is first read. + const loadStoredImages = async ( + attachments: readonly MessageAttachment[], + ): Promise => { const { loadInputImages } = await import("@/chat/attachments/images"); const storage = nextRun.environment?.attachmentStorage; if (!storage) throw new Error("Attachment storage is unavailable."); + return await loadInputImages({ + attachments, + conversationId: run.conversationId, + storage, + }); + }; + if (run.instruction.storedAttachments?.length) { nextRun.instruction = { ...run.instruction, - attachments: await loadInputImages({ - attachments: run.instruction.storedAttachments, - conversationId: run.conversationId, - storage, - }), + attachments: await loadStoredImages( + run.instruction.storedAttachments, + ), + }; + } + const drainSteeringMessages = nextRun.durability?.drainSteeringMessages; + if (drainSteeringMessages) { + // Steered web messages carry stored images like the instruction does. + nextRun.durability = { + ...nextRun.durability, + drainSteeringMessages: (accept) => + drainSteeringMessages(async (messages) => + accept( + await Promise.all( + messages.map(async ({ storedAttachments, ...message }) => + storedAttachments?.length + ? { + ...message, + attachments: + await loadStoredImages(storedAttachments), + } + : message, + ), + ), + ), + ), }; } return await execute(nextRun, streamFn); diff --git a/packages/junior/src/chat/task-execution/README.md b/packages/junior/src/chat/task-execution/README.md index 6f7aee5563..1bf6bf8618 100644 --- a/packages/junior/src/chat/task-execution/README.md +++ b/packages/junior/src/chat/task-execution/README.md @@ -77,8 +77,8 @@ old Redis state. Old mailbox, lease, and turn-cursor state can be lost. Committed SQL history remains. The code has no old-state reader, dual write, or rollback support. -`checkpoint.ts`, `turn-wake.ts`, `paused-turn.ts`, `state.ts`, `store.ts`, and -`worker.ts` define the execution surface. +`checkpoint.ts`, `turn-wake.ts`, `paused-turn.ts`, `state.ts`, `store.ts`, +`attempt-selection.ts`, and `worker.ts` define the execution surface. ## Execution @@ -111,6 +111,12 @@ and waits for the next turn. Slack `@` mentions use `interrupt` delivery; all other inbound messages use `defer`. +A dashboard steer changes the reader's own queued web messages from `defer` to +`interrupt`. A web Turn drains them only when they come from the Turn Actor. +A Slack Turn does not drain web input. While a yielded Turn waits to resume, a +steered web message does not start its own attempt. That Turn drains it, or it +starts the next Turn. + An active turn drains only messages with `interrupt` mailbox delivery. When a new turn starts, `interrupt` delivery is handled before queued `defer` delivery. diff --git a/packages/junior/src/chat/task-execution/attempt-selection.ts b/packages/junior/src/chat/task-execution/attempt-selection.ts new file mode 100644 index 0000000000..46f43c8eb7 --- /dev/null +++ b/packages/junior/src/chat/task-execution/attempt-selection.ts @@ -0,0 +1,55 @@ +import type { ConversationWorkState, InboundMessage } from "./store"; + +function selectContiguousTurnBatch( + messages: readonly InboundMessage[], +): InboundMessage[] { + const first = messages[0]; + if (!first) { + return []; + } + const nextTurnIndex = messages.findIndex( + (message) => + message.input.authorId !== first.input.authorId || + message.source !== first.source, + ); + return messages.slice( + 0, + nextTurnIndex === -1 ? messages.length : nextTurnIndex, + ); +} + +/** + * True when an interrupt can start its own attempt. + * + * A steered web message must not start a second Turn while a yielded Turn + * waits to resume. It stays pending for that Turn's drain or the next attempt. + */ +export function isAttemptInterrupt( + message: InboundMessage, + work: ConversationWorkState, +): boolean { + return ( + message.delivery === "interrupt" && + !(work.execution.status === "paused" && message.source === "web") + ); +} + +/** Prioritize interrupts while keeping each attempt to one Actor and Source. */ +export function selectAttemptMessages( + work: ConversationWorkState, +): InboundMessage[] { + const messages = work.messages; + const interrupts = messages.filter((message) => + isAttemptInterrupt(message, work), + ); + if (interrupts.length > 0) { + return selectContiguousTurnBatch(interrupts); + } + if (work.execution.status === "paused") return []; + const nonEventMessages = messages.filter( + (message) => message.source !== "event", + ); + return selectContiguousTurnBatch( + nonEventMessages.length > 0 ? nonEventMessages : messages, + ); +} diff --git a/packages/junior/src/chat/task-execution/conversation-turn.ts b/packages/junior/src/chat/task-execution/conversation-turn.ts index 4cff62e6a7..1997d0988e 100644 --- a/packages/junior/src/chat/task-execution/conversation-turn.ts +++ b/packages/junior/src/chat/task-execution/conversation-turn.ts @@ -83,6 +83,7 @@ import { type TurnInputFacts, } from "@/chat/task-execution/mailbox-turn"; import { joinMailboxText } from "@/chat/task-execution/mailbox-input"; +import { createWebSteeringDrain } from "@/chat/task-execution/web-steering"; import { resolveConversationDestination } from "@/chat/conversations/destination"; import { eventBatchMessageEvents, @@ -594,6 +595,17 @@ export function createConversationTurnWorker( }, delivery: deliverAssistantMessage, durability: { + ...(webActor + ? { + drainSteeringMessages: createWebSteeringDrain({ + actor: webActor, + author, + context, + conversation, + turnId, + }), + } + : undefined), onInputCommitted: acknowledge, shouldYield: context.shouldYield, onSandboxRefChanged: async (nextSandboxRef) => { diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index 4acb4a8d11..cec9fbe47d 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -743,8 +743,9 @@ export function createSlackConversationWorker( ); } }; - // Only explicit interruptions enter the active turn. Deferred work - // stays pending for a normal turn. + // Only explicit Slack interruptions enter the active turn. Deferred + // work and steered messages from other sources, such as the + // dashboard, stay pending for their own turn. const drainSteeringMessages = async ( accept: ( messages: SteeringCandidateMessage[], @@ -752,6 +753,7 @@ export function createSlackConversationWorker( ): Promise => { await context.attempt.drain(async (pendingRecords) => { const candidates = pendingRecords + .filter((record) => record.source === "slack") .map((record) => ({ inboundMessageId: record.inboundMessageId, message: restoreMessage({ diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 3083c1b5ca..31a6d890c0 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -1942,6 +1942,57 @@ export async function cancelHumanFacingPendingMessages(args: { }); } +/** + * Mark one author's queued web messages as interrupts without a worker lease. + * + * The active Turn drains them at its next model boundary when it belongs to the + * same author. Otherwise they start the next Turn ahead of deferred input. + */ +export async function steerWebPendingMessages(args: { + authorId: string; + conversationId: string; + inboundMessageIds: readonly string[]; + nowMs?: number; + state?: StateAdapter; +}): Promise<{ steeredInboundMessageIds: string[] }> { + const nowMs = args.nowMs ?? now(); + const requestedIds = new Set(args.inboundMessageIds); + return await withConversationMutation(args, async (state, lock) => { + const current = await readConversation(state, args.conversationId); + if (!current) { + return { steeredInboundMessageIds: [] }; + } + + const steeredInboundMessageIds: string[] = []; + const pendingMessages = current.execution.pendingMessages.map((message) => { + if ( + message.source !== "web" || + message.delivery !== "defer" || + message.input.authorId !== args.authorId || + !requestedIds.has(message.inboundMessageId) + ) { + return message; + } + steeredInboundMessageIds.push(message.inboundMessageId); + return { ...message, delivery: "interrupt" as const }; + }); + if (steeredInboundMessageIds.length === 0) { + return { steeredInboundMessageIds }; + } + + await writeConversation( + state, + lock, + withExecutionUpdate( + current, + { ...current.execution, pendingMessages }, + nowMs, + ), + ); + return { steeredInboundMessageIds }; + }); +} + /** Mark the leased conversation as needing another queue-delivered slice. */ export async function requestAnotherSlice(args: { conversationId: string; diff --git a/packages/junior/src/chat/task-execution/store.ts b/packages/junior/src/chat/task-execution/store.ts index 6cf063b84a..4a339a8c51 100644 --- a/packages/junior/src/chat/task-execution/store.ts +++ b/packages/junior/src/chat/task-execution/store.ts @@ -519,6 +519,18 @@ export async function cancelHumanFacingPendingMessages(args: { return result; } +/** Mark one author's queued web messages as interrupts. */ +export async function steerWebPendingMessages(args: { + authorId: string; + conversationId: string; + inboundMessageIds: readonly string[]; + nowMs?: number; + state?: StateAdapter; +}) { + // Delivery mode is not part of execution metadata, so there is nothing to record. + return await workState.steerWebPendingMessages(args); +} + /** Mark the leased conversation as needing another queue-delivered slice. */ export async function requestAnotherSlice(args: { conversationId: string; diff --git a/packages/junior/src/chat/task-execution/web-steering.ts b/packages/junior/src/chat/task-execution/web-steering.ts new file mode 100644 index 0000000000..55b2cf1359 --- /dev/null +++ b/packages/junior/src/chat/task-execution/web-steering.ts @@ -0,0 +1,109 @@ +import type { WebActor } from "@/chat/actor"; +import type { AgentSteeringMessage } from "@/chat/agent/types"; +import { + buildSteeringPiMessage, + withoutSavedSteeringMessages, +} from "@/chat/agent/prompt"; +import { readMessageAttachments } from "@/chat/attachments/input"; +import { persistConversationMessages } from "@/chat/conversations/messages"; +import { instructionProvenanceFor } from "@/chat/conversations/provenance"; +import { legacyWebMailboxMetadataSchema } from "@/chat/conversations/web-mailbox"; +import { + normalizeConversationText, + upsertConversationMessage, +} from "@/chat/services/conversation-memory"; +import type { + ConversationAuthor, + ThreadConversationState, +} from "@/chat/state/conversation"; +import { getTurnRecord } from "@/chat/task-execution/checkpoint"; +import type { InboundMessage } from "@/chat/task-execution/store"; +import type { ConversationWorkerContext } from "@/chat/task-execution/worker"; + +function steeringMessageFromWeb( + message: InboundMessage, + actor: WebActor, +): AgentSteeringMessage { + const storedAttachments = readMessageAttachments(message.input.attachments); + return { + actor: { + authorId: actor.userId, + ...(actor.fullName ? { authorName: actor.fullName } : undefined), + }, + provenance: instructionProvenanceFor(actor), + text: message.input.text.trim(), + timestampMs: message.createdAtMs, + ...(storedAttachments.length ? { storedAttachments } : undefined), + }; +} + +/** + * Drain steered web messages into the active web Turn. + * + * Only messages from the Turn Actor join. Steered messages from another + * participant stay pending and start their own Turn next, so they never act + * with this Actor's credentials. + */ +export function createWebSteeringDrain(args: { + actor: WebActor; + /** Transcript author of the Turn Actor. */ + author: ConversationAuthor; + context: ConversationWorkerContext; + conversation: ThreadConversationState; + turnId: string; +}) { + const { actor, author, context, conversation, turnId } = args; + return async ( + accept: (messages: AgentSteeringMessage[]) => Promise, + ): Promise => { + const accepted: AgentSteeringMessage[] = []; + await context.attempt.drain(async (pending) => { + const joining = pending.filter( + (message) => + message.source === "web" && + message.input.authorId === actor.userId && + legacyWebMailboxMetadataSchema.safeParse(message.input.metadata) + .success, + ); + if (joining.length === 0) return []; + + // Save the transcript Message first. A failed write leaves the input + // pending instead of missing from the transcript. + for (const message of joining) { + const attachments = readMessageAttachments(message.input.attachments); + upsertConversationMessage(conversation, { + id: message.inboundMessageId, + role: "user", + text: normalizeConversationText(message.input.text), + createdAtMs: message.createdAtMs, + author, + meta: { + ...(attachments.length ? { attachments } : undefined), + explicitMention: true, + source: "web", + }, + }); + } + await persistConversationMessages({ + conversation, + conversationId: context.conversationId, + }); + + // An earlier drain can save the messages in the Turn and fail before + // the mailbox ack. Do not give them to the Turn again. + const saved = + (await getTurnRecord(context.conversationId, turnId))?.piMessages ?? []; + const messages = withoutSavedSteeringMessages( + saved, + joining.map((message) => steeringMessageFromWeb(message, actor)), + buildSteeringPiMessage, + ); + if (messages.length > 0) { + await accept(messages); + accepted.push(...messages); + } + return joining.map((message) => message.inboundMessageId); + }); + return accepted; + }; +} diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 4fffce77f6..4c472587e5 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -42,6 +42,7 @@ import { type ConversationWorkState, type InboundMessage, } from "./store"; +import { isAttemptInterrupt, selectAttemptMessages } from "./attempt-selection"; export const CONVERSATION_WORK_DEFER_DELAY_MS = 15_000; const CONVERSATION_STOP_CHECK_INTERVAL_MS = 500; @@ -106,42 +107,6 @@ function now(options: ProcessConversationWorkOptions): number { return options.nowMs?.() ?? Date.now(); } -function selectContiguousTurnBatch( - messages: readonly InboundMessage[], -): InboundMessage[] { - const first = messages[0]; - if (!first) { - return []; - } - const nextTurnIndex = messages.findIndex( - (message) => - message.input.authorId !== first.input.authorId || - message.source !== first.source, - ); - return messages.slice( - 0, - nextTurnIndex === -1 ? messages.length : nextTurnIndex, - ); -} - -/** Prioritize interrupts while keeping each attempt to one Actor and Source. */ -function selectAttemptMessages(work: ConversationWorkState): InboundMessage[] { - const messages = work.messages; - const interrupts = messages.filter( - (message) => message.delivery === "interrupt", - ); - if (interrupts.length > 0) { - return selectContiguousTurnBatch(interrupts); - } - if (work.execution.status === "paused") return []; - const nonEventMessages = messages.filter( - (message) => message.source !== "event", - ); - return selectContiguousTurnBatch( - nonEventMessages.length > 0 ? nonEventMessages : messages, - ); -} - function nudgeIdempotencyKey( reason: string, conversationId: string, @@ -470,7 +435,7 @@ async function processConversationWorkInContext( getChatConfig().conversationWorkSoftYieldAfterMs); let attemptMessageIds: string[] = []; let attemptSelectedMessageIds = new Set(); - let attemptStartMessageIds = new Set(); + let attemptStartInterruptIds = new Set(); let failedWithoutProgress = false; let leaseLost = false; const markLeaseLost = (): void => { @@ -496,17 +461,17 @@ async function processConversationWorkInContext( leaseToken: lease.leaseToken, conversationStore: options.conversationStore, handle: async (messages) => { - // Pending work that was not selected when the attempt started belongs - // to a later actor-scoped attempt. Selected or newly arrived - // interrupts remain eligible for this attempt's drain. A `deferred` - // drain scopes the messages itself. + // Interrupts that were pending but not selected when the attempt + // started belong to a later actor-scoped attempt. Selected, newly + // arrived, and newly steered interrupts remain eligible for this + // attempt's drain. A `deferred` drain scopes the messages itself. const candidates = drainOptions?.deferred ? messages : messages.filter( (message) => message.delivery === "interrupt" && (attemptSelectedMessageIds.has(message.inboundMessageId) || - !attemptStartMessageIds.has(message.inboundMessageId)), + !attemptStartInterruptIds.has(message.inboundMessageId)), ); if (candidates.length === 0) { return []; @@ -601,8 +566,10 @@ async function processConversationWorkInContext( if (!runId) { throw new Error(`Conversation run is missing for ${conversationId}`); } - attemptStartMessageIds = new Set( - leasedWork.messages.map((message) => message.inboundMessageId), + attemptStartInterruptIds = new Set( + leasedWork.messages + .filter((message) => isAttemptInterrupt(message, leasedWork)) + .map((message) => message.inboundMessageId), ); if (hasRun && shouldYield()) { diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 63e3879adf..5547cff32c 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -26,6 +26,7 @@ import { requestConversationWork, releaseConversationWork, startConversationWork, + steerWebPendingMessages, stopConversationWork, type InboundMessage, } from "@/chat/task-execution/store"; @@ -36,6 +37,7 @@ import { import { CONVERSATION_WORK_DEFER_DELAY_MS, processConversationWork, + type ConversationWorkerContext, } from "@/chat/task-execution/worker"; import { processConversationQueueMessage } from "@/chat/task-execution/vercel-callback"; import { closeDb } from "@/chat/db"; @@ -1599,22 +1601,45 @@ describe("conversation work execution", () => { }), nowMs: 2_100, }); - const drained: string[][] = []; + await appendInboundMessage({ + message: inboundMessage("m3", { + createdAtMs: 3_000, + delivery: "defer", + input: { authorId: "dashboard:reader", text: "steer me" }, + receivedAtMs: 3_100, + source: "web", + }), + nowMs: 3_100, + }); + // Each entry is the run number and the ids that run drained. + const drained: string[] = []; + let runs = 0; + const drain = (context: ConversationWorkerContext) => + context.attempt.drain(async (messages) => { + const ids = messages.map((message) => message.inboundMessageId); + drained.push(`${runs}:${ids.join(",")}`); + return ids; + }); await expect( processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - await context.attempt.drain(async (messages) => { - drained.push(messages.map((message) => message.inboundMessageId)); - return messages.map((message) => message.inboundMessageId); + runs += 1; + await drain(context); + // A queued message steered after the attempt started joins it. + await steerWebPendingMessages({ + authorId: "dashboard:reader", + conversationId: CONVERSATION_ID, + inboundMessageIds: ["m3"], }); + await drain(context); return { status: "completed" }; }, }), ).resolves.toEqual({ status: "pending_requeued" }); - expect(drained).toEqual([["m1"]]); + expect(drained).toEqual(["1:m1", "1:m3"]); const state = await getConversationWorkState({ conversationId: CONVERSATION_ID, }); @@ -1626,6 +1651,53 @@ describe("conversation work execution", () => { ]); }); + it("resumes a yielded Turn before a newly steered web message", async () => { + const queue = createConversationWorkQueueTestAdapter(); + const webMessage = (id: string, authorId: string) => + inboundMessage(id, { + delivery: "defer", + input: { authorId, text: id }, + source: "web", + }); + await appendInboundMessage({ + message: webMessage("a1", "dashboard:a"), + nowMs: 1_000, + }); + const attempts: string[][] = []; + await processConversationWork(conversationQueueMessage(), { + queue, + run: async (context) => { + attempts.push( + context.attempt.messages.map((message) => message.inboundMessageId), + ); + await context.attempt.ack(); + await appendInboundMessage({ + message: webMessage("b1", "dashboard:b"), + }); + await steerWebPendingMessages({ + authorId: "dashboard:b", + conversationId: CONVERSATION_ID, + inboundMessageIds: ["b1"], + }); + return { status: "yielded" }; + }, + }); + + await processConversationWork(conversationQueueMessage(), { + queue, + run: async (context) => { + attempts.push( + context.attempt.messages.map((message) => message.inboundMessageId), + ); + await context.attempt.ack(); + return { status: "completed" }; + }, + }); + + // A second Turn must not start while the first waits to resume. + expect(attempts).toEqual([["a1"], [], ["b1"]]); + }); + it("does not rewrite mailbox state for an empty drain result", async () => { const events: string[] = []; const conversationStore = metadataEventsStore(events); 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 ed253a4b27..a06fd36d92 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 @@ -18,7 +18,11 @@ import { requestConversationWork, startConversationWork, } from "@/chat/task-execution/store"; -import { processConversationWork } from "@/chat/task-execution/worker"; +import { + processConversationWork, + type ConversationWorkerContext, + type ConversationWorkerResult, +} from "@/chat/task-execution/worker"; import { processConversationQueueMessage } from "@/chat/task-execution/vercel-callback"; import { createEventInboundMessage } from "@/chat/events/notification"; import { @@ -52,6 +56,7 @@ import { SLACK_BOT_USER_ID, createNoopSlackWebhookRuntime, createSlackAdapterFixture, + inboundMessage, type ConversationWorkQueueTestAdapter, handleSlackWebhookAndFlush, slackEnvelope, @@ -66,23 +71,34 @@ interface ProcessQueuedSlackWorkArgs { nowMs?: () => number; queue: ConversationWorkQueueTestAdapter; runNextPausedTurn?: SlackWorkerOptions["runNextPausedTurn"]; + /** Run a non-Slack attempt, as the app gives it to another worker. */ + runOtherSource?: ( + context: ConversationWorkerContext, + ) => Promise; softYieldAfterMs?: number; runtime: SlackWorkerOptions["runtime"]; state: StateAdapter; } function processNextQueuedSlackWork(args: ProcessQueuedSlackWorkArgs) { + const slackWorker = createSlackConversationWorker({ + getSlackAdapter: args.getSlackAdapter, + lookupSlackUser: args.lookupSlackUser, + runNextPausedTurn: args.runNextPausedTurn ?? (async () => false), + runtime: args.runtime, + state: args.state, + }); + const runOtherSource = args.runOtherSource; return processConversationQueueMessage(args.queue.takeMessage(), { nowMs: args.nowMs, queue: args.queue, softYieldAfterMs: args.softYieldAfterMs, - run: createSlackConversationWorker({ - getSlackAdapter: args.getSlackAdapter, - lookupSlackUser: args.lookupSlackUser, - runNextPausedTurn: args.runNextPausedTurn ?? (async () => false), - runtime: args.runtime, - state: args.state, - }), + run: runOtherSource + ? async (context) => + context.attempt.messages[0]?.source === "slack" + ? await slackWorker(context) + : await runOtherSource(context) + : slackWorker, state: args.state, }); } @@ -895,6 +911,69 @@ describe("Slack conversation work execution", () => { }); }); + it("leaves a steered web message pending during a Slack turn", async () => { + const queue = createConversationWorkQueueTestAdapter(); + const state = getStateAdapter(); + await state.connect(); + const slackAdapter = createSlackAdapterFixture(); + await handleSlackWebhookAndFlush({ + request: slackWebhookRequest( + slackEnvelope({ + text: `<@${SLACK_BOT_USER_ID}> first`, + ts: "1712345.0001", + }), + ), + services: { + getSlackAdapter: () => slackAdapter, + queue, + runtime: createNoopSlackWebhookRuntime(), + state, + }, + }); + + const injected: string[][] = []; + const runtime: SlackWorkerOptions["runtime"] = { + handleNewMention: async (_thread, _message, hooks) => { + await hooks.ack?.(); + // A dashboard reader steers their queued message during the Slack turn. + await appendInboundMessage({ + message: inboundMessage("web-steered", { + input: { authorId: "dashboard:reader", text: "steer from web" }, + source: "web", + }), + state, + }); + await hooks.drainSteeringMessages?.(async (steering) => { + injected.push(steering.map((candidate) => candidate.message.id)); + return steering.map((candidate) => candidate.inboundMessageId); + }); + }, + handleSubscribedMessage: async () => { + throw new Error("unexpected subscribed route"); + }, + }; + // The app gives web input to its own worker after the Slack turn. + const webAttempts: string[][] = []; + await expect( + processNextQueuedSlackWork({ + getSlackAdapter: () => slackAdapter, + queue, + runOtherSource: async (context) => { + webAttempts.push( + context.attempt.messages.map((message) => message.inboundMessageId), + ); + await context.attempt.ack(); + return { status: "completed" }; + }, + runtime, + state, + }), + ).resolves.toEqual({ status: "completed" }); + + expect(injected).toEqual([]); + expect(webAttempts).toEqual([["web-steered"]]); + }); + it("leaves events deferred during an active turn", async () => { const queue = createConversationWorkQueueTestAdapter(); let currentNowMs = 1_000; diff --git a/packages/junior/tests/integration/api/conversations/steer-pending-messages.test.ts b/packages/junior/tests/integration/api/conversations/steer-pending-messages.test.ts new file mode 100644 index 0000000000..5fc8bc19b3 --- /dev/null +++ b/packages/junior/tests/integration/api/conversations/steer-pending-messages.test.ts @@ -0,0 +1,79 @@ +import { afterEach, describe, expect, it } from "vitest"; +import { Hono } from "hono"; +import { createJuniorApi, type JuniorApiVariables } from "@/api"; +import { + conversationPendingMessagesReportSchema, + steerConversationPendingMessagesResponseSchema, +} from "@/api/schema"; +import { + appendAndEnqueueWebMessage, + createAndEnqueueConversation, +} from "@/chat/conversations/web-input"; +import { closeDb } from "@/chat/db"; +import { + closeConversationFixture, + createConversationFixture, +} from "../../../fixtures/conversation"; +import { testViewer } from "../../../fixtures/user"; + +function apiFor(email: string) { + const app = new Hono<{ Variables: JuniorApiVariables }>(); + app.use("*", async (context, next) => { + context.set("viewer", testViewer(email)); + await next(); + }); + app.route("/", createJuniorApi()); + return app; +} + +describe("conversation steer pending messages API", () => { + afterEach(async () => { + await closeConversationFixture(); + await closeDb(); + }); + + it("lets only the author's participant session steer a queued web message", async () => { + const { actor, conversationStore, queue, state } = + await createConversationFixture(); + const created = await createAndEnqueueConversation( + { actor, idempotencyKey: "steer-api-root", message: "first" }, + { conversationStore, queue, state }, + ); + const queued = await appendAndEnqueueWebMessage( + { + actor, + conversationId: created.conversationId, + idempotencyKey: "steer-api-second", + message: "second", + }, + { conversationStore, queue, state }, + ); + const path = `http://localhost/api/conversations/${encodeURIComponent(created.conversationId)}/pending-messages`; + const steer = (email: string) => + apiFor(email).request(`${path}/steer`, { + body: JSON.stringify({ inboundMessageIds: [queued.messageId] }), + headers: { "content-type": "application/json" }, + method: "POST", + }); + + expect((await steer("mallory@example.com")).status).toBe(403); + + const response = await steer(actor.email); + expect(response.status).toBe(200); + expect( + steerConversationPendingMessagesResponseSchema.parse( + await response.json(), + ).steeredInboundMessageIds, + ).toEqual([queued.messageId]); + + const report = conversationPendingMessagesReportSchema.parse( + await (await apiFor(actor.email).request(path)).json(), + ); + expect( + report.messages.map((message) => [message.text, message.delivery]), + ).toEqual([ + ["first", "defer"], + ["second", "interrupt"], + ]); + }); +});