diff --git a/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx b/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx index 1419709b72..5e53f35203 100644 --- a/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx +++ b/packages/junior-dashboard/src/client/conversations/ConversationPage.tsx @@ -18,6 +18,7 @@ import { useArchiveConversation, useCancelConversationPendingMessages, useConversationData, + useStopConversation, type PendingArchiveConversationUpdate, } from "./queries"; import type { ConversationMailboxMessage } from "./conversationOutbox"; @@ -284,6 +285,7 @@ export function ConversationPage(props: { // every 2s; a prop would bust footer memo while the reader types. pendingGeneratedAtRef={pendingGeneratedAtRef} pendingMessages={detail.pendingMessages} + running={live} /> ) : undefined } @@ -310,11 +312,13 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { pendingAuthorization?: ConversationPendingMessagesReport["authorization"]; pendingGeneratedAtRef: { current: string | undefined }; pendingMessages: readonly ConversationMailboxMessage[]; + running: boolean; }) { const appendMessage = useAppendConversationMessage(props.conversationId); const cancelPendingMessages = useCancelConversationPendingMessages( props.conversationId, ); + const stopConversation = useStopConversation(props.conversationId); // 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); @@ -419,6 +423,18 @@ const ConversationReplyFooter = memo(function ConversationReplyFooter(props: { > stopConversation.mutate()} + > + {stopConversation.isPending ? "Stopping…" : "Stop"} + + ) : undefined + } label="Continue this conversation" submitLabel="Send" onFocus={onComposerFocus} diff --git a/packages/junior-dashboard/src/client/conversations/queries.ts b/packages/junior-dashboard/src/client/conversations/queries.ts index 55c853a0c9..5f61c3bd22 100644 --- a/packages/junior-dashboard/src/client/conversations/queries.ts +++ b/packages/junior-dashboard/src/client/conversations/queries.ts @@ -21,6 +21,7 @@ import { conversationDetailReportSchema, conversationEventPageSchema, conversationPendingMessagesReportSchema, + stopConversationResponseSchema, } from "@sentry/junior/api/schema"; import { @@ -240,6 +241,25 @@ export function useAppendConversationMessage(conversationId: string) { }); } +/** Stop the active Turn for the open conversation. */ +export function useStopConversation(conversationId: string) { + const queryClient = useQueryClient(); + return useMutation({ + mutationFn: () => + post( + stopConversationResponseSchema, + `/api/conversations/${encodeURIComponent(conversationId)}/stop`, + {}, + ), + onSettled: async () => { + await queryClient.invalidateQueries({ + exact: true, + queryKey: conversationDetailQueryKey(conversationId), + }); + }, + }); +} + /** Cancel accepted human-facing mailbox rows for the open conversation. */ export function useCancelConversationPendingMessages(conversationId: string) { const queryClient = useQueryClient(); diff --git a/packages/junior/src/api/conversations/routes.ts b/packages/junior/src/api/conversations/routes.ts index 152b15c07b..7c5916c4d2 100644 --- a/packages/junior/src/api/conversations/routes.ts +++ b/packages/junior/src/api/conversations/routes.ts @@ -20,6 +20,7 @@ import { conversationStatsReportSchema, createConversationBodySchema, createConversationMessageBodySchema, + stopConversationResponseSchema, } from "../schema/conversation"; import { validateRequest } from "../validation"; import { requireViewer } from "../viewer"; @@ -38,6 +39,7 @@ import { readConversationFeed } from "./list"; import { cancelConversationPendingMessagesForViewer } from "./cancel-pending-messages"; import { requireConversationPendingMessages } from "./pending-messages"; import { readConversationStats } from "./stats"; +import { stopConversationForViewer } from "./stop"; /** Create the HTTP routes owned by the conversations API. */ export function createConversationRoutes(options: { @@ -113,6 +115,24 @@ export function createConversationRoutes(options: { }, ); + app.post( + "/:conversationId/stop", + requireViewer, + validateRequest( + "param", + conversationParamsSchema, + "Invalid route parameters.", + ), + async (context) => { + const viewer = context.get("viewer"); + const { conversationId } = context.req.valid("param"); + return jsonResponse( + stopConversationResponseSchema, + await stopConversationForViewer(viewer, conversationId), + ); + }, + ); + app.patch( "/:conversationId/archive", requireViewer, diff --git a/packages/junior/src/api/conversations/stop.ts b/packages/junior/src/api/conversations/stop.ts new file mode 100644 index 0000000000..45f9b32bd3 --- /dev/null +++ b/packages/junior/src/api/conversations/stop.ts @@ -0,0 +1,36 @@ +import type { User } from "@sentry/junior-plugin-api"; +import { stopConversationTurn } from "@/chat/conversations/stop"; +import { getConversationStore, getDb } from "@/chat/db"; +import { getVercelConversationWorkQueue } from "@/chat/task-execution/vercel-queue"; +import { throwApiError } from "../http"; +import type { StopConversationResponse } from "../schema/conversation"; +import { readConversationAccessFromSql } from "./access"; + +/** Stop the active Conversation Turn for one participant. */ +export async function stopConversationForViewer( + viewer: User, + conversationId: string, +): Promise { + const conversationStore = getConversationStore(); + if (!(await conversationStore.get({ conversationId }))) { + throwApiError(404, "Conversation not found."); + } + const access = await readConversationAccessFromSql( + getDb(), + [conversationId], + viewer, + ); + if (!access.get(conversationId)?.isParticipant) { + throwApiError(403, "Only conversation participants can stop this work."); + } + try { + const result = await stopConversationTurn({ + conversationId, + conversationStore, + queue: getVercelConversationWorkQueue(), + }); + return { conversationId, status: result.status }; + } catch (error) { + throwApiError(500, "Unable to stop this conversation.", error); + } +} diff --git a/packages/junior/src/api/schema.ts b/packages/junior/src/api/schema.ts index b970c115d8..6ed23cb01d 100644 --- a/packages/junior/src/api/schema.ts +++ b/packages/junior/src/api/schema.ts @@ -43,6 +43,7 @@ export { conversationSummaryReportSchema, createConversationBodySchema, createConversationMessageBodySchema, + stopConversationResponseSchema, } from "./schema/conversation"; export type { AcceptedConversationMessage, @@ -61,6 +62,7 @@ export type { ConversationPendingMessage, ConversationPendingMessageDelivery, ConversationPendingMessagesReport, + StopConversationResponse, ConversationReportEvent, ConversationReportEventData, ConversationReportStatus, diff --git a/packages/junior/src/api/schema/conversation.ts b/packages/junior/src/api/schema/conversation.ts index a2a3c245dc..36006d973b 100644 --- a/packages/junior/src/api/schema/conversation.ts +++ b/packages/junior/src/api/schema/conversation.ts @@ -174,6 +174,14 @@ export const cancelConversationPendingMessagesBodySchema = z }) .strict(); +/** Result of stopping the active Conversation Turn. */ +export const stopConversationResponseSchema = z + .object({ + conversationId: z.string().min(1), + status: z.enum(["no_work", "requested"]), + }) + .strict(); + /** Result of cancelling accepted human-facing mailbox rows. */ export const cancelConversationPendingMessagesResponseSchema = z .object({ @@ -906,6 +914,9 @@ export type ConversationPendingMessage = z.infer< export type ConversationPendingMessagesReport = z.infer< typeof conversationPendingMessagesReportSchema >; +export type StopConversationResponse = z.infer< + typeof stopConversationResponseSchema +>; export type CancelConversationPendingMessagesBody = z.infer< typeof cancelConversationPendingMessagesBodySchema >; diff --git a/packages/junior/src/chat/ingress/slack-webhook.ts b/packages/junior/src/chat/ingress/slack-webhook.ts index 43fee26636..44d4c26c04 100644 --- a/packages/junior/src/chat/ingress/slack-webhook.ts +++ b/packages/junior/src/chat/ingress/slack-webhook.ts @@ -12,6 +12,12 @@ import type { ConversationStore } from "@/chat/conversations/store"; import { getConversationEventStore, getConversationStore } from "@/chat/db"; import { appendConversationMessages } from "@/chat/conversations/messages"; import { stopConversationTurn } from "@/chat/conversations/stop"; +import { buildSteeringPiMessage } from "@/chat/agent/prompt"; +import { historyItemFromPiMessage } from "@/chat/pi/conversation-events"; +import { + slackMessageActor, + slackMessageProvenance, +} from "@/chat/providers/slack/input"; import { cancelSubscriptions } from "@/chat/events/store"; import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; import { appendAndEnqueueInboundMessage } from "@/chat/task-execution/store"; @@ -27,12 +33,14 @@ import { import { textMentionsBot } from "@/chat/ingress/bot-mention"; import { isExperimentalFeatureEnabled } from "@/chat/experimental"; import { recordSkippedConversationMessage } from "@/chat/runtime/conversation-message"; +import { stripLeadingBotMention } from "@/chat/runtime/thread-context"; import { getThreadStopDecision, SubscribedReplyReason, } from "@/chat/services/subscribed-decision"; import { coerceThreadConversationState } from "@/chat/state/conversation"; import { parseContent } from "@/chat/slack/message/content"; +import { stripLeadingSteeringOverride } from "@/chat/slack/message-control"; import { stopSlackThread } from "@/chat/slack/thread-stop"; import { extractMessageChangedMention, @@ -340,6 +348,31 @@ async function handleSlackThreadStop(args: { await cancelSubscriptions({ conversationId, state: args.state }); const content = parseContent(args.message); + const stopText = stripLeadingBotMention( + stripLeadingSteeringOverride(content.topLevelText), + { + botUserId: args.adapter.botUserId, + stripLeadingSlackMentionToken: Boolean(args.message.isMention), + }, + ); + const provenance = slackMessageProvenance( + args.message, + args.installation.teamId ?? "", + ); + const stopMessage = buildSteeringPiMessage({ + actor: slackMessageActor(args.message), + provenance, + text: stopText, + timestampMs: args.message.metadata.dateSent.getTime(), + }); + await getConversationEventStore().append(conversationId, [ + { + idempotencyKey: `slack-stop:${args.message.id}:agent`, + createdAtMs: args.message.metadata.dateSent.getTime(), + data: historyItemFromPiMessage(stopMessage, provenance), + }, + ]); + const conversation = coerceThreadConversationState(undefined); recordSkippedConversationMessage({ conversation, @@ -397,9 +430,10 @@ async function routeParsedMessage(args: { return; } + const stopText = stripLeadingSteeringOverride(args.event.text ?? ""); const stopDecision = getThreadStopDecision({ - rawText: args.event.text ?? "", - text: args.event.text ?? "", + rawText: stopText, + text: stopText, }); if (stopDecision) { await handleSlackThreadStop({ diff --git a/packages/junior/src/chat/providers/slack/input.ts b/packages/junior/src/chat/providers/slack/input.ts index 83c0b7e6d6..5c320cc332 100644 --- a/packages/junior/src/chat/providers/slack/input.ts +++ b/packages/junior/src/chat/providers/slack/input.ts @@ -82,15 +82,14 @@ export function appendRecentMessagesToContext( ); } -/** Return the actor stored with one inbound Slack message. */ -export function inboundMessageActor( - queued: QueuedTurnMessage, +/** Return the actor stored with one Slack message. */ +export function slackMessageActor( + message: Message, ): AgentSteeringMessage["actor"] { - const actor = getMessageActorIdentity(queued.message); - const authorId = - actor?.userId ?? parseActorUserId(queued.message.author.userId); + const actor = getMessageActorIdentity(message); + const authorId = actor?.userId ?? parseActorUserId(message.author.userId); const authorName = actor?.fullName ?? actor?.userName; - const slackTs = getMessageTimestamp(queued.message); + const slackTs = getMessageTimestamp(message); return { ...(authorId ? { authorId } : undefined), ...(authorName ? { authorName } : undefined), @@ -98,26 +97,42 @@ export function inboundMessageActor( }; } -/** Return the authority stored with one inbound Slack message. */ -export function inboundMessageProvenance( +/** Return the actor stored with one inbound Slack message. */ +export function inboundMessageActor( queued: QueuedTurnMessage, +): AgentSteeringMessage["actor"] { + return slackMessageActor(queued.message); +} + +/** Return the authority stored with one Slack message. */ +export function slackMessageProvenance( + message: Message, teamId: string, ): ConversationMessageProvenance { - const identity = getMessageActorIdentity(queued.message); + const identity = getMessageActorIdentity(message); + const userId = parseActorUserId(message.author.userId); const author = identity && "platform" in identity ? identity : createActor( - { userId: parseActorUserId(queued.message.author.userId) }, + { userId }, { platform: "slack", teamId, - userId: parseActorUserId(queued.message.author.userId), + userId, }, ); return instructionProvenanceFor(author); } +/** Return the authority stored with one inbound Slack message. */ +export function inboundMessageProvenance( + queued: QueuedTurnMessage, + teamId: string, +): ConversationMessageProvenance { + return slackMessageProvenance(queued.message, teamId); +} + /** Return the Slack channel name when it is available. */ export async function resolveChannelName( thread: Thread, diff --git a/packages/junior/tests/integration/slack/conversation-turn-steering-behavior.test.ts b/packages/junior/tests/integration/slack/conversation-turn-steering-behavior.test.ts index 612649759b..1830f4916a 100644 --- a/packages/junior/tests/integration/slack/conversation-turn-steering-behavior.test.ts +++ b/packages/junior/tests/integration/slack/conversation-turn-steering-behavior.test.ts @@ -712,9 +712,10 @@ describe("Slack behavior: durable turn steering", () => { handleSlackWebhookAndFlush({ request: slackWebhookRequest( makeMessageEvent({ - eventType: "message", - text: "stop", + eventType: "app_mention", + text: `<@${SLACK_BOT_USER_ID}>!! stop`, ts: "1712345.000500", + user: "U999", }), ), services, @@ -744,6 +745,10 @@ describe("Slack behavior: durable turn steering", () => { expect(await state.isSubscribed(conversationId)).toBe(false); await expect(listWatches({ conversationId, state })).resolves.toEqual([]); expect(agentRuns).toHaveLength(1); + await expect(loadMessageProvenance(conversationId, "stop")).resolves.toEqual({ + authority: "instruction", + actor: expect.objectContaining({ userId: "U999" }), + }); expect(reactionTargetsByName("eyes")).toEqual([ { @@ -766,7 +771,7 @@ describe("Slack behavior: durable turn steering", () => { }), }), expect.objectContaining({ - text: "stop", + text: `@${SLACK_BOT_USER_ID}!! stop`, meta: expect.objectContaining({ replied: false, skippedReason: "thread_opt_out:stop",