fix(runtime): Keep deferred messages out of active turns

TERMINOLOGY.md

@@ -13,10 +13,13 @@ Canonical words used across Junior's code and documentation.
 - **Agent input**: the inbound content, context, and runtime metadata selected
   for a turn.
 - **Steering message**: a user message that interrupts the active turn at the
-  next safe boundary and starts a new turn. Messages arriving together may be
-  batched into that turn.
+  next safe boundary and joins that turn. Messages arriving together may be
+  batched into the same turn.
 - **Follow-up message**: a user message that waits for the active turn to finish
   before starting the next turn.
+- **Mailbox delivery**: the durable instruction for an inbound message:
+  `interrupt` is eligible at the next safe boundary, while `defer` follows
+  normal ordering and waits when a turn is already active.
 - **Turn**: one request-to-final-response cycle. It may span multiple runs and
   execution slices; one model invocation is not a turn.
 - **Run**: one bounded attempt to advance a turn. A later run may resume the

packages/junior/src/chat/runtime/slack-runtime.ts

@@ -56,10 +56,7 @@ export interface AssistantLifecycleEvent {
   userId?: string;
 }
 
-type SteeringMode = "defer" | "interrupt";
-
 export interface SteeringCandidateMessage {
-  activeRequest: boolean;
   inboundMessageId: string;
   message: Message;
 }
@@ -69,7 +66,7 @@ export interface ReplyHooks {
   drainSteeringMessages?: (
     accept: (
       messages: SteeringCandidateMessage[],
-    ) => Promise<readonly string[] | void>,
+    ) => Promise<readonly string[]>,
   ) => Promise<void>;
   messageContext?: MessageContext;
   ack?: () => Promise<void>;
@@ -253,7 +250,6 @@ interface SteeringMessageDecision {
   context: TurnContext;
   decision: SubscribedReplyDecision;
   inboundMessageId: string;
-  mode: SteeringMode;
   message: Message;
   text: TurnMessageText;
 }
@@ -262,12 +258,11 @@ interface SteeringMessageSelection {
   accepted: Array<{
     inboundMessageId: string;
     message: Message;
-    mode: SteeringMode;
   }>;
   skipped: SteeringMessageDecision[];
 }
 
-/** Drain mailbox steering messages after classifying interrupt, defer, and skip. */
+/** Drain explicit mailbox interruptions after applying subscribed reply policy. */
 function createAcceptedSteeringDrain(
   hooks: ReplyHooks,
   options: {
@@ -295,18 +290,14 @@ function createAcceptedSteeringDrain(
     await hooks.drainSteeringMessages!(async (messages) => {
       const selection = await options.selectMessages(messages, context);
       await options.onSkipped?.(selection.skipped);
-      // Deferred accepted messages stay pending so a later worker slice handles
-      // them after the active answer is delivered.
-      const interrupted = selection.accepted
-        .filter((accepted) => accepted.mode === "interrupt")
-        .map((accepted) => accepted.message);
+      const interrupted = selection.accepted.map(
+        (accepted) => accepted.message,
+      );
       await accept(getQueuedMessagesFromSlackMessages(interrupted, options));
       interruptedMessages = interrupted;
       await options.onAcceptedForProcessing?.(interrupted);
       return [
-        ...selection.accepted
-          .filter((accepted) => accepted.mode === "interrupt")
-          .map((accepted) => accepted.inboundMessageId),
+        ...selection.accepted.map((accepted) => accepted.inboundMessageId),
         ...selection.skipped.map((skipped) => skipped.inboundMessageId),
       ];
     });
@@ -523,7 +514,6 @@ export function createSlackTurnRuntime<
   ): Promise<{
     context: TurnContext;
     decision: SubscribedReplyDecision;
-    mode: SteeringMode;
     text: TurnMessageText;
   }> => {
     const { message } = candidate;
@@ -541,22 +531,18 @@ export function createSlackTurnRuntime<
       rawText: appendSlackLegacyAttachmentText(message.text, message.raw),
       userText: appendSlackLegacyAttachmentText(strippedUserText, message.raw),
     };
-    const isActiveRequest =
-      candidate.activeRequest || Boolean(message.isMention);
-
     const decision = await deps.decideSubscribedReply({
       rawText: text.rawText,
       text: text.userText,
       conversationContext,
       hasAttachments:
         message.attachments.length > 0 || legacyAttachmentText !== "",
-      isExplicitMention: isActiveRequest,
+      isExplicitMention: true,
       context,
     });
     return {
       context,
       decision,
-      mode: isActiveRequest ? "interrupt" : "defer",
       text,
     };
   };
@@ -633,7 +619,6 @@ export function createSlackTurnRuntime<
         context: decision.context,
         decision: decision.decision,
         inboundMessageId: candidate.inboundMessageId,
-        mode: decision.mode,
         message: candidate.message,
         text: decision.text,
       });
