ref(slack): remove reply executor services

packages/junior-evals/src/behavior-harness.ts

@@ -56,7 +56,10 @@ import { completedAgentRun } from "@/chat/runtime/agent-run-outcome";
 import type { AgentRunner } from "@/chat/runtime/agent-runner";
 import { addAgentTurnUsage, type AgentTurnUsage } from "@/chat/usage";
 import { runNextPausedTurn } from "@/chat/task-execution/paused-turn";
-import { wakePausedTurn } from "@/chat/task-execution/turn-wake";
+import {
+  createPausedTurns,
+  wakePausedTurn,
+} from "@/chat/task-execution/turn-wake";
 import { ACTIVE_TURN_COMPACTION_SUMMARY_PREFIX } from "@/chat/services/context-compaction-marker";
 import { TURN_CONTEXT_TAG } from "@/chat/turn-context-tag";
 import { listIncompleteScheduledRuns } from "@/chat/scheduled-tasks/runs";
@@ -75,10 +78,6 @@ import {
   createAgentDispatchConversationWorker,
   createAgentDispatchWorkRouter,
 } from "@/chat/agent-dispatch/work";
-import {
-  ConversationTurnLifecycleService,
-  type ConversationTurnLifecycle,
-} from "@/chat/conversations/turn-lifecycle";
 import {
   getDispatchInputMessageId,
   getDispatchRecord,
@@ -1680,7 +1679,6 @@ function buildRuntimeServices(
   observations: RuntimeObservations,
   conversationWorkQueue: ConversationWorkQueueTestAdapter,
   steeringDelivery: SteeringDelivery,
-  turnLifecycle: ConversationTurnLifecycle,
   signal?: AbortSignal,
 ): JuniorRuntimeServiceOverrides {
   const replyTexts = scenario.overrides?.reply_texts ?? [];
@@ -2017,22 +2015,6 @@ function buildRuntimeServices(
         }
       },
     },
