diff --git a/packages/junior/src/chat/README.md b/packages/junior/src/chat/README.md index 7f272a1a4c..25669efdfa 100644 --- a/packages/junior/src/chat/README.md +++ b/packages/junior/src/chat/README.md @@ -415,7 +415,9 @@ check, and Slack Delivery read that field. They do not check the Source. each outcome and says if it is the creator's direct message or a channel. - The run has no Delivery port, so final assistant text is never delivered. - The run ends with one `finishAutomationRun` call. `send_message` posts the - declared message to the stored outcomes. `no_action` posts nothing. + declared message to the stored outcomes. Each of its optional + `threadReplies` posts as its own reply in the thread under that message. + `no_action` posts nothing. `misconfigured` records a blocked dispatch with the declared reason. A blocked dispatch sets the Scheduled automation or Event automation to `blocked` with that reason. It does not run again until its creator resumes diff --git a/packages/junior/src/chat/automation-result.ts b/packages/junior/src/chat/automation-result.ts index 9ad45423de..934aeb85d8 100644 --- a/packages/junior/src/chat/automation-result.ts +++ b/packages/junior/src/chat/automation-result.ts @@ -18,15 +18,24 @@ import { sanitizeAssistantText } from "@/chat/services/assistant-reply"; /** Model-facing tool that ends an Automation run. */ export const FINISH_AUTOMATION_RUN_TOOL_NAME = "finishAutomationRun"; +/** Most thread replies one declared message can carry. */ +export const MAX_AUTOMATION_THREAD_REPLIES = 10; + +const declaredText = z + .string() + .transform(sanitizeAssistantText) + .pipe(z.string().min(1)); + /** One declared result for an Automation run. */ export const automationResultSchema = z.discriminatedUnion("result", [ z.object({ result: z.literal("send_message"), // The declared message gets the same cleanup as a chat reply. - message: z - .string() - .transform(sanitizeAssistantText) - .pipe(z.string().min(1)), + message: declaredText, + threadReplies: z + .array(declaredText) + .max(MAX_AUTOMATION_THREAD_REPLIES) + .optional(), }), z.object({ result: z.literal("no_action"), @@ -55,11 +64,18 @@ export function deliversFinalText(dispatch: RunDispatch): boolean { return dispatch?.outcomes?.length !== 0 && !dispatch?.declaresResult; } +/** Text that a finished run posts, plus replies in the thread under it. */ +export type FinishedRunReply = { + text: string; + threadReplies: readonly string[]; +}; + /** - * Return the text to post after a run ends, if any. A chat Turn already + * Return what to post after a run ends, if anything. A chat Turn already * delivered a successful reply, so it posts only its failure reply. An - * Automation run posts only a declared message. It never posts a failure - * reply to its outcomes. The failed execution shows on the Automation. + * Automation run posts only a declared message and its thread replies. It + * never posts a failure reply to its outcomes. The failed execution shows on + * the Automation. */ export function finishedRunReply( result: { @@ -68,14 +84,19 @@ export function finishedRunReply( text: string; }, dispatch: RunDispatch, -): string | undefined { +): FinishedRunReply | undefined { if (!dispatch?.declaresResult) { - return result.diagnostics.outcome === "success" ? undefined : result.text; + return result.diagnostics.outcome === "success" + ? undefined + : { text: result.text, threadReplies: [] }; } return result.diagnostics.outcome === "success" && result.automation?.result === "send_message" && dispatch.outcomes?.length !== 0 - ? result.text + ? { + text: result.text, + threadReplies: result.automation.threadReplies ?? [], + } : undefined; } diff --git a/packages/junior/src/chat/providers/slack/resume.ts b/packages/junior/src/chat/providers/slack/resume.ts index 3acdd15b1e..c595d75835 100644 --- a/packages/junior/src/chat/providers/slack/resume.ts +++ b/packages/junior/src/chat/providers/slack/resume.ts @@ -499,6 +499,7 @@ async function resumeSlackTurnInContext( /** Post and record one completed assistant message for the resumed turn. */ const deliverAssistantMessage = async ( reply: AssistantMessage | string, + threadReplies: readonly string[] = [], ): Promise => { const message = typeof reply === "string" ? undefined : reply; const text = @@ -521,6 +522,7 @@ async function resumeSlackTurnInContext( conversationId: runArgs.conversationId, replyAttribution: runArgs.run?.dispatch?.replyAttribution, text, + threadReplies, threadTs: outcome.destination.threadTs, })), ); @@ -532,6 +534,7 @@ async function resumeSlackTurnInContext( conversationId: runArgs.conversationId, replyAttribution: runArgs.run?.dispatch?.replyAttribution, text, + threadReplies, threadTs: runArgs.threadTs, }); } @@ -549,7 +552,7 @@ async function resumeSlackTurnInContext( conversation: deliveryState.conversation, sessionId: deliveryState.sessionId, source: "slack", - text, + text: [text, ...threadReplies].join("\n\n"), userMessageId: deliveryState.userMessageId, }); if (messageTs) { @@ -639,9 +642,12 @@ async function resumeSlackTurnInContext( const dispatchErrorMessage = run.dispatch ? dispatchResult.errorMessage : undefined; - const replyText = finishedRunReply(reply, run.dispatch); - if (replyText !== undefined) { - await deliverAssistantMessage(replyText); + const finishedReply = finishedRunReply(reply, run.dispatch); + if (finishedReply !== undefined) { + await deliverAssistantMessage( + finishedReply.text, + finishedReply.threadReplies, + ); } runResultHandled = true; diff --git a/packages/junior/src/chat/providers/slack/turn.ts b/packages/junior/src/chat/providers/slack/turn.ts index d88ef77b17..289a5ca4af 100644 --- a/packages/junior/src/chat/providers/slack/turn.ts +++ b/packages/junior/src/chat/providers/slack/turn.ts @@ -807,6 +807,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { const deliverAssistantMessage = async ( reply: AssistantMessage | string, terminalDispatchOutcome?: "blocked" | "failed", + threadReplies: readonly string[] = [], ): Promise => { const agentMessage = typeof reply === "string" ? undefined : reply; const text = @@ -835,6 +836,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { replyAttribution: options.execution?.dispatch?.replyAttribution, text, + threadReplies, threadTs: outcome.destination.threadTs, }); slackMessageTs.push(...messageIds); @@ -852,6 +854,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { conversationId, replyAttribution: options.execution?.dispatch?.replyAttribution, text, + threadReplies, ...(threadTs ? { threadTs } : undefined), }); messageDestinations.push( @@ -887,7 +890,7 @@ export function createSlackTurn(deps: SlackTurnDeps) { conversation: preparedState.conversation, sessionId: turnId, source: "slack", - text, + text: [text, ...threadReplies].join("\n\n"), userMessageId: preparedState.userMessageId, }); if (slackTs) { @@ -1196,7 +1199,11 @@ export function createSlackTurn(deps: SlackTurnDeps) { options.execution?.dispatch, ); if (reply !== undefined) { - await deliverAssistantMessage(reply); + await deliverAssistantMessage( + reply.text, + undefined, + reply.threadReplies, + ); } const turnResult = runDispatchOutcome(finalResult); runResultHandled = true; diff --git a/packages/junior/src/chat/slack/reply.ts b/packages/junior/src/chat/slack/reply.ts index 316de6b419..91b93e1a7b 100644 --- a/packages/junior/src/chat/slack/reply.ts +++ b/packages/junior/src/chat/slack/reply.ts @@ -27,10 +27,17 @@ export async function sendSlackReply(args: { conversationId: string; replyAttribution?: ReplyAttribution; text: string; + /** Each item posts as its own reply in the thread under `text`. */ + threadReplies?: readonly string[]; threadTs?: string; }): Promise { - const posts: Array<{ text: string; cards: SlackCard[] }> = - splitSlackReplyText(args.text).map((text) => ({ text, cards: [] })); + // Split each part on its own so a thread reply always starts a new post. + const posts: Array<{ text: string; cards: SlackCard[] }> = [ + args.text, + ...(args.threadReplies ?? []), + ] + .flatMap((part) => splitSlackReplyText(part)) + .map((text) => ({ text, cards: [] })); const cards = (args.cards ?? []).map((card) => renderSlackCard(card, args.conversationId), ); diff --git a/packages/junior/src/chat/tools/runtime/finish-automation-run.ts b/packages/junior/src/chat/tools/runtime/finish-automation-run.ts index 78013dc9f7..d6d6c8c4d3 100644 --- a/packages/junior/src/chat/tools/runtime/finish-automation-run.ts +++ b/packages/junior/src/chat/tools/runtime/finish-automation-run.ts @@ -1,5 +1,8 @@ import { z } from "zod"; -import { automationResultSchema } from "@/chat/automation-result"; +import { + automationResultSchema, + MAX_AUTOMATION_THREAD_REPLIES, +} from "@/chat/automation-result"; import { isNoReplyMarker } from "@/chat/no-reply"; import { juniorToolOutputSchema } from "@/chat/tool-support/structured-result"; import { zodTool } from "@/chat/tool-support/zod-tool"; @@ -39,6 +42,13 @@ export function createFinishAutomationRunTool(options: { .string() .optional() .describe("Required for send_message. The text to post."), + threadReplies: z + .array(z.string()) + .max(MAX_AUTOMATION_THREAD_REPLIES) + .optional() + .describe( + "Optional for send_message. Each item posts as its own reply in the thread under the message, in order. Use it to keep the message short and put each section of a long digest in its own reply.", + ), } : undefined), reason: z @@ -55,9 +65,11 @@ export function createFinishAutomationRunTool(options: { const parsed = automationResultSchema.safeParse(input); if (!parsed.success) { throw new ToolInputError( - input.result === "send_message" - ? "send_message requires a non-empty message." - : `${input.result} requires a non-empty reason.`, + input.result !== "send_message" + ? `${input.result} requires a non-empty reason.` + : typeof input.message === "string" && input.message.trim() + ? "Every threadReplies item must be non-empty." + : "send_message requires a non-empty message.", ); } // A message that is only the old no-reply marker posts nothing. diff --git a/packages/junior/tests/integration/slack/reply-delivery.test.ts b/packages/junior/tests/integration/slack/reply-delivery.test.ts index 6861dbb945..36dbc06120 100644 --- a/packages/junior/tests/integration/slack/reply-delivery.test.ts +++ b/packages/junior/tests/integration/slack/reply-delivery.test.ts @@ -92,6 +92,27 @@ describe("sendSlackReply", () => { ); }); + it("posts each thread reply as its own post under the first message", async () => { + const messageTs = await sendSlackReply({ + channelId: "C123", + conversationId: "agent-dispatch:dispatch-1", + text: "Weekly digest", + threadReplies: ["**Cocoa**\nQuiet week", "**Java**\nOne fix"], + }); + + const posts = getCapturedSlackApiCalls("chat.postMessage"); + expect(posts.map((post) => post.params.text)).toEqual([ + "Weekly digest", + expect.stringContaining("**Cocoa**"), + expect.stringContaining("**Java**"), + ]); + expect(posts[0]?.params).not.toHaveProperty("thread_ts"); + expect(posts.slice(1).map((post) => post.params.thread_ts)).toEqual([ + messageTs[0], + messageTs[0], + ]); + }); + it("keeps the original thread for every chunk", async () => { const messageTs = await sendSlackReply({ channelId: "C123", diff --git a/packages/junior/tests/unit/chat/automation-result.test.ts b/packages/junior/tests/unit/chat/automation-result.test.ts index a5117550ad..2af9540b51 100644 --- a/packages/junior/tests/unit/chat/automation-result.test.ts +++ b/packages/junior/tests/unit/chat/automation-result.test.ts @@ -121,10 +121,32 @@ describe("automation run result", () => { finishedRunReply(result, { declaresResult: true, outcomes: [{ action: "send_message" }], - }), + })?.text, ).toBe("The digest."); }); + it("posts declared thread replies with the declared message", () => { + const result = automationTurnResult({ + newMessages: [ + declared({ + result: "send_message", + message: "Weekly digest", + threadReplies: ["Cocoa: quiet week", "Java: one fix"], + }), + ], + }); + + expect( + finishedRunReply(result, { + declaresResult: true, + outcomes: [{ action: "send_message" }], + }), + ).toEqual({ + text: "Weekly digest", + threadReplies: ["Cocoa: quiet week", "Java: one fix"], + }); + }); + it("keeps a result that an earlier slice saved when a later slice errs", () => { const saved = declared({ result: "send_message", message: "The digest." }); @@ -173,7 +195,7 @@ describe("automation run result", () => { finishedRunReply(turn, { declaresResult: true, outcomes: [{ action: "send_message" }], - }), + })?.text, ).toBe(posted); expect(runDispatchOutcome(turn)).toEqual(dispatch); }, @@ -185,7 +207,10 @@ describe("automation run result", () => { text: "I ran into an internal error.", }; - expect(finishedRunReply(failed, undefined)).toBe(failed.text); + expect(finishedRunReply(failed, undefined)).toEqual({ + text: failed.text, + threadReplies: [], + }); expect(finishedRunReply(failed, { declaresResult: true })).toBeUndefined(); }); }); @@ -201,6 +226,14 @@ describe("finishAutomationRun", () => { await expect( finish(true, { result: "send_message" }), ).rejects.toMatchObject({ name: "ToolInputError" }); + // A blank thread reply names the reply, not the valid message. + await expect( + finish(true, { + result: "send_message", + message: "Weekly digest", + threadReplies: ["Cocoa: quiet week", " "], + }), + ).rejects.toThrow("Every threadReplies item must be non-empty."); await expect(finish(true, { result: "no_action" })).rejects.toMatchObject({ name: "ToolInputError", });