fix(slack): treat steering stop as hard stop

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: {
     >
       <ConversationComposer
         draftId={props.conversationId}
+        footerStart={
+          props.running ? (
+            <button
+              className="rounded border border-red-300/30 px-2 py-1 font-mono text-xs text-red-200 hover:bg-red-300/10 disabled:opacity-50"
+              disabled={stopConversation.isPending}
+              type="button"
+              onClick={() => stopConversation.mutate()}
+            >
+              {stopConversation.isPending ? "Stopping…" : "Stop"}
+            </button>
+          ) : undefined
+        }
         label="Continue this conversation"
         submitLabel="Send"
         onFocus={onComposerFocus}

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();

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,

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<StopConversationResponse> {
+  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);
+  }
+}

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,

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
 >;

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({

packages/junior/src/chat/providers/slack/input.ts

@@ -82,42 +82,57 @@ 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),
     ...(slackTs ? { slackTs } : undefined),
   };
 }
 
-/** 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,

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",