@@ -660,7 +645,6 @@ export function createSlackTurnRuntime<
         .map((message) => ({
           inboundMessageId: message.inboundMessageId,
           message: message.message,
-          mode: message.mode,
         })),
       skipped: selected.filter((message) => !message.decision.shouldReply),
     };

packages/junior/src/chat/task-execution/README.md

@@ -6,14 +6,18 @@ source of truth.
 
 ## State Model
 
-- A conversation mailbox contains normalized pending user work.
+- A conversation mailbox contains normalized pending work with a durable
+  delivery mode: `interrupt` or `defer`.
 - A queue payload identifies the conversation to wake; persisted conversation
-  work owns destination and routing.
+  work owns destination and delivery.
 - A lease grants one worker temporary execution ownership.
 - Check-ins extend active ownership and allow heartbeat recovery to distinguish
   slow work from abandoned work.
 - Delivery state prevents a completed turn from being posted twice.
 
+Schema-v1 mailbox entries migrate to deferred delivery. Schema-v2 entries
+require a valid delivery value and reject invalid pending work.
+
 `state.ts`, `store.ts`, and their runtime schemas define the persisted shapes.
 
 ## Execution
@@ -30,8 +34,15 @@ source of truth.
 6. Terminal delivery or intentional no-reply completion records the delivered
    turn before acknowledging work.
 
-New messages that arrive during a run remain durable and are drained at the
-next safe boundary or subsequent wake-up.
+New messages that arrive during a run remain durable. `interrupt` work is
+eligible at the next safe boundary, while `defer` work follows normal ordering
+and waits for the next turn.
+
+Slack `@` mentions use `interrupt` delivery; all other inbound messages use
+`defer`.
+
+An active turn drains only `interrupt` messages. When a new turn starts,
+interrupt messages are handled before queued deferred work.
 
 ## Queue And Lease Rules
 

packages/junior/src/chat/task-execution/slack-work.ts

@@ -566,6 +566,7 @@ export function createSlackResourceEventInboundMessage(
     createdAtMs: input.event.occurredAtMs,
     destination,
     inboundMessageId: `resource-event:${input.subscription.id}:${input.event.eventKey}`,
+    delivery: "defer",
     source: "resource_event",
     receivedAtMs: Date.now(),
     input: {
@@ -627,18 +628,6 @@ function routeForRecords(records: InboundMessage[]): SlackConversationRoute {
     : "subscribed";
 }
 
-function isSlackAssistantThreadUserMessage(message: Message): boolean {
-  const raw =
-    message.raw && typeof message.raw === "object"
-      ? (message.raw as Record<string, unknown>)
-      : undefined;
-  return (
-    raw?.channel_type === "im" &&
-    typeof raw.thread_ts === "string" &&
-    raw.thread_ts.trim().length > 0
-  );
-}
-
 function isResourceEventNotificationMessage(message: Message): boolean {
   const raw =
     message.raw && typeof message.raw === "object"
@@ -837,31 +826,20 @@ export function createSlackConversationWorker(
             );
           }
         };
-        // Restore stored mailbox entries as Slack steering candidates; the
-        // runtime returns only the inbound ids it handled durably.
+        // Only explicit interruptions enter the active turn. Deferred work
+        // stays pending for a normal turn.
         const drainSteeringMessages = async (
           accept: (
             messages: SteeringCandidateMessage[],
-          ) => Promise<readonly string[] | void>,
+          ) => Promise<readonly string[]>,
         ): Promise<void> => {
           await context.attempt.drain(async (pendingRecords) => {
-            const messages = pendingRecords.map((record) => {
-              const metadata = parseSlackMetadata(record.input.metadata);
-              if (!metadata) {
-                throw new Error(
-                  "Conversation mailbox record is not Slack metadata",
-                );
-              }
-              const message = restoreMessage({ adapter, record });
-              return {
-                activeRequest:
-                  metadata.route === "mention" ||
-                  isSlackAssistantThreadUserMessage(message),
+            return await accept(
+              pendingRecords.map((record) => ({
                 inboundMessageId: record.inboundMessageId,
-                message,
-              };
-            });
-            return await accept(messages);
+                message: restoreMessage({ adapter, record }),
+              })),
+            );
           });
         };
 
@@ -938,6 +916,7 @@ export function buildSlackInboundMessage(args: {
       args.conversationId,
       args.message.id,
     ].join(":"),
+    delivery: args.message.isMention ? "interrupt" : "defer",
     source: "slack",
     createdAtMs: args.message.metadata.dateSent.getTime(),
     receivedAtMs: args.receivedAtMs,