diff --git a/apps/api/src/handlers/discord/index.ts b/apps/api/src/handlers/discord/index.ts index fa0eecf9b..c2c831756 100644 --- a/apps/api/src/handlers/discord/index.ts +++ b/apps/api/src/handlers/discord/index.ts @@ -24,6 +24,7 @@ import { setLatestInboundMessageId, } from '@roomote/communication/messages'; import { reactionEmojiMatches } from '@roomote/communication/reaction-emoji'; +import { parseGoalCommand } from '@roomote/communication/goal-command'; import { getTaskUrl } from '@roomote/cloud-agents/server'; import { MANAGED_DEPLOYMENT_READ_ONLY_MESSAGE, @@ -53,6 +54,7 @@ import { findTaskBackedAutomationReportRun, releaseCommunicationOutOfBandClaim, resumeCommunicationTaskFromSnapshot, + activateCommunicationGoal, } from '@roomote/sdk/server/communication'; import { tryHandleDiscordRequestUserInputMessage } from './request-user-input.js'; import { retireDiscordPrReviewOffersBestEffort } from './pr-review-action.js'; @@ -154,6 +156,7 @@ const DISCORD_HELP_MESSAGE = [ '', '**Available commands**', '`/new request:` — start a fresh task.', + '`/goal request:` — keep working toward an objective across multiple turns.', '`/link code:` — link this Discord account in a DM with me.', '`/help` — show this message.', '', @@ -508,16 +511,22 @@ async function processDiscordGatewayEvent( return { ok: true, linked: true }; } - if (command && command.name !== 'new') { + if (command && command.name !== 'new' && command.name !== 'goal') { return { ok: true, ignored: 'unsupported_command' }; } - if (command?.name === 'new' && !command.request) { + if ( + (command?.name === 'new' || command?.name === 'goal') && + !command.request + ) { await replyToDiscordEvent({ provider: resolved.provider, applicationId: resolved.applicationId, channel, interaction: interactionReplyContext(event), - text: 'Add what you want Roomote to do in the `request` field.', + text: + command.name === 'goal' + ? 'Add the objective Roomote should complete in the `request` field.' + : 'Add what you want Roomote to do in the `request` field.', }); return { ok: true, started: false, reason: 'missing_request' }; } @@ -531,6 +540,10 @@ async function processDiscordGatewayEvent( : {}), }; const forceNewTask = command?.name === 'new'; + const goalCommand = + command?.name === 'goal' && command.request + ? parseGoalCommand(`/goal ${command.request}`) + : null; const repliedToAutomationReport = !forceNewTask && message?.message_reference?.message_id ? await findTaskBackedAutomationReportRun({ @@ -699,17 +712,27 @@ async function processDiscordGatewayEvent( for (const warning of processedAttachments.warnings) { apiLogger.warn(`[discord] Attachment warning: ${warning}`); } - const queuedMessage = discordEventToQueuedCommunicationMessage(event, { - botUserId: resolved.botUserId, - userId: senderUserId, - isTaskThread: isRoomoteThread, - parentChannelId: channel.parentChannelId, - attachmentImages: processedAttachments.images, - attachmentText: processedAttachments.attachmentTexts, - }); - if (!queuedMessage) { + const normalizedQueuedMessage = discordEventToQueuedCommunicationMessage( + event, + { + botUserId: resolved.botUserId, + userId: senderUserId, + isTaskThread: isRoomoteThread, + parentChannelId: channel.parentChannelId, + attachmentImages: processedAttachments.images, + attachmentText: processedAttachments.attachmentTexts, + }, + ); + if (!normalizedQueuedMessage) { return { ok: true, ignored: 'empty_task_entry' }; } + const queuedMessage = goalCommand?.goal + ? { + ...normalizedQueuedMessage, + text: goalCommand.objective, + goal: goalCommand.goal, + } + : normalizedQueuedMessage; if (pendingRoutingReply) { let routingReplyAckPinned = false; @@ -868,11 +891,24 @@ async function processDiscordGatewayEvent( message: messageForQueue, }); try { - const queued = await queueCommunicationMessageOnce( - 'discord', - activeRun.id, - messageWithOutOfBand, - ); + const queueMessage = ( + goalContext?: typeof messageWithOutOfBand.goalContext, + ) => + queueCommunicationMessageOnce('discord', activeRun.id, { + ...messageWithOutOfBand, + ...(goalContext ? { goalContext } : {}), + }); + const queued = goalCommand?.goal + ? await activateCommunicationGoal({ + taskId: activeRun.taskId, + goal: goalCommand.goal, + deliver: async (goalContext) => { + if (!(await queueMessage(goalContext))) { + throw new Error('Discord goal command was already queued'); + } + }, + }).then((result) => result.success) + : await queueMessage(); // A typed reply supersedes any pending PR review offers here. retireDiscordPrReviewOffersBestEffort({ channelId: metadata.communicationChannelId, diff --git a/apps/api/src/handlers/discord/routing-confirmation.ts b/apps/api/src/handlers/discord/routing-confirmation.ts index 316dc1fbf..a2ec1dd88 100644 --- a/apps/api/src/handlers/discord/routing-confirmation.ts +++ b/apps/api/src/handlers/discord/routing-confirmation.ts @@ -648,6 +648,7 @@ async function launchPendingDiscordRoute(input: { provider: input.provider, launchOwnerUserId: input.pending.launchOwnerUserId, queuedMessage: input.pending.queuedMessage, + goal: input.pending.queuedMessage.goal, metadata: input.pending.metadata, channel: input.pending.channel, workspace, diff --git a/apps/api/src/handlers/discord/task-launch.ts b/apps/api/src/handlers/discord/task-launch.ts index c6bf57f60..79ce9365e 100644 --- a/apps/api/src/handlers/discord/task-launch.ts +++ b/apps/api/src/handlers/discord/task-launch.ts @@ -2,6 +2,7 @@ import { ALL_REPOSITORIES, TaskPayloadKind, type QueuedCommunicationMessage, + type TaskGoalInput, type TaskInitiator, type TaskSpec, } from '@roomote/types'; @@ -361,6 +362,7 @@ export async function launchDiscordTask(input: { metadata: DiscordEventCommunicationMetadata; channel: DiscordChannelContext; workspace: DiscordWorkspaceSelection; + goal?: TaskGoalInput; /** `/new` in an existing task thread creates a sibling, never a second run in-place. */ forceNewThread?: boolean; /** @@ -493,6 +495,7 @@ export async function launchDiscordTask(input: { workflow: 'standard', surface: 'discord', trigger: 'message', + ...(input.goal ? { goal: input.goal } : {}), }, { // Automation initiators derive the 'automation' launch class; forcing diff --git a/apps/api/src/handlers/discord/task-orchestration.ts b/apps/api/src/handlers/discord/task-orchestration.ts index a0040155d..32499f82e 100644 --- a/apps/api/src/handlers/discord/task-orchestration.ts +++ b/apps/api/src/handlers/discord/task-orchestration.ts @@ -15,6 +15,7 @@ import { ALL_REPOSITORIES, type QueuedCommunicationMessage, type TaskInitiator, + type TaskGoalInput, } from '@roomote/types'; import type { DiscordEventCommunicationMetadata } from '@roomote/communication/discord-event'; @@ -83,6 +84,7 @@ export async function startNewDiscordTask(input: { /** Absent only for automation-owned channel auto-start launches. */ launchOwnerUserId?: string; queuedMessage: QueuedCommunicationMessage; + goal?: TaskGoalInput; metadata: DiscordEventCommunicationMetadata; channel: DiscordChannelContext; interaction?: { @@ -366,6 +368,7 @@ export async function startNewDiscordTask(input: { channel: input.channel, workspace, forceNewThread: input.forceNewThread, + goal: input.goal ?? input.queuedMessage.goal, ...(kickoffMessage ? { kickoffMessage } : {}), ...(input.intakeAckPinned ? { intakeAckPinned: true } : {}), }); diff --git a/apps/api/src/handlers/slack/events/active-run.ts b/apps/api/src/handlers/slack/events/active-run.ts index 81fcd16ff..0fac04252 100644 --- a/apps/api/src/handlers/slack/events/active-run.ts +++ b/apps/api/src/handlers/slack/events/active-run.ts @@ -31,6 +31,8 @@ import { stripLeadingSlackProductMention, } from '@roomote/cloud-agents'; import { setTrustedRunActingUserOnSuccess } from '@roomote/db/server'; +import { parseGoalCommand } from '@roomote/communication/goal-command'; +import { activateCommunicationGoal } from '@roomote/sdk/server/communication'; import { apiLogger } from '../../../logging.js'; import { retireSlackPrReviewOffersBestEffort } from '../pr-review-retire.js'; @@ -446,6 +448,15 @@ export async function processActiveRunMessage( getLatestSlackBotReply(event.channel, threadId), ]); const messageText = stripLeadingSlackProductMention(normalizedMessageText); + const goalCommand = parseGoalCommand(messageText); + if (goalCommand && !goalCommand.goal) { + await slack.postMessage({ + channel: event.channel, + thread_ts: threadId, + text: 'Send an objective after the command — for example, `/goal ship the release`.', + }); + return; + } const currentMessageFiles = resolveCurrentSlackMessageFiles({ currentMessageTs: deliveryTs, eventFiles: event.files, @@ -521,15 +532,40 @@ export async function processActiveRunMessage( runId: activeRun.id, senderUserId: userId, }); - await queueSlackMessage(activeRun.id, { - text: messageTextWithVideoDescriptions, + const queuedGoalText = goalCommand?.goal + ? goalCommand.objective + : messageTextWithVideoDescriptions; + const queuedMessage = { + text: queuedGoalText, user: event.user, userId, ts: event.ts, images: allImages.length > 0 ? allImages : undefined, - formattedPrompt, + formattedPrompt: goalCommand?.goal ? undefined : formattedPrompt, turnPolicy, - }); + }; + if (goalCommand?.goal && activeRun.taskId) { + const activated = await activateCommunicationGoal({ + taskId: activeRun.taskId, + goal: goalCommand.goal, + deliver: async (goalContext) => { + await queueSlackMessage(activeRun.id, { + ...queuedMessage, + goalContext, + }); + }, + }); + if (!activated.success) { + await slack.postMessage({ + channel: event.channel, + thread_ts: threadId, + text: 'Goal Mode could not be enabled because another activation is already in progress. Try again in a moment.', + }); + return; + } + } else { + await queueSlackMessage(activeRun.id, queuedMessage); + } await clearLatestUserMessage(activeRun.id); // A typed reply supersedes any pending PR review offers in the thread. retireSlackPrReviewOffersBestEffort({ diff --git a/apps/api/src/handlers/teams/index.ts b/apps/api/src/handlers/teams/index.ts index 2f07d8738..82a7acb57 100644 --- a/apps/api/src/handlers/teams/index.ts +++ b/apps/api/src/handlers/teams/index.ts @@ -18,6 +18,7 @@ import { teamsActivityToQueuedCommunicationMessage, } from '@roomote/communication/teams-activity'; import { queueCommunicationMessage } from '@roomote/communication/messages'; +import { parseGoalCommand } from '@roomote/communication/goal-command'; import { buildAccountLinkPromptText, buildAccountLinkThreadReplyText, @@ -26,6 +27,7 @@ import { } from '@roomote/communication/chat-messages'; import type { TeamsCommunicationProvider } from '@roomote/communication/teams-provider'; import { createTeamsCommunicationProviderFromRuntimeCredentials } from '@roomote/sdk/server'; +import { activateCommunicationGoal } from '@roomote/sdk/server/communication'; import { exchangeMicrosoftDelegatedGraphToken, extractTeamsGraphHostedContentIds, @@ -55,6 +57,7 @@ import { TaskPayloadKind, PRODUCT_NAME, type QueuedCommunicationMessage, + type TaskGoalInput, isDeploymentReadOnlyError, populateSnapshotResumeCommunicationMetadata, restoreSnapshotResumeVisiblePromptFields, @@ -1338,6 +1341,7 @@ async function startNewTeamsTask(input: { queuedMessage: QueuedTeamsCommunicationMessage; metadata: TeamsActivityCommunicationMetadata; workspaceOverride?: TeamsWorkspaceSelection; + goal?: TaskGoalInput; }) { const launchUserId = input.mappedUserId; const threadHistory = await fetchTeamsThreadMessagesBestEffort({ @@ -1428,6 +1432,7 @@ async function startNewTeamsTask(input: { workflow: 'standard', surface: 'teams', trigger: 'message', + ...(input.goal ? { goal: input.goal } : {}), }, { launchClass: 'human', @@ -1516,7 +1521,6 @@ async function resumePendingTeamsAuthToken( conversationId: metadata.communicationChannelId, threadId: metadata.communicationThreadId, }); - if (activeRun) { // Trusted pre-queue actor switch; see acting-user-sync.ts. await syncActingUserForInboundMessage({ @@ -1950,6 +1954,23 @@ teams.post('/', async (c) => { conversationId: metadata.communicationChannelId, threadId: metadata.communicationThreadId, }); + const goalCommand = parseGoalCommand(queuedMessage.text); + if (goalCommand && !goalCommand.goal) { + await postTeamsMessageBestEffort({ + conversationId: metadata.communicationChannelId, + threadId: metadata.communicationThreadId, + serviceUrl: metadata.communicationServiceUrl, + text: 'Send an objective after the command — for example, `/goal ship the release`.', + }); + return c.json({ ok: true, queued: false, repliedInline: true }); + } + if (goalCommand?.goal) { + queuedMessage = { + ...queuedMessage, + text: goalCommand.objective, + goal: goalCommand.goal, + }; + } if (!activeRun) { if (!isTeamsTaskEntryActivity(activity)) { @@ -2182,6 +2203,7 @@ teams.post('/', async (c) => { mappedUserId, queuedMessage, metadata, + goal: queuedMessage.goal, }); } catch (error) { if (isDeploymentReadOnlyError(error)) { @@ -2267,7 +2289,30 @@ teams.post('/', async (c) => { outOfBandClaim = attached.claim; } try { - await queueCommunicationMessage('teams', activeRun.id, activeFollowUp); + if (goalCommand?.goal) { + const activated = await activateCommunicationGoal({ + taskId: activeRun.taskId, + goal: goalCommand.goal, + deliver: async (goalContext) => { + await queueCommunicationMessage('teams', activeRun.id, { + ...activeFollowUp, + goalContext, + }); + }, + }); + if (!activated.success) { + await releaseCommunicationOutOfBandClaim(outOfBandClaim); + await postTeamsMessageBestEffort({ + conversationId: metadata.communicationChannelId, + threadId: metadata.communicationThreadId, + serviceUrl: metadata.communicationServiceUrl, + text: 'Goal Mode could not be enabled because another activation is already in progress. Try again in a moment.', + }); + return c.json({ ok: true, queued: false, repliedInline: true }); + } + } else { + await queueCommunicationMessage('teams', activeRun.id, activeFollowUp); + } } catch (error) { await releaseCommunicationOutOfBandClaim(outOfBandClaim); throw error; diff --git a/apps/api/src/handlers/telegram/__tests__/task-launch.test.ts b/apps/api/src/handlers/telegram/__tests__/task-launch.test.ts index 159271222..92e410317 100644 --- a/apps/api/src/handlers/telegram/__tests__/task-launch.test.ts +++ b/apps/api/src/handlers/telegram/__tests__/task-launch.test.ts @@ -57,6 +57,36 @@ describe('Telegram task topic launch', () => { rememberTelegramImplicitTopicMock.mockResolvedValue(undefined); }); + it('passes Goal Mode into a fresh launch', async () => { + await launchTelegramTask({ + launchOwnerUserId: 'user-1', + queuedMessage: { + provider: 'telegram', + text: 'ship the release', + user: 'Ada', + userId: 'user-1', + ts: '1', + channel: '10', + }, + metadata: { + communicationProvider: 'telegram', + communicationChannelId: '10', + }, + workspace: { + repoForPayload: 'acme/repo', + workspaceDisplayName: 'repo', + }, + goal: { objective: 'ship the release', maxContinuations: 5 }, + }); + + expect(enqueueTaskMock).toHaveBeenCalledWith( + expect.objectContaining({ + goal: { objective: 'ship the release', maxContinuations: 5 }, + }), + expect.anything(), + ); + }); + it('uses a newly created topic as the task conversation', async () => { createTelegramForumTopicBestEffortMock.mockResolvedValue({ threadId: '77', diff --git a/apps/api/src/handlers/telegram/index.ts b/apps/api/src/handlers/telegram/index.ts index 4ac032da0..87265408c 100644 --- a/apps/api/src/handlers/telegram/index.ts +++ b/apps/api/src/handlers/telegram/index.ts @@ -18,6 +18,7 @@ import { getTelegramUpdateMessage, getTelegramUpdateMessageReaction, getTelegramNewTaskCommand, + getTelegramGoalCommand, isTelegramImplicitTopicCreatedMessage, isTelegramPrivateChat, isTelegramStartCommand, @@ -26,6 +27,8 @@ import { parseTelegramUpdate, telegramUpdateToQueuedCommunicationMessage, } from '@roomote/communication/telegram-update'; +import { GOAL_COMMAND_USAGE } from '@roomote/communication/goal-command'; +import { activateCommunicationGoal } from '@roomote/sdk/server/communication'; import { apiLogger } from '../../logging.js'; import { syncActingUserForInboundMessage } from '../tasks/acting-user-sync.js'; @@ -60,6 +63,7 @@ const TELEGRAM_COMMAND_HELP = [ '*Available commands*', '`/start` — show this welcome message.', '`/new ` — start a fresh task instead of resuming the previous one; when topics are available, it opens a new topic.', + '`/goal ` — keep working toward an objective across multiple turns.', ].join('\n'); const TELEGRAM_WELCOME_MESSAGE = [ @@ -400,6 +404,26 @@ telegram.post('/', async (c) => { const newTaskCommand = getTelegramNewTaskCommand(update, { botUsername: botUsername ?? undefined, }); + const goalCommand = getTelegramGoalCommand(update, { + botUsername: botUsername ?? undefined, + }); + if (goalCommand && !goalCommand.goal) { + await postTelegramMessageBestEffort({ + chatId: metadata.communicationChannelId, + threadId: metadata.communicationThreadId, + replyToMessageId: metadata.communicationMessageId, + text: `Send an objective after the command — for example, \`${GOAL_COMMAND_USAGE.replace('', 'ship the release')}\`.`, + textFormat: 'markdown', + }); + return c.json({ ok: true, queued: false, repliedInline: true }); + } + if (goalCommand?.goal && queuedMessage) { + queuedMessage = { + ...queuedMessage, + text: goalCommand.objective, + goal: goalCommand.goal, + }; + } if (!queuedMessage && !newTaskCommand) { return c.json({ ok: true, ignored: 'unsupported_update' }); @@ -512,7 +536,29 @@ telegram.post('/', async (c) => { runId: activeRun.id, senderUserId: queuedMessage.userId, }); - await queueCommunicationMessage('telegram', activeRun.id, queuedMessage); + if (goalCommand?.goal) { + const activated = await activateCommunicationGoal({ + taskId: activeRun.taskId, + goal: goalCommand.goal, + deliver: async (goalContext) => { + await queueCommunicationMessage('telegram', activeRun.id, { + ...queuedMessage, + goalContext, + }); + }, + }); + if (!activated.success) { + await postTelegramMessageBestEffort({ + chatId: metadata.communicationChannelId, + threadId: metadata.communicationThreadId, + replyToMessageId: metadata.communicationMessageId, + text: 'Goal Mode could not be enabled because another activation is already in progress. Try again in a moment.', + }); + return c.json({ ok: true, queued: false, repliedInline: true }); + } + } else { + await queueCommunicationMessage('telegram', activeRun.id, queuedMessage); + } // A typed reply supersedes any pending PR review offers in the chat. retireTelegramPrReviewOffersBestEffort({ chatId: conversation.chatId, diff --git a/apps/api/src/handlers/telegram/routing-confirmation.ts b/apps/api/src/handlers/telegram/routing-confirmation.ts index d9f53aad3..08934b601 100644 --- a/apps/api/src/handlers/telegram/routing-confirmation.ts +++ b/apps/api/src/handlers/telegram/routing-confirmation.ts @@ -518,6 +518,7 @@ async function launchPendingTelegramRoute(input: { await launchTelegramTask({ launchOwnerUserId: input.pending.launchOwnerUserId, queuedMessage: input.pending.queuedMessage, + goal: input.pending.queuedMessage.goal, metadata: input.pending.metadata, workspace, createTopicForTask: input.pending.createTopicForTask, diff --git a/apps/api/src/handlers/telegram/task-launch.ts b/apps/api/src/handlers/telegram/task-launch.ts index 4fd33e575..bac83603f 100644 --- a/apps/api/src/handlers/telegram/task-launch.ts +++ b/apps/api/src/handlers/telegram/task-launch.ts @@ -3,6 +3,7 @@ import { ALL_REPOSITORIES, buildTelegramMessagePermalink, TaskPayloadKind, + type TaskGoalInput, type TaskSpec, } from '@roomote/types'; import { db, environments, eq } from '@roomote/db/server'; @@ -90,6 +91,7 @@ export async function launchTelegramTask(input: { metadata: TelegramUpdateCommunicationMetadata; workspace: TelegramWorkspaceSelection; createTopicForTask?: boolean; + goal?: TaskGoalInput; }) { const topicName = buildTelegramTaskTopicName(input.queuedMessage.text); const createdTopic = input.createTopicForTask @@ -140,6 +142,7 @@ export async function launchTelegramTask(input: { workflow: 'standard', surface: 'telegram', trigger: 'message', + ...(input.goal ? { goal: input.goal } : {}), }, { launchClass: 'human', diff --git a/apps/api/src/handlers/telegram/task-orchestration.ts b/apps/api/src/handlers/telegram/task-orchestration.ts index a78a286aa..489b71366 100644 --- a/apps/api/src/handlers/telegram/task-orchestration.ts +++ b/apps/api/src/handlers/telegram/task-orchestration.ts @@ -3,7 +3,7 @@ import type { TelegramUpdateCommunicationMetadata, } from '@roomote/communication/telegram-update'; import { Env } from '@roomote/env'; -import { ALL_REPOSITORIES } from '@roomote/types'; +import { ALL_REPOSITORIES, type TaskGoalInput } from '@roomote/types'; import { buildTelegramRoutingContext, getTaskUrl, @@ -81,6 +81,7 @@ export async function startNewTelegramTask(input: { /** Start a new task topic even when the command came from an older topic. */ forceNewTopic?: boolean; workspaceOverride?: TelegramWorkspaceSelection; + goal?: TaskGoalInput; }) { const needsPrivateTopicCapability = input.message.chat.type.toLowerCase() === 'private' && @@ -173,6 +174,7 @@ export async function startNewTelegramTask(input: { metadata: input.metadata, workspace, createTopicForTask, + goal: input.goal ?? input.queuedMessage.goal, }); return { diff --git a/apps/worker/src/run-task/polling/communication.ts b/apps/worker/src/run-task/polling/communication.ts index 717e133c1..4162cd7e2 100644 --- a/apps/worker/src/run-task/polling/communication.ts +++ b/apps/worker/src/run-task/polling/communication.ts @@ -254,6 +254,7 @@ export function createCommunicationMessageInterval({ // The delivered sender always equals the server-side acting user. userId: msgPrep.effectiveUserId ?? undefined, clientMessageId: getCommunicationClientMessageId(provider, message), + goalContext: message.goalContext, }); if (!sent) { diff --git a/apps/worker/src/run-task/polling/slack.ts b/apps/worker/src/run-task/polling/slack.ts index 3e30e6cd5..f289230c0 100644 --- a/apps/worker/src/run-task/polling/slack.ts +++ b/apps/worker/src/run-task/polling/slack.ts @@ -251,6 +251,7 @@ export function createSlackMessageInterval({ // The delivered sender always equals the server-side acting user. userId: msgPrep.effectiveUserId ?? undefined, clientMessageId: getSlackClientMessageId(msg), + goalContext: msg.goalContext, }); logger.log( diff --git a/apps/worker/src/run-task/run-task.ts b/apps/worker/src/run-task/run-task.ts index 5fcefa1e9..efd419e54 100644 --- a/apps/worker/src/run-task/run-task.ts +++ b/apps/worker/src/run-task/run-task.ts @@ -14,6 +14,7 @@ import { RunStatus, TaskPayloadKind, type QueuedCommunicationMessage, + type TaskGoal, getSlackChannelFromTaskPayload, getSlackThreadTsFromTaskPayload, isCommunicationProvider, @@ -1596,6 +1597,7 @@ export const runTask = async ({ workflowPhase?: string; source?: string; clientMessageId?: string; + goalContext?: TaskGoal; userId?: string; }) => { if (deferredResumePromptRetryTimer) { diff --git a/apps/worker/src/run-task/types.ts b/apps/worker/src/run-task/types.ts index 950c608c9..d433a47e7 100644 --- a/apps/worker/src/run-task/types.ts +++ b/apps/worker/src/run-task/types.ts @@ -6,6 +6,7 @@ import type { CommunicationProvider, EnvironmentConfig, RequestedWorkKind, + TaskGoal, } from '@roomote/types'; import type { TaskRun, DequeuedTaskRun } from '@roomote/sdk/client'; @@ -293,6 +294,7 @@ export interface ListenerOptions { userName?: string; userImageUrl?: string; clientMessageId?: string; + goalContext?: TaskGoal; }) => boolean; slackReplySatisfactionStateFile?: string; answerUserInputRequest: (options: { diff --git a/packages/communication/package.json b/packages/communication/package.json index f61e762f8..a23c2a7b9 100644 --- a/packages/communication/package.json +++ b/packages/communication/package.json @@ -9,6 +9,7 @@ "./discord-event": "./src/discord-event.ts", "./discord-provider": "./src/discord-provider.ts", "./discord-request-user-input": "./src/discord-request-user-input.ts", + "./goal-command": "./src/goal-command.ts", "./messages": "./src/messages.ts", "./mock-discord-server": "./src/mock-discord-server.ts", "./provider": "./src/provider.ts", diff --git a/packages/communication/src/__tests__/goal-command.test.ts b/packages/communication/src/__tests__/goal-command.test.ts new file mode 100644 index 000000000..f7b152b19 --- /dev/null +++ b/packages/communication/src/__tests__/goal-command.test.ts @@ -0,0 +1,21 @@ +import { describe, expect, it } from 'vitest'; + +import { parseGoalCommand } from '../goal-command'; + +describe('parseGoalCommand', () => { + it('parses the canonical command case-insensitively', () => { + expect(parseGoalCommand('/GOAL ship the release ')).toEqual({ + objective: 'ship the release', + goal: { objective: 'ship the release', maxContinuations: 5 }, + }); + }); + + it('returns an empty command result so providers can show usage', () => { + expect(parseGoalCommand('/goal')).toEqual({ objective: '', goal: null }); + }); + + it('does not intercept ordinary messages', () => { + expect(parseGoalCommand('please /goal ship the release')).toBeNull(); + expect(parseGoalCommand('/goalkeeper notes')).toBeNull(); + }); +}); diff --git a/packages/communication/src/__tests__/telegram-provider.test.ts b/packages/communication/src/__tests__/telegram-provider.test.ts index 832b13744..7745c3ad2 100644 --- a/packages/communication/src/__tests__/telegram-provider.test.ts +++ b/packages/communication/src/__tests__/telegram-provider.test.ts @@ -29,6 +29,10 @@ describe('TelegramCommunicationProvider', () => { commands: [ { command: 'start', description: 'Show welcome and command help' }, { command: 'new', description: 'Start a fresh task' }, + { + command: 'goal', + description: 'Keep working toward an objective across turns', + }, ], }); }); diff --git a/packages/communication/src/__tests__/telegram-update.test.ts b/packages/communication/src/__tests__/telegram-update.test.ts index 4bb6bf0ba..2ac1c9b90 100644 --- a/packages/communication/src/__tests__/telegram-update.test.ts +++ b/packages/communication/src/__tests__/telegram-update.test.ts @@ -2,6 +2,7 @@ import { describe, expect, it } from 'vitest'; import { getTelegramNewTaskCommand, + getTelegramGoalCommand, getTelegramUpdateCallbackQuery, getTelegramUpdateCommunicationMetadata, getTelegramUpdateMessageReaction, @@ -700,4 +701,26 @@ describe('Telegram update helpers', () => { ).toBeNull(); }); }); + + describe('getTelegramGoalCommand', () => { + it('parses a private-chat goal command', () => { + const parsed = parseTelegramUpdate({ + update_id: 700, + message: { + message_id: 701, + date: 1, + chat: { id: 1, type: 'private' }, + text: '/goal ship the release', + entities: [{ type: 'bot_command', offset: 0, length: 5 }], + }, + }); + expect(parsed.success).toBe(true); + if (!parsed.success) return; + + expect(getTelegramGoalCommand(parsed.data)).toEqual({ + objective: 'ship the release', + goal: { objective: 'ship the release', maxContinuations: 5 }, + }); + }); + }); }); diff --git a/packages/communication/src/discord-event.ts b/packages/communication/src/discord-event.ts index efd363093..cbaf3506f 100644 --- a/packages/communication/src/discord-event.ts +++ b/packages/communication/src/discord-event.ts @@ -573,7 +573,8 @@ export function isDiscordTaskEntryEvent( isDiscordBotMentioned(message, options.botUserId)) ); } - return getDiscordInteractionCommand(event)?.name === 'new'; + const commandName = getDiscordInteractionCommand(event)?.name; + return commandName === 'new' || commandName === 'goal'; } function formatDiscordUser(input: { @@ -637,7 +638,8 @@ export function discordEventToQueuedCommunicationMessage( const interaction = getDiscordInteractionCreate(event); const command = getDiscordInteractionCommand(event); - if (!interaction || command?.name !== 'new') return null; + if (!interaction || (command?.name !== 'new' && command?.name !== 'goal')) + return null; const user = getDiscordInteractionUser(interaction); return { provider: 'discord', diff --git a/packages/communication/src/discord-provider.ts b/packages/communication/src/discord-provider.ts index ed60dddfa..28d0ed1aa 100644 --- a/packages/communication/src/discord-provider.ts +++ b/packages/communication/src/discord-provider.ts @@ -1204,6 +1204,19 @@ export class DiscordCommunicationProvider implements CommunicationProviderAdapte }, ], }, + { + name: 'goal', + description: 'Keep working toward an objective across multiple turns', + type: 1, + options: [ + { + type: 3, + name: 'request', + description: 'What objective should Roomote complete?', + required: true, + }, + ], + }, { name: 'link', description: 'Link this Discord account to Roomote', diff --git a/packages/communication/src/goal-command.ts b/packages/communication/src/goal-command.ts new file mode 100644 index 000000000..3f9e105b4 --- /dev/null +++ b/packages/communication/src/goal-command.ts @@ -0,0 +1,30 @@ +import { + DEFAULT_TASK_GOAL_MAX_CONTINUATIONS, + type TaskGoalInput, +} from '@roomote/types'; + +export const GOAL_COMMAND_NAME = 'goal'; +export const GOAL_COMMAND_USAGE = '/goal '; + +export type ParsedGoalCommand = { + objective: string; + goal: TaskGoalInput | null; +}; + +export function parseGoalCommand(text: string): ParsedGoalCommand | null { + const match = /^\/goal(?:\s+([\s\S]*))?$/iu.exec(text.trim()); + if (!match) { + return null; + } + + const objective = (match[1] ?? '').trim(); + return { + objective, + goal: objective + ? { + objective, + maxContinuations: DEFAULT_TASK_GOAL_MAX_CONTINUATIONS, + } + : null, + }; +} diff --git a/packages/communication/src/index.ts b/packages/communication/src/index.ts index 41fddcbad..ac1a89a61 100644 --- a/packages/communication/src/index.ts +++ b/packages/communication/src/index.ts @@ -2,6 +2,7 @@ export * from './chat-messages'; export * from './discord-event'; export * from './discord-provider'; export * from './discord-request-user-input'; +export * from './goal-command'; export * from './messages'; export * from './provider'; export * from './reaction-emoji'; diff --git a/packages/communication/src/telegram-provider.ts b/packages/communication/src/telegram-provider.ts index c9551e21e..2f472ce15 100644 --- a/packages/communication/src/telegram-provider.ts +++ b/packages/communication/src/telegram-provider.ts @@ -527,6 +527,10 @@ export class TelegramCommunicationProvider implements CommunicationProviderAdapt commands: [ { command: 'start', description: 'Show welcome and command help' }, { command: 'new', description: 'Start a fresh task' }, + { + command: 'goal', + description: 'Keep working toward an objective across turns', + }, ], }); } diff --git a/packages/communication/src/telegram-update.ts b/packages/communication/src/telegram-update.ts index 19f2e0965..1f4c2763b 100644 --- a/packages/communication/src/telegram-update.ts +++ b/packages/communication/src/telegram-update.ts @@ -1,6 +1,7 @@ import { z } from 'zod'; import type { QueuedCommunicationMessage } from '@roomote/types'; +import { parseGoalCommand, type ParsedGoalCommand } from './goal-command'; const telegramUserSchema = z .object({ @@ -610,6 +611,58 @@ export function getTelegramNewTaskCommand( return null; } +export function getTelegramGoalCommand( + update: TelegramUpdate, + options: TelegramBotMentionOptions = {}, +): ParsedGoalCommand | null { + const message = getTelegramUpdateMessage(update); + const text = message?.text; + + if (!text) { + return null; + } + + const botUsername = normalizeTelegramBotUsername(options.botUsername); + const entities = message.entities ?? []; + for (const entity of entities) { + if (entity.type !== 'bot_command') { + continue; + } + + const parsed = parseTelegramBotCommand(readEntityText(text, entity)); + if (!parsed || parsed.command !== 'goal') { + continue; + } + if (botUsername && parsed.botSuffix && parsed.botSuffix !== botUsername) { + continue; + } + if (!isLeadingBotCommand(text, entities, entity, options)) { + continue; + } + + const mentionPrefix = findLeadingBotMentionBefore( + text, + entities, + entity.offset, + options, + ); + if ( + botUsername && + !isTelegramPrivateChat(message) && + !parsed.botSuffix && + !mentionPrefix + ) { + continue; + } + + return parseGoalCommand( + `/goal ${normalizeWhitespace(text.slice(entity.offset + entity.length))}`, + ); + } + + return null; +} + export function stripTelegramBotInvocation( text: string, message: TelegramMessage, diff --git a/packages/sdk/src/server/lib/communication/__tests__/communication-goal.test.ts b/packages/sdk/src/server/lib/communication/__tests__/communication-goal.test.ts new file mode 100644 index 000000000..07377408c --- /dev/null +++ b/packages/sdk/src/server/lib/communication/__tests__/communication-goal.test.ts @@ -0,0 +1,63 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; + +const { prepareTaskGoalActivation, commit, rollback } = vi.hoisted(() => ({ + prepareTaskGoalActivation: vi.fn(), + commit: vi.fn(), + rollback: vi.fn(), +})); + +vi.mock('@roomote/db/server', () => ({ prepareTaskGoalActivation })); + +import { activateCommunicationGoal } from '../communication-goal'; + +describe('activateCommunicationGoal', () => { + beforeEach(() => { + vi.clearAllMocks(); + rollback.mockResolvedValue(true); + commit.mockResolvedValue({ + objective: 'ship the release', + generation: 'goal-generation:1', + status: 'active', + maxContinuations: 5, + continuationsUsed: 0, + blockedReason: null, + completedAt: null, + }); + prepareTaskGoalActivation.mockResolvedValue({ + generation: 'goal-generation:1', + commit, + rollback, + }); + }); + + it('delivers trusted context before committing activation', async () => { + const deliver = vi.fn().mockResolvedValue(undefined); + await expect( + activateCommunicationGoal({ + taskId: 'task-1', + goal: { objective: 'ship the release', maxContinuations: 5 }, + deliver, + }), + ).resolves.toMatchObject({ success: true }); + + expect(deliver).toHaveBeenCalledWith( + expect.objectContaining({ generation: 'goal-generation:1' }), + ); + expect(deliver.mock.invocationCallOrder[0]).toBeLessThan( + commit.mock.invocationCallOrder[0]!, + ); + }); + + it('rolls back when durable delivery fails', async () => { + const error = new Error('queue unavailable'); + await expect( + activateCommunicationGoal({ + taskId: 'task-1', + goal: { objective: 'ship the release', maxContinuations: 5 }, + deliver: vi.fn().mockRejectedValue(error), + }), + ).rejects.toBe(error); + expect(commit).not.toHaveBeenCalled(); + expect(rollback).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/sdk/src/server/lib/communication/communication-goal.ts b/packages/sdk/src/server/lib/communication/communication-goal.ts new file mode 100644 index 000000000..944ecdcd8 --- /dev/null +++ b/packages/sdk/src/server/lib/communication/communication-goal.ts @@ -0,0 +1,47 @@ +import { prepareTaskGoalActivation } from '@roomote/db/server'; +import type { TaskGoal, TaskGoalInput } from '@roomote/types'; + +export type ActivateCommunicationGoalResult = + | { success: true; goal: TaskGoal } + | { + success: false; + reason: 'activation_pending' | 'activation_superseded'; + }; + +export async function activateCommunicationGoal(input: { + taskId: string; + goal: TaskGoalInput; + deliver: (goalContext: TaskGoal) => Promise; +}): Promise { + const activation = await prepareTaskGoalActivation({ + taskId: input.taskId, + goal: input.goal, + }); + if (!activation) { + return { success: false, reason: 'activation_pending' }; + } + + const goalContext: TaskGoal = { + ...input.goal, + generation: activation.generation, + status: 'active', + continuationsUsed: 0, + blockedReason: null, + completedAt: null, + }; + + try { + await input.deliver(goalContext); + } catch (error) { + await activation.rollback().catch(() => false); + throw error; + } + + const goal = await activation.commit(); + if (!goal) { + await activation.rollback(); + return { success: false, reason: 'activation_superseded' }; + } + + return { success: true, goal }; +} diff --git a/packages/sdk/src/server/lib/communication/index.ts b/packages/sdk/src/server/lib/communication/index.ts index acc7db901..b72c8af9d 100644 --- a/packages/sdk/src/server/lib/communication/index.ts +++ b/packages/sdk/src/server/lib/communication/index.ts @@ -1,3 +1,4 @@ export * from './communication-snapshot-resume'; +export * from './communication-goal'; export * from './communication-task-run-lookup'; export * from './communication-out-of-band-context'; diff --git a/packages/slack/src/block-kit.ts b/packages/slack/src/block-kit.ts index 2504974f1..6109e395b 100644 --- a/packages/slack/src/block-kit.ts +++ b/packages/slack/src/block-kit.ts @@ -14,7 +14,9 @@ import { isDeploymentReadOnlyError, MANAGED_DEPLOYMENT_READ_ONLY_MESSAGE, getTaskModelDisplayName, + type TaskGoalInput, } from '@roomote/types'; +import { parseGoalCommand } from '@roomote/communication/goal-command'; import { Env } from '@roomote/env'; import { getRedis, REDIS_KEYS } from '@roomote/redis'; import { @@ -375,6 +377,7 @@ interface RoutingPrefillData { routingDurationMs?: number; userRoute?: string; warningText?: string; + goal?: TaskGoalInput; } interface RoutingConfirmActionValue { @@ -1203,6 +1206,11 @@ export async function showTaskConfiguration({ }, }); + const parsedGoal = parseGoalCommand( + stripLeadingSlackProductMention( + stripLeadingRawSlackMention(event.text), + ), + ); const prefillData: RoutingPrefillData = { agentName, workspaceOnly, @@ -1228,6 +1236,7 @@ export async function showTaskConfiguration({ routingDebug: routingResult.debug, routingDurationMs: suggestedRoutingDurationMs, warningText, + ...(parsedGoal?.goal ? { goal: parsedGoal.goal } : {}), }; await getRedis().set( @@ -1966,10 +1975,14 @@ async function startImmediateSlackTask({ const messageText = stripLeadingSlackProductMention( await slack.normalizeIncomingText(stripLeadingRawSlackMention(event.text)), ); + const parsedGoal = parseGoalCommand(messageText); + const taskDescription = parsedGoal?.goal + ? parsedGoal.goal.objective + : messageText; const images = event.processedImages || []; const taskText = appendSlackVideoDescriptionsToText({ text: appendAttachmentTextsToPromptText({ - text: messageText, + text: taskDescription, attachmentTexts: event.processedAttachmentTexts, }), videoDescriptions: event.processedVideoDescriptions, @@ -2002,6 +2015,7 @@ async function startImmediateSlackTask({ channel: event.channel, messageTs: threadId, })) ?? undefined, + goal: parsedGoal?.goal ?? undefined, ...(replaceMessageTs ? { queuedStartedMessage: { @@ -2141,10 +2155,16 @@ async function createRunFromPrefill({ stripLeadingRawSlackMention(originalEvent.text), ), ); + const parsedGoal = prefill.goal + ? { goal: prefill.goal } + : parseGoalCommand(messageText); + const taskDescription = parsedGoal?.goal + ? parsedGoal.goal.objective + : messageText; const images = originalEvent.processedImages || []; const taskText = appendSlackVideoDescriptionsToText({ text: appendAttachmentTextsToPromptText({ - text: messageText, + text: taskDescription, attachmentTexts: originalEvent.processedAttachmentTexts, }), videoDescriptions: originalEvent.processedVideoDescriptions, @@ -2177,6 +2197,7 @@ async function createRunFromPrefill({ channel: originalEvent.channel, messageTs: threadId, })) ?? undefined, + goal: parsedGoal?.goal ?? undefined, ...(prefill.confirmMessageTs ? { queuedStartedMessage: { diff --git a/packages/slack/src/start-auto-routed-slack-task.ts b/packages/slack/src/start-auto-routed-slack-task.ts index 2c8cd8f6c..ee6461abd 100644 --- a/packages/slack/src/start-auto-routed-slack-task.ts +++ b/packages/slack/src/start-auto-routed-slack-task.ts @@ -19,6 +19,7 @@ import { stripLeadingRawSlackMention, stripLeadingSlackProductMention, } from '@roomote/cloud-agents'; +import { parseGoalCommand } from '@roomote/communication/goal-command'; import { buildSlackRoutingContext, detectSlackMcpSetupRequirement, @@ -251,9 +252,13 @@ export async function startAutoRoutedSlackTask({ } } - const taskDescription = stripLeadingSlackProductMention( + const taskDescriptionWithCommand = stripLeadingSlackProductMention( await slack.normalizeIncomingText(stripLeadingRawSlackMention(prompt)), ); + const goalCommand = parseGoalCommand(taskDescriptionWithCommand); + const taskDescription = goalCommand?.goal + ? goalCommand.objective + : taskDescriptionWithCommand; const warningText = buildStatuspageSlackWarning( await getStatuspageIncident(), ); @@ -517,6 +522,7 @@ export async function startAutoRoutedSlackTask({ webPath, slackConversationUrl: slackConversationUrl ?? undefined, skipInitialActingUser: !initiatorLinkedUserId, + goal: goalCommand?.goal ?? undefined, ...(existingMessageTs ? { queuedStartedMessage: { diff --git a/packages/slack/src/start-slack-app-mention.ts b/packages/slack/src/start-slack-app-mention.ts index e53d136c2..227342c62 100644 --- a/packages/slack/src/start-slack-app-mention.ts +++ b/packages/slack/src/start-slack-app-mention.ts @@ -11,6 +11,7 @@ import { type ReasoningEffort, type SlackAppMentionTask, type TaskInitiator, + type TaskGoalInput, type TaskTrigger, type TaskVisibility, type TaskWorkflow, @@ -155,6 +156,7 @@ export async function startSlackAppMentionTask(input: { webPath?: string; slackConversationUrl?: string; skipInitialActingUser?: boolean; + goal?: TaskGoalInput; /** * Started-message metadata callers persist themselves via * setSlackStartedMessageTs after the launch. Accepted here so call sites @@ -326,6 +328,7 @@ export async function startSlackAppMentionTask(input: { slackChannelId: input.channel, slackThreadTs: input.threadTs, }, + ...(input.goal ? { goal: input.goal } : {}), }, input.skipInitialActingUser ? { skipInitialActingUser: true } : {}, ); diff --git a/packages/types/src/communication.ts b/packages/types/src/communication.ts index 1054c067a..320331374 100644 --- a/packages/types/src/communication.ts +++ b/packages/types/src/communication.ts @@ -1,5 +1,7 @@ import { z } from 'zod'; +import { taskGoalInputSchema, taskGoalSchema } from './task-goal'; + export const communicationProviders = [ 'slack', 'teams', @@ -85,6 +87,8 @@ export const queuedCommunicationMessageSchema = z.object({ threadTs: z.string().optional(), images: z.array(z.string()).optional(), formattedPrompt: z.string().optional(), + goalContext: taskGoalSchema.optional(), + goal: taskGoalInputSchema.optional(), turnPolicy: z .object({ reactionsAllowed: z.boolean().optional(), diff --git a/packages/types/src/task-goal.ts b/packages/types/src/task-goal.ts new file mode 100644 index 000000000..61d6fda0b --- /dev/null +++ b/packages/types/src/task-goal.ts @@ -0,0 +1,34 @@ +import { z } from 'zod'; + +export const TASK_GOAL_STATUSES = [ + 'active', + 'complete', + 'blocked', + 'budget_limited', +] as const; + +export type TaskGoalStatus = (typeof TASK_GOAL_STATUSES)[number]; + +export const DEFAULT_TASK_GOAL_MAX_CONTINUATIONS = 5; + +export const taskGoalInputSchema = z.object({ + objective: z.string().trim().min(1).max(10_000), + maxContinuations: z + .number() + .int() + .min(1) + .max(20) + .default(DEFAULT_TASK_GOAL_MAX_CONTINUATIONS), +}); + +export type TaskGoalInput = z.infer; + +export const taskGoalSchema = taskGoalInputSchema.extend({ + generation: z.string().nullable(), + status: z.enum(TASK_GOAL_STATUSES), + continuationsUsed: z.number().int().min(0), + blockedReason: z.string().nullable(), + completedAt: z.coerce.date().nullable(), +}); + +export type TaskGoal = z.infer; diff --git a/packages/types/src/task-runs.ts b/packages/types/src/task-runs.ts index 516591b0b..cc792cfa1 100644 --- a/packages/types/src/task-runs.ts +++ b/packages/types/src/task-runs.ts @@ -16,6 +16,15 @@ import { prActions } from './cloud-agents'; import { ALL_REPOSITORIES } from './constants'; import { sourceControlProviderSchema } from './source-control'; import { resolveTaskModelIdAlias } from './task-models'; +export { + DEFAULT_TASK_GOAL_MAX_CONTINUATIONS, + TASK_GOAL_STATUSES, + taskGoalInputSchema, + taskGoalSchema, + type TaskGoal, + type TaskGoalInput, + type TaskGoalStatus, +} from './task-goal'; /** * Task classification vocabulary. @@ -114,37 +123,6 @@ export const TASK_STATES = [ export type TaskState = (typeof TASK_STATES)[number]; -export const TASK_GOAL_STATUSES = [ - 'active', - 'complete', - 'blocked', - 'budget_limited', -] as const; - -export type TaskGoalStatus = (typeof TASK_GOAL_STATUSES)[number]; - -export const DEFAULT_TASK_GOAL_MAX_CONTINUATIONS = 5; - -export const taskGoalInputSchema = z.object({ - objective: z.string().trim().min(1).max(10_000), - maxContinuations: z - .number() - .int() - .min(1) - .max(20) - .default(DEFAULT_TASK_GOAL_MAX_CONTINUATIONS), -}); - -export type TaskGoalInput = z.infer; - -export type TaskGoal = TaskGoalInput & { - generation: string | null; - status: TaskGoalStatus; - continuationsUsed: number; - blockedReason: string | null; - completedAt: Date | null; -}; - export const RUN_KINDS = ['fresh', 'resume'] as const; export type RunKind = (typeof RUN_KINDS)[number];