-    replyExecutor: {
-      turnLifecycle,
-      wakePausedTurn: async (request) => {
-        await wakePausedTurn(request, {
-          queue: conversationWorkQueue,
-          state: env.stateAdapter,
-        });
-      },
-      scheduleSessionCompletedPluginTasks: async (params) => {
-        await scheduleSessionCompletedPluginTasks(params, {
-          send: async (message) => {
-            await processEvalPluginTask(message);
-          },
-        });
-      },
-    },
     visionContext: {
       listThreadReplies: async ({ channelId, threadTs, targetMessageTs }) => {
         const threadId = buildRuntimeThreadId({
@@ -2703,17 +2685,13 @@ export async function runEvalScenario(
 
     const conversationWorkQueue = createConversationWorkQueueTestAdapter();
     const steeringDelivery: SteeringDelivery = {};
-    const turnLifecycle = new ConversationTurnLifecycleService(
-      getConversationEventStore(),
-    );
     const services = buildRuntimeServices(
       scenario,
       env,
       threadRecordsById,
       observations,
       conversationWorkQueue,
       steeringDelivery,
-      turnLifecycle,
       options.signal,
     );
     const evalAgentRunner = services.agentRunner;
@@ -2723,6 +2701,11 @@ export async function runEvalScenario(
 
     const slackRuntime = createSlackRuntime({
       getSlackAdapter: () => slackAdapter as any,
+      pausedTurns: createPausedTurns({
+        queue: conversationWorkQueue,
+        state: env.stateAdapter,
+      }),
+      sendPluginTask: processEvalPluginTask,
       services,
     });
 

packages/junior/src/chat/app/conversation-work.ts

@@ -6,10 +6,7 @@ import type { ConversationWorkQueue } from "@/chat/task-execution/queue";
 import type { VercelConversationWorkCallbackOptions } from "@/chat/task-execution/vercel-callback";
 import { createSlackConversationWorker } from "@/chat/task-execution/slack-work";
 import { runNextPausedTurn } from "@/chat/task-execution/paused-turn";
-import {
-  getPausedTurnRequest,
-  wakePausedTurn,
-} from "@/chat/task-execution/turn-wake";
+import { createPausedTurns } from "@/chat/task-execution/turn-wake";
 import {
   buildDispatchRoutingContext,
   createAgentDispatchConversationWorker,
@@ -56,30 +53,18 @@ export function createConversationWork(
   runtime: ReturnType<typeof createSlackRuntime>;
 } {
   const apiTurnCancellation = createApiTurnCancellation();
-  const services: JuniorRuntimeServiceOverrides = {
-    ...options.services,
-    agentRunner: options.agentRunner,
-    replyExecutor: {
-      ...options.services?.replyExecutor,
-      getPausedTurnRequest:
-        options.services?.replyExecutor?.getPausedTurnRequest ??
-        (async (request) =>
-          await getPausedTurnRequest({
-            ...request,
-            conversationStore: options.conversationStore,
-          })),
-      wakePausedTurn:
-        options.services?.replyExecutor?.wakePausedTurn ??
-        (async (request) =>
-          await wakePausedTurn(request, {
-            queue: options.queue,
-            state: options.state,
-          })),
-    },
-  };
+  const pausedTurns = createPausedTurns({
+    conversationStore: options.conversationStore,
+    queue: options.queue,
+    ...(options.state ? { state: options.state } : undefined),
+  });
   const runtime = createSlackRuntime({
     getSlackAdapter: options.getSlackAdapter,
-    services,
+    pausedTurns,
+    services: {
+      ...options.services,
+      agentRunner: options.agentRunner,
+    },
   });
   const slackWorker = createSlackConversationWorker({
     getSlackAdapter: options.getSlackAdapter,
@@ -89,9 +74,7 @@ export function createConversationWork(
         conversationId,
         {
           agentRunner: options.agentRunner,
-          scheduleSessionCompletedPluginTasks:
-            services.replyExecutor?.scheduleSessionCompletedPluginTasks,
-          wakePausedTurn: services.replyExecutor?.wakePausedTurn,
+          wakePausedTurn: pausedTurns.wake,
         },
         runOptions,
       ),
@@ -106,9 +89,7 @@ export function createConversationWork(
           agentRunner: options.agentRunner,
           inputMessageIds: [getDispatchInputMessageId(dispatch.id)],
           routingContext: buildDispatchRoutingContext(dispatch),
-          scheduleSessionCompletedPluginTasks:
-            services.replyExecutor?.scheduleSessionCompletedPluginTasks,
-          wakePausedTurn: services.replyExecutor?.wakePausedTurn,
+          wakePausedTurn: pausedTurns.wake,
         },
         { shouldYield: hooks.shouldYield },
       );

packages/junior/src/chat/app/factory.ts

@@ -37,10 +37,20 @@ import {
   ensureSlackMessageActorIdentity,
   getMessageActorIdentity,
 } from "@/chat/services/message-actor-identity";
+import { lookupSlackUser } from "@/chat/slack/user";
+import { ConversationTurnLifecycleService } from "@/chat/conversations/turn-lifecycle";
+import { getConversationEventStore } from "@/chat/db";
+import type { ScheduleSessionCompletedPluginTasksOptions } from "@/chat/plugins/task-runner";
+import {
+  createPausedTurns,
+  type PausedTurns,
+} from "@/chat/task-execution/turn-wake";
 
 export interface CreateSlackRuntimeOptions {
   getSlackAdapter: () => SlackAdapter;
   now?: () => number;
+  pausedTurns?: PausedTurns;
+  sendPluginTask?: ScheduleSessionCompletedPluginTasksOptions["send"];
   services?: JuniorRuntimeServiceOverrides;
 }
 
@@ -80,6 +90,9 @@ function upsertSkippedConversationMessage(
 
 export function createSlackRuntime(options: CreateSlackRuntimeOptions) {
   const services = createJuniorRuntimeServices(options.services);
+  const turnLifecycle = new ConversationTurnLifecycleService(
+    getConversationEventStore(),
+  );
   const prepareTurnState = createPrepareTurnState({
     compactConversationIfNeeded:
       services.conversationMemory.compactConversationIfNeeded,
@@ -104,18 +117,21 @@ export function createSlackRuntime(options: CreateSlackRuntimeOptions) {
             ensureSlackMessageActorIdentity(
               message,
               destination.teamId,
-              services.replyExecutor.lookupSlackUser,
+              lookupSlackUser,
             ),
           ),
       );
     },
   });
   const executeSlackTurn = createSlackTurn({
+    contextCompactor: services.contextCompactor,
     executeTurn: services.executeTurn,
     getSlackAdapter: options.getSlackAdapter,
+    pausedTurns: options.pausedTurns ?? createPausedTurns(),
     prepareTurnState,
     resolveUserAttachments: services.visionContext.resolveUserAttachments,
-    services: services.replyExecutor,
+    sendPluginTask: options.sendPluginTask,
+    turnLifecycle,
   });
 
   const runtime = createSlackTurnRuntime<
@@ -127,8 +143,7 @@ export function createSlackRuntime(options: CreateSlackRuntimeOptions) {
     getBotUserId: () => options.getSlackAdapter().botUserId,
     modelId: defaultModelId(botConfig),
     now: options.now ?? (() => Date.now()),
-    failConversationTurn: (input) =>
-      services.replyExecutor.turnLifecycle.fail(input),
+    failConversationTurn: (input) => turnLifecycle.fail(input),
     prepareTurnState,
     persistPreparedState: async ({ thread, preparedState }) => {
       await persistThreadState(thread, {

packages/junior/src/chat/app/services.ts

@@ -1,11 +1,6 @@
 import { completeObject, completeText } from "@/chat/pi/client";
 import { executeAgentRun as executeAgentRunImpl } from "@/chat/agent";
 import type { SandboxEgressTracePropagationConfig } from "@/chat/sandbox/egress/tracing";
-import {
-  getPausedTurnRequest,
-  wakePausedTurn,
-} from "@/chat/task-execution/turn-wake";
-import { scheduleSessionCompletedPluginTasks } from "@/chat/plugins/task-runner";
 import {
   createConversationMemoryService,
   type ConversationMemoryDeps,
@@ -18,13 +13,11 @@ import {
 } from "@/chat/services/context-compaction";
 import { downloadPrivateSlackFile } from "@/chat/slack/client";
 import { listThreadReplies } from "@/chat/slack/channel";
-import { lookupSlackUser } from "@/chat/slack/user";
 import {
   createSubscribedReplyPolicy,
   type SubscribedReplyPolicy,
   type SubscribedReplyPolicyDeps,
 } from "@/chat/services/subscribed-reply-policy";
-import type { SlackTurnServices } from "@/chat/providers/slack/turn";
 import {
   createVisionContextService,
   type VisionContextDeps,
@@ -35,16 +28,13 @@ import {
   type AgentRunner,
 } from "@/chat/runtime/agent-runner";
 import { executeTurn, type ExecuteTurn } from "@/chat/runtime/turn-execution";
-import { ConversationTurnLifecycleService } from "@/chat/conversations/turn-lifecycle";
-import { getConversationEventStore } from "@/chat/db";
 import { bindSpawnAgent } from "@/chat/agent-invocations/spawn";
 import { getVercelConversationWorkQueue } from "@/chat/task-execution/vercel-queue";
 
 export interface JuniorRuntimeServices {
   conversationMemory: ConversationMemoryService;
   contextCompactor: ContextCompactor;
   executeTurn: ExecuteTurn;
-  replyExecutor: SlackTurnServices;
   subscribedReplyPolicy: SubscribedReplyPolicy;
   visionContext: VisionContextService;
 }
@@ -53,7 +43,6 @@ export interface JuniorRuntimeServiceOverrides {
   agentRunner?: AgentRunner;
   conversationMemory?: Partial<ConversationMemoryDeps>;
   contextCompactor?: Partial<ContextCompactorDeps>;
-  replyExecutor?: Partial<SlackTurnServices>;
   subscribedReplyPolicy?: Partial<SubscribedReplyPolicyDeps>;
   sandbox?: {
     tracePropagation?: SandboxEgressTracePropagationConfig;
@@ -92,23 +81,6 @@ export function createJuniorRuntimeServices(
     contextCompactor,
     executeTurn: async (run, saveResult, timeoutMs) =>
       await executeTurn(agentRunner, run, saveResult, timeoutMs),
-    replyExecutor: {
-      contextCompactor:
-        overrides.replyExecutor?.contextCompactor ?? contextCompactor,
-      getPausedTurnRequest:
-        overrides.replyExecutor?.getPausedTurnRequest ?? getPausedTurnRequest,
-      lookupSlackUser:
-        overrides.replyExecutor?.lookupSlackUser ?? lookupSlackUser,
-      wakePausedTurn: overrides.replyExecutor?.wakePausedTurn ?? wakePausedTurn,
-      scheduleSessionCompletedPluginTasks:
-        overrides.replyExecutor?.scheduleSessionCompletedPluginTasks ??
-        (async (params) => {
-          await scheduleSessionCompletedPluginTasks(params);
-        }),
-      turnLifecycle:
-        overrides.replyExecutor?.turnLifecycle ??
-        new ConversationTurnLifecycleService(getConversationEventStore()),
-    },
     subscribedReplyPolicy: createSubscribedReplyPolicy({
       completeObject:
         overrides.subscribedReplyPolicy?.completeObject ?? completeObject,