From 302b0d484ef2a601d5d89e4dfba53d2c5dfc0653 Mon Sep 17 00:00:00 2001 From: David Cramer Date: Tue, 25 Aug 2026 18:39:17 -0700 Subject: [PATCH] ref(slack): remove reply executor services --- packages/junior-evals/src/behavior-harness.ts | 35 ++------ .../junior/src/chat/app/conversation-work.ts | 45 +++------- packages/junior/src/chat/app/factory.ts | 23 ++++- packages/junior/src/chat/app/services.ts | 28 ------ .../junior/src/chat/providers/slack/turn.ts | 86 +++++++++---------- .../junior/src/chat/services/persist-retry.ts | 4 +- .../src/chat/task-execution/turn-wake.ts | 23 +++++ .../junior/tests/fixtures/chat-runtime.ts | 15 ++++ .../integration/slack/bot-handlers.test.ts | 48 ++--------- .../message-changed-reply-contract.test.ts | 17 +--- .../slack/message-content-behavior.test.ts | 60 ------------- 11 files changed, 134 insertions(+), 250 deletions(-) diff --git a/packages/junior-evals/src/behavior-harness.ts b/packages/junior-evals/src/behavior-harness.ts index ca0d440db7..5f31ebe2a8 100644 --- a/packages/junior-evals/src/behavior-harness.ts +++ b/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,9 +2685,6 @@ export async function runEvalScenario( const conversationWorkQueue = createConversationWorkQueueTestAdapter(); const steeringDelivery: SteeringDelivery = {}; - const turnLifecycle = new ConversationTurnLifecycleService( - getConversationEventStore(), - ); const services = buildRuntimeServices( scenario, env, @@ -2713,7 +2692,6 @@ export async function runEvalScenario( 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, }); diff --git a/packages/junior/src/chat/app/conversation-work.ts b/packages/junior/src/chat/app/conversation-work.ts index 11596fe83d..74207581c1 100644 --- a/packages/junior/src/chat/app/conversation-work.ts +++ b/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; } { 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 }, ); diff --git a/packages/junior/src/chat/app/factory.ts b/packages/junior/src/chat/app/factory.ts index d81a7710a8..5aa77c2e56 100644 --- a/packages/junior/src/chat/app/factory.ts +++ b/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, { diff --git a/packages/junior/src/chat/app/services.ts b/packages/junior/src/chat/app/services.ts index e4dcbb2543..cfca55037e 100644 --- a/packages/junior/src/chat/app/services.ts +++ b/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,8 +28,6 @@ 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"; @@ -44,7 +35,6 @@ 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; contextCompactor?: Partial; - replyExecutor?: Partial; subscribedReplyPolicy?: Partial; 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, diff --git a/packages/junior/src/chat/providers/slack/turn.ts b/packages/junior/src/chat/providers/slack/turn.ts index 6b3660bc28..17cbd92a3e 100644 --- a/packages/junior/src/chat/providers/slack/turn.ts +++ b/packages/junior/src/chat/providers/slack/turn.ts @@ -104,7 +104,7 @@ import { isResourceEventSlackMessage, RESOURCE_EVENT_SYSTEM_ACTOR, } from "@/chat/resource-events/actor"; -import type { PausedTurnRequest } from "@/chat/task-execution/turn-wake"; +import type { PausedTurns } from "@/chat/task-execution/turn-wake"; import { ConversationTurnBoundaryError, CooperativeTurnYieldError, @@ -147,6 +147,10 @@ import { import { requireSlackDestination } from "@/chat/destination"; import { persistConversationMessages } from "@/chat/conversations/messages"; import type { ConversationTurnLifecycle } from "@/chat/conversations/turn-lifecycle"; +import { + scheduleSessionCompletedPluginTasks, + type ScheduleSessionCompletedPluginTasksOptions, +} from "@/chat/plugins/task-runner"; import type { AgentRunResult } from "@/chat/services/turn-result"; import type { DispatchTurnContext, @@ -220,24 +224,11 @@ async function loadPiMessagesForTurn(args: { return {}; } -export interface SlackTurnServices { - contextCompactor: ContextCompactor; - getPausedTurnRequest: (args: { - conversationId: string; - turnId: string; - }) => Promise; - lookupSlackUser: typeof lookupSlackUser; - turnLifecycle: ConversationTurnLifecycle; - wakePausedTurn: (request: PausedTurnRequest) => Promise; - scheduleSessionCompletedPluginTasks: (params: { - conversationId: string; - sessionId: string; - }) => Promise; -} - interface SlackTurnDeps { + contextCompactor: ContextCompactor; executeTurn: ExecuteTurn; getSlackAdapter: () => SlackAdapter; + pausedTurns: PausedTurns; resolveUserAttachments: ( attachments: Message["attachments"] | undefined, context: { @@ -257,7 +248,8 @@ interface SlackTurnDeps { }> >; prepareTurnState: (args: PrepareTurnStateInput) => Promise; - services: SlackTurnServices; + sendPluginTask?: ScheduleSessionCompletedPluginTasksOptions["send"]; + turnLifecycle: Pick; } /** Return whether the Slack caller should publish destination output. */ @@ -372,7 +364,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { ensureSlackMessageActorIdentity( queued.message, teamId, - deps.services.lookupSlackUser, + lookupSlackUser, ), ), ); @@ -392,7 +384,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { executionActor = await ensureSlackMessageActorIdentity( message, teamId, - deps.services.lookupSlackUser, + lookupSlackUser, ); if (executionActor) { credentialContext = credentialContextForActor(executionActor); @@ -561,7 +553,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { return; } if (conversationId && activeTurnId) { - const pausedTurn = await deps.services.getPausedTurnRequest({ + const pausedTurn = await deps.pausedTurns.get({ conversationId, turnId: activeTurnId, }); @@ -574,7 +566,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { throw new TurnInputDeferredError(); } try { - await deps.services.wakePausedTurn(pausedTurn); + await deps.pausedTurns.wake(pausedTurn); } catch (error) { logException(error, "agent.continue.schedule.failed", { "app.ai.resume_session_version": pausedTurn.expectedVersion, @@ -682,7 +674,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { nextTurnId: turnId, }); if (conversationId && preparedState.userMessageId) { - await deps.services.turnLifecycle.start({ + await deps.turnLifecycle.start({ conversationId, createdAtMs: Date.now(), inputMessageIds: [ @@ -966,24 +958,23 @@ export function createSlackTurn(deps: SlackTurnDeps) { loadedPiMessages.canCompact && piMessages?.length ) { - const compaction = - await deps.services.contextCompactor.maybeCompact({ - conversation: preparedState.conversation, - conversationContext: preparedState.conversationContext, - conversationId, - metadata: { - threadId, - actorId: slackActorId, - channelId, - runId, - }, - onCompactionStart: () => status.update(compactingStatus), - piMessages, - modelId: modelIdForProfile( - botConfig, - loadedPiMessages.modelProfile ?? botConfig.defaultProfile, - ), - }); + const compaction = await deps.contextCompactor.maybeCompact({ + conversation: preparedState.conversation, + conversationContext: preparedState.conversationContext, + conversationId, + metadata: { + threadId, + actorId: slackActorId, + channelId, + runId, + }, + onCompactionStart: () => status.update(compactingStatus), + piMessages, + modelId: modelIdForProfile( + botConfig, + loadedPiMessages.modelProfile ?? botConfig.defaultProfile, + ), + }); if (compaction.compacted) { piMessages = compaction.piMessages; await persistThreadState(thread, { @@ -1327,7 +1318,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { conversation: preparedState.conversation, }); if (conversationId) { - await deps.services.turnLifecycle.fail({ + await deps.turnLifecycle.fail({ conversationId, createdAtMs: Date.now(), ...(authFailureEventId @@ -1375,7 +1366,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { ); } try { - await deps.services.wakePausedTurn({ + await deps.pausedTurns.wake({ conversationId, destination, turnId: turnId, @@ -1414,10 +1405,13 @@ export function createSlackTurn(deps: SlackTurnDeps) { conversationId ) { try { - await deps.services.scheduleSessionCompletedPluginTasks({ - conversationId, - sessionId: turnId, - }); + await scheduleSessionCompletedPluginTasks( + { + conversationId, + sessionId: turnId, + }, + deps.sendPluginTask ? { send: deps.sendPluginTask } : undefined, + ); } catch (error) { logException( error, diff --git a/packages/junior/src/chat/services/persist-retry.ts b/packages/junior/src/chat/services/persist-retry.ts index f185f8e38d..1bd6766ed6 100644 --- a/packages/junior/src/chat/services/persist-retry.ts +++ b/packages/junior/src/chat/services/persist-retry.ts @@ -2,8 +2,8 @@ * Shared post-delivery persist retry. * * Delivered outcomes the user already saw must not be lost to a transient - * state-write failure, so both the Slack reply executor and the dispatch - * runner retry these persists with the same short linear backoff. + * state-write failure, so both Slack turns and dispatch turns retry these + * persists with the same short linear backoff. */ import { sleep } from "@/chat/sleep"; diff --git a/packages/junior/src/chat/task-execution/turn-wake.ts b/packages/junior/src/chat/task-execution/turn-wake.ts index 8b70f8b585..b5176b59d9 100644 --- a/packages/junior/src/chat/task-execution/turn-wake.ts +++ b/packages/junior/src/chat/task-execution/turn-wake.ts @@ -34,6 +34,15 @@ interface TurnWakeOptions { state?: StateAdapter; } +/** Look up and wake paused Turns without exposing queue or storage details. */ +export interface PausedTurns { + get(args: { + conversationId: string; + turnId: string; + }): Promise; + wake(request: PausedTurnRequest): Promise; +} + /** Build the worker input for a paused turn. */ export async function getPausedTurnRequest(args: { conversationId: string; @@ -106,3 +115,17 @@ export async function wakePausedTurn( state: options.state, }); } + +/** Create paused Turn operations for one app. */ +export function createPausedTurns( + options: TurnWakeOptions & { conversationStore?: ConversationStore } = {}, +): PausedTurns { + return { + get: (args) => + getPausedTurnRequest({ + ...args, + conversationStore: options.conversationStore, + }), + wake: (request) => wakePausedTurn(request, options), + }; +} diff --git a/packages/junior/tests/fixtures/chat-runtime.ts b/packages/junior/tests/fixtures/chat-runtime.ts index 9d25de282a..d9906852d6 100644 --- a/packages/junior/tests/fixtures/chat-runtime.ts +++ b/packages/junior/tests/fixtures/chat-runtime.ts @@ -3,13 +3,22 @@ import { type CreateSlackRuntimeOptions, } from "@/chat/app/factory"; import type { JuniorRuntimeServiceOverrides } from "@/chat/app/services"; +import type { ConversationStore } from "@/chat/conversations/store"; +import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import type { ScheduleSessionCompletedPluginTasksOptions } from "@/chat/plugins/task-runner"; +import { createPausedTurns } from "@/chat/task-execution/turn-wake"; +import type { StateAdapter } from "chat"; import { FakeSlackAdapter } from "./slack-harness"; export function createTestChatRuntime( args: { + conversationStore?: ConversationStore; now?: CreateSlackRuntimeOptions["now"]; + queue?: ConversationWorkQueue; + sendPluginTask?: ScheduleSessionCompletedPluginTasksOptions["send"]; services?: JuniorRuntimeServiceOverrides; slackAdapter?: FakeSlackAdapter; + state?: StateAdapter; } = {}, ) { const slackAdapter = args.slackAdapter ?? new FakeSlackAdapter(); @@ -19,6 +28,12 @@ export function createTestChatRuntime( slackRuntime: createSlackRuntime({ getSlackAdapter: () => slackAdapter, now: args.now, + pausedTurns: createPausedTurns({ + conversationStore: args.conversationStore, + queue: args.queue, + state: args.state, + }), + sendPluginTask: args.sendPluginTask, services: args.services, }), }; diff --git a/packages/junior/tests/integration/slack/bot-handlers.test.ts b/packages/junior/tests/integration/slack/bot-handlers.test.ts index 20b3cedceb..4faafad5e2 100644 --- a/packages/junior/tests/integration/slack/bot-handlers.test.ts +++ b/packages/junior/tests/integration/slack/bot-handlers.test.ts @@ -23,10 +23,6 @@ import { getTurnRecord, upsertTurnRecord, } from "@/chat/task-execution/turn-cursor"; -import { - wakePausedTurn as schedulePausedTurnWake, - type PausedTurnRequest, -} from "@/chat/task-execution/turn-wake"; import { resetSlackApiMockState } from "../../msw/handlers/slack-api"; import { FakeSlackAdapter, @@ -110,12 +106,14 @@ async function loadTurnLifecycleEvents(conversationId: string) { function createRuntime( args: { + queue?: ConversationWorkQueueTestAdapter; services?: JuniorRuntimeServiceOverrides; slackAdapter?: FakeSlackAdapter; } = {}, ) { const services = args.services ?? {}; return createTestChatRuntime({ + queue: args.queue, slackAdapter: args.slackAdapter, services: { ...services, @@ -145,12 +143,6 @@ function slackDestination(channelId: string) { } satisfies Destination; } -function bindPausedTurnQueue(queue: ConversationWorkQueueTestAdapter) { - return async (request: PausedTurnRequest): Promise => { - await schedulePausedTurnWake(request, { queue }); - }; -} - function createAwaitingContinuationState(args: { activeSessionId: string; replied?: boolean; @@ -222,15 +214,11 @@ describe("bot handlers (integration)", () => { }); it("handleNewMention: posts reply from executeAgentRun", async () => { - const scheduleSessionCompletedPluginTasks = vi.fn(async () => undefined); const { slackRuntime } = createTestChatRuntime({ services: { agentRunner: createModelAgentRunner( createModelStream([{ type: "text", text: "Hello from the bot!" }]), ), - replyExecutor: { - scheduleSessionCompletedPluginTasks, - }, visionContext: { listThreadReplies: async () => [], }, @@ -265,10 +253,6 @@ describe("bot handlers (integration)", () => { return false; }); expect(hasReply).toBe(true); - expect(scheduleSessionCompletedPluginTasks).toHaveBeenCalledWith({ - conversationId: "slack:C0INT:1700000000.000", - sessionId: "turn_msg-new-mention", - }); }); it("does not replay a message that already has a delivered reply", async () => { @@ -682,11 +666,9 @@ describe("bot handlers (integration)", () => { const queueSendEntered = queue.holdNextSendUntil(finishQueueSend.promise); const ack = vi.fn(); const { slackRuntime } = createRuntime({ + queue, services: { agentRunner: neverRunAgentRunner(), - replyExecutor: { - wakePausedTurn: bindPausedTurnQueue(queue), - }, }, }); @@ -767,11 +749,9 @@ describe("bot handlers (integration)", () => { }); const queue = createConversationWorkQueueTestAdapter(); const { slackRuntime } = createRuntime({ + queue, services: { agentRunner: neverRunAgentRunner(), - replyExecutor: { - wakePausedTurn: bindPausedTurnQueue(queue), - }, }, }); const thread = await createTestThread({ @@ -824,11 +804,9 @@ describe("bot handlers (integration)", () => { }); const queue = createConversationWorkQueueTestAdapter(); const { slackRuntime } = createRuntime({ + queue, services: { agentRunner: neverRunAgentRunner(), - replyExecutor: { - wakePausedTurn: bindPausedTurnQueue(queue), - }, }, }); const thread = await createTestThread({ @@ -974,11 +952,9 @@ describe("bot handlers (integration)", () => { const queue = createConversationWorkQueueTestAdapter(); const ack = vi.fn(); const { slackRuntime } = createRuntime({ + queue, services: { agentRunner: neverRunAgentRunner(), - replyExecutor: { - wakePausedTurn: bindPausedTurnQueue(queue), - }, }, }); const thread = await createTestThread({ @@ -1138,11 +1114,9 @@ describe("bot handlers (integration)", () => { }); const queue = createConversationWorkQueueTestAdapter(); const { slackRuntime } = createRuntime({ + queue, services: { agentRunner: neverRunAgentRunner(), - replyExecutor: { - wakePausedTurn: bindPausedTurnQueue(queue), - }, }, }); @@ -1193,11 +1167,9 @@ describe("bot handlers (integration)", () => { const queue = createConversationWorkQueueTestAdapter(); const onTurnStatePersisted = vi.fn(); const { slackRuntime } = createRuntime({ + queue, services: { agentRunner: neverRunAgentRunner(), - replyExecutor: { - wakePausedTurn: bindPausedTurnQueue(queue), - }, }, }); @@ -1257,11 +1229,9 @@ describe("bot handlers (integration)", () => { const queue = createConversationWorkQueueTestAdapter(); queue.rejectSends(); const { slackRuntime } = createRuntime({ + queue, services: { agentRunner: neverRunAgentRunner(), - replyExecutor: { - wakePausedTurn: bindPausedTurnQueue(queue), - }, }, }); diff --git a/packages/junior/tests/integration/slack/message-changed-reply-contract.test.ts b/packages/junior/tests/integration/slack/message-changed-reply-contract.test.ts index 005e912787..2985304823 100644 --- a/packages/junior/tests/integration/slack/message-changed-reply-contract.test.ts +++ b/packages/junior/tests/integration/slack/message-changed-reply-contract.test.ts @@ -14,8 +14,8 @@ import { createModelAgentRunner } from "../../fixtures/agent-runner"; import { createModelStream } from "../../fixtures/model-stream"; import { queueSlackApiError } from "../../msw/handlers/slack-api"; import { + createPausedTurns, getPausedTurnRequest, - wakePausedTurn as schedulePausedTurnWake, } from "@/chat/task-execution/turn-wake"; import { buildDeterministicTurnId } from "@/chat/runtime/turn"; import { @@ -81,20 +81,11 @@ async function createEditedDmBot(args: { }); const slackRuntime = createSlackRuntime({ getSlackAdapter: () => bot.getAdapter("slack"), + ...(args.queue + ? { pausedTurns: createPausedTurns({ queue: args.queue, state }) } + : undefined), services: { agentRunner: args.agentRunner, - replyExecutor: { - ...(args.queue - ? { - wakePausedTurn: async (request) => { - await schedulePausedTurnWake(request, { - queue: args.queue, - state, - }); - }, - } - : undefined), - }, }, }); diff --git a/packages/junior/tests/integration/slack/message-content-behavior.test.ts b/packages/junior/tests/integration/slack/message-content-behavior.test.ts index 75e09ba8b7..85d15c2011 100644 --- a/packages/junior/tests/integration/slack/message-content-behavior.test.ts +++ b/packages/junior/tests/integration/slack/message-content-behavior.test.ts @@ -14,7 +14,6 @@ import { commitMessages } from "@/chat/conversations/projection"; import { historyItemFromPiMessage } from "@/chat/pi/conversation-events"; import { upsertTurnRecord } from "@/chat/task-execution/turn-cursor"; import { getConversationEventStore } from "@/chat/db"; -import { botConfig } from "@/chat/config"; import type { AgentRun } from "@/chat/agent/types"; import { createTestChatRuntime } from "../../fixtures/chat-runtime"; import { @@ -23,7 +22,6 @@ import { createTestDestination, } from "../../fixtures/slack-harness"; import { - createModelAgentRunner, createModelAgentRunnerForRun, neverRunAgentRunner, } from "../../fixtures/agent-runner"; @@ -488,64 +486,6 @@ describe("Slack behavior: message content", () => { ); }); - it("uses the projected handoff model for turn-start context limits", async () => { - const modelIds: string[] = []; - const priorMessages = [ - { - role: "user", - content: [{ type: "text", text: "Continue after handoff." }], - timestamp: 1, - }, - ] as PiMessage[]; - const thread = await createTestThread({ - id: "slack:C0BEHAVIOR:1700005005.500", - }); - await getConversationEventStore().replaceHistory(thread.id, { - createdAtMs: 1, - data: { - type: "handoff", - modelProfile: "handoff", - modelId: botConfig.profiles.handoff!.modelId, - replacementHistory: priorMessages.map((message) => ({ - item: historyItemFromPiMessage(message, { authority: "context" }), - })), - }, - }); - await persistThreadState(thread, { - conversation: coerceThreadConversationState({}), - }); - - const { slackRuntime } = createTestChatRuntime({ - services: { - agentRunner: createModelAgentRunner( - createModelStream([{ type: "text", text: "Done." }]), - ), - replyExecutor: { - contextCompactor: { - maybeCompact: async (args) => { - modelIds.push(args.modelId); - return { compacted: false, reason: "below_threshold" }; - }, - }, - }, - }, - }); - - await slackRuntime.handleNewMention( - thread, - createTestMessage({ - id: "m-content-handoff-model", - text: "<@U0APP> continue", - isMention: true, - threadId: thread.id, - author: { userId: "U0TESTER" }, - }), - { destination: createTestDestination(thread) }, - ); - - expect(modelIds).toEqual([botConfig.profiles.handoff!.modelId]); - }); - it("rejects active-turn history that conflicts with committed conversation history", async () => { const calls: CapturedCall[] = []; const activeMessages: PiMessage[] = [