diff --git a/.docker/caddy/Caddyfile b/.docker/caddy/Caddyfile index 0b119f678..dbc240292 100644 --- a/.docker/caddy/Caddyfile +++ b/.docker/caddy/Caddyfile @@ -1,6 +1,7 @@ # Local development edge that mirrors the production Caddy routing in -# deploy/caddy/Caddyfile: a single origin serves the web app, and the -# reserved /_roomote-api/* prefix is stripped and proxied to the API. +# deploy/caddy/Caddyfile: a single origin serves the web app, public webhook +# paths go directly to the API, and the reserved /_roomote-api/* prefix is +# stripped and proxied to the API. # # `pnpm dev` points the ngrok web tunnel at this listener instead of the # web app, and derives TRPC_URL as /_roomote-api — the same @@ -29,6 +30,20 @@ } } + # Webhook URLs are public integration contracts. Send them directly to the + # API so a missing or stale web-app proxy route cannot acknowledge an event + # with the Next.js HTML fallback without actually processing it. + @api_webhooks { + path /api/webhooks /api/webhooks/* + } + + handle @api_webhooks { + reverse_proxy host.docker.internal:13001 { + lb_try_duration 10s + lb_try_interval 250ms + } + } + @api { path /_roomote-api /_roomote-api/* } diff --git a/.gitignore b/.gitignore index 11830053d..829da6c82 100644 --- a/.gitignore +++ b/.gitignore @@ -84,3 +84,6 @@ artifacts/ci-e2e/ # Scheduled-task runner lock (machine-local runtime state) .agents/scheduled_tasks.lock + +# Editor swap files +*.swp diff --git a/apps/api/src/handlers/slack/events/fast-agent-processing.test.ts b/apps/api/src/handlers/slack/events/fast-agent-processing.test.ts index 6d62cae4f..bb82ebe61 100644 --- a/apps/api/src/handlers/slack/events/fast-agent-processing.test.ts +++ b/apps/api/src/handlers/slack/events/fast-agent-processing.test.ts @@ -108,25 +108,17 @@ describe('processFastAgentMessage', () => { teamId: 'T123', }); - expect(slack.addReaction).toHaveBeenNthCalledWith(1, { - channel: 'D123', - timestamp: '100.001', - name: 'eyes', - }); - expect(slack.addReaction).toHaveBeenNthCalledWith(2, { + expect(slack.addReaction).toHaveBeenCalledOnce(); + expect(slack.addReaction).toHaveBeenCalledWith({ channel: 'D123', timestamp: '100.001', name: 'thumbsup', }); - expect(slack.removeReaction).toHaveBeenCalledWith({ - channel: 'D123', - timestamp: '100.001', - name: 'eyes', - }); + expect(slack.removeReaction).not.toHaveBeenCalled(); expect(mocks.postThreadMessage).not.toHaveBeenCalled(); }); - it('keeps the processing reaction when it becomes the visible closeout', async () => { + it('can intentionally use eyes as an emoji-only closeout', async () => { mocks.answerQuestion.mockImplementationOnce( async ({ postSlackReaction, @@ -172,7 +164,7 @@ describe('processFastAgentMessage', () => { expect(mocks.postThreadMessage).not.toHaveBeenCalled(); }); - it('clears a same-name processing reaction after an intermediate acknowledgement', async () => { + it('can intentionally use eyes as an intermediate acknowledgement', async () => { mocks.answerQuestion.mockImplementationOnce( async ({ postSlackReaction, @@ -215,11 +207,7 @@ describe('processFastAgentMessage', () => { }); expect(slack.addReaction).toHaveBeenCalledOnce(); - expect(slack.removeReaction).toHaveBeenCalledWith({ - channel: 'D123', - timestamp: '100.001', - name: 'eyes', - }); + expect(slack.removeReaction).not.toHaveBeenCalled(); expect(mocks.postThreadMessage).toHaveBeenCalledWith( expect.objectContaining({ text: 'I found the answer.' }), ); @@ -348,10 +336,14 @@ describe('processFastAgentMessage', () => { ).rejects.toThrow('Slack did not accept the Fast parent reply.'); }); - it('shows the task-processing reaction until the fast response is loaded', async () => { + it('does not set an assistant thread status or automatic reaction', async () => { + // Slack replaces custom status text with its own rotating placeholders + // ("Generating response…"), so Fast turns deliberately post no status + // and no reaction; the task card is the progress surface. const slack = { addReaction: vi.fn().mockResolvedValue(true), removeReaction: vi.fn().mockResolvedValue(true), + setAssistantThreadStatus: vi.fn().mockResolvedValue(true), normalizeIncomingText: vi.fn(async (text: string) => text), fetchThreadMessages: vi.fn(async () => [ { @@ -378,22 +370,8 @@ describe('processFastAgentMessage', () => { teamId: 'T123', }); - expect(slack.addReaction).toHaveBeenCalledWith({ - channel: 'D123', - timestamp: '100.001', - name: 'eyes', - }); - expect(slack.removeReaction).toHaveBeenCalledWith({ - channel: 'D123', - timestamp: '100.001', - name: 'eyes', - }); - expect(slack.addReaction.mock.invocationCallOrder[0]).toBeLessThan( - mocks.answerQuestion.mock.invocationCallOrder[0]!, - ); - expect(mocks.answerQuestion.mock.invocationCallOrder[0]).toBeLessThan( - slack.removeReaction.mock.invocationCallOrder[0]!, - ); + expect(slack.setAssistantThreadStatus).not.toHaveBeenCalled(); + expect(slack.addReaction).not.toHaveBeenCalled(); expect(mocks.answerQuestion).toHaveBeenCalledOnce(); expect(mocks.answerQuestion).toHaveBeenCalledWith( expect.objectContaining({ @@ -450,11 +428,12 @@ describe('processFastAgentMessage', () => { ); }); - it('clears the task-processing reaction when fast processing fails', async () => { + it('clears the native thinking status when fast processing fails', async () => { mocks.answerQuestion.mockRejectedValueOnce(new Error('model unavailable')); const slack = { addReaction: vi.fn().mockResolvedValue(true), removeReaction: vi.fn().mockResolvedValue(true), + setAssistantThreadStatus: vi.fn().mockResolvedValue(true), normalizeIncomingText: vi.fn(async (text: string) => text), fetchThreadMessages: vi.fn(async () => []), }; @@ -475,11 +454,8 @@ describe('processFastAgentMessage', () => { }), ).rejects.toThrow('model unavailable'); - expect(slack.removeReaction).toHaveBeenCalledWith({ - channel: 'D123', - timestamp: '100.001', - name: 'eyes', - }); + expect(slack.setAssistantThreadStatus).not.toHaveBeenCalled(); + expect(slack.addReaction).not.toHaveBeenCalled(); expect(mocks.releaseLock).toHaveBeenCalledOnce(); }); diff --git a/apps/api/src/handlers/slack/events/fast-agent-task-launcher.test.ts b/apps/api/src/handlers/slack/events/fast-agent-task-launcher.test.ts index 9bc167740..b596041a1 100644 --- a/apps/api/src/handlers/slack/events/fast-agent-task-launcher.test.ts +++ b/apps/api/src/handlers/slack/events/fast-agent-task-launcher.test.ts @@ -1,6 +1,10 @@ const mocks = vi.hoisted(() => ({ enqueueTask: vi.fn(), getTaskUrl: vi.fn(() => 'https://roomote.example/task/task-1'), + getSlackLiveTaskStreamData: vi.fn(), + setSlackLiveTaskStreamData: vi.fn(), + startTaskStream: vi.fn(), + appendTaskStream: vi.fn(), })); vi.mock('@roomote/cloud-agents/server', () => ({ @@ -8,6 +12,18 @@ vi.mock('@roomote/cloud-agents/server', () => ({ getTaskUrl: mocks.getTaskUrl, })); +vi.mock('@roomote/slack', () => ({ + buildSlackLiveTaskTitle: (prompt: string) => prompt, + getSlackLiveTaskStreamData: mocks.getSlackLiveTaskStreamData, + setSlackLiveTaskStreamData: mocks.setSlackLiveTaskStreamData, +})); + +vi.mock('@roomote/db/server', () => ({ + db: { query: { tasks: { findFirst: vi.fn().mockResolvedValue(null) } } }, + eq: vi.fn(), + tasks: { id: 'tasks.id', title: 'tasks.title' }, +})); + import { ALL_REPOSITORIES, TaskPayloadKind } from '@roomote/types'; import { createFastAgentTaskLauncher } from './fast-agent-task-launcher.js'; @@ -19,13 +35,20 @@ describe('createFastAgentTaskLauncher', () => { async ( _input: unknown, options: { - beforeEnqueue: (taskRun: { taskId: string }) => Promise; + beforeEnqueue: (taskRun: { + id: number; + taskId: string; + }) => Promise; }, ) => { - await options.beforeEnqueue({ taskId: 'task-1' }); + await options.beforeEnqueue({ id: 42, taskId: 'task-1' }); return { taskId: 'task-1' }; }, ); + mocks.getSlackLiveTaskStreamData.mockResolvedValue(null); + mocks.startTaskStream.mockResolvedValue('stream-ts'); + mocks.appendTaskStream.mockResolvedValue(true); + mocks.setSlackLiveTaskStreamData.mockResolvedValue(undefined); }); it('launches a communication-isolated child owned by the Fast parent', async () => { @@ -47,6 +70,10 @@ describe('createFastAgentTaskLauncher', () => { } as never, userId: 'user-1', teamId: 'T123', + slack: { + startTaskStream: mocks.startTaskStream, + appendTaskStream: mocks.appendTaskStream, + } as never, }); const order: string[] = []; const postKickoff = vi.fn(async () => { @@ -56,10 +83,13 @@ describe('createFastAgentTaskLauncher', () => { async ( _input: unknown, options: { - beforeEnqueue: (taskRun: { taskId: string }) => Promise; + beforeEnqueue: (taskRun: { + id: number; + taskId: string; + }) => Promise; }, ) => { - await options.beforeEnqueue({ taskId: 'task-1' }); + await options.beforeEnqueue({ id: 42, taskId: 'task-1' }); order.push('queued'); return { taskId: 'task-1' }; }, @@ -77,6 +107,26 @@ describe('createFastAgentTaskLauncher', () => { taskId: 'task-1', taskUrl: 'https://roomote.example/task/task-1', }); + expect(mocks.startTaskStream).toHaveBeenCalledWith( + expect.objectContaining({ + channel: 'C123', + threadTs: '100.001', + recipientTeamId: 'T123', + recipientUserId: 'U123', + task: expect.objectContaining({ + id: 'roomote-task-task-1', + title: 'Add a regression test', + status: 'in_progress', + }), + }), + ); + expect(mocks.setSlackLiveTaskStreamData).toHaveBeenCalledWith( + 'task-1', + expect.objectContaining({ + messageTs: 'stream-ts', + taskId: 'task-1', + }), + ); expect(mocks.enqueueTask).toHaveBeenCalledWith( { task: { @@ -97,6 +147,7 @@ describe('createFastAgentTaskLauncher', () => { slackChannel: 'C123', slackThreadTs: '100.001', }, + liveTaskStream: true, environmentId: 'env-1', }, }, @@ -128,6 +179,10 @@ describe('createFastAgentTaskLauncher', () => { userMapping: { slackUserId: 'U123' } as never, userId: 'user-1', teamId: 'T123', + slack: { + startTaskStream: mocks.startTaskStream, + appendTaskStream: mocks.appendTaskStream, + } as never, }); const postKickoff = vi.fn().mockRejectedValue(new Error('Slack failed')); let queued = false; diff --git a/apps/api/src/handlers/slack/events/fast-agent-task-launcher.ts b/apps/api/src/handlers/slack/events/fast-agent-task-launcher.ts index a7c3f7548..46cde01cc 100644 --- a/apps/api/src/handlers/slack/events/fast-agent-task-launcher.ts +++ b/apps/api/src/handlers/slack/events/fast-agent-task-launcher.ts @@ -3,8 +3,17 @@ import { getTaskUrl, type LaunchFastAgentSlackTask, } from '@roomote/cloud-agents/server'; -import { type SlackEvent } from '@roomote/slack'; import { + buildSlackLiveTaskTitle, + getSlackLiveTaskStreamData, + setSlackLiveTaskStreamData, + type SlackEvent, + type SlackNotifier, +} from '@roomote/slack'; +import { + db, + eq, + tasks, type SlackInstallation, type SlackUserMapping, } from '@roomote/db/server'; @@ -18,6 +27,7 @@ export function createFastAgentTaskLauncher(params: { event: SlackEvent; slackInstallation: SlackInstallation; userMapping: SlackUserMapping; + slack: SlackNotifier; userId: string; teamId: string; }): LaunchFastAgentSlackTask { @@ -42,12 +52,138 @@ export function createFastAgentTaskLauncher(params: { slackChannel: params.event.channel, slackThreadTs: threadId, }, + liveTaskStream: true, ...(environmentId && environmentId !== ALL_REPOSITORIES ? { environmentId } : {}), }, }; let taskUrl: string | undefined; + + const startLiveTaskStream = async (taskRun: { + id: number; + taskId: string; + }): Promise => { + try { + // A card for this task already exists (for example an idempotent + // relaunch of the same task); keep updating it instead of starting + // a second stream in the thread. + if (await getSlackLiveTaskStreamData(taskRun.taskId)) { + return; + } + + const resolvedTaskUrl = + taskUrl ?? + getTaskUrl({ + taskId: taskRun.taskId, + utm: { source: 'slack', campaign: 'fast-delegation' }, + }); + const title = buildSlackLiveTaskTitle(prompt); + const taskUpdateId = `roomote-task-${taskRun.taskId}`; + // One entry whose title always shows the CURRENT step (the worker + // title-swaps it per todo; only title/status replace on append). + // The task link is sent exactly once (Slack accumulates sources), + // and the settled card returns to the task title with the output. + const initialTask = { + id: taskUpdateId, + title, + // A pending-only stream does not render; start in_progress so the + // card is visible with the kickoff instead of materializing only + // at the worker's first update. + status: 'in_progress' as const, + sources: [ + { type: 'url' as const, url: resolvedTaskUrl, text: 'View task' }, + ], + }; + const messageTs = await params.slack.startTaskStream({ + channel: params.event.channel, + threadTs: threadId, + recipientTeamId: params.teamId, + recipientUserId: params.event.user ?? params.userMapping.slackUserId, + task: initialTask, + }); + + if (messageTs) { + // The Slack client does not paint a stream whose only content is + // the opening chunk; re-append the entry so the card renders + // immediately instead of waiting for the worker's first update. + // Sources are deliberately omitted: Slack appends them per chunk + // instead of replacing, so the link is sent exactly once. + await params.slack.appendTaskStream({ + channel: params.event.channel, + messageTs, + task: { + id: initialTask.id, + title: initialTask.title, + status: initialTask.status, + }, + }); + } + + if (messageTs) { + await setSlackLiveTaskStreamData(taskRun.taskId, { + channel: params.event.channel, + messageTs, + taskId: taskRun.taskId, + taskUpdateId, + threadTs: threadId, + title, + taskUrl: resolvedTaskUrl, + }); + } + } catch (error) { + console.error( + `[Fast Agent] Failed to start Slack live task stream for run ${taskRun.id}: ${error instanceof Error ? error.message : String(error)}`, + ); + } + }; + + // The generated task title usually lands right after enqueue, well + // before the worker's first event; refresh the card's opening + // (prompt-derived) title as soon as it exists. Bounded to the + // pre-worker window so it never overwrites a step title. + const refreshLiveTaskCardTitle = async (taskId: string): Promise => { + try { + for (const delayMs of [0, 5_000]) { + if (delayMs > 0) { + await new Promise((resolve) => setTimeout(resolve, delayMs)); + } + + const [data, taskRow] = await Promise.all([ + getSlackLiveTaskStreamData(taskId), + db.query.tasks.findFirst({ + where: eq(tasks.id, taskId), + columns: { title: true }, + }), + ]); + const generatedTitle = taskRow?.title?.trim(); + if (!data) { + return; + } + if (!generatedTitle) { + continue; + } + + const title = buildSlackLiveTaskTitle(generatedTitle); + if (title === data.title) { + return; + } + + await params.slack.appendTaskStream({ + channel: data.channel, + messageTs: data.messageTs, + task: { id: data.taskUpdateId, title, status: 'in_progress' }, + }); + await setSlackLiveTaskStreamData(taskId, { ...data, title }); + return; + } + } catch (error) { + console.error( + `[Fast Agent] Failed to refresh live task card title for ${taskId}: ${error instanceof Error ? error.message : String(error)}`, + ); + } + }; + const launch = await enqueueTask( { task, @@ -63,6 +199,7 @@ export function createFastAgentTaskLauncher(params: { utm: { source: 'slack', campaign: 'fast-delegation' }, }); await postKickoff({ taskId: taskRun.taskId, taskUrl }); + await startLiveTaskStream(taskRun); }, }, ); @@ -74,10 +211,8 @@ export function createFastAgentTaskLauncher(params: { }; } - return { - success: true, - taskId: launch.taskId, - taskUrl, - }; + void refreshLiveTaskCardTitle(launch.taskId); + + return { success: true, taskId: launch.taskId, taskUrl }; }; } diff --git a/apps/api/src/handlers/slack/events/fast-agent.ts b/apps/api/src/handlers/slack/events/fast-agent.ts index 1af5da0c2..26059e7a9 100644 --- a/apps/api/src/handlers/slack/events/fast-agent.ts +++ b/apps/api/src/handlers/slack/events/fast-agent.ts @@ -65,7 +65,6 @@ export async function processFastAgentMessage(params: { continuation = false, activeTaskId = null, launchTask, - processingReactionName = 'eyes', } = params; const threadId = event.thread_ts || event.ts; const releaseFastAgentLock = await acquireFastAgentTurnLock({ @@ -81,21 +80,16 @@ export async function processFastAgentMessage(params: { return; } - const normalizedText = stripLeadingSlackProductMention( - await slack.normalizeIncomingText( - stripLeadingFastCommandMention(event.authoredText ?? event.text), - ), - ); - const question = extractFastQuestion(normalizedText, continuation); - - let didAddProcessingReaction = false; - try { - didAddProcessingReaction = await slack.addReaction({ - channel: event.channel, - timestamp: event.ts, - name: processingReactionName, - }); + // Deliberately no assistant thread status here: Slack replaces the + // custom status text with its own rotating "Generating response…" + // placeholders, which read as noise next to the task card. + const normalizedText = stripLeadingSlackProductMention( + await slack.normalizeIncomingText( + stripLeadingFastCommandMention(event.authoredText ?? event.text), + ), + ); + const question = extractFastQuestion(normalizedText, continuation); if (!question) { await postSlackThreadMarkdownMessage({ @@ -187,19 +181,7 @@ export async function processFastAgentMessage(params: { // aborted mid-flight. didSendVisibleResponse = true; }, - postSlackReaction: async ({ name, purpose, slackMessageTs }) => { - if ( - didAddProcessingReaction && - name === processingReactionName && - slackMessageTs === event.ts - ) { - if (purpose === 'closeout') { - didAddProcessingReaction = false; - } - didSendVisibleResponse = true; - return; - } - + postSlackReaction: async ({ name, slackMessageTs }) => { const added = await slack.addReaction({ channel: event.channel, timestamp: slackMessageTs, @@ -227,15 +209,6 @@ export async function processFastAgentMessage(params: { }); } } finally { - if (didAddProcessingReaction) { - await slack - .removeReaction({ - channel: event.channel, - timestamp: event.ts, - name: processingReactionName, - }) - .catch(() => {}); - } await releaseFastAgentLock().catch(() => {}); } } diff --git a/apps/dev/src/index.ts b/apps/dev/src/index.ts index 64081e4de..1af42880b 100644 --- a/apps/dev/src/index.ts +++ b/apps/dev/src/index.ts @@ -19,8 +19,8 @@ import { // Host port of the local Caddy edge (the `caddy-dev` compose service). The // public tunnel targets this port so one origin serves the web app and, under -// the reserved /_roomote-api/* prefix, the API — mirroring the production -// routing in deploy/caddy/Caddyfile. +// public webhook paths and the reserved /_roomote-api/* prefix to the API — +// mirroring the production routing in deploy/caddy/Caddyfile. const CADDY_DEV_PORT = 18080; const DEVELOPMENT_WORKER_IMAGE_REPOSITORY = 'ghcr.io/roocodeinc/roomote-worker'; @@ -73,8 +73,8 @@ class LocalDevStarter { await DockerService.stopSelfHostAppContainers(options.verbose); // The tunnel targets the local Caddy edge, which serves the web app - // and proxies /_roomote-api/* to the API — the same routing deployed - // environments get from deploy/caddy/Caddyfile. + // and proxies public webhook paths plus /_roomote-api/* to the API — the + // same routing deployed environments get from deploy/caddy/Caddyfile. const publicUrlResolution = await NgrokService.resolvePublicUrl({ port: CADDY_DEV_PORT, verbose: options.verbose, diff --git a/apps/worker/src/callbacks/__tests__/slack-live-task-stream.test.ts b/apps/worker/src/callbacks/__tests__/slack-live-task-stream.test.ts new file mode 100644 index 000000000..c93719ac9 --- /dev/null +++ b/apps/worker/src/callbacks/__tests__/slack-live-task-stream.test.ts @@ -0,0 +1,315 @@ +const mocks = vi.hoisted(() => ({ + appendTaskStream: vi.fn(), + stopTaskStream: vi.fn(), + sdkGetStreamData: vi.fn(), + sdkClearStreamData: vi.fn(), + sdkFindFirstInstallation: vi.fn(), +})); + +vi.mock('@roomote/slack/client', () => ({ + SlackNotifier: class { + appendTaskStream = mocks.appendTaskStream; + stopTaskStream = mocks.stopTaskStream; + }, +})); + +vi.mock('@roomote/sdk/client', () => ({ + sdk: { + taskRuns: { + getSlackLiveTaskStreamData: mocks.sdkGetStreamData, + clearSlackLiveTaskStreamData: mocks.sdkClearStreamData, + }, + slackInstallations: { + findFirst: mocks.sdkFindFirstInstallation, + }, + }, +})); + +import { RunStatus, TaskPayloadKind } from '@roomote/types'; +import type { TaskRun } from '@roomote/sdk/client'; + +import { + finishSlackLiveTaskStream, + getSlackLiveTaskStreamRunTaskCallbacks, + startSlackLiveTaskStream, + updateSlackLiveTaskStream, +} from '../slack-live-task-stream'; + +const streamData = { + channel: 'C123', + messageTs: 'stream-ts', + taskId: 'task-1', + taskUpdateId: 'roomote-task-task-1', + threadTs: '100.001', + title: 'Fix the button', + taskUrl: 'https://roomote.example/task/task-1', +}; + +// The module caches stream data per run id for the process lifetime, so +// every test uses a fresh run id. +let nextRunId = 100; + +function createTaskRun( + overrides: { payloadKind?: TaskPayloadKind; payload?: unknown } = {}, +): TaskRun { + return { + id: nextRunId++, + taskId: 'task-1', + payloadKind: overrides.payloadKind ?? TaskPayloadKind.StandardTask, + payload: overrides.payload ?? { + description: 'Fix the button', + liveTaskStream: true, + }, + } as unknown as TaskRun; +} + +describe('Slack live task stream', () => { + beforeEach(() => { + vi.clearAllMocks(); + mocks.sdkGetStreamData.mockResolvedValue(streamData); + // Model the server: once the data is cleared, later fetches miss. + mocks.sdkClearStreamData.mockImplementation(async () => { + mocks.sdkGetStreamData.mockResolvedValue(null); + }); + mocks.sdkFindFirstInstallation.mockResolvedValue({ + botAccessToken: 'xoxb-test', + }); + mocks.appendTaskStream.mockResolvedValue(true); + mocks.stopTaskStream.mockResolvedValue(true); + }); + + it('keeps the task title on start, only warming the data cache', async () => { + const taskRun = createTaskRun(); + await startSlackLiveTaskStream(taskRun); + + expect(mocks.sdkGetStreamData).toHaveBeenCalledWith({ runId: taskRun.id }); + expect(mocks.appendTaskStream).not.toHaveBeenCalled(); + }); + + it('does not re-send the View task source on updates', async () => { + const taskRun = createTaskRun(); + await updateSlackLiveTaskStream( + taskRun, + { type: 'text', ts: 1000, text: 'First update.' }, + {}, + ); + + const task = mocks.appendTaskStream.mock.calls[0]?.[0]?.task; + expect(task.sources).toBeUndefined(); + }); + + it('updates the card for resumed runs through the same run-scoped lookup', async () => { + const resumed = createTaskRun({ + payloadKind: TaskPayloadKind.SnapshotResume, + payload: { sourceRunId: 42, liveTaskStream: true }, + }); + + await updateSlackLiveTaskStream( + resumed, + { type: 'text', ts: 1000, text: 'Resumed update.' }, + {}, + ); + + expect(mocks.sdkGetStreamData).toHaveBeenCalledWith({ runId: resumed.id }); + expect(mocks.appendTaskStream).toHaveBeenCalledWith({ + channel: 'C123', + messageTs: 'stream-ts', + task: expect.objectContaining({ status: 'in_progress' }), + }); + }); + + it('replaces the entry title with the current todo', async () => { + const taskRun = createTaskRun(); + const context = {}; + const event = { + type: 'todo_update' as const, + ts: 1000, + todos: [ + { id: '1', content: 'Inspect the code', status: 'completed' as const }, + { id: '2', content: 'Make the change', status: 'in_progress' as const }, + { id: '3', content: 'Verify the change', status: 'pending' as const }, + ], + }; + + await updateSlackLiveTaskStream(taskRun, event, context); + await updateSlackLiveTaskStream(taskRun, event, context); + + expect(mocks.appendTaskStream).toHaveBeenCalledOnce(); + expect(mocks.appendTaskStream).toHaveBeenCalledWith({ + channel: 'C123', + messageTs: 'stream-ts', + task: expect.objectContaining({ + title: 'Make the change (1/3)', + status: 'in_progress', + }), + }); + }); + + it('accumulates narrative text into the output body', async () => { + const taskRun = createTaskRun(); + await updateSlackLiveTaskStream( + taskRun, + { type: 'text', ts: 1000, text: 'Wiring the new model into spawning.' }, + {}, + ); + + expect(mocks.appendTaskStream).toHaveBeenCalledWith({ + channel: 'C123', + messageTs: 'stream-ts', + task: expect.objectContaining({ + title: 'Fix the button', + output: '\nWiring the new model into spawning.', + }), + }); + }); + + it('filters transient status lines and limits narration per step', async () => { + const taskRun = createTaskRun(); + const context = {}; + + await updateSlackLiveTaskStream( + taskRun, + { type: 'text', ts: 1000, text: 'Provider error: Bad Gateway' }, + context, + ); + await updateSlackLiveTaskStream( + taskRun, + { type: 'text', ts: 1001, text: 'Retrying now.' }, + context, + ); + await updateSlackLiveTaskStream( + taskRun, + { type: 'text', ts: 1002, text: 'Inspecting the registry.' }, + context, + ); + await updateSlackLiveTaskStream( + taskRun, + { type: 'text', ts: 1003, text: 'A second narration for this step.' }, + context, + ); + + expect(mocks.appendTaskStream).toHaveBeenCalledOnce(); + expect(mocks.appendTaskStream).toHaveBeenCalledWith( + expect.objectContaining({ + task: expect.objectContaining({ + output: '\nInspecting the registry.', + }), + }), + ); + + // A new step reopens the narration budget. + await updateSlackLiveTaskStream( + taskRun, + { + type: 'todo_update', + ts: 1004, + todos: [ + { + id: '1', + content: 'Make the change', + status: 'in_progress' as const, + }, + ], + }, + context, + ); + await updateSlackLiveTaskStream( + taskRun, + { type: 'text', ts: 1005, text: 'Editing the selector metadata.' }, + context, + ); + + expect(mocks.appendTaskStream).toHaveBeenCalledTimes(3); + }); + + it('does not expose reasoning events in Slack', async () => { + await updateSlackLiveTaskStream( + createTaskRun(), + { type: 'reasoning', ts: 1000, text: 'private reasoning' }, + {}, + ); + + expect(mocks.sdkGetStreamData).not.toHaveBeenCalled(); + expect(mocks.appendTaskStream).not.toHaveBeenCalled(); + }); + + it('settles the card with the completion output and clears its state', async () => { + const taskRun = createTaskRun(); + await updateSlackLiveTaskStream( + taskRun, + { type: 'completion', ts: 1000, text: 'Ready for review.' }, + {}, + ); + + expect(mocks.stopTaskStream).toHaveBeenCalledWith({ + channel: 'C123', + messageTs: 'stream-ts', + task: expect.objectContaining({ + title: 'Fix the button', + status: 'complete', + output: '\nReady for review.', + }), + }); + expect(mocks.sdkClearStreamData).toHaveBeenCalledWith({ + runId: taskRun.id, + }); + }); + + it('marks a canceled run as an error when no completion event settled it', async () => { + await finishSlackLiveTaskStream(createTaskRun(), RunStatus.Canceled); + + expect(mocks.stopTaskStream).toHaveBeenCalledWith({ + channel: 'C123', + messageTs: 'stream-ts', + task: expect.objectContaining({ + status: 'error', + output: '\nTask canceled.', + }), + }); + }); + + it('settles a completed run as a fallback when the completion event was lost', async () => { + await finishSlackLiveTaskStream(createTaskRun(), RunStatus.Completed); + + expect(mocks.stopTaskStream).toHaveBeenCalledWith({ + channel: 'C123', + messageTs: 'stream-ts', + task: expect.objectContaining({ + status: 'complete', + output: '\nTask completed.', + }), + }); + }); + + it('does not settle the completion fallback twice after the real completion', async () => { + const taskRun = createTaskRun(); + await updateSlackLiveTaskStream( + taskRun, + { type: 'completion', ts: 1000, text: 'Ready for review.' }, + {}, + ); + await finishSlackLiveTaskStream(taskRun, RunStatus.Completed); + + expect(mocks.stopTaskStream).toHaveBeenCalledOnce(); + }); + + it('retains the stream for idle runs awaiting a resume', async () => { + await finishSlackLiveTaskStream(createTaskRun(), RunStatus.Idle); + + expect(mocks.stopTaskStream).not.toHaveBeenCalled(); + expect(mocks.sdkClearStreamData).not.toHaveBeenCalled(); + }); + + it('wires callbacks only for runs that opted into a card', async () => { + expect( + getSlackLiveTaskStreamRunTaskCallbacks( + createTaskRun({ payload: { description: 'no card' } }), + ), + ).toEqual({}); + + const callbacks = getSlackLiveTaskStreamRunTaskCallbacks(createTaskRun()); + expect(callbacks.onStart).toBeDefined(); + expect(callbacks.onMessage).toBeDefined(); + expect(callbacks.onExit).toBeDefined(); + }); +}); diff --git a/apps/worker/src/callbacks/slack-live-task-stream.ts b/apps/worker/src/callbacks/slack-live-task-stream.ts new file mode 100644 index 000000000..1514c9387 --- /dev/null +++ b/apps/worker/src/callbacks/slack-live-task-stream.ts @@ -0,0 +1,397 @@ +import { RunStatus } from '@roomote/types'; +import { + SlackNotifier, + type SlackLiveTaskStreamData, + type SlackTaskStreamStatus, +} from '@roomote/slack/client'; +import { sdk, type TaskRun } from '@roomote/sdk/client'; + +import type { + CallbackEvent, + RunTaskCallbacks, + RunTaskContext, +} from '../run-task'; +import { captureWorkerException } from '../monitoring/sentry'; +import { getCallbackEventKey } from './utils'; + +const updateQueues = new Map>(); + +/** Harness status noise that reads as an error but resolves on its own. */ +const TRANSIENT_NARRATION_PATTERN = /^(provider error|retrying)\b/i; + +function usesSlackLiveTaskStream(taskRun: TaskRun): boolean { + return ( + taskRun.payload !== null && + typeof taskRun.payload === 'object' && + 'liveTaskStream' in taskRun.payload && + taskRun.payload.liveTaskStream === true + ); +} + +/** One card entry. `title` is the only replaceable text slot, so it carries + * the CURRENT step (todo) and swaps cleanly on every change — no history, + * just the latest step; on settle it returns to the task title. `output` + * accumulates the narrative beneath it as newline-prefixed deltas. The + * 'View task' source is sent once by the launcher (Slack appends sources + * and details instead of replacing them). */ +function buildCardUpdate( + data: SlackLiveTaskStreamData, + status: SlackTaskStreamStatus, + content: { title?: string; output?: string }, +) { + return { + id: data.taskUpdateId, + title: content.title ?? data.title, + status, + ...(content.output ? { output: content.output } : {}), + }; +} + +/** The card's current (rotating) title, so narrative appends don't reset it + * back to the task title between todo changes. */ +function getCardTitle(context: RunTaskContext): string | undefined { + const existing = context.slackLiveTaskCardTitle; + return typeof existing === 'string' && existing ? existing : undefined; +} + +function setCardTitle(context: RunTaskContext, title: string): void { + context.slackLiveTaskCardTitle = title; +} + +async function enqueueStreamUpdate( + runId: number, + update: () => Promise, +): Promise { + const previous = updateQueues.get(runId) ?? Promise.resolve(); + const next = previous.catch(() => {}).then(update); + updateQueues.set(runId, next); + + try { + await next; + } finally { + if (updateQueues.get(runId) === next) { + updateQueues.delete(runId); + } + } +} + +function shouldProcessEvent( + event: CallbackEvent, + context: RunTaskContext, +): boolean { + // Match the Linear agent callback semantics (linear-agent.ts): ignore + // older events, allow distinct same-timestamp events, and suppress + // exact duplicates. + const lastProcessedTs = + (context.slackLiveTaskLastProcessedTs as number | undefined) ?? 0; + + if (event.ts < lastProcessedTs) { + return false; + } + + const processedEventKeys = + (context.slackLiveTaskProcessedEventKeys as Set | undefined) ?? + new Set(); + context.slackLiveTaskProcessedEventKeys = processedEventKeys; + + if (event.ts > lastProcessedTs) { + context.slackLiveTaskLastProcessedTs = event.ts; + processedEventKeys.clear(); + } + + const eventKey = getCallbackEventKey(event); + if (processedEventKeys.has(eventKey)) { + return false; + } + + processedEventKeys.add(eventKey); + return true; +} + +// The stream data is written once per task by the launcher; the worker reads +// it through the platform API (control-plane Redis is unreachable from the +// sandbox) and caches per run for the process lifetime. +const streamDataCache = new Map< + number, + Promise +>(); + +function resolveStreamData( + taskRun: TaskRun, +): Promise { + const cached = streamDataCache.get(taskRun.id); + if (cached) { + return cached; + } + + const lookup = sdk.taskRuns + .getSlackLiveTaskStreamData({ runId: taskRun.id }) + .catch((error) => { + // Do not cache transport failures; the next event retries. + streamDataCache.delete(taskRun.id); + throw error; + }); + streamDataCache.set(taskRun.id, lookup); + return lookup; +} + +let slack: SlackNotifier | undefined = undefined; + +async function getSlackNotifier(): Promise { + if (!slack) { + const slackInstallation = await sdk.slackInstallations.findFirst(); + + if (!slackInstallation) { + throw new Error('Slack installation not found.'); + } + + slack = new SlackNotifier(slackInstallation.botAccessToken); + } + + return slack; +} + +async function appendCardUpdate(params: { + taskRun: TaskRun; + context: RunTaskContext; + /** Replaces the entry title (the current step). */ + title?: string; + /** Appended to the entry's output body as a newline-prefixed delta. */ + output?: string; +}): Promise { + const outputLine = params.output?.trim(); + const title = params.title ?? getCardTitle(params.context); + + if (params.title && params.title !== getCardTitle(params.context)) { + // A new step opens a fresh narration budget for the body. + params.context.slackLiveTaskStepNarrated = false; + } + + // Skip no-op updates: nothing new to append and the title is unchanged. + if ( + !outputLine && + (!params.title || getCardTitle(params.context) === params.title) + ) { + return; + } + if (outputLine && params.context.slackLiveTaskLastOutput === outputLine) { + return; + } + + if (params.title) { + setCardTitle(params.context, params.title); + } + if (outputLine) { + params.context.slackLiveTaskLastOutput = outputLine; + } + + await enqueueStreamUpdate(params.taskRun.id, async () => { + const data = await resolveStreamData(params.taskRun); + if (!data) { + return; + } + + const notifier = await getSlackNotifier(); + await notifier.appendTaskStream({ + channel: data.channel, + messageTs: data.messageTs, + task: buildCardUpdate(data, 'in_progress', { + ...(title ? { title } : {}), + ...(outputLine ? { output: `\n${outputLine}` } : {}), + }), + }); + }); +} + +async function stopStream(params: { + taskRun: TaskRun; + status: Extract; + output: string; +}): Promise { + await enqueueStreamUpdate(params.taskRun.id, async () => { + // Fetch fresh data for the final state so the settled card carries the + // task's latest generated title. + streamDataCache.delete(params.taskRun.id); + const data = await resolveStreamData(params.taskRun); + if (!data) { + return; + } + + const notifier = await getSlackNotifier(); + const stopped = await notifier.stopTaskStream({ + channel: data.channel, + messageTs: data.messageTs, + task: buildCardUpdate(data, params.status, { + output: `\n${params.output}`, + }), + }); + + if (stopped) { + await sdk.taskRuns.clearSlackLiveTaskStreamData({ + runId: params.taskRun.id, + }); + streamDataCache.set(params.taskRun.id, Promise.resolve(null)); + } + }); +} + +export async function startSlackLiveTaskStream( + taskRun: TaskRun, +): Promise { + // The card keeps the task title until the first real step arrives; this + // just warms the stream-data cache so the first event updates instantly. + await resolveStreamData(taskRun).catch(() => {}); +} + +export async function updateSlackLiveTaskStream( + taskRun: TaskRun, + event: CallbackEvent, + context: RunTaskContext, +): Promise { + // Internal reasoning is deliberately not exposed in Slack; the card + // gets the safe semantic event stream without chain-of-thought content. + if (event.type === 'reasoning' || !shouldProcessEvent(event, context)) { + return; + } + + if (event.type === 'completion') { + await stopStream({ + taskRun, + status: 'complete', + output: event.text, + }); + return; + } + + if (event.type === 'text') { + // Appended body text is permanent, so transient status lines (provider + // retries) never enter it, and each step contributes at most one + // narration line to keep the card readable on long runs. + if ( + TRANSIENT_NARRATION_PATTERN.test(event.text.trim()) || + context.slackLiveTaskStepNarrated === true + ) { + return; + } + context.slackLiveTaskStepNarrated = true; + await appendCardUpdate({ taskRun, context, output: event.text }); + return; + } + + if (event.type === 'todo_update') { + const completedCount = event.todos.filter( + (todo) => todo.status === 'completed', + ).length; + const progress = `${completedCount}/${event.todos.length}`; + const current = + event.todos.find((todo) => todo.status === 'in_progress') ?? + event.todos.find((todo) => todo.status === 'pending'); + + await appendCardUpdate({ + taskRun, + context, + title: current + ? `${current.content} (${progress})` + : `${progress} steps complete`, + }); + return; + } + + if (event.type === 'request_user_input' || event.type === 'followup') { + await appendCardUpdate({ + taskRun, + context, + title: 'Waiting for your input…', + }); + return; + } + + if (event.type === 'request_user_input_response') { + await appendCardUpdate({ + taskRun, + context, + title: 'Continuing with your answer…', + }); + } +} + +export async function finishSlackLiveTaskStream( + taskRun: TaskRun, + status: RunStatus, +): Promise { + // Idle runs retain the stream for a later resume. + if (status === RunStatus.Idle) { + return; + } + + if (status === RunStatus.Completed) { + // Usually a no-op: the completion CallbackEvent already settled the + // stream (and cleared its data) with the real output. This fallback + // guarantees the card cannot stay spinning when that event is lost or + // its single Slack call failed. + await stopStream({ + taskRun, + status: 'complete', + output: 'Task completed.', + }); + return; + } + + await stopStream({ + taskRun, + status: 'error', + output: + status === RunStatus.Canceled + ? 'Task canceled.' + : 'The task stopped because of an error.', + }); +} + +function reportStreamCallbackError( + error: unknown, + stage: string, + runId: number, +): void { + captureWorkerException(error, { runId, stage }); +} + +/** + * Card updates for any run whose payload opted into liveTaskStream, + * independent of payload kind: Fast children run as StandardTask and + * resumes run as SnapshotResume, and all of them own a card. + */ +export function getSlackLiveTaskStreamRunTaskCallbacks( + taskRun: TaskRun, +): RunTaskCallbacks { + if (!usesSlackLiveTaskStream(taskRun)) { + return {}; + } + + return { + onStart: async (run) => { + try { + await startSlackLiveTaskStream(run); + } catch (error) { + reportStreamCallbackError(error, 'slackLiveTaskStream.onStart', run.id); + } + }, + onMessage: async (run, _taskId, event, context) => { + try { + await updateSlackLiveTaskStream(run, event, context); + } catch (error) { + reportStreamCallbackError( + error, + 'slackLiveTaskStream.onMessage', + run.id, + ); + } + }, + onExit: async (run, status) => { + try { + await finishSlackLiveTaskStream(run, status); + } catch (error) { + reportStreamCallbackError(error, 'slackLiveTaskStream.onExit', run.id); + } + }, + }; +} diff --git a/apps/worker/src/callbacks/slack-mention.ts b/apps/worker/src/callbacks/slack-mention.ts index 2cd78e40f..acfe948b5 100644 --- a/apps/worker/src/callbacks/slack-mention.ts +++ b/apps/worker/src/callbacks/slack-mention.ts @@ -311,7 +311,7 @@ export const slackMentionCallbacks: RunTaskCallbacks = { ); } }, - onExit: async (taskRun: TaskRun, _status, context: RunTaskContext) => { + onExit: async (taskRun: TaskRun, status, context: RunTaskContext) => { try { const threadIds = new Set( [...getSlackRequestUserInputReplyTargets(context).values()].map( diff --git a/apps/worker/src/commands/resume.ts b/apps/worker/src/commands/resume.ts index 39ca43573..13ff4af48 100644 --- a/apps/worker/src/commands/resume.ts +++ b/apps/worker/src/commands/resume.ts @@ -13,6 +13,8 @@ import { runTask } from '../run-task'; import { getLinearSessionIdFromResumePayload } from '../run-task/linear-resume-payload'; import { linearAgentCallbacks } from '../callbacks/linear-agent'; import { slackMentionCallbacks } from '../callbacks/slack-mention'; +import { mergeRunTaskCallbacks } from '../callbacks/communication'; +import { getSlackLiveTaskStreamRunTaskCallbacks } from '../callbacks/slack-live-task-stream'; import { buildWorkspaceConfig, executeTaskRun } from './utils'; @@ -75,10 +77,19 @@ export async function resume(runId: number): Promise { getSlackThreadTsFromTaskPayload(jobContext.taskRun.payload), ); + // defaultCallbacks (from executeTaskRun) already merge the live + // task-card callbacks; the linear/slack overrides must re-add them so + // a resumed run with a card keeps updating it. const callbacks = isLinearResume - ? linearAgentCallbacks + ? mergeRunTaskCallbacks( + linearAgentCallbacks, + getSlackLiveTaskStreamRunTaskCallbacks(jobContext.taskRun), + ) : isSlackResume - ? slackMentionCallbacks + ? mergeRunTaskCallbacks( + slackMentionCallbacks, + getSlackLiveTaskStreamRunTaskCallbacks(jobContext.taskRun), + ) : defaultCallbacks; // Seed callback context for resumed integrations before runTask starts. diff --git a/apps/worker/src/commands/utils/execute-task-run.ts b/apps/worker/src/commands/utils/execute-task-run.ts index 1f92976d8..fe032d761 100644 --- a/apps/worker/src/commands/utils/execute-task-run.ts +++ b/apps/worker/src/commands/utils/execute-task-run.ts @@ -23,6 +23,7 @@ import { import type { WorkspaceConfig } from '../../workspace'; import type { RepoLocalSkill } from '../../workspace/repo-local-skills'; import { callbackMap } from '../../callbacks'; +import { getSlackLiveTaskStreamRunTaskCallbacks } from '../../callbacks/slack-live-task-stream'; import { getCommunicationRunTaskCallbacks, mergeRunTaskCallbacks, @@ -346,6 +347,7 @@ export async function executeTaskRun({ callbacks = mergeRunTaskCallbacks( callbackMap[taskRun.payloadKind as TaskPayloadKind] ?? {}, getCommunicationRunTaskCallbacks(taskRun), + getSlackLiveTaskStreamRunTaskCallbacks(taskRun), ); workerEnv = WorkerEnv.fromProcessEnv(process.env); diff --git a/deploy/README.md b/deploy/README.md index 4e9669aa5..1c9c2bc26 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -257,9 +257,10 @@ To let Terraform create these records in DigitalOcean DNS, pass that zone. Caddy serves web and worker-facing API traffic on the app domain. The app -domain routes the reserved `/_roomote-api/*` prefix to the API container after -stripping that prefix, routes the configured artifact bucket path to MinIO for -presigned S3 requests, and sends other app-domain paths to the web container. +domain routes public `/api/webhooks/*` requests directly to the API, routes the +reserved `/_roomote-api/*` prefix to the API after stripping that prefix, +routes the configured artifact bucket path to MinIO for presigned S3 requests, +and sends other app-domain paths to the web container. ## Create A Deployment diff --git a/deploy/caddy/Caddyfile b/deploy/caddy/Caddyfile index ee8cb874c..cb4f2aed0 100644 --- a/deploy/caddy/Caddyfile +++ b/deploy/caddy/Caddyfile @@ -33,6 +33,16 @@ } } + # Webhook URLs are public integration contracts. Route them directly to the + # API rather than relying on the web container's compatibility proxy. + @api_webhooks { + path /api/webhooks /api/webhooks/* + } + + handle @api_webhooks { + import roomote_proxy api:3001 + } + @local_sandbox path_regexp local_sandbox ^/_roomote-sandbox/([a-z0-9]+)(/.*)$ handle @local_sandbox { diff --git a/deploy/ci/validate-deployment-artifacts.mjs b/deploy/ci/validate-deployment-artifacts.mjs index be18f5554..000850b01 100644 --- a/deploy/ci/validate-deployment-artifacts.mjs +++ b/deploy/ci/validate-deployment-artifacts.mjs @@ -307,6 +307,16 @@ assert( 'caddy: Caddyfile must allow internal mode to remove wildcard on-demand TLS', ); const caddyfile = read('deploy/caddy/Caddyfile'); +assert( + caddyfile.includes('path /api/webhooks /api/webhooks/*'), + 'caddy: app domain must route public webhooks directly to the API', +); +assert( + caddyfile.includes( + 'handle @api_webhooks {\n\t\timport roomote_proxy api:3001', + ), + 'caddy: public webhooks must bypass the web application', +); assert( caddyfile.includes( 'path_regexp local_sandbox ^/_roomote-sandbox/([a-z0-9]+)(/.*)$', diff --git a/docker-compose.yml b/docker-compose.yml index 74509c7cb..92d88a8d1 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -65,7 +65,8 @@ services: mc mb --ignore-existing "roomote/$$S3_BUCKET_ARTIFACTS" # Local dev edge mirroring the production Caddy routing: one origin serves - # the web app and proxies the reserved /_roomote-api/* prefix to the API. + # the web app, routes public webhooks directly to the API, and proxies the + # reserved /_roomote-api/* prefix to the API. # The dev CLI points the public tunnel at this port so hosted compute # workers can reach the API through the public app URL. caddy-dev: diff --git a/packages/cloud-agents/src/server/task-run-queue.ts b/packages/cloud-agents/src/server/task-run-queue.ts index bfc4ccdd9..6522fef94 100644 --- a/packages/cloud-agents/src/server/task-run-queue.ts +++ b/packages/cloud-agents/src/server/task-run-queue.ts @@ -2155,6 +2155,17 @@ function inheritSnapshotResumeFastAgentContext( ) { payload.communicationContextInherited = true; } + + if ( + sourcePayload && + typeof sourcePayload === 'object' && + !Array.isArray(sourcePayload) && + (sourcePayload as Record).liveTaskStream === true + ) { + // The card in the Slack thread belongs to the task; every resumed run + // must keep updating it. + payload.liveTaskStream = true; + } } async function enqueueSnapshotResume( diff --git a/packages/sdk/src/server/lib/task-runs/slack-live-task-stream.ts b/packages/sdk/src/server/lib/task-runs/slack-live-task-stream.ts new file mode 100644 index 000000000..b51e83765 --- /dev/null +++ b/packages/sdk/src/server/lib/task-runs/slack-live-task-stream.ts @@ -0,0 +1,56 @@ +import { + buildSlackLiveTaskTitle, + clearSlackLiveTaskStreamData, + getSlackLiveTaskStreamData, + type SlackLiveTaskStreamData, +} from '@roomote/slack'; +import { db, eq, taskRuns, tasks } from '@roomote/db/server'; + +/** + * Serve the run's live task-card data to workers. The data lives in + * control-plane Redis keyed by task id (stable across snapshot resumes); + * sandboxed workers can only reach it through this API. + * + * The card title tracks the task's generated title once one exists (the + * launcher only had the raw prompt at stream-start time); task_update + * titles replace on append, so the worker's next update renames the card. + */ +export async function getSlackLiveTaskStreamDataForRun( + runId: number, +): Promise { + const run = await db.query.taskRuns.findFirst({ + where: eq(taskRuns.id, runId), + columns: { taskId: true }, + }); + if (!run) { + return null; + } + + const data = await getSlackLiveTaskStreamData(run.taskId); + if (!data) { + return null; + } + + const task = await db.query.tasks.findFirst({ + where: eq(tasks.id, run.taskId), + columns: { title: true }, + }); + const taskTitle = task?.title?.trim(); + + return taskTitle + ? { ...data, title: buildSlackLiveTaskTitle(taskTitle) } + : data; +} + +export async function clearSlackLiveTaskStreamDataForRun( + runId: number, +): Promise { + const run = await db.query.taskRuns.findFirst({ + where: eq(taskRuns.id, runId), + columns: { taskId: true }, + }); + + if (run) { + await clearSlackLiveTaskStreamData(run.taskId); + } +} diff --git a/packages/sdk/src/server/routers/task-runs.ts b/packages/sdk/src/server/routers/task-runs.ts index f35471e4a..d0524e97f 100644 --- a/packages/sdk/src/server/routers/task-runs.ts +++ b/packages/sdk/src/server/routers/task-runs.ts @@ -78,6 +78,10 @@ import { } from '@roomote/linear'; import { publishCommunicationRequestUserInput } from '../lib/communication-request-user-input'; import { publishFastAgentRequestUserInput } from '../lib/task-runs/publish-fast-agent-request-user-input'; +import { + clearSlackLiveTaskStreamDataForRun, + getSlackLiveTaskStreamDataForRun, +} from '../lib/task-runs/slack-live-task-stream'; import { authenticatedProcedure, isRunToken, @@ -541,6 +545,14 @@ export const taskRunsRouter = router({ getMessageSources: runScoped(z.object({ runId: z.number() }), 'runId').query( ({ input }) => getMessageSources(input.runId), ), + getSlackLiveTaskStreamData: runScoped( + z.object({ runId: z.number() }), + 'runId', + ).query(({ input }) => getSlackLiveTaskStreamDataForRun(input.runId)), + clearSlackLiveTaskStreamData: runScoped( + z.object({ runId: z.number() }), + 'runId', + ).mutation(({ input }) => clearSlackLiveTaskStreamDataForRun(input.runId)), getResolvedGitAuthor: runScoped( z.object({ runId: z.number() }), 'runId', diff --git a/packages/sdk/src/task-runs.ts b/packages/sdk/src/task-runs.ts index 4ac056a24..e70d5ccf9 100644 --- a/packages/sdk/src/task-runs.ts +++ b/packages/sdk/src/task-runs.ts @@ -256,6 +256,14 @@ export const getMessageSources = ( options: AppRouterInput['taskRuns']['getMessageSources'], ) => client.taskRuns.getMessageSources.query(options); +export const getSlackLiveTaskStreamData = ( + options: AppRouterInput['taskRuns']['getSlackLiveTaskStreamData'], +) => client.taskRuns.getSlackLiveTaskStreamData.query(options); + +export const clearSlackLiveTaskStreamData = ( + options: AppRouterInput['taskRuns']['clearSlackLiveTaskStreamData'], +) => client.taskRuns.clearSlackLiveTaskStreamData.mutate(options); + export const getResolvedGitAuthor = ( options: AppRouterInput['taskRuns']['getResolvedGitAuthor'], ) => client.taskRuns.getResolvedGitAuthor.query(options); diff --git a/packages/slack/scripts/run-mock-slack.ts b/packages/slack/scripts/run-mock-slack.ts index 698175db3..33c42af76 100644 --- a/packages/slack/scripts/run-mock-slack.ts +++ b/packages/slack/scripts/run-mock-slack.ts @@ -61,6 +61,28 @@ const configSchema = z.object({ attachments: z.array(z.unknown()).optional(), ephemeral: z.boolean().optional(), reactions: z.array(z.string()).optional(), + chunks: z.array(z.unknown()).optional(), + streaming_state: z.enum(['in_progress', 'completed']).optional(), + }), + ) + .optional(), + assistantThreadStatuses: z + .array( + z.object({ + channel: z.string().min(1), + threadTs: z.string().min(1), + status: z.string(), + }), + ) + .optional(), + streamEvents: z + .array( + z.object({ + method: z.enum(['start', 'append', 'stop']), + channel: z.string().min(1), + messageTs: z.string().min(1), + threadTs: z.string().optional(), + chunks: z.array(z.unknown()), }), ) .optional(), diff --git a/packages/slack/src/__tests__/mock-slack-server.test.ts b/packages/slack/src/__tests__/mock-slack-server.test.ts index c186c4743..051a38674 100644 --- a/packages/slack/src/__tests__/mock-slack-server.test.ts +++ b/packages/slack/src/__tests__/mock-slack-server.test.ts @@ -806,4 +806,119 @@ describe('MockSlackServer', () => { await server.stop(); } }); + + it('records assistant status and native task stream updates', async () => { + const server = new MockSlackServer({ + state: { + team: { id: 'T1', domain: 'mock-roomote' }, + acceptedBotTokens: ['xoxb-mock-token'], + channels: [{ id: 'C1', name: 'product-debug', isMember: true }], + users: [{ id: 'U1', name: 'alex', displayName: 'Alex' }], + messages: [ + { + channel: 'C1', + ts: '1710000000.000100', + text: 'thread root', + user: 'U1', + type: 'message', + }, + ], + }, + }); + + const post = (method: string, body: Record) => + fetch(`${server.baseUrl}/api/${method}`, { + method: 'POST', + headers: { + authorization: 'Bearer xoxb-mock-token', + 'content-type': 'application/json', + }, + body: JSON.stringify(body), + }); + + try { + await server.start(); + + await post('assistant.threads.setStatus', { + channel_id: 'C1', + thread_ts: '1710000000.000100', + status: 'is thinking…', + }); + expect(server.getState().assistantThreadStatuses).toEqual([ + { + channel: 'C1', + threadTs: '1710000000.000100', + status: 'is thinking…', + }, + ]); + + const startResponse = await post('chat.startStream', { + channel: 'C1', + thread_ts: '1710000000.000100', + chunks: [ + { + type: 'task_update', + id: 'task-1', + title: 'Fix the button', + status: 'pending', + }, + ], + }); + const started = (await startResponse.json()) as { ts: string }; + + await post('chat.appendStream', { + channel: 'C1', + ts: started.ts, + chunks: [ + { + type: 'task_update', + id: 'task-1', + title: 'Fix the button', + status: 'in_progress', + details: 'Updating the component', + }, + ], + }); + await post('chat.stopStream', { + channel: 'C1', + ts: started.ts, + chunks: [ + { + type: 'task_update', + id: 'task-1', + title: 'Fix the button', + status: 'complete', + output: 'Ready for review', + }, + ], + }); + await post('assistant.threads.setStatus', { + channel_id: 'C1', + thread_ts: '1710000000.000100', + status: '', + }); + + const state = server.getState(); + expect(state.assistantThreadStatuses).toEqual([]); + expect(state.streamEvents?.map((event) => event.method)).toEqual([ + 'start', + 'append', + 'stop', + ]); + expect( + state.messages?.find((message) => message.ts === started.ts), + ).toMatchObject({ + streaming_state: 'completed', + chunks: [ + expect.objectContaining({ + id: 'task-1', + status: 'complete', + output: 'Ready for review', + }), + ], + }); + } finally { + await server.stop(); + } + }); }); diff --git a/packages/slack/src/__tests__/slack-notifier.test.ts b/packages/slack/src/__tests__/slack-notifier.test.ts index f3f57152a..509b1e120 100644 --- a/packages/slack/src/__tests__/slack-notifier.test.ts +++ b/packages/slack/src/__tests__/slack-notifier.test.ts @@ -240,6 +240,127 @@ describe('SlackNotifier', () => { }); }); + describe('native agent surfaces', () => { + it('sets assistant status and drives a task stream', async () => { + getGlobalWithFetch().fetch = vi + .fn() + .mockResolvedValueOnce({ + ok: true, + json: async () => ({ ok: true }), + }) + .mockResolvedValueOnce({ + ok: true, + json: async () => ({ ok: true, ts: 'stream-ts' }), + }) + .mockResolvedValueOnce({ + ok: true, + json: async () => ({ ok: true, ts: 'stream-ts' }), + }) + .mockResolvedValueOnce({ + ok: true, + json: async () => ({ ok: true, ts: 'stream-ts' }), + }); + + await expect( + notifier.setAssistantThreadStatus({ + channel: 'C123', + threadTs: '100.001', + status: 'is thinking…', + }), + ).resolves.toBe(true); + + const initialTask = { + id: 'task-1', + title: 'Fix the button', + status: 'pending' as const, + }; + await expect( + notifier.startTaskStream({ + channel: 'C123', + threadTs: '100.001', + recipientTeamId: 'T123', + recipientUserId: 'U123', + task: initialTask, + }), + ).resolves.toBe('stream-ts'); + await expect( + notifier.appendTaskStream({ + channel: 'C123', + messageTs: 'stream-ts', + task: { ...initialTask, status: 'in_progress' }, + }), + ).resolves.toBe(true); + await expect( + notifier.stopTaskStream({ + channel: 'C123', + messageTs: 'stream-ts', + task: { + ...initialTask, + status: 'complete', + output: 'Ready for review', + }, + }), + ).resolves.toBe(true); + + expect(getGlobalWithFetch().fetch).toHaveBeenNthCalledWith( + 1, + 'https://slack.com/api/assistant.threads.setStatus', + expect.objectContaining({ + body: JSON.stringify({ + channel_id: 'C123', + thread_ts: '100.001', + status: 'is thinking…', + }), + }), + ); + expect(getGlobalWithFetch().fetch).toHaveBeenNthCalledWith( + 2, + 'https://slack.com/api/chat.startStream', + expect.objectContaining({ + body: JSON.stringify({ + channel: 'C123', + thread_ts: '100.001', + task_display_mode: 'timeline', + recipient_team_id: 'T123', + recipient_user_id: 'U123', + chunks: [{ type: 'task_update', ...initialTask }], + }), + }), + ); + expect(getGlobalWithFetch().fetch).toHaveBeenNthCalledWith( + 3, + 'https://slack.com/api/chat.appendStream', + expect.objectContaining({ + body: JSON.stringify({ + channel: 'C123', + ts: 'stream-ts', + chunks: [ + { type: 'task_update', ...initialTask, status: 'in_progress' }, + ], + }), + }), + ); + expect(getGlobalWithFetch().fetch).toHaveBeenNthCalledWith( + 4, + 'https://slack.com/api/chat.stopStream', + expect.objectContaining({ + body: JSON.stringify({ + channel: 'C123', + ts: 'stream-ts', + chunks: [ + { + type: 'task_update', + ...initialTask, + status: 'complete', + output: 'Ready for review', + }, + ], + }), + }), + ); + }); + }); + describe('postMessageDetailed', () => { it('returns the message ts on success', async () => { getGlobalWithFetch().fetch = vi.fn().mockResolvedValue({ diff --git a/packages/slack/src/__tests__/start-slack-app-mention.test.ts b/packages/slack/src/__tests__/start-slack-app-mention.test.ts index c734178a0..e86ab16f0 100644 --- a/packages/slack/src/__tests__/start-slack-app-mention.test.ts +++ b/packages/slack/src/__tests__/start-slack-app-mention.test.ts @@ -123,6 +123,34 @@ describe('startSlackAppMentionTask', () => { ); }); + it('runs a surface hook before dispatching a fresh task run', async () => { + const beforeTaskRunDispatch = vi.fn(); + enqueueTaskMock.mockImplementationOnce(async (_input, options) => { + await options.beforeEnqueue({ id: 42, taskId: 'task_123' }); + return { id: 42, taskId: 'task_123' }; + }); + const { startSlackAppMentionTask } = + await import('../start-slack-app-mention'); + + await startSlackAppMentionTask({ + initiator: { kind: 'user', userId: 'user_123' }, + trigger: 'message', + channel: 'C123', + teamId: 'T123', + slackUserId: 'U123', + text: 'hello', + ts: '111.000', + threadTs: '111.000', + repo: 'owner/repo', + beforeTaskRunDispatch, + }); + + expect(beforeTaskRunDispatch).toHaveBeenCalledWith({ + id: 42, + taskId: 'task_123', + }); + }); + it('persists an exact Slack conversation permalink onto a reused active task run', async () => { findActiveSlackTaskRunMock.mockResolvedValueOnce({ id: 99, @@ -214,4 +242,40 @@ describe('startSlackAppMentionTask', () => { }), ); }); + + it('runs a surface hook before dispatching an active-task follow-up', async () => { + findActiveSlackTaskRunMock.mockResolvedValueOnce({ + id: 99, + taskId: 'task_existing', + payload: { + channel: 'C123', + text: 'earlier text', + thread_ts: '111.000', + }, + }); + const beforeTaskRunDispatch = vi.fn(); + const { startSlackAppMentionTask } = + await import('../start-slack-app-mention'); + + await startSlackAppMentionTask({ + initiator: { kind: 'user', userId: 'user_123' }, + trigger: 'message', + channel: 'C123', + teamId: 'T123', + slackUserId: 'U123', + text: 'hello again', + ts: '111.001', + threadTs: '111.000', + repo: 'owner/repo', + beforeTaskRunDispatch, + }); + + expect(beforeTaskRunDispatch).toHaveBeenCalledWith({ + id: 99, + taskId: 'task_existing', + }); + expect(beforeTaskRunDispatch.mock.invocationCallOrder[0]).toBeLessThan( + queueSlackMessageMock.mock.invocationCallOrder[0]!, + ); + }); }); diff --git a/packages/slack/src/client.ts b/packages/slack/src/client.ts index 3cf1e1be5..d1692048a 100644 --- a/packages/slack/src/client.ts +++ b/packages/slack/src/client.ts @@ -1,4 +1,15 @@ export { SlackNotifier } from './slack-notifier'; +export type { + SlackTaskStreamStatus, + SlackTaskStreamUpdate, +} from './slack-notifier'; +export { + buildSlackLiveTaskTitle, + clearSlackLiveTaskStreamData, + getSlackLiveTaskStreamData, + setSlackLiveTaskStreamData, +} from './live-task-stream'; +export type { SlackLiveTaskStreamData } from './live-task-stream'; export { convertMarkdownToSlack, diff --git a/packages/slack/src/index.ts b/packages/slack/src/index.ts index fee29fb25..2fdff26ad 100644 --- a/packages/slack/src/index.ts +++ b/packages/slack/src/index.ts @@ -10,6 +10,7 @@ export * from './automation-root-footer'; export * from './automation-result-blocks'; export * from './handle-followup-answer'; export * from './interactive-response'; +export * from './live-task-stream'; export * from './markdown-converter'; export * from './mcp-recommendations'; export * from './mcp-setup-suggestion'; diff --git a/packages/slack/src/live-task-stream.ts b/packages/slack/src/live-task-stream.ts new file mode 100644 index 000000000..27f62856a --- /dev/null +++ b/packages/slack/src/live-task-stream.ts @@ -0,0 +1,77 @@ +import { getRedis } from '@roomote/redis'; + +const SLACK_LIVE_TASK_STREAM_TTL_SECONDS = 7 * 24 * 60 * 60; +const SLACK_LIVE_TASK_TITLE_MAX_LENGTH = 160; + +export interface SlackLiveTaskStreamData { + channel: string; + messageTs: string; + taskId: string; + taskUpdateId: string; + threadTs: string; + title: string; + taskUrl?: string; +} + +// Keyed by task id: runs are replaced on snapshot resume, but the card in the +// Slack thread belongs to the task for its whole lifetime. +function getSlackLiveTaskStreamKey(taskId: string): string { + return `slack:live_task_stream:task:${taskId}`; +} + +export function buildSlackLiveTaskTitle(prompt: string): string { + const normalized = prompt.replace(/\s+/g, ' ').trim(); + + if (normalized.length <= SLACK_LIVE_TASK_TITLE_MAX_LENGTH) { + return normalized; + } + + return `${normalized.slice(0, SLACK_LIVE_TASK_TITLE_MAX_LENGTH - 1).trimEnd()}…`; +} + +export async function setSlackLiveTaskStreamData( + taskId: string, + data: SlackLiveTaskStreamData, +): Promise { + await getRedis().set( + getSlackLiveTaskStreamKey(taskId), + JSON.stringify(data), + 'EX', + SLACK_LIVE_TASK_STREAM_TTL_SECONDS, + ); +} + +export async function getSlackLiveTaskStreamData( + taskId: string, +): Promise { + const raw = await getRedis().get(getSlackLiveTaskStreamKey(taskId)); + + if (!raw) { + return null; + } + + try { + const parsed = JSON.parse(raw) as Partial; + + if ( + typeof parsed.channel !== 'string' || + typeof parsed.messageTs !== 'string' || + typeof parsed.taskId !== 'string' || + typeof parsed.taskUpdateId !== 'string' || + typeof parsed.threadTs !== 'string' || + typeof parsed.title !== 'string' + ) { + return null; + } + + return parsed as SlackLiveTaskStreamData; + } catch { + return null; + } +} + +export async function clearSlackLiveTaskStreamData( + taskId: string, +): Promise { + await getRedis().del(getSlackLiveTaskStreamKey(taskId)); +} diff --git a/packages/slack/src/mock-slack-server.ts b/packages/slack/src/mock-slack-server.ts index 9e0cdfd43..2a5323e49 100644 --- a/packages/slack/src/mock-slack-server.ts +++ b/packages/slack/src/mock-slack-server.ts @@ -32,6 +32,22 @@ type MockSlackStoredMessage = { files?: SlackFile[]; ephemeral?: boolean; reactions?: string[]; + chunks?: unknown[]; + streaming_state?: 'in_progress' | 'completed'; +}; + +export type MockSlackAssistantThreadStatus = { + channel: string; + threadTs: string; + status: string; +}; + +export type MockSlackStreamEvent = { + method: 'start' | 'append' | 'stop'; + channel: string; + messageTs: string; + threadTs?: string; + chunks: unknown[]; }; export type MockSlackUser = { @@ -99,6 +115,8 @@ export type MockSlackState = { manifestCredentials?: MockSlackManifestCredentials; /** Apps created through `apps.manifest.create`, oldest first. */ createdManifests?: MockSlackCreatedManifest[]; + assistantThreadStatuses?: MockSlackAssistantThreadStatus[]; + streamEvents?: MockSlackStreamEvent[]; }; export type MockSlackRoomoteTarget = { @@ -153,6 +171,8 @@ function normalizeState(state: MockSlackState): MockSlackState { reactions: [], ...message, })), + assistantThreadStatuses: cloneState(state.assistantThreadStatuses ?? []), + streamEvents: cloneState(state.streamEvents ?? []), }; } @@ -436,6 +456,49 @@ export class MockSlackServer { return `${seconds}.${microseconds}`; } + private mergeStreamChunks( + existing: unknown[] | undefined, + incoming: unknown[], + ): unknown[] { + const merged = [...(existing ?? [])]; + + for (const chunk of incoming) { + if ( + chunk && + typeof chunk === 'object' && + !Array.isArray(chunk) && + (chunk as JsonRecord).type === 'task_update' && + typeof (chunk as JsonRecord).id === 'string' + ) { + const taskId = (chunk as JsonRecord).id; + const existingIndex = merged.findIndex( + (candidate) => + candidate !== null && + typeof candidate === 'object' && + !Array.isArray(candidate) && + (candidate as JsonRecord).type === 'task_update' && + (candidate as JsonRecord).id === taskId, + ); + + if (existingIndex >= 0) { + merged.splice(existingIndex, 1, cloneState(chunk)); + continue; + } + } + + merged.push(cloneState(chunk)); + } + + return merged; + } + + private recordStreamEvent(event: MockSlackStreamEvent): void { + this.state.streamEvents = [ + ...(this.state.streamEvents ?? []), + cloneState(event), + ]; + } + private async handleRequest( request: IncomingMessage, response: ServerResponse, @@ -779,6 +842,80 @@ export class MockSlackServer { return; } + case 'POST assistant.threads.setStatus': { + const channel = String(jsonBody.channel_id ?? ''); + const threadTs = String(jsonBody.thread_ts ?? ''); + const status = String(jsonBody.status ?? ''); + const remaining = (this.state.assistantThreadStatuses ?? []).filter( + (entry) => entry.channel !== channel || entry.threadTs !== threadTs, + ); + + this.state.assistantThreadStatuses = status + ? [...remaining, { channel, threadTs, status }] + : remaining; + json(response, 200, { ok: true }); + return; + } + + case 'POST chat.startStream': { + const ts = this.nextTs(); + const chunks = Array.isArray(jsonBody.chunks) ? jsonBody.chunks : []; + const message = this.storeOutgoingMessage({ + ts, + payload: jsonBody, + ephemeral: false, + }); + message.chunks = cloneState(chunks); + message.streaming_state = 'in_progress'; + this.recordStreamEvent({ + method: 'start', + channel: message.channel, + messageTs: message.ts, + ...(message.thread_ts ? { threadTs: message.thread_ts } : {}), + chunks, + }); + json(response, 200, this.successResponse(message)); + return; + } + + case 'POST chat.appendStream': + case 'POST chat.stopStream': { + const channel = String(jsonBody.channel ?? ''); + const ts = String(jsonBody.ts ?? ''); + const chunks = Array.isArray(jsonBody.chunks) ? jsonBody.chunks : []; + const message = (this.state.messages ?? []).find( + (entry) => entry.channel === channel && entry.ts === ts, + ); + + if (!message || !message.streaming_state) { + json(response, 200, { ok: false, error: 'message_not_found' }); + return; + } + + if (message.streaming_state === 'completed') { + json(response, 200, { + ok: false, + error: 'streaming_state_conflict', + }); + return; + } + + message.chunks = this.mergeStreamChunks(message.chunks, chunks); + const stopping = path === 'chat.stopStream'; + if (stopping) { + message.streaming_state = 'completed'; + } + this.recordStreamEvent({ + method: stopping ? 'stop' : 'append', + channel, + messageTs: ts, + ...(message.thread_ts ? { threadTs: message.thread_ts } : {}), + chunks, + }); + json(response, 200, this.successResponse(message)); + return; + } + case 'POST chat.postEphemeral': { const ts = this.nextTs(); const message = this.storeOutgoingMessage({ diff --git a/packages/slack/src/slack-notifier.ts b/packages/slack/src/slack-notifier.ts index d8f320818..7fb0d9777 100644 --- a/packages/slack/src/slack-notifier.ts +++ b/packages/slack/src/slack-notifier.ts @@ -83,6 +83,80 @@ type SlackUsersListResponse = { }; }; +export type SlackTaskStreamStatus = + | 'pending' + | 'in_progress' + | 'complete' + | 'error'; + +export interface SlackTaskStreamUpdate { + id: string; + title: string; + status: SlackTaskStreamStatus; + details?: string; + output?: string; + sources?: Array<{ + type: 'url'; + url: string; + text: string; + }>; +} + +// The documented chunk guidance is 256 chars, but the surface demonstrably +// demonstrably renders far larger details/output content. +// callSlackAgentApi logs Slack's response warnings, which is where a dropped +// oversized chunk would surface. +const SLACK_TASK_STREAM_CHUNK_MAX_CHARS = 256; +const SLACK_TASK_STREAM_DETAILS_MAX_CHARS = 3000; +const SLACK_TASK_STREAM_OUTPUT_MAX_CHARS = 4000; + +function truncateWithEllipsis(text: string, maxLength: number): string { + const normalized = text.trim(); + + if (normalized.length <= maxLength) { + return normalized; + } + + return `${normalized.slice(0, maxLength - 1).trimEnd()}…`; +} + +/** Length cap that preserves whitespace: details/output are append deltas + * whose leading newlines separate them from already-streamed content. */ +function truncatePreservingWhitespace(text: string, maxLength: number): string { + if (text.length <= maxLength) { + return text; + } + + return `${text.slice(0, maxLength - 1)}…`; +} + +/** Cap each task_update field at its rendering-safe budget. */ +function fitTaskStreamUpdate( + task: SlackTaskStreamUpdate, +): Record { + return { + type: 'task_update', + ...task, + title: truncateWithEllipsis(task.title, SLACK_TASK_STREAM_CHUNK_MAX_CHARS), + ...(task.details + ? { + details: truncatePreservingWhitespace( + task.details, + SLACK_TASK_STREAM_DETAILS_MAX_CHARS, + ), + } + : {}), + ...(task.output + ? { + output: truncatePreservingWhitespace( + task.output, + SLACK_TASK_STREAM_OUTPUT_MAX_CHARS, + ), + } + : {}), + }; +} + const SLACK_USERS_LIST_LIMIT = 999; const MAX_SLACK_CONVERSATIONS_REPLIES_RATE_LIMIT_RETRIES = 3; const MAX_SLACK_UPDATE_RETRIES = 2; @@ -927,6 +1001,154 @@ export class SlackNotifier { } } + private async callSlackAgentApi( + endpoint: + | 'assistant.threads.setStatus' + | 'chat.startStream' + | 'chat.appendStream' + | 'chat.stopStream', + payload: Record, + ): Promise { + try { + const response = await slackFetch(buildSlackApiUrl(endpoint), { + method: 'POST', + headers: { + 'Content-Type': 'application/json; charset=utf-8', + Authorization: `Bearer ${this.token}`, + }, + body: JSON.stringify(payload), + }); + + if (!response.ok) { + console.error( + `[callSlackAgentApi] Slack ${endpoint} failed: ${response.status} ${response.statusText}`, + ); + return null; + } + + const result = (await response.json()) as SlackResponse; + + if (!result.ok) { + console.error( + `[callSlackAgentApi] Slack ${endpoint} error: ${result.error ?? 'unknown_error'}`, + ); + } + + // Slack reports dropped chunks (for example an oversized task_update) + // as warnings on an ok:true response; surface them or the stream + // renders as an empty message with no trace of why. + const warnings = [ + result.warning, + ...(result.response_metadata?.warnings ?? []), + ...(result.response_metadata?.messages ?? []), + ].filter(Boolean); + if (warnings.length > 0) { + console.warn( + `[callSlackAgentApi] Slack ${endpoint} warnings: ${JSON.stringify(warnings)}`, + ); + } + + return result; + } catch (error) { + console.error( + `[callSlackAgentApi] Failed to call Slack ${endpoint}: ${error instanceof Error ? error.message : String(error)}`, + ); + return null; + } + } + + /** + * Shows Slack's native app/thread status (for example, "Roomote is + * thinking…"). Passing an empty status clears the indicator. + */ + public async setAssistantThreadStatus({ + channel, + threadTs, + status, + }: { + channel: string; + threadTs: string; + status: string; + }): Promise { + const response = await this.callSlackAgentApi( + 'assistant.threads.setStatus', + { + channel_id: channel, + thread_ts: threadTs, + status, + }, + ); + + return response?.ok === true; + } + + /** Starts a native Slack task stream and returns its message timestamp. */ + public async startTaskStream({ + channel, + threadTs, + recipientTeamId, + recipientUserId, + task, + }: { + channel: string; + threadTs: string; + recipientTeamId?: string; + recipientUserId?: string; + task: SlackTaskStreamUpdate; + }): Promise { + const response = await this.callSlackAgentApi('chat.startStream', { + channel, + thread_ts: threadTs, + // A single entry renders as one card in timeline mode (plan mode + // would duplicate the title as both header and row). NOTE: 'dense' is + // documented but the API rejects it as an invalid enum value. + task_display_mode: 'timeline', + ...(recipientTeamId ? { recipient_team_id: recipientTeamId } : {}), + ...(recipientUserId ? { recipient_user_id: recipientUserId } : {}), + chunks: [fitTaskStreamUpdate(task)], + }); + + return response?.ok ? response.ts : undefined; + } + + /** Updates the task card inside an existing native Slack stream. */ + public async appendTaskStream({ + channel, + messageTs, + task, + }: { + channel: string; + messageTs: string; + task: SlackTaskStreamUpdate; + }): Promise { + const response = await this.callSlackAgentApi('chat.appendStream', { + channel, + ts: messageTs, + chunks: [fitTaskStreamUpdate(task)], + }); + + return response?.ok === true; + } + + /** Settles a native Slack task stream with its final task-card state. */ + public async stopTaskStream({ + channel, + messageTs, + task, + }: { + channel: string; + messageTs: string; + task: SlackTaskStreamUpdate; + }): Promise { + const response = await this.callSlackAgentApi('chat.stopStream', { + channel, + ts: messageTs, + chunks: [fitTaskStreamUpdate(task)], + }); + + return response?.ok === true; + } + public async postMessage(message: SlackMessage) { return (await this.postMessageDetailed(message)).ts; } diff --git a/packages/slack/src/start-slack-app-mention.ts b/packages/slack/src/start-slack-app-mention.ts index 657a4ae15..224cd2870 100644 --- a/packages/slack/src/start-slack-app-mention.ts +++ b/packages/slack/src/start-slack-app-mention.ts @@ -153,6 +153,7 @@ export async function startSlackAppMentionTask(input: { environmentId?: string; reasoningEffort?: ReasoningEffort; readinessMessage?: string; + liveTaskStream?: boolean; images?: string[]; threadMessages?: SlackThreadMessage[]; latestOwnBotReplyText?: string; @@ -160,6 +161,15 @@ export async function startSlackAppMentionTask(input: { webPath?: string; slackConversationUrl?: string; skipInitialActingUser?: boolean; + /** + * Runs after the task run exists but before its work is dispatched. Throwing + * aborts the dispatch; callers that use this for best-effort UI should catch + * their own errors. + */ + beforeTaskRunDispatch?: (taskRun: { + id: number; + taskId: string; + }) => Promise; /** * Started-message metadata callers persist themselves via * setSlackStartedMessageTs after the launch. Accepted here so call sites @@ -236,6 +246,11 @@ export async function startSlackAppMentionTask(input: { } } + await input.beforeTaskRunDispatch?.({ + id: activeRun.id, + taskId: activeRun.taskId, + }); + await queueSlackMessage(activeRun.id, { text: input.text, user: input.slackUserId, @@ -308,6 +323,7 @@ export async function startSlackAppMentionTask(input: { ...(input.readinessMessage ? { readinessMessage: input.readinessMessage } : {}), + ...(input.liveTaskStream ? { liveTaskStream: true } : {}), ...(input.images?.length ? { images: input.images } : {}), ...(promptRelevantThreadMessages?.length ? { threadMessages: promptRelevantThreadMessages } @@ -338,7 +354,12 @@ export async function startSlackAppMentionTask(input: { slackThreadTs: input.threadTs, }, }, - input.skipInitialActingUser ? { skipInitialActingUser: true } : {}, + { + ...(input.skipInitialActingUser ? { skipInitialActingUser: true } : {}), + ...(input.beforeTaskRunDispatch + ? { beforeEnqueue: input.beforeTaskRunDispatch } + : {}), + }, ); return { diff --git a/packages/slack/src/types.ts b/packages/slack/src/types.ts index e1fc24c58..c4482d495 100644 --- a/packages/slack/src/types.ts +++ b/packages/slack/src/types.ts @@ -27,6 +27,11 @@ export interface SlackResponse { ts?: string; message_ts?: string; error?: string; + warning?: string; + response_metadata?: { + warnings?: string[]; + messages?: string[]; + }; message?: Record; } diff --git a/packages/types/src/__tests__/task-runs.test.ts b/packages/types/src/__tests__/task-runs.test.ts index edb2150a6..ba46c575b 100644 --- a/packages/types/src/__tests__/task-runs.test.ts +++ b/packages/types/src/__tests__/task-runs.test.ts @@ -1068,6 +1068,7 @@ describe('taskSpecSchema', () => { sourcePayload: { slackChannel: 'C123', slackTeamDomain: 'acme-team', + liveTaskStream: true, }, threadTs: '111.222', }); @@ -1077,6 +1078,7 @@ describe('taskSpecSchema', () => { slackChannel: 'C123', teamDomain: 'acme-team', thread_ts: '111.222', + liveTaskStream: true, }); expect(payload).not.toHaveProperty('slackTeamDomain'); }); diff --git a/packages/types/src/task-runs.ts b/packages/types/src/task-runs.ts index 1ca703208..9fe9e8fb9 100644 --- a/packages/types/src/task-runs.ts +++ b/packages/types/src/task-runs.ts @@ -1048,6 +1048,8 @@ const sharedTaskPayloadSchema = z.object({ communicationContextInherited: z.boolean().optional(), /** Runless Fast parent that owns this task's user-visible lifecycle. */ fastAgentParent: fastAgentParentSchema.optional(), + /** Native Slack task stream enabled for a Fast-mode delegation. */ + liveTaskStream: z.boolean().optional(), /** Provider event that caused this fresh launch; used for idempotent retries. */ communicationSourceEventId: z.string().optional(), /** @@ -2009,6 +2011,15 @@ export function populateSnapshotResumeSlackMetadata( if (conversationUrl) { payload.slackConversationUrl = conversationUrl; } + + if ( + options.sourcePayload && + typeof options.sourcePayload === 'object' && + (options.sourcePayload as { liveTaskStream?: unknown }).liveTaskStream === + true + ) { + payload.liveTaskStream = true; + } } export function populateSnapshotResumeCommunicationMetadata(