From 0f5b3ce7638f749268f63f690d31015945d75593 Mon Sep 17 00:00:00 2001 From: David Cramer Date: Fri, 28 Aug 2026 21:46:40 -0700 Subject: [PATCH] ref(chat)!: remove per-Turn publish choice --- TERMINOLOGY.md | 10 +-- packages/junior/src/api/acp/conversations.ts | 2 +- packages/junior/src/chat/README.md | 33 ++++----- .../junior/src/chat/agent-dispatch/work.ts | 1 - .../junior/src/chat/agent-invocations/work.ts | 2 - packages/junior/src/chat/agent/index.ts | 5 -- packages/junior/src/chat/agent/resume.ts | 2 - packages/junior/src/chat/agent/types.ts | 3 - .../junior/src/chat/app/conversation-work.ts | 4 +- .../src/chat/conversations/web-input.ts | 3 - .../junior/src/chat/providers/slack/resume.ts | 18 ++--- .../src/chat/providers/slack/runtime.ts | 35 +++------ .../src/chat/providers/slack/system-turn.ts | 15 ++-- .../junior/src/chat/providers/slack/turn.ts | 49 +++++-------- .../junior/src/chat/resource-events/README.md | 4 +- .../src/chat/resource-events/notification.ts | 4 -- .../junior/src/chat/slack/dispatch-turn.ts | 4 +- .../junior/src/chat/task-execution/README.md | 26 +++---- .../chat/task-execution/assistant-message.ts | 20 +++--- .../src/chat/task-execution/checkpoint.ts | 2 - .../chat/task-execution/conversation-turn.ts | 71 +++++++++---------- .../src/chat/task-execution/paused-turn.ts | 5 -- .../src/chat/task-execution/slack-work.ts | 8 --- .../junior/src/chat/task-execution/state.ts | 48 +++++++++++-- .../src/chat/task-execution/turn-cursor.ts | 26 +++++-- .../junior/src/chat/task-execution/worker.ts | 9 +-- .../component/agent-dispatch-worker.test.ts | 1 - .../component/agent-invocation-worker.test.ts | 3 - .../resource-events/resource-events.test.ts | 1 - .../component/runtime/agent-resume.test.ts | 2 - .../task-execution/checkpoint.test.ts | 4 +- .../task-execution/conversation-work.test.ts | 49 +++++++++---- .../slack-conversation-work.test.ts | 8 +-- .../tests/component/vercel-queue-dev.test.ts | 1 - .../tests/fixtures/conversation-work.ts | 1 - .../junior/tests/integration/acp-http.test.ts | 1 - .../integration/agent-continue-slack.test.ts | 1 - .../integration/agent-dispatch-work.test.ts | 12 ++++ .../integration/agent-invocation-work.test.ts | 1 - .../cancel-pending-messages.test.ts | 1 - .../conversations/pending-messages.test.ts | 1 - .../conversation-turn-work.test.ts | 65 +++++++++++------ .../tests/integration/durable-queue.test.ts | 1 - 43 files changed, 277 insertions(+), 285 deletions(-) diff --git a/TERMINOLOGY.md b/TERMINOLOGY.md index ee538842b3..2400f77832 100644 --- a/TERMINOLOGY.md +++ b/TERMINOLOGY.md @@ -28,14 +28,8 @@ Canonical words used across Junior's code and documentation. such as a Slack channel or thread. A Conversation has zero or one Location. A Run carries this same Location when the agent or tools need it. Location does not allow output to be sent. Conversation visibility is separate. -- **Delivery**: an optional function that sends Run output to the Conversation - Location. Only Delivery allows output to be sent there. Storing a completed - assistant Message in the Conversation does not depend on Delivery. -- **publish**: whether one Turn also sends assistant output to the Conversation - Location through Delivery. The Conversation always stores each completed - assistant Message. `publishExternally` is the legacy mailbox, Turn checkpoint, - and Run field for this fact until those interfaces use `publish` or Delivery - alone. +- **Delivery**: a function that sends Run output to the Conversation Location. + A Conversation without Delivery stores completed assistant Messages only. - **User**: one person-level record. A user may have several linked identities. - **Identity**: one provider account, such as a Slack account in one workspace, optionally linked to a user. diff --git a/packages/junior/src/api/acp/conversations.ts b/packages/junior/src/api/acp/conversations.ts index 41fe2f5dae..902a80dd37 100644 --- a/packages/junior/src/api/acp/conversations.ts +++ b/packages/junior/src/api/acp/conversations.ts @@ -93,7 +93,7 @@ async function hasConversationAccess( return Boolean(access?.isParticipant); } -/** Return whether one Turn can publish its terminal result without racing cleanup. */ +/** Return whether one Turn can return its terminal result without racing cleanup. */ async function turnTerminalIsReady(args: { conversationId: string; eventStore: ConversationEventStore; diff --git a/packages/junior/src/chat/README.md b/packages/junior/src/chat/README.md index 77e9c867d3..1c96791e82 100644 --- a/packages/junior/src/chat/README.md +++ b/packages/junior/src/chat/README.md @@ -93,10 +93,6 @@ singleton. - **Destination**: explicit target for output or a side effect. Current uses as a Conversation Location are migration debt. A feature may use Destination before it creates a Conversation at that target. -- **publish**: Turn fact that says whether assistant output is also sent to the - Conversation Location through Delivery. The Conversation always stores each - completed assistant Message. `publishExternally` is the legacy field for this - fact. ## Target Interface @@ -126,14 +122,12 @@ type InboundMessage = { source: Source; actor?: Actor; input: AgentInput; - publish: boolean; }; type Turn = { turnId: string; source: Source; actor?: Actor; - publish: boolean; }; type Delivery = (message: AssistantMessage) => void | Promise; @@ -148,11 +142,15 @@ type AgentRun = { }; ``` -`Source.kind` states what produced the input. The worker copies Source and -publish from the selected input to the Turn. It loads Location from the -Conversation. Before every new or resumed Run, the provider supplies Delivery -when the Turn publishes. A feature may use Destination to select a target before -it creates a Conversation. That target becomes the new Conversation Location. +`Source.kind` states what produced the input. The worker copies Source from the +selected input to the Turn. It loads Location from the Conversation. Before +every new or resumed Run, the work owner supplies Delivery. Slack input gets +Slack Delivery. Web and local input do not get provider Delivery. Resource +events get Delivery for the Conversation Location. Scheduled, Event task, and +plugin dispatch work gets Delivery for its explicit Destination. Agent +invocation does not get Delivery. A feature may use Destination to select a +target before it creates a Conversation. That target becomes the new +Conversation Location. Attribution does not grant authority. `run.actors` records participating actors; credential issuance still requires the current actor or an explicit delegated @@ -204,19 +202,16 @@ delegation without becoming the execution actor or a general task owner. separate. - The final Run interface has Source, optional Location, and optional Delivery. Source does not contain Location. Delivery is created for the Location and - does not repeat it. A dashboard continuation may therefore carry a Slack - Location without getting Slack Delivery. + does not repeat it. A dashboard continuation in a Slack Conversation carries + its Location for tools but does not get Slack Delivery. - The final interface uses Conversation, Source, Location, and Delivery. Do not add another type, routing object, or wrapper for the same values. - A Conversation may have one parent Conversation. It stores that relation as `parentConversationId`. Location is independent and is not copied from the parent. A Run may read the parent Conversation when it needs that Location. -- Each Turn stores `publish`. The worker gets Source from the selected input and - Location from the Conversation. Before every new or resumed Run, the owning - provider supplies Delivery only when the Turn publishes. Source, Actor, - Destination, and Location do not invent Delivery. The legacy - `publishExternally` field remains only until mailbox and Turn checkpoint data - use the final fact. +- Before each new or resumed Run, the work owner supplies optional Delivery. + Source, Actor, and Location do not select Delivery. A child Conversation does + not get Delivery from its parent's Location. - Host-owned runtime context and the actor's current instruction are separate user messages. The context message immediately precedes the instruction, remains context-authority on resume, and may be replaced before a later model diff --git a/packages/junior/src/chat/agent-dispatch/work.ts b/packages/junior/src/chat/agent-dispatch/work.ts index b6dffd813f..5c609dfbf2 100644 --- a/packages/junior/src/chat/agent-dispatch/work.ts +++ b/packages/junior/src/chat/agent-dispatch/work.ts @@ -134,7 +134,6 @@ export function buildAgentDispatchInboundMessage( }, }, receivedAtMs: nowMs, - publishExternally: true, source: "plugin", }; } diff --git a/packages/junior/src/chat/agent-invocations/work.ts b/packages/junior/src/chat/agent-invocations/work.ts index e2fbc96c4c..6db7ffa5ff 100644 --- a/packages/junior/src/chat/agent-invocations/work.ts +++ b/packages/junior/src/chat/agent-invocations/work.ts @@ -85,7 +85,6 @@ export function buildAgentInvocationInboundMessage( }, }, receivedAtMs: nowMs, - publishExternally: false, source: "internal", }; } @@ -437,7 +436,6 @@ export function createAgentInvocationWorker(agentRunner: AgentRunner) { actor: invocation.actor, credentialContext: invocation.credentialContext, destination: invocation.destination, - publishExternally: context.publishExternally, source: invocation.source, ...(location ? { location } : undefined), surface: "internal", diff --git a/packages/junior/src/chat/agent/index.ts b/packages/junior/src/chat/agent/index.ts index 9f55c812d3..caaf8f790c 100644 --- a/packages/junior/src/chat/agent/index.ts +++ b/packages/junior/src/chat/agent/index.ts @@ -299,7 +299,6 @@ async function executeAgentRunInPrivacyContext( slackConversation: run.slackConversation, slackActionToken: run.slackActionToken, destination: run.destination, - publishExternally: run.publishExternally, surface: run.surface, dispatch: run.dispatch, toolChannelId: run.toolChannelId, @@ -512,10 +511,6 @@ async function executeAgentRunInPrivacyContext( : undefined), durability, recordActiveMcpProviders, - publishExternally: - checkpoint.record?.publishExternally ?? - routing.publishExternally ?? - false, actor, runSource, conversationId, diff --git a/packages/junior/src/chat/agent/resume.ts b/packages/junior/src/chat/agent/resume.ts index 5a6771065c..89e4987a85 100644 --- a/packages/junior/src/chat/agent/resume.ts +++ b/packages/junior/src/chat/agent/resume.ts @@ -42,7 +42,6 @@ interface ResumeStateArgs { dispatchId?: string; durability: AgentDurability; recordActiveMcpProviders: () => Promise; - publishExternally: boolean; actor?: Actor; runSource: Source; conversationId: string; @@ -118,7 +117,6 @@ export function createResumeState(args: ResumeStateArgs) { channelName: args.channelName, destination: args.destination, dispatchId: args.dispatchId, - publishExternally: args.publishExternally, source: args.runSource, actor: args.actor, surface: args.surface, diff --git a/packages/junior/src/chat/agent/types.ts b/packages/junior/src/chat/agent/types.ts index 62b47c7439..4c088477d6 100644 --- a/packages/junior/src/chat/agent/types.ts +++ b/packages/junior/src/chat/agent/types.ts @@ -218,9 +218,6 @@ export type AgentRun = { // TODO(dcramer): Remove AgentRun.destination after tool side effects use // feature-owned targets and place context comes from Location. destination: Destination; - // TODO(dcramer): Remove AgentRun.publishExternally after Turn checkpoints - // store publish and each provider uses it to supply Delivery before every Run. - publishExternally?: boolean; surface?: AgentTurnSurface; dispatch?: AgentDispatch; diff --git a/packages/junior/src/chat/app/conversation-work.ts b/packages/junior/src/chat/app/conversation-work.ts index c71ce8f3e8..49c4f7a091 100644 --- a/packages/junior/src/chat/app/conversation-work.ts +++ b/packages/junior/src/chat/app/conversation-work.ts @@ -28,7 +28,7 @@ import { } from "@/chat/agent-dispatch/store"; import { createSlackRuntime } from "./factory"; import type { JuniorRuntimeServiceOverrides } from "./services"; -import { createSlackSystemTurnPublisher } from "@/chat/providers/slack/system-turn"; +import { createSlackSystemTurnDelivery } from "@/chat/providers/slack/system-turn"; import { scheduleSessionCompletedPluginTasks, type ScheduleSessionCompletedPluginTasksOptions, @@ -134,7 +134,7 @@ export function createConversationWork( const invocationWorker = createAgentInvocationWorker(options.agentRunner); const conversationTurnWorker = createConversationTurnWorker( options.agentRunner, - createSlackSystemTurnPublisher({ + createSlackSystemTurnDelivery({ getSlackAdapter: options.getSlackAdapter, state: options.state, }), diff --git a/packages/junior/src/chat/conversations/web-input.ts b/packages/junior/src/chat/conversations/web-input.ts index 7e575c60d5..e3e739ca13 100644 --- a/packages/junior/src/chat/conversations/web-input.ts +++ b/packages/junior/src/chat/conversations/web-input.ts @@ -174,9 +174,6 @@ export function buildWebInboundMessage(args: { } satisfies LegacyWebMailboxMetadata, }, receivedAtMs: nowMs, - // TODO(dcramer): Rename this stored field to publish after deployed - // mailbox readers and writers use the new name. - publishExternally: false, // TODO(dcramer): Replace this string after deployed mailbox readers and // writers use a complete web Source. source: "web", diff --git a/packages/junior/src/chat/providers/slack/resume.ts b/packages/junior/src/chat/providers/slack/resume.ts index 41d5bd1237..40d2780eff 100644 --- a/packages/junior/src/chat/providers/slack/resume.ts +++ b/packages/junior/src/chat/providers/slack/resume.ts @@ -565,17 +565,13 @@ async function resumeSlackTurnInContext( const deliveryState = await getDeliveryConversation(); let slackMessageTs: string[] = []; try { - // TODO(dcramer): Remove this missing-as-publish fallback after no stored - // Turn checkpoint can omit publishExternally. - if (runArgs.run?.publishExternally !== false) { - slackMessageTs = await sendSlackReply({ - channelId: runArgs.channelId, - conversationId: runArgs.conversationId, - replyAttribution: runArgs.run?.dispatch?.replyAttribution, - text, - threadTs: runArgs.threadTs, - }); - } + slackMessageTs = await sendSlackReply({ + channelId: runArgs.channelId, + conversationId: runArgs.conversationId, + replyAttribution: runArgs.run?.dispatch?.replyAttribution, + text, + threadTs: runArgs.threadTs, + }); } catch (error) { if (isRetryableSlackPostError(error)) { throw new RetryableDeliveryError(error); diff --git a/packages/junior/src/chat/providers/slack/runtime.ts b/packages/junior/src/chat/providers/slack/runtime.ts index 1e60891ec8..740a3f12cc 100644 --- a/packages/junior/src/chat/providers/slack/runtime.ts +++ b/packages/junior/src/chat/providers/slack/runtime.ts @@ -100,7 +100,6 @@ interface SteeringDrainContext { export interface SlackTurnOptions extends ReplyHooks { conversationId?: string; destination: Destination; - publishExternally?: boolean; } const THREAD_OPTOUT_ACK = @@ -189,7 +188,6 @@ export interface SlackTurnRuntimeDependencies { onTurnStatePersisted?: () => Promise; preparedState?: TPreparedState; queuedMessages?: QueuedTurnMessage[]; - publishExternally?: boolean; drainSteeringMessages?: ( accept: (messages: QueuedTurnMessage[]) => Promise, context?: SteeringDrainContext, @@ -354,11 +352,6 @@ function actorUserName(message: Message): string | undefined { } /** Build the Slack event runtime that routes mentions and subscribed messages. */ -/** Slack surfaces publish unless a caller opts out. */ -function shouldPublishExternally(publishExternally?: boolean): boolean { - return publishExternally !== false; -} - export function createSlackTurnRuntime< TPreparedState, TAssistantEvent extends AssistantLifecycleEvent = AssistantLifecycleEvent, @@ -778,7 +771,6 @@ export function createSlackTurnRuntime< conversationId: hooks.conversationId, destination: hooks.destination, queuedMessages, - publishExternally: shouldPublishExternally(hooks.publishExternally), ack, onToolInvocation: toolInvocationHook, onTurnCompleted, @@ -843,13 +835,11 @@ export function createSlackTurnRuntime< lifecycleError = error; } await hooks.beforeFirstResponsePost?.(); - if (shouldPublishExternally(hooks.publishExternally)) { - await postFallbackErrorReplyWithLogging({ - thread, - eventId, - postFailureEventName: "mention.handler.failure_reply_post.failed", - }); - } + await postFallbackErrorReplyWithLogging({ + thread, + eventId, + postFailureEventName: "mention.handler.failure_reply_post.failed", + }); if (lifecycleError) throw lifecycleError; } finally { if (completed) { @@ -1111,7 +1101,6 @@ export function createSlackTurnRuntime< conversationId: hooks.conversationId, destination: hooks.destination, preparedState, - publishExternally: shouldPublishExternally(hooks.publishExternally), beforeFirstResponsePost: hooks.beforeFirstResponsePost, queuedMessages, ack, @@ -1178,14 +1167,12 @@ export function createSlackTurnRuntime< lifecycleError = error; } await hooks.beforeFirstResponsePost?.(); - if (shouldPublishExternally(hooks.publishExternally)) { - await postFallbackErrorReplyWithLogging({ - thread, - eventId, - postFailureEventName: - "subscribed_message.handler.failure_reply_post.failed", - }); - } + await postFallbackErrorReplyWithLogging({ + thread, + eventId, + postFailureEventName: + "subscribed_message.handler.failure_reply_post.failed", + }); if (lifecycleError) throw lifecycleError; } finally { if (completed) { diff --git a/packages/junior/src/chat/providers/slack/system-turn.ts b/packages/junior/src/chat/providers/slack/system-turn.ts index d7b1a70dc6..4a451f4aa7 100644 --- a/packages/junior/src/chat/providers/slack/system-turn.ts +++ b/packages/junior/src/chat/providers/slack/system-turn.ts @@ -1,20 +1,25 @@ import type { SlackAdapter } from "@chat-adapter/slack"; import type { StateAdapter } from "chat"; -import type { PublishMessage } from "@/chat/task-execution/assistant-message"; +import type { DeliverMessage } from "@/chat/task-execution/assistant-message"; import { RetryableDeliveryError } from "@/chat/agent/types"; import { runWithSlackInstallation } from "@/chat/slack/adapter-context"; import { isRetryableSlackPostError } from "@/chat/slack/errors"; import { sendSlackReply } from "@/chat/slack/reply"; -/** Create Slack publishing for system Turns that have no webhook Message. */ -export function createSlackSystemTurnPublisher(args: { +/** + * Deliver system Turn output to Slack without a webhook Message. + * + * TODO(dcramer): Replace this Location-taking function with Delivery bound to one + * Location after Resource event work supplies Delivery before the Run. + */ +export function createSlackSystemTurnDelivery(args: { getSlackAdapter: () => SlackAdapter; state?: StateAdapter; -}): PublishMessage { +}): DeliverMessage { return async ({ conversationId, location, text }) => { if (location.provider !== "slack") { throw new Error( - `Slack system Turn cannot publish to ${location.provider} Location`, + `Slack system Turn cannot deliver to ${location.provider} Location`, ); } let messageIds: string[]; diff --git a/packages/junior/src/chat/providers/slack/turn.ts b/packages/junior/src/chat/providers/slack/turn.ts index 84602c925b..5ba88e23e8 100644 --- a/packages/junior/src/chat/providers/slack/turn.ts +++ b/packages/junior/src/chat/providers/slack/turn.ts @@ -248,11 +248,6 @@ interface SlackTurnDeps { sendPluginTask?: ScheduleSessionCompletedPluginTasksOptions["send"]; } -/** Return whether the Slack caller should publish destination output. */ -function shouldPublishExternally(publishExternally?: boolean): boolean { - return publishExternally !== false; -} - /** Build the Slack caller that prepares input and delivers output for a Turn. */ export function createSlackTurn(deps: SlackTurnDeps) { const turnLifecycle = getTurnLifecycle(); @@ -273,7 +268,6 @@ export function createSlackTurn(deps: SlackTurnDeps) { onTurnStatePersisted?: () => Promise; preparedState?: PreparedTurnState; queuedMessages?: QueuedTurnMessage[]; - publishExternally?: boolean; execution?: DispatchTurnContext; skipBackfill?: boolean; drainSteeringMessages?: ( @@ -445,9 +439,6 @@ export function createSlackTurn(deps: SlackTurnDeps) { ); try { await beforeFirstResponsePost(); - if (!shouldPublishExternally(options.publishExternally)) { - return; - } if (channelId && threadTs) { await sendSlackReply({ channelId, @@ -636,9 +627,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { }); if (configReply) { await beforeFirstResponsePost(); - if (shouldPublishExternally(options.publishExternally)) { - await thread.post(buildSlackOutputMessage(configReply.text)); - } + await thread.post(buildSlackOutputMessage(configReply.text)); markConversationMessage( preparedState.conversation, preparedState.userMessageId, @@ -824,24 +813,21 @@ export function createSlackTurn(deps: SlackTurnDeps) { // classified as retryable delivery errors. await beforeFirstResponsePost(); try { - if (shouldPublishExternally(options.publishExternally)) { - if (channelId && thread.adapter.name === "slack") { - slackMessageTs = await sendSlackReply({ - channelId, - conversationId, - replyAttribution: - options.execution?.dispatch?.replyAttribution, - text, - ...(threadTs ? { threadTs } : undefined), - }); - } else { - for (const part of splitSlackReplyText(text)) { - const postedMessageTs = ( - await thread.post(buildSlackOutputMessage(part)) - ).id; - if (postedMessageTs) { - slackMessageTs.push(postedMessageTs); - } + if (channelId && thread.adapter.name === "slack") { + slackMessageTs = await sendSlackReply({ + channelId, + conversationId, + replyAttribution: options.execution?.dispatch?.replyAttribution, + text, + ...(threadTs ? { threadTs } : undefined), + }); + } else { + for (const part of splitSlackReplyText(text)) { + const postedMessageTs = ( + await thread.post(buildSlackOutputMessage(part)) + ).id; + if (postedMessageTs) { + slackMessageTs.push(postedMessageTs); } } } @@ -1101,9 +1087,6 @@ export function createSlackTurn(deps: SlackTurnDeps) { source, ...(location ? { location } : undefined), destination, - publishExternally: shouldPublishExternally( - options.publishExternally, - ), surface: options.execution?.surface ?? "slack", dispatch: options.execution?.dispatch, toolChannelId, diff --git a/packages/junior/src/chat/resource-events/README.md b/packages/junior/src/chat/resource-events/README.md index f8827bddd6..3635bf5f5a 100644 --- a/packages/junior/src/chat/resource-events/README.md +++ b/packages/junior/src/chat/resource-events/README.md @@ -16,8 +16,8 @@ conversation. before the Slack thread is marked unsubscribed. - Plugin route code validates and normalizes incoming events before calling the ingestion boundary. -- Every conversation can hold a resource-event watch. Delivery wakes that - conversation mailbox; the conversation destination chooses the worker. +- Every conversation can hold a resource-event watch. A matching event wakes + that conversation mailbox; the conversation destination chooses the worker. - Plugin-owned routes publish normalized events through the route-hook resource event publisher; core binds the plugin namespace and never needs the raw provider webhook. Publication requires an active registration that declares diff --git a/packages/junior/src/chat/resource-events/notification.ts b/packages/junior/src/chat/resource-events/notification.ts index a0ee637838..3e623e13da 100644 --- a/packages/junior/src/chat/resource-events/notification.ts +++ b/packages/junior/src/chat/resource-events/notification.ts @@ -119,10 +119,6 @@ export function createResourceEventInboundMessage(input: { delivery: "defer", source: "resource_event", receivedAtMs: input.receivedAtMs ?? Date.now(), - // TODO(dcramer): Store the final publish fact here after resource-event - // ingress loads the Conversation. Then remove the worker's Location-based - // publish default. Resource events only wake the mailbox until that change. - publishExternally: false, input: { text: input.text, authorId: RESOURCE_EVENT_AUTHOR_ID, diff --git a/packages/junior/src/chat/slack/dispatch-turn.ts b/packages/junior/src/chat/slack/dispatch-turn.ts index b512a932f8..d50ded5f21 100644 --- a/packages/junior/src/chat/slack/dispatch-turn.ts +++ b/packages/junior/src/chat/slack/dispatch-turn.ts @@ -21,7 +21,6 @@ type ExecuteSlackTurn = ( conversationId?: string; destination: DispatchRecord["destination"]; execution: DispatchTurnContext; - publishExternally?: boolean; onTurnDeliveryAccepted?: (messageId?: string) => void; onTurnOutcome?: (result: DispatchTurnResult) => void; shouldYield?: () => boolean; @@ -48,6 +47,8 @@ export function createSlackDispatchTurnRunner(options: { await state.connect(); const conversationId = getDispatchConversationId(dispatch); const adapter = options.getSlackAdapter(); + // TODO(dcramer): Remove this synthetic Slack Message and Thread after + // dispatch work supplies Slack Delivery to the shared Turn path. const message = new Message({ id: getDispatchInputMessageId(dispatch.id), threadId: conversationId, @@ -92,7 +93,6 @@ export function createSlackDispatchTurnRunner(options: { ack: hooks.ack, conversationId, destination: dispatch.destination, - publishExternally: true, execution: { disabledFeatures: ["interactive-auth"], locationConfiguration: options.getLocationConfiguration( diff --git a/packages/junior/src/chat/task-execution/README.md b/packages/junior/src/chat/task-execution/README.md index 6c6b3ab51f..c403a13b06 100644 --- a/packages/junior/src/chat/task-execution/README.md +++ b/packages/junior/src/chat/task-execution/README.md @@ -39,29 +39,31 @@ Runtime and Redis status is `paused`. SQL free-text / enum rows may still say ## State Model -- A Conversation mailbox contains pending work. Each item has a Source, an - `interrupt` or `defer` mailbox delivery, and the legacy - `publishExternally` field. The worker keeps different publish choices in - separate Turns and saves the selected choice as the Turn's `publish` fact. +- A Conversation mailbox contains pending work. Each item has a Source and an + `interrupt` or `defer` mailbox delivery. - A Turn cursor saves the Source and Actor selected from the input that started the Turn. Steering input keeps its own Actor in message provenance. - A queue message identifies the conversation to wake. The stored work controls delivery. A provider conversation stores its destination. Child work without a destination gets its authority from its stored agent invocation. -- `publish` means the Turn also sends assistant output to the Conversation - Location. The Conversation always stores each completed assistant Message. - Dashboard and destinationless work do not publish. A resource event can - publish to a Slack Location without becoming a Slack Source. Destination or - Location presence must not invent publish. +- The Conversation always stores each completed assistant Message. Slack input + delivers to Slack. Web and local input stay in Junior. Resource events use the + Conversation Location. Scheduled, Event task, and plugin dispatch work uses + its explicit Destination. Agent invocation has no Delivery. A child + Conversation does not use its parent's Location for Delivery. +- The shared web and Resource event worker still uses Run Delivery to store the + completed assistant Message. It checks Resource event Source before it also + calls Slack. This is temporary. The owning code has a removal TODO for the + Source check and for storing the Message through Run Delivery. - A lease grants one worker temporary execution ownership. - Dispatch projection updates take a short dispatch lock only while the conversation lease is already held. They never wait for conversation work, which keeps lock ordering one-way. - 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. The Turn - checkpoint keeps the publish choice across pause and yield. Its stored field - remains `publishExternally` until the durable data contract is migrated. +- Delivery state prevents a completed Turn from being posted twice. Redis + mailbox and Turn cursor records still write `publishExternally` for deployed + readers. Current workers ignore that compatibility field. Redis execution state uses the new v2 keys. This release does not read or move old Redis state. Old mailbox, lease, and turn-cursor state can be lost. diff --git a/packages/junior/src/chat/task-execution/assistant-message.ts b/packages/junior/src/chat/task-execution/assistant-message.ts index 5d9ba1cd0c..6d099082ca 100644 --- a/packages/junior/src/chat/task-execution/assistant-message.ts +++ b/packages/junior/src/chat/task-execution/assistant-message.ts @@ -10,31 +10,31 @@ import { import { persistWithRetry } from "@/chat/services/persist-retry"; import type { ThreadConversationState } from "@/chat/state/conversation"; -/** Facts returned after a provider publishes one Message. */ -export type PublishedMessage = { +/** Facts returned after provider Delivery. */ +export type DeliveryResult = { providerMessageId?: string; providerConversationBindings?: ProviderConversationReference[]; }; /** - * Publish one Message through the provider that owns its Location. + * Deliver one Message through the provider that owns its Location. * - * TODO(dcramer): Delete PublishedMessage and PublishMessage after the core Turn + * TODO(dcramer): Delete DeliveryResult and DeliverMessage after the core Turn * lifecycle stores completed assistant Messages and the mailbox worker can * accept provider Delivery directly. */ -export type PublishMessage = (input: { +export type DeliverMessage = (input: { conversationId: string; location: Location; text: string; -}) => Promise; +}) => Promise; /** Store one assistant Message and its Agent history. */ export async function commitAssistantMessage(args: { agentMessage?: AssistantMessage; conversation: ThreadConversationState; conversationId: string; - publishedMessage?: PublishedMessage; + deliveryResult?: DeliveryResult; sessionId: string; source?: "slack" | "web"; text: string; @@ -47,11 +47,11 @@ export async function commitAssistantMessage(args: { text: args.text, userMessageId: args.userMessageId, }); - if (args.publishedMessage?.providerMessageId) { + if (args.deliveryResult?.providerMessageId) { // TODO(dcramer): Remove this Slack-only branch after Message update can // store a provider Message ID. markConversationMessage(args.conversation, conversationMessageId, { - slackTs: args.publishedMessage.providerMessageId, + slackTs: args.deliveryResult.providerMessageId, }); } try { @@ -64,7 +64,7 @@ export async function commitAssistantMessage(args: { conversationMessageId, conversationId: args.conversationId, providerConversationBindings: - args.publishedMessage?.providerConversationBindings, + args.deliveryResult?.providerConversationBindings, }), ); } catch (error) { diff --git a/packages/junior/src/chat/task-execution/checkpoint.ts b/packages/junior/src/chat/task-execution/checkpoint.ts index 18baaca535..3a4227ecb6 100644 --- a/packages/junior/src/chat/task-execution/checkpoint.ts +++ b/packages/junior/src/chat/task-execution/checkpoint.ts @@ -80,7 +80,6 @@ interface TurnCheckpointWrite { destinationVisibility?: ConversationPrivacy; dispatchId?: string; dispatchOutcome?: AgentDispatchOutcome; - publishExternally?: boolean; source?: Source; surface?: AgentTurnSurface; turnStartMessageIndex?: number; @@ -184,7 +183,6 @@ function sharedWrite(args: TurnCheckpointWrite, latest?: TurnRecord) { destination: args.destination, destinationVisibility: args.destinationVisibility, dispatchId: args.dispatchId ?? latest?.dispatchId, - publishExternally: args.publishExternally ?? latest?.publishExternally, source: args.source, surface: args.surface ?? latest?.surface, traceId: getActiveTraceId() ?? latest?.traceId, diff --git a/packages/junior/src/chat/task-execution/conversation-turn.ts b/packages/junior/src/chat/task-execution/conversation-turn.ts index 7d2bd5767a..b714ee2545 100644 --- a/packages/junior/src/chat/task-execution/conversation-turn.ts +++ b/packages/junior/src/chat/task-execution/conversation-turn.ts @@ -12,7 +12,8 @@ import { } from "@/chat/conversations/messages"; import { commitAssistantMessage, - type PublishMessage, + type DeliverMessage, + type DeliveryResult, } from "@/chat/task-execution/assistant-message"; import { ConversationTurnLifecycleService } from "@/chat/conversations/turn-lifecycle"; import type { ConversationTurnFailureCode } from "@/chat/conversations/history"; @@ -160,7 +161,7 @@ async function completeCancelledConversationTurn(args: { */ export function createConversationTurnWorker( agentRunner: AgentRunner, - publishMessage?: PublishMessage, + deliverMessage?: DeliverMessage, ) { return async ( context: ConversationWorkerContext, @@ -215,7 +216,7 @@ export function createConversationTurnWorker( const savedTurn = isResume ? await getTurnRecord(context.conversationId, turnId) : undefined; - let resumedResourceEvent = false; + let savedMessageIsResourceEvent = false; if (isResume) { const userMessage = getTurnUserMessage(conversation, turnId); if (!userMessage) { @@ -223,7 +224,8 @@ export function createConversationTurnWorker( `Unable to locate the persisted user message for Turn "${turnId}"`, ); } - resumedResourceEvent = isResourceEventConversationMessage(userMessage); + savedMessageIsResourceEvent = + isResourceEventConversationMessage(userMessage); // Resume has no new input. Restore Source and Actor from the Turn. // TODO(dcramer): Remove the saved Message fallback after no deployed Turn // cursor can omit Source or Actor. @@ -251,18 +253,18 @@ export function createConversationTurnWorker( visibility: storedConversation?.visibility, }); const webActor = actor.platform === "web" ? actor : undefined; - // TODO(dcramer): Copy publish from the selected Inbound message after - // resource-event input stores the final publish fact. Then remove this - // Location default and the checkpoint fallback. - const publish = - savedTurn?.publishExternally ?? - Boolean( - storedConversation?.location && - (source.kind === "resource_event" || resumedResourceEvent), - ); - // TODO(dcramer): Remove this legacy surface choice after Turn checkpoints - // store Source.kind and publish, and resume reads both fields. - const surface = publish ? ("slack" as const) : ("api" as const); + const conversationLocation = storedConversation?.location; + // TODO(dcramer): Remove the saved Message check after every deployed Turn + // cursor stores Resource event Source. + // TODO(dcramer): Remove this Source-based Delivery choice after the core + // Turn lifecycle stores assistant Messages and web and Resource event work + // each supplies optional provider Delivery. Source must not select Delivery. + const deliverToProvider = + Boolean(conversationLocation) && + (source.kind === "resource_event" || savedMessageIsResourceEvent); + // TODO(dcramer): Stop deriving surface from Delivery after active Turn + // lookup and reporting read Source for web and Resource event Turns. + const surface = deliverToProvider ? ("slack" as const) : ("api" as const); return await withLogContext( { @@ -373,27 +375,26 @@ export function createConversationTurnWorker( return; } failureCode = "delivery_failed"; - const location = storedConversation?.location; - if (publish && (!location || !publishMessage)) { - throw new Error( - `Conversation ${context.conversationId} cannot publish its system Turn`, - ); + let deliveryResult: DeliveryResult | undefined; + if (deliverToProvider) { + if (!conversationLocation || !deliverMessage) { + throw new Error( + `Conversation ${context.conversationId} cannot deliver to its Location`, + ); + } + deliveryResult = await deliverMessage({ + conversationId: context.conversationId, + location: conversationLocation, + text: replyText, + }); } - const publishedMessage = - publish && location && publishMessage - ? await publishMessage({ - conversationId: context.conversationId, - location, - text: replyText, - }) - : undefined; await commitAssistantMessage({ ...(agentMessage ? { agentMessage } : undefined), conversation, conversationId: context.conversationId, - ...(publishedMessage ? { publishedMessage } : undefined), + ...(deliveryResult ? { deliveryResult } : undefined), sessionId: turnId, - ...(publishedMessage + ...(deliveryResult ? { source: "slack" as const } : source.kind === "web" ? { source: "web" as const } @@ -482,12 +483,9 @@ export function createConversationTurnWorker( // TODO(dcramer): Remove AgentRun.destination after agent and tool // code reads AgentRun.location and no Run consumer needs it. destination, - // TODO(dcramer): Remove AgentRun.publishExternally after the - // saved Turn publish choice controls optional Delivery. - publishExternally: publish, source, - ...(storedConversation?.location - ? { location: storedConversation.location } + ...(conversationLocation + ? { location: conversationLocation } : undefined), surface, ...(stopSignal ? { signal: stopSignal } : undefined), @@ -572,7 +570,6 @@ export function createConversationTurnWorker( // TODO(dcramer): Remove checkpoint Destination after resume // reads the Conversation Location. destination, - publishExternally: publish, source, actor, surface, diff --git a/packages/junior/src/chat/task-execution/paused-turn.ts b/packages/junior/src/chat/task-execution/paused-turn.ts index fbe67aa993..a0bde08b8a 100644 --- a/packages/junior/src/chat/task-execution/paused-turn.ts +++ b/packages/junior/src/chat/task-execution/paused-turn.ts @@ -534,11 +534,6 @@ async function runPausedTurnInContext( destination: routingDestination, ...(dispatch ? { dispatch } : undefined), ...(routing.location ? { location: routing.location } : undefined), - // Slack resume publishes unless the checkpoint opted out. - // Missing means legacy/in-flight Slack turns still post. - // TODO(dcramer): Remove this missing-as-publish fallback after no - // stored Turn checkpoint can omit publishExternally. - publishExternally: activeTurn.publishExternally !== false, source, ...(surface ? { surface } : undefined), toolChannelId: destination.channelId, diff --git a/packages/junior/src/chat/task-execution/slack-work.ts b/packages/junior/src/chat/task-execution/slack-work.ts index 4730256f6f..90798d80eb 100644 --- a/packages/junior/src/chat/task-execution/slack-work.ts +++ b/packages/junior/src/chat/task-execution/slack-work.ts @@ -626,8 +626,6 @@ export function createSlackConversationWorker( if (!latestRecord) { return { status: "completed" }; } - const publishExternally = - latestRecord.publishExternally ?? context.publishExternally; const latestMetadata = parseSlackMetadata(latestRecord.input.metadata); if (!latestMetadata) { throw new Error( @@ -700,9 +698,6 @@ export function createSlackConversationWorker( ): Promise => { await context.attempt.drain(async (pendingRecords) => { const candidates = pendingRecords - .filter( - (record) => record.publishExternally === publishExternally, - ) .map((record) => ({ inboundMessageId: record.inboundMessageId, message: restoreMessage({ @@ -724,7 +719,6 @@ export function createSlackConversationWorker( await options.runtime.handleNewMention(thread, latestMessage, { conversationId: context.conversationId, destination, - publishExternally, messageContext, drainSteeringMessages, ack, @@ -738,7 +732,6 @@ export function createSlackConversationWorker( { conversationId: context.conversationId, destination, - publishExternally, messageContext, drainSteeringMessages, ack, @@ -805,7 +798,6 @@ export function buildSlackInboundMessage(args: { source: "slack", createdAtMs: args.message.metadata.dateSent.getTime(), receivedAtMs: args.receivedAtMs, - publishExternally: true, input: { text: args.message.text || " ", authorId, diff --git a/packages/junior/src/chat/task-execution/state.ts b/packages/junior/src/chat/task-execution/state.ts index 9822a292b6..a1307281e0 100644 --- a/packages/junior/src/chat/task-execution/state.ts +++ b/packages/junior/src/chat/task-execution/state.ts @@ -99,9 +99,6 @@ export type AgentInput = z.output; /** Durable delivery modes for pending inbound mailbox work. */ export const inboundMessageDeliverySchema = z.enum(["defer", "interrupt"]); -/** Whether this turn also publishes assistant output to the conversation destination. */ -export const publishExternallySchema = z.boolean(); - export type InboundMessageDelivery = z.output< typeof inboundMessageDeliverySchema >; @@ -118,13 +115,19 @@ export const inboundMessageSchema = z injectedAtMs: z.number().finite().optional(), input: agentInputSchema, receivedAtMs: z.number().finite(), - publishExternally: publishExternallySchema, source: inboundMessageSourceSchema, }) .strict(); export type InboundMessage = z.output; +/** Redis mailbox shape kept for deployed workers that still require this field. */ +const storedInboundMessageSchema = inboundMessageSchema.extend({ + publishExternally: z.boolean(), +}); + +type StoredInboundMessage = z.output; + export interface Lease { acquiredAtMs: number; expiresAtMs: number; @@ -363,8 +366,19 @@ function normalizeExecutionStatus(value: unknown): ExecutionStatus | undefined { } function normalizeMessage(value: unknown): InboundMessage | undefined { - const parsed = inboundMessageSchema.safeParse(value); - return parsed.success ? parsed.data : undefined; + const parsed = storedInboundMessageSchema.safeParse(value); + if (!parsed.success) { + return undefined; + } + const { publishExternally: _publishExternally, ...message } = parsed.data; + return message; +} + +function storeMessage( + message: InboundMessage, + publishExternally: boolean, +): StoredInboundMessage { + return { ...message, publishExternally }; } /** Whether this is the final attempt before an unacked message is dead-lettered. */ @@ -1004,9 +1018,29 @@ async function writeConversation( if (!fenced) { throw new ConversationMutationFencedError(next.conversationId); } + // TODO(dcramer): Remove the stored publishExternally field after no deployed + // mailbox reader requires it. Destination is used only to write this old + // Redis shape. Current workers ignore the field. + const hasSlackDestination = next.destination?.platform === "slack"; + const stored = { + ...next, + execution: { + ...next.execution, + pendingMessages: next.execution.pendingMessages.map((message) => + storeMessage( + message, + hasSlackDestination && + (message.source === "slack" || + message.source === "resource_event" || + message.source === "plugin" || + message.source === "scheduler"), + ), + ), + }, + }; await state.set( conversationKey(next.conversationId), - next, + stored, JUNIOR_THREAD_STATE_TTL_MS, ); await upsertIndexEntry({ diff --git a/packages/junior/src/chat/task-execution/turn-cursor.ts b/packages/junior/src/chat/task-execution/turn-cursor.ts index c0af467ef9..29e8101b05 100644 --- a/packages/junior/src/chat/task-execution/turn-cursor.ts +++ b/packages/junior/src/chat/task-execution/turn-cursor.ts @@ -138,7 +138,6 @@ export interface TurnRecord { */ actors: Actor[]; resumeReason?: TurnPauseReason; - publishExternally?: boolean; /** Input that started this Turn; absent only on stored legacy cursors. */ source?: Source; resumedFromSliceId?: number; @@ -178,6 +177,8 @@ interface StoredTurnRecord extends Omit< | "piMessageProvenance" | "turnStartMessageIndex" > { + /** Redis field kept for deployed workers that still read publishExternally. */ + publishExternally?: boolean; /** * `seq` of the last event in `junior_conversation_events` whose projection reproduces * this record's committed Pi messages; -1 when nothing was committed. @@ -457,7 +458,6 @@ function materializeTurnRecord( dispatchOutcome: stored.dispatchOutcome, errorMessage: stored.errorMessage, resumeReason: stored.resumeReason, - publishExternally: stored.publishExternally, source: stored.source, resultMessageId: stored.resultMessageId, resumedFromSliceId: stored.resumedFromSliceId, @@ -779,7 +779,7 @@ async function updateTurnState(args: { errorMessage: args.errorMessage ?? args.existing.errorMessage, historyVersion: parsed.historyVersion, resumeReason: args.existing.resumeReason, - publishExternally: args.existing.publishExternally, + publishExternally: parsed.publishExternally, source: args.existing.source, resultMessageId: args.resultMessageId ?? args.existing.resultMessageId, resumedFromSliceId: args.existing.resumedFromSliceId, @@ -815,7 +815,6 @@ export async function upsertTurnRecord(args: { trailingMessageProvenance?: ConversationMessageProvenance[]; actor?: Actor; resumeReason?: TurnPauseReason; - publishExternally?: boolean; errorMessage?: string; resumedFromSliceId?: number; traceId?: string; @@ -909,6 +908,20 @@ async function upsertTurnRecordLocked( ? undefined : commit.messageSeqs.filter((seq) => seq <= turnStartSeq).length); + // TODO(dcramer): Remove the stored publishExternally field after no deployed + // Turn cursor reader requires it. Current workers ignore the field. + const turnSource = args.source ?? existingRecord?.source; + const dispatchId = args.dispatchId ?? existingRecord?.dispatchId; + const hasProviderLocation = + Boolean(conversation?.location) || args.destination?.platform === "slack"; + const publishExternally = + dispatchId || turnSource + ? hasProviderLocation && + (Boolean(dispatchId) || + turnSource?.kind === "slack" || + turnSource?.kind === "resource_event") + : (existingRecord?.publishExternally ?? false); + return await setStoredRecord({ actor: existingRecord?.actor ?? args.actor, fence, @@ -941,14 +954,13 @@ async function upsertTurnRecordLocked( historyVersion: commit.historyVersion, previousVersion: existingRecord?.version, ...definedProps({ - dispatchId: args.dispatchId ?? existingRecord?.dispatchId, + dispatchId, dispatchOutcome: args.dispatchOutcome ?? existingRecord?.dispatchOutcome, errorMessage: args.errorMessage, lastProgressAtMs: args.lastProgressAtMs, resumeReason: args.resumeReason, - publishExternally: - args.publishExternally ?? existingRecord?.publishExternally, + publishExternally, source: args.source ?? existingRecord?.source, resultMessageId: args.resultMessageId ?? existingRecord?.resultMessageId, diff --git a/packages/junior/src/chat/task-execution/worker.ts b/packages/junior/src/chat/task-execution/worker.ts index 8fb1c1a60e..03937b019f 100644 --- a/packages/junior/src/chat/task-execution/worker.ts +++ b/packages/junior/src/chat/task-execution/worker.ts @@ -44,7 +44,6 @@ export interface ConversationWorkerContext { checkIn(): Promise; conversationId: string; destination?: Destination; - publishExternally: boolean; /** True when the current execution slice must stop at its next safe boundary. */ shouldYield(): boolean; /** Return an AbortSignal backed by the durable Conversation stop request. */ @@ -102,7 +101,7 @@ function selectContiguousTurnBatch( const nextTurnIndex = messages.findIndex( (message) => message.input.authorId !== first.input.authorId || - message.publishExternally !== first.publishExternally, + message.source !== first.source, ); return messages.slice( 0, @@ -110,7 +109,7 @@ function selectContiguousTurnBatch( ); } -/** Prioritize interrupts while keeping each attempt scoped to one actor. */ +/** Prioritize interrupts while keeping each attempt to one Actor and Source. */ function selectAttemptMessages(work: ConversationWorkState): InboundMessage[] { const messages = work.messages; const interrupts = messages.filter( @@ -615,9 +614,6 @@ async function processConversationWorkInContext( attemptMessageIds = attemptMessages.map( (message) => message.inboundMessageId, ); - // Empty batches are resume-only. Adapters read the checkpoint flag; do - // not set publish from destination presence alone. - const publishExternally = attemptMessages[0]?.publishExternally ?? false; attemptSelectedMessageIds = new Set(attemptMessageIds); const ack = async (): Promise => { const acknowledged = await ackMessages({ @@ -654,7 +650,6 @@ async function processConversationWorkInContext( }, conversationId, destination, - publishExternally, shouldYield, stopSignal: stop.signal, checkIn, diff --git a/packages/junior/tests/component/agent-dispatch-worker.test.ts b/packages/junior/tests/component/agent-dispatch-worker.test.ts index 38d05d91e7..f044ffa2b5 100644 --- a/packages/junior/tests/component/agent-dispatch-worker.test.ts +++ b/packages/junior/tests/component/agent-dispatch-worker.test.ts @@ -66,7 +66,6 @@ function createContext( checkIn: vi.fn(async () => true), conversationId: message.conversationId, destination, - publishExternally: true, shouldYield: () => false, ...overrides, }; diff --git a/packages/junior/tests/component/agent-invocation-worker.test.ts b/packages/junior/tests/component/agent-invocation-worker.test.ts index 4065d14133..f0aff0ebba 100644 --- a/packages/junior/tests/component/agent-invocation-worker.test.ts +++ b/packages/junior/tests/component/agent-invocation-worker.test.ts @@ -115,7 +115,6 @@ describe("agent invocation worker", () => { }, checkIn: vi.fn(), conversationId: created.childConversationId, - publishExternally: false, shouldYield: () => false, } satisfies ConversationWorkerContext; @@ -174,7 +173,6 @@ describe("agent invocation worker", () => { }, checkIn: vi.fn(), conversationId: created.childConversationId, - publishExternally: false, shouldYield: () => false, }) satisfies ConversationWorkerContext; const firstAck = vi.fn(async () => {}); @@ -227,7 +225,6 @@ describe("agent invocation worker", () => { checkIn: vi.fn(), conversationId: created.childConversationId, destination: DESTINATION, - publishExternally: false, shouldYield: () => false, } satisfies ConversationWorkerContext; diff --git a/packages/junior/tests/component/resource-events/resource-events.test.ts b/packages/junior/tests/component/resource-events/resource-events.test.ts index bf32a37ecf..9520326746 100644 --- a/packages/junior/tests/component/resource-events/resource-events.test.ts +++ b/packages/junior/tests/component/resource-events/resource-events.test.ts @@ -136,7 +136,6 @@ describe("resource event delivery", () => { expect(work?.messages).toHaveLength(1); expect(work?.messages[0]).toMatchObject({ source: "resource_event", - publishExternally: false, input: { text: expect.stringContaining("CI failed on workflow test."), metadata: { diff --git a/packages/junior/tests/component/runtime/agent-resume.test.ts b/packages/junior/tests/component/runtime/agent-resume.test.ts index 8c679dedde..54a093f7d6 100644 --- a/packages/junior/tests/component/runtime/agent-resume.test.ts +++ b/packages/junior/tests/component/runtime/agent-resume.test.ts @@ -30,7 +30,6 @@ async function resumeState(conversationId: string, turnId: string) { destination: { platform: "local", conversationId }, durability: {}, recordActiveMcpProviders: async () => undefined, - publishExternally: true, runSource: createLocalSource(conversationId), conversationId, turnId, @@ -65,7 +64,6 @@ describe("agent resume", () => { recordActiveMcpProviders: async () => { throw new Error("provider metadata unavailable"); }, - publishExternally: true, runSource: createLocalSource(conversationId), conversationId, turnId, diff --git a/packages/junior/tests/component/task-execution/checkpoint.test.ts b/packages/junior/tests/component/task-execution/checkpoint.test.ts index c68ad70669..9a2a104473 100644 --- a/packages/junior/tests/component/task-execution/checkpoint.test.ts +++ b/packages/junior/tests/component/task-execution/checkpoint.test.ts @@ -265,7 +265,6 @@ describe("turn checkpoint", () => { source: SLACK_SOURCE, piMessages: priorMessages, resumeReason: "auth", - publishExternally: false, errorMessage: "initial auth pause", }); @@ -287,7 +286,6 @@ describe("turn checkpoint", () => { sliceId: 2, resumedFromSliceId: 1, resumeReason: "auth", - publishExternally: false, source: SLACK_SOURCE, errorMessage: "plugin auth pause", piMessages: [priorMessages[0]], @@ -458,7 +456,6 @@ describe("turn checkpoint", () => { destination: SLACK_DESTINATION, inboundMessageId: "turn-activity-message", receivedAtMs: 9_000, - publishExternally: true, delivery: "defer", source: "slack", }, @@ -545,6 +542,7 @@ describe("turn checkpoint", () => { expect(stored).not.toHaveProperty("destination"); expect(stored).toMatchObject({ actor: SLACK_ACTOR, + publishExternally: true, source: SLACK_SOURCE, }); diff --git a/packages/junior/tests/component/task-execution/conversation-work.test.ts b/packages/junior/tests/component/task-execution/conversation-work.test.ts index 0b24518824..534c7494a3 100644 --- a/packages/junior/tests/component/task-execution/conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/conversation-work.test.ts @@ -576,7 +576,7 @@ describe("conversation work execution", () => { execution: { inboundMessageIds: [pendingMessage.inboundMessageId], pendingCount: 1, - pendingMessages: [pendingMessage], + pendingMessages: [{ ...pendingMessage, publishExternally: true }], status: "idle", updatedAtMs: 1_000, }, @@ -639,6 +639,7 @@ describe("conversation work execution", () => { it("rejects pending messages with a different conversation destination", async () => { const state = getStateAdapter(); await state.connect(); + const pendingMessage = inboundMessage("m1"); await state.set(CONVERSATION_WORK_STATE_KEY, { schemaVersion: 2, conversationId: CONVERSATION_ID, @@ -649,8 +650,9 @@ describe("conversation work execution", () => { pendingCount: 1, pendingMessages: [ { - ...inboundMessage("m1"), + ...pendingMessage, destination: OTHER_SLACK_DESTINATION, + publishExternally: true, }, ], status: "pending", @@ -1032,9 +1034,9 @@ describe("conversation work execution", () => { expect(queue.sentRecords()).toEqual([]); }); - it("keeps different publishExternally values in separate attempts", async () => { + it("keeps publishExternally out of worker input", async () => { const queue = createConversationWorkQueueTestAdapter(); - const attempts: Array<{ ids: string[]; publishExternally: boolean }> = []; + const attempts: string[][] = []; await appendInboundMessage({ message: inboundMessage("m1", { delivery: "defer" }), nowMs: 1_000, @@ -1042,33 +1044,50 @@ describe("conversation work execution", () => { await appendInboundMessage({ message: inboundMessage("m2", { createdAtMs: 2_000, + destination: undefined, delivery: "defer", receivedAtMs: 2_000, - publishExternally: false, + source: "web", }), nowMs: 2_000, }); + await appendInboundMessage({ + message: inboundMessage("m3", { + createdAtMs: 3_000, + delivery: "defer", + receivedAtMs: 3_000, + source: "plugin", + }), + nowMs: 3_000, + }); + + const state = getStateAdapter(); + const stored = (await state.get(CONVERSATION_WORK_STATE_KEY)) as { + execution: { pendingMessages: Array> }; + }; + expect(stored.execution.pendingMessages).toEqual([ + expect.objectContaining({ publishExternally: true }), + expect.objectContaining({ publishExternally: false }), + expect.objectContaining({ publishExternally: true }), + ]); await expect( processConversationWork(conversationQueueMessage(), { queue, run: async (context) => { - attempts.push({ - ids: context.attempt.messages.map( - (message) => message.inboundMessageId, - ), - publishExternally: context.publishExternally, - }); + attempts.push( + context.attempt.messages.map((message) => message.inboundMessageId), + ); + expect(context.attempt.messages[0]).not.toHaveProperty( + "publishExternally", + ); await context.attempt.ack(); return { status: "completed" }; }, }), ).resolves.toEqual({ status: "completed" }); - expect(attempts).toEqual([ - { ids: ["m1"], publishExternally: true }, - { ids: ["m2"], publishExternally: false }, - ]); + expect(attempts).toEqual([["m1"], ["m2"], ["m3"]]); }); it("resumes a paused turn before defer delivery after requeue", async () => { diff --git a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts index acb145d905..f4301d4d1a 100644 --- a/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts +++ b/packages/junior/tests/component/task-execution/slack-conversation-work.test.ts @@ -224,7 +224,6 @@ describe("Slack conversation work execution", () => { isDM: false, }); const handleNewMention = vi.fn(async (_thread, restored, hooks) => { - expect(hooks.publishExternally).toBe(true); expect(restored.text).toBe(""); expect(restored.attachments).toHaveLength(1); expect(restored.formatted.children).toHaveLength(2); @@ -327,10 +326,6 @@ describe("Slack conversation work execution", () => { createdAtMs: 1_000, receivedAtMs: 1_100, }; - const malformedWithDelivery = { - ...malformed, - publishExternally: true as const, - }; const worker = createSlackConversationWorker({ getSlackAdapter: () => slackAdapter, runNextPausedTurn: async () => false, @@ -353,12 +348,11 @@ describe("Slack conversation work execution", () => { destination: SLACK_DESTINATION, drain: async () => [], isFinalAttempt: false, - messages: [malformedWithDelivery], + messages: [malformed], }, checkIn: async () => true, conversationId: CONVERSATION_ID, destination: SLACK_DESTINATION, - publishExternally: true, shouldYield: () => false, }), ).rejects.toThrow( diff --git a/packages/junior/tests/component/vercel-queue-dev.test.ts b/packages/junior/tests/component/vercel-queue-dev.test.ts index 150c7e897f..81c18c520d 100644 --- a/packages/junior/tests/component/vercel-queue-dev.test.ts +++ b/packages/junior/tests/component/vercel-queue-dev.test.ts @@ -466,7 +466,6 @@ describe("registerVercelConversationWorkDevConsumer", () => { source: "slack", createdAtMs: 1_000, receivedAtMs: 1_100, - publishExternally: true, input: { authorId: "U123", text: "message m1", diff --git a/packages/junior/tests/fixtures/conversation-work.ts b/packages/junior/tests/fixtures/conversation-work.ts index 16a6aa0848..f35ec04ef2 100644 --- a/packages/junior/tests/fixtures/conversation-work.ts +++ b/packages/junior/tests/fixtures/conversation-work.ts @@ -223,7 +223,6 @@ export function inboundMessage( source: "slack", createdAtMs: 1_000, receivedAtMs: 1_100, - publishExternally: true, input: { text: `message ${inboundMessageId}`, authorId: "U123", diff --git a/packages/junior/tests/integration/acp-http.test.ts b/packages/junior/tests/integration/acp-http.test.ts index f2affb8ec3..b377650f8a 100644 --- a/packages/junior/tests/integration/acp-http.test.ts +++ b/packages/junior/tests/integration/acp-http.test.ts @@ -300,7 +300,6 @@ describe("remote ACP HTTP", () => { ]); expect(harness.agentRuns).toHaveLength(1); expect(harness.agentRuns[0]).toMatchObject({ - publishExternally: false, source: { kind: "web", visibility: "private" }, }); await expect(harness.historyTexts(sessionId)).resolves.toEqual([ diff --git a/packages/junior/tests/integration/agent-continue-slack.test.ts b/packages/junior/tests/integration/agent-continue-slack.test.ts index 1302a8640e..bfc333f83f 100644 --- a/packages/junior/tests/integration/agent-continue-slack.test.ts +++ b/packages/junior/tests/integration/agent-continue-slack.test.ts @@ -260,7 +260,6 @@ describe("paused turn Slack integration", () => { channelId: "C123", threadTs: "1712345.0001", }, - publishExternally: true, source: storedSource, toolChannelId: "C123", state: expect.objectContaining({ diff --git a/packages/junior/tests/integration/agent-dispatch-work.test.ts b/packages/junior/tests/integration/agent-dispatch-work.test.ts index fc57456366..2cf4a45870 100644 --- a/packages/junior/tests/integration/agent-dispatch-work.test.ts +++ b/packages/junior/tests/integration/agent-dispatch-work.test.ts @@ -8,6 +8,7 @@ import { import { enqueueAgentDispatch } from "@/chat/agent-dispatch/work"; import { disconnectStateAdapter } from "@/chat/state/adapter"; import { processConversationQueueMessage } from "@/chat/task-execution/vercel-callback"; +import { turnCursorKey } from "@/chat/task-execution/turn-cursor-keys"; import { persistConversationMessages } from "@/chat/conversations/messages"; import { coerceThreadConversationState } from "@/chat/state/conversation"; import { getUserMessageInstructionText } from "@/chat/pi/transcript"; @@ -67,6 +68,17 @@ describe("agent dispatch conversation work", () => { params: expect.objectContaining({ text: "Done" }), }), ]); + await expect( + state.get( + turnCursorKey( + getDispatchConversationId(dispatch), + getDispatchTurnId(dispatch.id), + ), + ), + ).resolves.toMatchObject({ + dispatchId: dispatch.id, + publishExternally: true, + }); expect(modelStream).toHaveBeenCalledOnce(); const instruction = modelStream.mock.calls[0]?.[1].messages.at(-1); if (!instruction) { diff --git a/packages/junior/tests/integration/agent-invocation-work.test.ts b/packages/junior/tests/integration/agent-invocation-work.test.ts index fd88f28487..441319e83b 100644 --- a/packages/junior/tests/integration/agent-invocation-work.test.ts +++ b/packages/junior/tests/integration/agent-invocation-work.test.ts @@ -422,7 +422,6 @@ describe("agent invocation conversation work", () => { channelId: "C123", threadTs: "1712345.0001", }, - publishExternally: false, source: slackInvocationInput.source, surface: "internal", runId: created.invocationId, diff --git a/packages/junior/tests/integration/api/conversations/cancel-pending-messages.test.ts b/packages/junior/tests/integration/api/conversations/cancel-pending-messages.test.ts index 8aa7f95dc8..cb899e1320 100644 --- a/packages/junior/tests/integration/api/conversations/cancel-pending-messages.test.ts +++ b/packages/junior/tests/integration/api/conversations/cancel-pending-messages.test.ts @@ -239,7 +239,6 @@ describe("conversation cancel pending messages API", () => { text: "internal wake", }, receivedAtMs: Date.now(), - publishExternally: false, source: "internal", }, state, diff --git a/packages/junior/tests/integration/api/conversations/pending-messages.test.ts b/packages/junior/tests/integration/api/conversations/pending-messages.test.ts index ae5734e5c4..53f168b12b 100644 --- a/packages/junior/tests/integration/api/conversations/pending-messages.test.ts +++ b/packages/junior/tests/integration/api/conversations/pending-messages.test.ts @@ -140,7 +140,6 @@ describe("conversation pending messages API", () => { text: "slack interrupt", }, receivedAtMs: 3_100, - publishExternally: true, source: "slack", }, nowMs: 3_100, diff --git a/packages/junior/tests/integration/conversation-turn-work.test.ts b/packages/junior/tests/integration/conversation-turn-work.test.ts index 8fa0f79ae2..def2309257 100644 --- a/packages/junior/tests/integration/conversation-turn-work.test.ts +++ b/packages/junior/tests/integration/conversation-turn-work.test.ts @@ -1,6 +1,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { createResourceEventSource, + createSlackSource, createWebSource, } from "@sentry/junior-plugin-api"; import { @@ -157,7 +158,6 @@ describe("Conversation mailbox Turn work", () => { messageId: accepted.messageId, }); expect(inbound).toMatchObject({ - publishExternally: false, source: "web", }); @@ -184,7 +184,6 @@ describe("Conversation mailbox Turn work", () => { expect(agentRuns).toHaveLength(1); expect(agentRuns[0]).toEqual( expect.objectContaining({ - publishExternally: false, source: createWebSource(accepted.conversationId, "public"), actor: expect.objectContaining({ platform: "web" }), }), @@ -231,7 +230,6 @@ describe("Conversation mailbox Turn work", () => { conversationTurnIdForMessage(accepted.messageId), ), ).resolves.toMatchObject({ - publishExternally: false, state: "completed", surface: "api", }); @@ -357,7 +355,6 @@ describe("Conversation mailbox Turn work", () => { checkIn: async () => true, conversationId: accepted.conversationId, destination, - publishExternally: false, shouldYield: () => false, stopSignal: () => stop.signal, }), @@ -395,7 +392,6 @@ describe("Conversation mailbox Turn work", () => { }, ], destination, - publishExternally: false, source: createWebSource(accepted.conversationId), actor, surface: "api", @@ -408,7 +404,6 @@ describe("Conversation mailbox Turn work", () => { }), conversationId: accepted.conversationId, destination, - publishExternally: false, shouldYield: () => false, checkIn: async () => true, }); @@ -596,7 +591,6 @@ describe("Conversation mailbox Turn work", () => { ).resolves.toEqual({ status: "yielded" }); await expect(getTurnRecord(conversationId, turnId)).resolves.toMatchObject({ actor: RESOURCE_EVENT_SYSTEM_ACTOR, - publishExternally: false, resumeReason: "yield", source, state: "paused", @@ -620,7 +614,6 @@ describe("Conversation mailbox Turn work", () => { credentialContext: { actor: RESOURCE_EVENT_SYSTEM_ACTOR }, destination, disabledFeatures: ["interactive-auth"], - publishExternally: false, source, turnId, }), @@ -629,7 +622,6 @@ describe("Conversation mailbox Turn work", () => { } await expect(getTurnRecord(conversationId, turnId)).resolves.toMatchObject({ actor: RESOURCE_EVENT_SYSTEM_ACTOR, - publishExternally: false, source, state: "completed", }); @@ -660,7 +652,7 @@ describe("Conversation mailbox Turn work", () => { ]); }, 10_000); - it("keeps legacy resource event publishing on after resume", async () => { + it("delivers a resumed resource event from the Conversation Location", async () => { const { conversationStore, queue, state } = await createConversationFixture(); const conversationId = "slack:C123"; @@ -707,7 +699,7 @@ describe("Conversation mailbox Turn work", () => { }); const agentRuns: AgentRun[] = []; let firstRun = true; - const publishMessage = vi.fn(async () => ({ + const deliverMessage = vi.fn(async () => ({ providerMessageId: "1712346.0001", })); const worker = createConversationTurnWorker( @@ -724,7 +716,7 @@ describe("Conversation mailbox Turn work", () => { { type: "text", text: "Review request handled." }, ]); }), - publishMessage, + deliverMessage, ); const run = requireConversationTurn(worker); @@ -737,11 +729,12 @@ describe("Conversation mailbox Turn work", () => { state, }), ).resolves.toEqual({ status: "yielded" }); - await expect(getTurnRecord(conversationId, turnId)).resolves.toMatchObject({ - publishExternally: true, + const pausedTurn = await getTurnRecord(conversationId, turnId); + expect(pausedTurn).toMatchObject({ source, state: "paused", }); + expect(pausedTurn).not.toHaveProperty("publishExternally"); const storedCursor = await state.get(turnCursorKey(conversationId, turnId)); if (!storedCursor || typeof storedCursor !== "object") { @@ -772,17 +765,15 @@ describe("Conversation mailbox Turn work", () => { expect(agentRuns).toHaveLength(2); expect(agentRuns[0]).toEqual( expect.objectContaining({ - publishExternally: true, source, }), ); expect(agentRuns[1]).toEqual( expect.objectContaining({ - publishExternally: true, source: createWebSource(conversationId), }), ); - expect(publishMessage).toHaveBeenCalledOnce(); + expect(deliverMessage).toHaveBeenCalledOnce(); }, 10_000); it("does not claim dispatch resume wakes that share surface api", async () => { @@ -808,7 +799,6 @@ describe("Conversation mailbox Turn work", () => { }, ], destination, - publishExternally: true, dispatchId: "dispatch_shared_surface", surface: "api", }); @@ -817,14 +807,13 @@ describe("Conversation mailbox Turn work", () => { attempt: emptyConversationTurnAttempt({ conversationId, destination }), conversationId, destination, - publishExternally: true, shouldYield: () => false, checkIn: async () => true, }); expect(resolved).toBeUndefined(); }); - it("continues a Slack-rooted conversation without publishing externally", async () => { + it("keeps dashboard input in Junior without abandoning a Slack auth pause", async () => { const { actor, conversationStore, queue, state } = await createConversationFixture(); const conversationId = "slack:C1200:1712345.1200"; @@ -854,6 +843,29 @@ describe("Conversation mailbox Turn work", () => { }, visibility: "public", }); + const slackAuthTurnId = "turn-slack-auth"; + await saveTurnCheckpoint({ + mode: "paused", + conversationId, + turnId: slackAuthTurnId, + sliceId: 1, + reason: "auth", + messages: [ + { + role: "user", + content: [{ type: "text", text: "Slack Turn waiting for auth." }], + timestamp: 1, + }, + ], + destination: slackDestination, + source: createSlackSource({ + channelId: "C1200", + teamId: "T1200", + threadTs: "1712345.1200", + visibility: "public", + }), + surface: "slack", + }); const accepted = await appendAndEnqueueWebMessage( { @@ -897,11 +909,11 @@ describe("Conversation mailbox Turn work", () => { }); expect(inbound).toMatchObject({ destination: slackDestination, - publishExternally: false, source: "web", }); const agentRuns: AgentRun[] = []; + const deliverMessage = vi.fn(async () => ({})); const worker = createConversationTurnWorker( createModelAgentRunnerForRun((run) => { agentRuns.push(run); @@ -909,6 +921,7 @@ describe("Conversation mailbox Turn work", () => { { type: "text", text: "Dashboard-only reply." }, ]); }), + deliverMessage, ); const run = requireConversationTurn(worker); @@ -926,10 +939,10 @@ describe("Conversation mailbox Turn work", () => { expect.objectContaining({ destination: expect.objectContaining({ platform: "slack" }), location: storedConversation?.location, - publishExternally: false, source: expect.objectContaining({ kind: "web" }), }), ); + expect(deliverMessage).not.toHaveBeenCalled(); const messages = ( await getConversationEventStore().loadMessageHistory(conversationId) @@ -951,9 +964,15 @@ describe("Conversation mailbox Turn work", () => { conversationTurnIdForMessage(accepted.messageId), ), ).resolves.toMatchObject({ - publishExternally: false, state: "completed", surface: "api", }); + await expect( + getTurnRecord(conversationId, slackAuthTurnId), + ).resolves.toMatchObject({ + resumeReason: "auth", + state: "paused", + surface: "slack", + }); }); }); diff --git a/packages/junior/tests/integration/durable-queue.test.ts b/packages/junior/tests/integration/durable-queue.test.ts index 979746dcee..e97e9d5a56 100644 --- a/packages/junior/tests/integration/durable-queue.test.ts +++ b/packages/junior/tests/integration/durable-queue.test.ts @@ -455,7 +455,6 @@ describe("durable queue contract", () => { getTurnRecord(CONVERSATION_ID, turnId), ).resolves.toMatchObject({ actors: [{ platform: "system", name: "resource-event" }], - publishExternally: true, state: "completed", surface: "slack", });