diff --git a/cloud/apps/connector/src/connector-server.ts b/cloud/apps/connector/src/connector-server.ts index 6166e37dbd7..c68b43cc59e 100644 --- a/cloud/apps/connector/src/connector-server.ts +++ b/cloud/apps/connector/src/connector-server.ts @@ -17,7 +17,7 @@ import { queueDelivery } from './queue-delivery.js' import { limitByClientIp, RateLimiter } from './rate-limiter.js' import { RequestDrain } from './request-drain.js' import { createSlackEdge } from './slack-edge.js' -import type { QueueInTransaction } from './slack-routing.js' +import type { QueueInTransaction } from './queue-delivery.js' const UNAUTHENTICATED_PER_MINUTE_PER_IP = 30 const AUTHENTICATED_PER_MINUTE_PER_IP = 6_000 diff --git a/cloud/apps/connector/src/owner-controls.ts b/cloud/apps/connector/src/owner-controls.ts new file mode 100644 index 00000000000..596a392370a --- /dev/null +++ b/cloud/apps/connector/src/owner-controls.ts @@ -0,0 +1,143 @@ +import { createHash } from 'node:crypto' +import type { CloudDatabase, SqlRow } from '@orca-cloud/cloud-database' +import { + ConnectorDeliveryResultSchema, + ConnectorPendingPromptSchema, + type ConnectorHelloRequest, + type ConnectorPendingPrompt +} from '@orca-cloud/connector-contract' +import { lockConnectorHost } from './connector-hosts.js' +import type { QueueInTransaction } from './queue-delivery.js' + +// What an owner can do to their conversation from any connected app: answer the agent's prompt +// (decisions §1: only the owner). It goes to the machine through the inbox. +export type OwnerAction = { kind: 'answer'; promptId: string; optionId: string } + +export type OwnerActionOutcome = + | { kind: 'not_owner'; ownerUserId: string } + // The agent no longer shows that prompt, or it was already answered. + | { kind: 'moved_on' } + | { kind: 'answered'; ownerUserId: string; label: string } + +// Where the owner's answer to the current prompt stands, read from its inbox row. +export type OwnerActionState = + | { kind: 'on_its_way' } + | { kind: 'refused'; reason: string } + | { kind: 'unknown' } + | { kind: 'unreached' } + +export type AnswerablePrompt = Extract + +// One answer per prompt, so repeated presses are inbox duplicates. Hashed, so any contract-valid +// conversation and prompt ids still make a delivery id within its 128 limit. +export const answerDeliveryId = (externalId: string, promptId: string) => + `answer:${createHash('sha256').update(`${externalId}\0${promptId}`).digest('hex')}` + +export function pendingPrompt(thread: SqlRow): ConnectorPendingPrompt | undefined { + return typeof thread.pending_prompt === 'string' + ? ConnectorPendingPromptSchema.safeParse(JSON.parse(thread.pending_prompt)).data + : undefined +} + +// The thread's pending prompt when its machine takes answers of that kind. +export function answerablePrompt( + thread: SqlRow, + hello: ConnectorHelloRequest | undefined +): AnswerablePrompt | null { + const prompt = pendingPrompt(thread) + return prompt && + prompt.kind !== 'unsupported' && + hello?.features.includes('answer') && + hello.promptKinds.includes(prompt.kind) + ? prompt + : null +} + +// The owner's answer to this prompt, if they sent one: matched by its own delivery id, so a newer +// prompt starts clean. +export async function ownerActionState( + transaction: CloudDatabase, + hostId: string, + deliveryId: string +): Promise { + const [row] = await transaction.query( + 'SELECT state, result FROM connector_inbox WHERE host_id = ? AND delivery_id = ?', + [hostId, deliveryId] + ) + if (!row) { + return null + } + if (row.state === 'expired') { + return { kind: 'unreached' } + } + if (row.state !== 'done') { + return { kind: 'on_its_way' } + } + const result = ConnectorDeliveryResultSchema.safeParse(JSON.parse(String(row.result))).data + switch (result?.outcome) { + case 'accepted': + return { kind: 'on_its_way' } + case 'refused': + return { kind: 'refused', reason: result.reason } + case 'expired': + return { kind: 'unreached' } + default: + return { kind: 'unknown' } + } +} + +// Decided under the host lock the machine's events are applied under, so the prompt read here is the +// one the machine last reported. +export async function applyOwnerAction( + transaction: CloudDatabase, + queue: QueueInTransaction, + externalId: string, + actorUserId: string, + action: OwnerAction +): Promise { + const [found] = await transaction.query( + 'SELECT host_id, owner_user_id FROM connector_threads WHERE external_id = ?', + [externalId] + ) + if (!found) { + return null + } + const ownerUserId = String(found.owner_user_id) + if (actorUserId !== ownerUserId) { + return { kind: 'not_owner', ownerUserId } + } + const hostId = String(found.host_id) + await lockConnectorHost(transaction, hostId) + const [thread] = await transaction.query( + 'SELECT * FROM connector_threads WHERE external_id = ? AND host_id = ?', + [externalId, hostId] + ) + if (!thread) { + return { kind: 'moved_on' } + } + const target = { + externalId, + teamId: String(thread.team_id), + channelId: String(thread.channel_id), + threadTs: String(thread.thread_ts), + ownerUserId, + hostId, + placeId: String(thread.place_id) + } + const prompt = pendingPrompt(thread) + const option = + prompt?.kind !== 'unsupported' && prompt?.promptId === action.promptId + ? prompt.options.find((candidate) => candidate.id === action.optionId) + : undefined + if (!option) { + return { kind: 'moved_on' } + } + const queued = await queue(transaction, { + thread: target, + deliveryId: answerDeliveryId(externalId, action.promptId), + body: { kind: 'answer', promptId: action.promptId, optionId: action.optionId } + }) + return queued === 'queued' + ? { kind: 'answered', ownerUserId, label: option.label } + : { kind: 'moved_on' } +} diff --git a/cloud/apps/connector/src/queue-delivery.ts b/cloud/apps/connector/src/queue-delivery.ts index 1b3f1876bca..cebcfaf778f 100644 --- a/cloud/apps/connector/src/queue-delivery.ts +++ b/cloud/apps/connector/src/queue-delivery.ts @@ -5,6 +5,12 @@ import type { ConnectorThreads, NewThread } from './connector-threads.js' export type QueueResult = EnqueueResult | 'unknown_host' | 'not_this_hosts_thread' +// queueDelivery inside a caller's transaction. +export type QueueInTransaction = ( + transaction: CloudDatabase, + input: Parameters[1] +) => ReturnType + // One host-locked transaction finds the host, binds the thread and queues, so per-host seq is // assigned in order. export async function queueDelivery( diff --git a/cloud/apps/connector/src/slack-clicks.test.ts b/cloud/apps/connector/src/slack-clicks.test.ts new file mode 100644 index 00000000000..c225d4a61fe --- /dev/null +++ b/cloud/apps/connector/src/slack-clicks.test.ts @@ -0,0 +1,268 @@ +import { ConnectorPollResponseSchema } from '@orca-cloud/connector-contract' +import { afterEach, describe, expect, it, vi } from 'vitest' +import { cleanupConnectorHarnesses } from './connector-server-harness.test-fixture.js' +import { answerDeliveryId } from './owner-controls.js' +import { + CHANNEL, + conversation, + EXTERNAL_ID, + OWNER, + startedThread, + TEAMMATE, + turnEnded, + turnStarted +} from './slack-harness.test-fixture.js' + +afterEach(async () => { + await cleanupConnectorHarnesses() + vi.unstubAllGlobals() +}) + +const CARD = `prompt:${EXTERNAL_ID}:p1` +const approval = ( + options = [ + { id: 'allow_once', label: 'Allow once' }, + { id: 'reject_once', label: 'Deny' } + ] +) => ({ + kind: 'prompt', + conversation, + paneKey: 'pane-1', + prompt: { kind: 'approval', promptId: 'p1', tool: 'Bash', summary: 'rm -rf ', options } +}) + +// A started thread whose agent is working and now asks for approval; its card is in Slack. +async function asked(prompt: unknown = approval()) { + const harness = await startedThread() + await harness.machine.ack([{ deliveryId: harness.start?.deliveryId, step: 'received' }]) + await harness.events(turnStarted, prompt) + // The reaction, the status and the card, one per pass in the channel. + await harness.pass() + await harness.pass() + await harness.pass() + const card = async () => { + const [row] = await harness.database.query( + 'SELECT body, slack_ts FROM connector_slack_posts WHERE post_id = ?', + [CARD] + ) + return row + ? { ...JSON.parse(String(row.body)), ts: row.slack_ts === null ? null : String(row.slack_ts) } + : undefined + } + const deliveries = async () => + ConnectorPollResponseSchema.parse((await harness.machine.poll()).body).deliveries + const ephemerals = async () => + (await harness.posts()) + .filter((post) => post.kind === 'ephemeral') + .map((post) => [String(post.post_id), JSON.parse(String(post.body)).text]) + return { ...harness, card, deliveries, ephemerals } +} + +describe('Slack prompt cards', () => { + it('posts the prompt with a button per option, and says the answer is wanted here', async () => { + const harness = await asked() + const posted = harness.fake + .take('chat.postMessage') + .find((call) => JSON.stringify(call.args).includes(CARD)) + expect(posted?.args).toMatchObject({ + channel: CHANNEL, + text: '*Bash* wants approval\nrm -rf <dist>', + blocks: [ + { type: 'section' }, + { + type: 'actions', + elements: [ + { action_id: 'answer:0', text: { text: 'Allow once' }, value: 'allow_once' }, + { action_id: 'answer:1', text: { text: 'Deny' }, value: 'reject_once' } + ] + } + ] + }) + expect(await harness.status()).toBe('Waiting for your answer.') + }) + + it('sends the owner’s click to the machine once, and closes the card with their answer', async () => { + const harness = await asked() + const { ts } = await harness.card() + expect((await harness.click({ messageTs: ts, value: 'reject_once' })).status).toBe(200) + expect((await harness.click({ messageTs: ts, value: 'allow_once' })).status).toBe(200) + expect(await harness.deliveries()).toEqual([ + expect.objectContaining({ + kind: 'answer', + deliveryId: answerDeliveryId(EXTERNAL_ID, 'p1'), + promptId: 'p1', + optionId: 'reject_once' + }) + ]) + const card = await harness.card() + expect(card.text).toBe(`<@${OWNER}> answered: Deny.`) + // The status follows at once, before the machine confirms anything. + expect(await harness.status()).toBe('Working on build-box…') + expect(card.blocks.map((block: { type: string }) => block.type)).toEqual(['section', 'context']) + expect(await harness.ephemerals()).toEqual([ + [`moved-on:${CARD}:${OWNER}`, 'Already answered or moved on.'] + ]) + // The machine's next report still shows the prompt: the answered card stays closed. + await harness.events(approval()) + expect((await harness.card()).text).toBe(`<@${OWNER}> answered: Deny.`) + }) + + it('tells anyone but the owner, privately, that only the owner can answer', async () => { + const harness = await asked() + const { ts } = await harness.card() + await harness.click({ user: TEAMMATE, messageTs: ts, value: 'allow_once' }) + expect(await harness.deliveries()).toEqual([]) + expect(await harness.ephemerals()).toEqual([ + [`only-owner:${CARD}:${TEAMMATE}`, `Only <@${OWNER}> can do this.`] + ]) + expect((await harness.card()).blocks).toHaveLength(2) + }) + + it('closes the card when the agent moves on, and a late click sends nothing', async () => { + const harness = await asked() + const { ts } = await harness.card() + await harness.events(turnEnded) + const card = await harness.card() + expect(card.text).toBe('No longer waiting for this answer.') + expect(card.blocks.map((block: { type: string }) => block.type)).toEqual(['section', 'context']) + await harness.click({ messageTs: ts, value: 'allow_once' }) + expect(await harness.deliveries()).toEqual([]) + expect(await harness.ephemerals()).toEqual([ + [`moved-on:${CARD}:${OWNER}`, 'Already answered or moved on.'] + ]) + }) + + it('reads an answer that never arrived as offline, until the agent shows a newer prompt', async () => { + const harness = await asked() + await harness.click({ messageTs: (await harness.card()).ts, value: 'allow_once' }) + harness.advance(5 * 60_000 + 1) + await harness.runtime.inbox.expireDue(harness.edge.onExpired) + expect(await harness.status()).toMatch(/^Couldn’t reach/) + await harness.events({ ...approval(), prompt: { ...approval().prompt, promptId: 'p2' } }) + const [row] = await harness.database.query( + 'SELECT body FROM connector_slack_posts WHERE post_id = ?', + [`status:${EXTERNAL_ID}`] + ) + expect(JSON.parse(String(row?.body)).text).toBe('Waiting for your answer.') + }) + + it('leaves prompts Slack can’t answer to Orca: no card, the status says where', async () => { + for (const prompt of [ + approval([{ id: 'allow_once', label: ' ' }]), + approval([]), + { + ...approval(), + prompt: { + kind: 'question', + promptId: 'p1', + question: 'Which?', + options: [{ id: 'a', label: 'A' }] + } + }, + { ...approval(), prompt: { kind: 'unsupported', promptId: 'p1' } } + ]) { + const harness = await asked(prompt) + expect(await harness.card()).toBeUndefined() + expect(await harness.status()).toBe('Waiting for your answer in Orca on build-box.') + } + }) + + it('ignores a click on a message that isn’t a card, and refuses an unsigned one', async () => { + const harness = await asked() + expect( + (await harness.click({ messageTs: '1800000999.000001', value: 'allow_once' })).status + ).toBe(200) + const response = await harness.request('/slack/interactivity', { + method: 'POST', + headers: { 'content-type': 'application/x-www-form-urlencoded' }, + body: 'payload=%7B%7D' + }) + expect(response.status).toBe(401) + expect(await harness.deliveries()).toEqual([]) + expect(await harness.ephemerals()).toEqual([]) + }) + + it('answers nothing while paused, and says so to the clicker', async () => { + const harness = await asked() + await harness.operator('pause', '--team', 'T1') + await harness.click({ messageTs: (await harness.card()).ts, value: 'allow_once' }) + expect(await harness.deliveries()).toEqual([]) + expect((await harness.ephemerals()).map(([, text]) => text)).toEqual([ + 'Orca for Slack is paused.' + ]) + }) + + it('shows an answer’s refusal or unknown outcome as soon as it is settled', async () => { + for (const [result, text] of [ + [{ outcome: 'refused', reason: 'prompt_not_pending' }, 'Already answered or moved on.'], + [ + { outcome: 'unknown', reason: 'writer_failed' }, + 'This may not have reached the agent; check Orca.' + ] + ] as const) { + const harness = await asked() + await harness.click({ messageTs: (await harness.card()).ts, value: 'allow_once' }) + const [answer] = await harness.deliveries() + await harness.machine.ack([{ deliveryId: answer?.deliveryId, step: 'done', result }]) + expect(await harness.status()).toBe(text) + } + }) + + it('closes a card when a snapshot no longer lists its conversation', async () => { + const harness = await asked() + await harness.events({ + kind: 'snapshot', + conversations: [ + { + conversation, + paneKey: 'pane-1', + agent: 'claude', + working: true, + prompt: approval().prompt + } + ] + }) + expect((await harness.card()).blocks).toHaveLength(2) + await harness.events({ kind: 'snapshot', conversations: [] }) + expect((await harness.card()).text).toBe('No longer waiting for this answer.') + }) + + it('keeps an answered card’s note within Slack’s limits for any label', async () => { + const label = '&'.repeat(3_100) + const harness = await asked(approval([{ id: 'allow_once', label }])) + await harness.click({ messageTs: (await harness.card()).ts, value: 'allow_once' }) + const note = (await harness.card()).blocks[1].elements[0].text + expect(note).toBe(`<@${OWNER}> answered: ${'&'.repeat(75)}.`) + }) + + it('leaves an answered card final after the inbox forgets its answer', async () => { + const harness = await asked() + const { ts } = await harness.card() + await harness.click({ messageTs: ts, value: 'allow_once' }) + const [answer] = await harness.deliveries() + await harness.machine.ack([ + { deliveryId: answer?.deliveryId, step: 'done', result: { outcome: 'unknown', reason: 'x' } } + ]) + harness.advance(8 * 24 * 60 * 60_000) + await harness.runtime.inbox.pruneSettled(7 * 24 * 60 * 60_000) + await harness.events(approval()) + expect((await harness.card()).blocks).toHaveLength(2) + expect((await harness.card()).text).toBe(`<@${OWNER}> answered: Allow once.`) + expect(await harness.status()).toBe('Waiting for your answer in Orca on build-box.') + await harness.click({ messageTs: ts, value: 'allow_once' }) + expect(await harness.deliveries()).toEqual([]) + }) + + it('delivers an answer to a prompt id at the contract’s 128-character limit', async () => { + const promptId = 'p'.repeat(128) + const harness = await asked({ ...approval(), prompt: { ...approval().prompt, promptId } }) + const [card] = await harness.database.query( + 'SELECT slack_ts FROM connector_slack_posts WHERE post_id = ?', + [`prompt:${EXTERNAL_ID}:${promptId}`] + ) + await harness.click({ messageTs: String(card?.slack_ts), value: 'allow_once' }) + expect(await harness.deliveries()).toEqual([ + expect.objectContaining({ kind: 'answer', promptId, optionId: 'allow_once' }) + ]) + }) +}) diff --git a/cloud/apps/connector/src/slack-clicks.ts b/cloud/apps/connector/src/slack-clicks.ts new file mode 100644 index 00000000000..f38f5797b08 --- /dev/null +++ b/cloud/apps/connector/src/slack-clicks.ts @@ -0,0 +1,149 @@ +import type { CloudDatabase } from '@orca-cloud/cloud-database' +import { z } from 'zod' +import { lockConnectorHost } from './connector-hosts.js' +import { applyOwnerAction, type OwnerAction } from './owner-controls.js' +import type { SlackConfig } from './slack-config.js' +import type { SlackInstalls } from './slack-installs.js' +import type { SlackPosts } from './slack-posts.js' +import { isOpenCard, promptCardId, settledCard } from './slack-prompt-cards.js' +import type { QueueInTransaction } from './queue-delivery.js' +import { slackText } from './slack-text.js' +import { refreshSlackStatus } from './slack-thread-posts.js' + +const BlockActionsSchema = z.looseObject({ + type: z.literal('block_actions'), + team: z.looseObject({ id: z.string().min(1) }), + user: z.looseObject({ id: z.string().min(1) }), + channel: z.looseObject({ id: z.string().min(1) }), + container: z.looseObject({ message_ts: z.string().min(1) }), + actions: z + .array(z.looseObject({ action_id: z.string(), value: z.string().min(1).optional() })) + .min(1) +}) + +// A press of an answer button on a prompt card. What it answers is found from the message it is +// on; the button only says which option. +export type SlackClick = { + teamId: string + channelId: string + userId: string + messageTs: string + optionId: string +} + +// Slack sends interactions form-encoded, the JSON in `payload`. Other interactions are ignored. +export function readSlackClick(rawBody: string): SlackClick | 'ignored' | 'invalid' { + let payload: unknown + try { + payload = JSON.parse(new URLSearchParams(rawBody).get('payload') ?? '') + } catch { + return 'invalid' + } + const parsed = BlockActionsSchema.safeParse(payload) + if (!parsed.success) { + return typeof payload === 'object' && payload !== null ? 'ignored' : 'invalid' + } + const [action] = parsed.data.actions + if (!action?.action_id.startsWith('answer:') || action.value === undefined) { + return 'ignored' + } + return { + teamId: parsed.data.team.id, + channelId: parsed.data.channel.id, + userId: parsed.data.user.id, + messageTs: parsed.data.container.message_ts, + optionId: action.value + } +} + +// Decisions §1: only the conversation's owner answers; anyone else is told so privately. +export async function handleSlackClick( + deps: { + config: SlackConfig + database: CloudDatabase + installs: SlackInstalls + posts: SlackPosts + queue: QueueInTransaction + }, + click: SlackClick +): Promise { + const { config, installs, posts, queue } = deps + await deps.database.transaction(async (transaction) => { + const install = await installs.find(transaction, click.teamId) + const [message] = await transaction.query( + `SELECT post_id, thread_ts, body FROM connector_slack_posts + WHERE team_id = ? AND channel_id = ? AND slack_ts = ? AND kind = 'status'`, + [click.teamId, click.channelId, click.messageTs] + ) + if (!install || !message) { + return + } + const postId = String(message.post_id) + const place = { + teamId: click.teamId, + channelId: click.channelId, + threadTs: String(message.thread_ts) + } + const threads = await transaction.query( + `SELECT external_id, host_id FROM connector_threads + WHERE team_id = ? AND channel_id = ? AND thread_ts = ?`, + [place.teamId, place.channelId, place.threadTs] + ) + const thread = threads.find((row) => + postId.startsWith(promptCardId(String(row.external_id), '')) + ) + if (!thread) { + return + } + const externalId = String(thread.external_id) + const tell = (id: string, text: string) => + posts.queue(transaction, { + ...place, + postId: id, + kind: 'ephemeral', + body: { text, user: click.userId, threadTs: place.threadTs } + }) + if (config.paused || install.paused) { + await tell( + `paused:${place.teamId}:${place.channelId}:${place.threadTs}:${click.userId}`, + slackText.paused + ) + return + } + // Cards close only under the host lock (events, clicks), so this read holds until commit. + await lockConnectorHost(transaction, String(thread.host_id)) + const [card] = await transaction.query( + 'SELECT body FROM connector_slack_posts WHERE post_id = ?', + [postId] + ) + if (!isOpenCard(JSON.parse(String(card?.body ?? '{}')))) { + await tell(`moved-on:${postId}:${click.userId}`, slackText.alreadyAnswered) + return + } + const action: OwnerAction = { + kind: 'answer', + promptId: postId.slice(promptCardId(externalId, '').length), + optionId: click.optionId + } + const outcome = await applyOwnerAction(transaction, queue, externalId, click.userId, action) + switch (outcome?.kind) { + case 'not_owner': + await tell(`only-owner:${postId}:${click.userId}`, slackText.onlyOwner(outcome.ownerUserId)) + return + case 'moved_on': + await tell(`moved-on:${postId}:${click.userId}`, slackText.alreadyAnswered) + return + case 'answered': + await transaction.lockQuotaScope(`orca-connector-slack-status:${externalId}`) + await posts.setStatus(transaction, { + ...place, + postId, + ...settledCard( + JSON.parse(String(message.body)), + slackText.promptAnswered(outcome.ownerUserId, outcome.label) + ) + }) + await refreshSlackStatus(transaction, posts, String(thread.host_id), externalId) + } + }) +} diff --git a/cloud/apps/connector/src/slack-edge.ts b/cloud/apps/connector/src/slack-edge.ts index a649c37a513..0bce63c042e 100644 --- a/cloud/apps/connector/src/slack-edge.ts +++ b/cloud/apps/connector/src/slack-edge.ts @@ -5,7 +5,7 @@ import { type ConnectorEvent, type ConnectorReplyResponse } from '@orca-cloud/connector-contract' -import type { Hono, MiddlewareHandler } from 'hono' +import type { Context, Hono, MiddlewareHandler } from 'hono' import { bodyLimit } from 'hono/body-limit' import { every } from './connector-background.js' import type { ConnectorInbox, ExpiredThread } from './connector-inbox.js' @@ -15,8 +15,10 @@ import { readSlackEnvelope } from './slack-events.js' import { SlackInstalls } from './slack-installs.js' import { SlackMembers } from './slack-members.js' import { SlackPostWorker } from './slack-post-worker.js' +import { handleSlackClick, readSlackClick } from './slack-clicks.js' import { SlackPosts } from './slack-posts.js' -import { routeSlackMessage, type QueueInTransaction } from './slack-routing.js' +import type { QueueInTransaction } from './queue-delivery.js' +import { routeSlackMessage } from './slack-routing.js' import { verifySlackSignature } from './slack-signature.js' import { SlackThreadStarts } from './slack-thread-start.js' import { queueAgentReply, queueTurnFallback, refreshSlackStatus } from './slack-thread-posts.js' @@ -63,9 +65,14 @@ export function createSlackEdge(deps: { } } - function registerRoutes(app: Hono, limitMachineBody: MiddlewareHandler): void { + // Slack's own requests: signed over the raw body, size-capped. + const fromSlack = ( + app: Hono, + path: string, + handle: (context: Context, rawBody: string) => Promise + ) => app.post( - '/slack/events', + path, bodyLimit({ maxSize: MAX_SLACK_BODY_BYTES, onError: (context) => context.json({ error: 'request_too_large' }, 413) @@ -79,33 +86,49 @@ export function createSlackEdge(deps: { rawBody, nowMs: now() }) - if (!verified) { - return context.json({ error: 'invalid_signature' }, 401) - } - let payload: unknown - try { - payload = JSON.parse(rawBody) - } catch { - return context.json({ error: 'invalid_request' }, 400) - } - const envelope = readSlackEnvelope(payload) - if (envelope.kind === 'invalid') { - return context.json({ error: 'invalid_request' }, 400) - } - if (envelope.kind === 'url_verification') { - return context.json({ challenge: envelope.challenge }) - } - // Answered before any Slack call; a database failure is a 500, so Slack retries. - if (envelope.kind === 'message' && config.teamIds.has(envelope.message.teamId)) { - await routeSlackMessage( - { config, database, installs, posts, queue: deps.queue, now }, - envelope.message - ) - } - return context.body(null, 200) + return verified + ? await handle(context, rawBody) + : context.json({ error: 'invalid_signature' }, 401) } ) + function registerRoutes(app: Hono, limitMachineBody: MiddlewareHandler): void { + fromSlack(app, '/slack/events', async (context, rawBody) => { + let payload: unknown + try { + payload = JSON.parse(rawBody) + } catch { + return context.json({ error: 'invalid_request' }, 400) + } + const envelope = readSlackEnvelope(payload) + if (envelope.kind === 'invalid') { + return context.json({ error: 'invalid_request' }, 400) + } + if (envelope.kind === 'url_verification') { + return context.json({ challenge: envelope.challenge }) + } + // Answered before any Slack call; a database failure is a 500, so Slack retries. + if (envelope.kind === 'message' && config.teamIds.has(envelope.message.teamId)) { + await routeSlackMessage( + { config, database, installs, posts, queue: deps.queue, now }, + envelope.message + ) + } + return context.body(null, 200) + }) + + // Answer presses on prompt cards; answered before any Slack call, like events. + fromSlack(app, '/slack/interactivity', async (context, rawBody) => { + const click = readSlackClick(rawBody) + if (click === 'invalid') { + return context.json({ error: 'invalid_request' }, 400) + } + if (click !== 'ignored' && config.teamIds.has(click.teamId)) { + await handleSlackClick({ config, database, installs, posts, queue: deps.queue }, click) + } + return context.body(null, 200) + }) + app.post(CONNECTOR_PATHS.replies, limitMachineBody, async (context) => { const body = ConnectorReplyRequestSchema.safeParse(await context.req.json().catch(() => null)) if (!body.success) { @@ -144,17 +167,15 @@ export function createSlackEdge(deps: { ? event.conversations.map((entry) => entry.conversation.externalId) : [event.conversation.externalId] ) - // A snapshot can mark threads it doesn't list unverifiable. - const unverifiable = events.some((event) => event.kind === 'snapshot') - ? await transaction.query( - `SELECT external_id FROM connector_threads - WHERE host_id = ? AND turn_state = 'unverifiable'`, - [hostId] - ) + // A snapshot also changes the threads it doesn't list (unverifiable turn, no prompt). + const unlisted = events.some((event) => event.kind === 'snapshot') + ? await transaction.query('SELECT external_id FROM connector_threads WHERE host_id = ?', [ + hostId + ]) : [] await refresh(transaction, hostId, [ ...externalIds, - ...unverifiable.map((row) => String(row.external_id)) + ...unlisted.map((row) => String(row.external_id)) ]) for (const event of events) { if (event.kind === 'turn-ended' && event.lastAssistantMessage !== undefined) { diff --git a/cloud/apps/connector/src/slack-harness.test-fixture.ts b/cloud/apps/connector/src/slack-harness.test-fixture.ts index 1ecea3a50fa..c35d40eb183 100644 --- a/cloud/apps/connector/src/slack-harness.test-fixture.ts +++ b/cloud/apps/connector/src/slack-harness.test-fixture.ts @@ -1,6 +1,10 @@ import { createHmac, randomBytes } from 'node:crypto' -import { CONNECTOR_PATHS, type ConnectorProject } from '@orca-cloud/connector-contract' -import { vi } from 'vitest' +import { + CONNECTOR_PATHS, + ConnectorPollResponseSchema, + type ConnectorProject +} from '@orca-cloud/connector-contract' +import { expect, vi } from 'vitest' import { createConnectorHarness } from './connector-server-harness.test-fixture.js' import type { SlackConfig } from './slack-config.js' import { runSlackOperator } from './slack-operator.js' @@ -77,11 +81,16 @@ export async function createSlackHarness(slack: Partial = {}) { }) await operator('install', '--team', TEAM, '--bot-user', BOT) - const signedPost = (body: string, timestamp = Math.floor(harness.now() / 1000)) => - harness.request('/slack/events', { + const signedPost = ( + body: string, + timestamp = Math.floor(harness.now() / 1000), + path = '/slack/events', + contentType = 'application/json' + ) => + harness.request(path, { method: 'POST', headers: { - 'content-type': 'application/json', + 'content-type': contentType, 'x-slack-request-timestamp': String(timestamp), 'x-slack-signature': `v0=${createHmac('sha256', SIGNING_SECRET).update(`v0:${timestamp}:${body}`).digest('hex')}` }, @@ -114,6 +123,24 @@ export async function createSlackHarness(slack: Partial = {}) { }) ) + // A button press, form-encoded as Slack sends interactions. + const click = (fields: { user?: string; messageTs: string; value: string; actionId?: string }) => + signedPost( + new URLSearchParams({ + payload: JSON.stringify({ + type: 'block_actions', + team: { id: TEAM }, + user: { id: fields.user ?? OWNER }, + channel: { id: CHANNEL }, + container: { type: 'message', message_ts: fields.messageTs, channel_id: CHANNEL }, + actions: [{ action_id: fields.actionId ?? 'answer:0', value: fields.value }] + }) + }).toString(), + undefined, + '/slack/interactivity', + 'application/x-www-form-urlencoded' + ) + // A connected machine of `account` listing `projects`. async function machineWith( account: string, @@ -148,5 +175,72 @@ export async function createSlackHarness(slack: Partial = {}) { 'SELECT post_id, kind, body, state, next_at, slack_ts FROM connector_slack_posts ORDER BY post_id' ) - return { ...harness, edge, fake, tokenKey, operator, signedPost, message, machineWith, posts } + return { + ...harness, + edge, + fake, + tokenKey, + operator, + signedPost, + message, + click, + machineWith, + posts + } +} + +export const THREAD = '1800000000.000100' +export const EXTERNAL_ID = `${TEAM}:${CHANNEL}:${THREAD}:${OWNER}` +export const conversation = { source: 'slack', externalId: EXTERNAL_ID } +export const TURN = 1_800_000_001_000 + +// A linked owner's conversation, its start handed to their machine. +export async function startedThread() { + const harness = await createSlackHarness() + await harness.operator( + 'link', + '--team', + TEAM, + '--user', + OWNER, + '--account', + 'user-1', + '--default-project', + 'orca' + ) + const machine = await harness.machineWith('user-1', [project('orca')]) + await harness.message({ text: `<@${BOT}> fix it`, ts: THREAD }) + const [start] = ConnectorPollResponseSchema.parse((await machine.poll()).body).deliveries + let seq = 0 + const events = async (...batch: unknown[]) => { + const response = await machine.call(CONNECTOR_PATHS.events, { + v: 1, + helloRev: 1, + seq: ++seq, + events: batch + }) + expect(response.status).toBe(204) + } + const status = async () => { + const [row] = await harness.database.query( + 'SELECT body FROM connector_slack_posts WHERE post_id = ?', + [`status:${EXTERNAL_ID}`] + ) + return JSON.parse(String(row?.body)).text + } + const reply = (replyId: string, text: string, by = machine) => + by.call(CONNECTOR_PATHS.replies, { v: 1, replyId, conversation, turnId: TURN, text }) + const pass = () => harness.edge.worker.sendDue() + return { ...harness, machine, start, events, status, reply, pass } +} + +export const turnStarted = { kind: 'turn-started', conversation, paneKey: 'pane-1', at: TURN } +export const turnEnded = { + kind: 'turn-ended', + conversation, + paneKey: 'pane-1', + turnId: TURN, + at: TURN + 5_000, + stopReason: 'end_turn', + lastAssistantMessage: 'Fixed & done' } diff --git a/cloud/apps/connector/src/slack-post-worker.ts b/cloud/apps/connector/src/slack-post-worker.ts index f210e946b5f..1f765074c62 100644 --- a/cloud/apps/connector/src/slack-post-worker.ts +++ b/cloud/apps/connector/src/slack-post-worker.ts @@ -4,7 +4,10 @@ import type { SlackInstalls } from './slack-installs.js' import type { ClaimedPost, SlackPostKind, SlackPosts } from './slack-posts.js' import type { SlackResult, SlackWebApi } from './slack-web-api.js' -const TextBodySchema = z.object({ text: z.string().min(1) }) +const TextBodySchema = z.object({ + text: z.string().min(1), + blocks: z.array(z.looseObject({ type: z.string() })).optional() +}) const ReactionBodySchema = z.object({ name: z.string().min(1), timestamp: z.string().min(1) }) const EphemeralBodySchema = z.object({ text: z.string().min(1), @@ -102,12 +105,13 @@ export class SlackPostWorker { await this.finish(post, result, 'retry') return } - const { text } = TextBodySchema.parse(post.body) + const { text, blocks } = TextBodySchema.parse(post.body) if (post.kind === 'status' && post.slackTs !== null) { const result = await api(token, 'chat.update', { channel: post.channelId, ts: post.slackTs, - text + text, + ...(blocks ? { blocks } : {}) }) await this.finish(post, result, 'retry') return @@ -129,6 +133,7 @@ export class SlackPostWorker { channel: post.channelId, thread_ts: post.threadTs, text, + ...(blocks ? { blocks } : {}), unfurl_links: false, unfurl_media: false, metadata: { event_type: POST_METADATA_TYPE, event_payload: { post_id: post.postId } } diff --git a/cloud/apps/connector/src/slack-posting.test.ts b/cloud/apps/connector/src/slack-posting.test.ts index 19989b980e0..a511df8bf87 100644 --- a/cloud/apps/connector/src/slack-posting.test.ts +++ b/cloud/apps/connector/src/slack-posting.test.ts @@ -6,8 +6,15 @@ import { CHANNEL, createSlackHarness, OWNER, + conversation, + EXTERNAL_ID, project, - TEAM + startedThread, + TEAM, + THREAD, + TURN, + turnEnded, + turnStarted } from './slack-harness.test-fixture.js' import type { CloudDatabase } from '@orca-cloud/cloud-database' import { SlackInstalls } from './slack-installs.js' @@ -20,61 +27,6 @@ afterEach(async () => { vi.unstubAllGlobals() }) -const THREAD = '1800000000.000100' -const EXTERNAL_ID = `${TEAM}:${CHANNEL}:${THREAD}:${OWNER}` -const conversation = { source: 'slack', externalId: EXTERNAL_ID } -const TURN = 1_800_000_001_000 - -async function startedThread() { - const harness = await createSlackHarness() - await harness.operator( - 'link', - '--team', - TEAM, - '--user', - OWNER, - '--account', - 'user-1', - '--default-project', - 'orca' - ) - const machine = await harness.machineWith('user-1', [project('orca')]) - await harness.message({ text: `<@${BOT}> fix it`, ts: THREAD }) - const [start] = ConnectorPollResponseSchema.parse((await machine.poll()).body).deliveries - let seq = 0 - const events = async (...batch: unknown[]) => { - const response = await machine.call(CONNECTOR_PATHS.events, { - v: 1, - helloRev: 1, - seq: ++seq, - events: batch - }) - expect(response.status).toBe(204) - } - const status = async () => { - const [row] = await harness.database.query( - 'SELECT body FROM connector_slack_posts WHERE post_id = ?', - [`status:${EXTERNAL_ID}`] - ) - return JSON.parse(String(row?.body)).text - } - const reply = (replyId: string, text: string, by = machine) => - by.call(CONNECTOR_PATHS.replies, { v: 1, replyId, conversation, turnId: TURN, text }) - const pass = () => harness.edge.worker.sendDue() - return { ...harness, machine, start, events, status, reply, pass } -} - -const turnStarted = { kind: 'turn-started', conversation, paneKey: 'pane-1', at: TURN } -const turnEnded = { - kind: 'turn-ended', - conversation, - paneKey: 'pane-1', - turnId: TURN, - at: TURN + 5_000, - stopReason: 'end_turn', - lastAssistantMessage: 'Fixed & done' -} - describe('Slack thread status and posts', () => { it('edits one status message as the turn moves, and posts the last message when the agent didn’t reply', async () => { const harness = await startedThread() @@ -130,9 +82,25 @@ describe('Slack thread status and posts', () => { result: { outcome: 'refused', reason: 'no_such_project' } } ]) - expect(await harness.status()).toBe( - 'Your Orca on build-box couldn’t take this (no_such_project).' - ) + // A code with no line of its own reads generically; the code itself stays out of the channel. + expect(await harness.status()).toBe('Your Orca on build-box couldn’t take this.') + }) + + it('records a refusal whose code is an inherited property name, and reads it generically', async () => { + for (const reason of ['__proto__', 'toString']) { + const harness = await startedThread() + const response = await harness.machine.ack([ + { + deliveryId: harness.start?.deliveryId, + step: 'done', + result: { outcome: 'refused', reason } + } + ]) + expect(response.status).toBe(204) + const [row] = await harness.database.query('SELECT state FROM connector_inbox') + expect(row?.state).toBe('done') + expect(await harness.status()).toBe('Your Orca on build-box couldn’t take this.') + } }) it('shows a turn the machine can no longer vouch for', async () => { @@ -168,7 +136,11 @@ describe('Slack thread status and posts', () => { expect(harness.fake.take()).toEqual([ { method: 'chat.update', - args: { channel: CHANNEL, ts: expect.any(String), text: 'Working on build-box…' } + args: { + channel: CHANNEL, + ts: expect.any(String), + text: 'Working on build-box…' + } } ]) }) diff --git a/cloud/apps/connector/src/slack-posts.ts b/cloud/apps/connector/src/slack-posts.ts index 7ef15e2fe48..cd6a6458811 100644 --- a/cloud/apps/connector/src/slack-posts.ts +++ b/cloud/apps/connector/src/slack-posts.ts @@ -5,6 +5,8 @@ import type { CloudDatabase, SqlRow } from '@orca-cloud/cloud-database' export type SlackPostKind = 'message' | 'status' | 'reaction' | 'ephemeral' export type SlackPlace = { teamId: string; channelId: string; threadTs: string } +// A Block Kit block, built in slack-prompt-cards.ts and passed to Slack as is. +export type SlackBlock = Record export type ClaimedPost = SlackPlace & { postId: string @@ -76,7 +78,7 @@ export class SlackPosts { // A text change never erases a delivery outcome: an uncertain status stays so, and stops editing. async setStatus( transaction: CloudDatabase, - status: SlackPlace & { postId: string; text: string } + status: SlackPlace & { postId: string; text: string; blocks?: SlackBlock[] } ): Promise { const now = this.now() await transaction.query( @@ -100,7 +102,7 @@ export class SlackPosts { status.teamId, status.channelId, status.threadTs, - JSON.stringify({ text: status.text }), + JSON.stringify({ text: status.text, ...(status.blocks ? { blocks: status.blocks } : {}) }), now, now, now diff --git a/cloud/apps/connector/src/slack-prompt-cards.ts b/cloud/apps/connector/src/slack-prompt-cards.ts new file mode 100644 index 00000000000..97879002dcc --- /dev/null +++ b/cloud/apps/connector/src/slack-prompt-cards.ts @@ -0,0 +1,100 @@ +import type { CloudDatabase, SqlRow } from '@orca-cloud/cloud-database' +import type { ConnectorHelloRequest } from '@orca-cloud/connector-contract' +import { z } from 'zod' +import { answerablePrompt, type AnswerablePrompt } from './owner-controls.js' +import type { SlackBlock, SlackPlace, SlackPosts } from './slack-posts.js' +import { MAX_BUTTON_LABEL_CHARS, slackText } from './slack-text.js' + +// Slack's limit on buttons per actions block. +const MAX_BUTTONS = 25 + +const CardBodySchema = z.object({ + blocks: z.array(z.looseObject({ type: z.string() })).optional() +}) + +export const promptCardId = (externalId: string, promptId: string) => + `prompt:${externalId}:${promptId}` + +// An answerable prompt whose options all fit on buttons; anything else is answered in Orca. +export function slackAnswerablePrompt( + thread: SqlRow, + hello: ConnectorHelloRequest | undefined +): AnswerablePrompt | null { + const prompt = answerablePrompt(thread, hello) + const fits = + prompt !== null && + prompt.options.length > 0 && + prompt.options.length <= MAX_BUTTONS && + prompt.options.every((option) => option.label.trim().length > 0) && + slackText.promptCard(prompt).length > 0 + return fits ? prompt : null +} + +function openCard(prompt: AnswerablePrompt): { text: string; blocks: SlackBlock[] } { + const text = slackText.promptCard(prompt) + return { + text, + blocks: [ + { type: 'section', text: { type: 'mrkdwn', text } }, + { + type: 'actions', + elements: prompt.options.map((option, index) => ({ + type: 'button', + action_id: `answer:${index}`, + text: { type: 'plain_text', text: option.label.slice(0, MAX_BUTTON_LABEL_CHARS) }, + value: option.id + })) + } + ] + } +} + +// What the card asked, with its buttons replaced by a closing note. +export function settledCard(body: unknown, note: string): { text: string; blocks: SlackBlock[] } { + const asked = CardBodySchema.safeParse(body).data?.blocks?.[0] + return { + text: note, + blocks: [ + ...(asked ? [asked] : []), + { type: 'context', elements: [{ type: 'mrkdwn', text: note }] } + ] + } +} + +// A card still showing its buttons; an answered or closed one is final. +export function isOpenCard(body: unknown): boolean { + return ( + CardBodySchema.safeParse(body).data?.blocks?.some((block) => block.type === 'actions') ?? false + ) +} + +// The conversation's prompt cards as a projection, beside its status and under the same lock: the +// shown prompt gets a card once; any other open card closes. A final card is never reopened. True +// when the shown prompt's card offers its buttons. +export async function refreshPromptCards( + transaction: CloudDatabase, + posts: SlackPosts, + thread: SlackPlace & { externalId: string }, + prompt: AnswerablePrompt | null +): Promise { + const cards = await transaction.query( + `SELECT post_id, body FROM connector_slack_posts WHERE kind = 'status' AND post_id LIKE ?`, + [`${promptCardId(thread.externalId, '')}%`] + ) + const shownId = prompt ? promptCardId(thread.externalId, prompt.promptId) : null + const shown = cards.find((card) => card.post_id === shownId) + if (prompt && shownId && !shown) { + await posts.setStatus(transaction, { ...thread, postId: shownId, ...openCard(prompt) }) + } + for (const card of cards) { + const body: unknown = JSON.parse(String(card.body)) + if (card.post_id !== shownId && isOpenCard(body)) { + await posts.setStatus(transaction, { + ...thread, + postId: String(card.post_id), + ...settledCard(body, slackText.promptClosed) + }) + } + } + return prompt !== null && (!shown || isOpenCard(JSON.parse(String(shown.body)))) +} diff --git a/cloud/apps/connector/src/slack-routing.ts b/cloud/apps/connector/src/slack-routing.ts index c62fc0e1aa2..17ec1251e55 100644 --- a/cloud/apps/connector/src/slack-routing.ts +++ b/cloud/apps/connector/src/slack-routing.ts @@ -1,7 +1,7 @@ import type { CloudDatabase } from '@orca-cloud/cloud-database' import { CONNECTOR_AGENTS, ConnectorDeliveryResultSchema } from '@orca-cloud/connector-contract' import type { DeliveryBody } from './connector-inbox.js' -import type { queueDelivery } from './queue-delivery.js' +import type { QueueInTransaction } from './queue-delivery.js' import type { SlackConfig } from './slack-config.js' import { mentionsBot, promptText, type SlackMessage } from './slack-events.js' import { findSlackIdentity } from './slack-identities.js' @@ -13,12 +13,6 @@ import { slackText } from './slack-text.js' import { reactSeen, replyToWriter } from './slack-writer-replies.js' import { refreshSlackStatus } from './slack-thread-posts.js' -// queueDelivery inside the routing transaction. -export type QueueInTransaction = ( - transaction: CloudDatabase, - input: Parameters[1] -) => ReturnType - export type SlackRoutingDeps = { config: SlackConfig database: CloudDatabase diff --git a/cloud/apps/connector/src/slack-text.ts b/cloud/apps/connector/src/slack-text.ts index 873a72d85d8..ef17c2a4197 100644 --- a/cloud/apps/connector/src/slack-text.ts +++ b/cloud/apps/connector/src/slack-text.ts @@ -1,4 +1,5 @@ import type { PlaceOption } from './slack-machine-choice.js' +import type { AnswerablePrompt } from './owner-controls.js' // Every line Orca posts in Slack, in one place for review. Replies meant only for the person who // wrote (paused, unlinked, update, which project) go out ephemeral. Inputs are escaped here, so @@ -8,6 +9,35 @@ export function escapeSlackText(text: string): string { } const MAX_LISTED_OPTIONS = 10 +// Slack's limits on a section's text and a button's label. +const MAX_SECTION_CHARS = 3_000 +export const MAX_BUTTON_LABEL_CHARS = 75 + +// The machine's refusal reasons (its CONNECTOR_REFUSALS). +// A Map, so only these own entries match: a code like `toString` falls to the generic line. +const REFUSALS = new Map string>([ + ['connectors_disabled', (machine) => `Connectors are turned off in Orca on ${machine}.`], + ['agent_not_reachable', (machine) => `Couldn’t reach the agent in Orca on ${machine}.`], + ['unsupported_agent', (machine) => `Orca on ${machine} can’t run that agent from Slack.`], + ['delivery_id_reused', (machine) => `Your Orca on ${machine} couldn’t take this.`], + ['delivery_backlog_full', (machine) => `Orca on ${machine} is busy. Send again in a moment.`], + [ + 'agent_asleep_in_window', + (machine) => `The agent in Orca on ${machine} is asleep. Wake it in Orca, then send again.` + ], + [ + 'no_session_to_reopen', + (machine) => `The earlier session in Orca on ${machine} is gone. Mention me to start a new one.` + ], + ['reopen_refused', (machine) => `Orca on ${machine} couldn’t reopen the earlier session.`], + ['prompt_not_pending', () => 'Already answered or moved on.'], + ['prompt_unsupported', (machine) => `Answer this one in Orca on ${machine}.`], + [ + 'option_not_offered', + (machine) => `That option isn’t offered any more. Answer in Orca on ${machine}.` + ], + ['conversation_cancelled', () => 'Stopped before it reached the agent.'] +]) export const slackText = { paused: 'Orca for Slack is paused.', @@ -27,15 +57,44 @@ export const slackText = { working: (hostName: string) => `Working on ${escapeSlackText(hostName)}…`, needsAnswer: (hostName: string) => `Waiting for your answer in Orca on ${escapeSlackText(hostName)}.`, + // The prompt's card in the thread has the buttons. + needsAnswerHere: 'Waiting for your answer.', done: (hostName: string) => `Done on ${escapeSlackText(hostName)}.`, // Q18, in the public status: the thread sees who it was waiting on (Slack renders <@U…> as a name). offline: (ownerUserId: string, hostName: string) => `Couldn’t reach <@${ownerUserId}>’s Orca on ${escapeSlackText(hostName)}. Send again when it’s back.`, - refused: (hostName: string, reason: string) => - `Your Orca on ${escapeSlackText(hostName)} couldn’t take this (${escapeSlackText(reason)}).`, + // A machine's refusal, in words; a code we don't know shows the generic line and is logged. + refused: (hostName: string, reason: string) => { + const machine = escapeSlackText(hostName) + const line = REFUSALS.get(reason)?.(machine) + if (line === undefined) { + console.warn(JSON.stringify({ event: 'orca_connector_slack_unknown_refusal', reason })) + } + return line ?? `Your Orca on ${machine} couldn’t take this.` + }, + // The machine couldn't read what was sent: it predates that kind of delivery. + unsupported: (hostName: string) => + `Your Orca on ${escapeSlackText(hostName)} can’t take this yet. Update Orca.`, unknown: 'This may not have reached the agent; check Orca.', noUpdate: 'No update from your Orca; check Orca.' }, + // A prompt's card: what the agent asks, then its state once it stops taking answers. + promptCard: (prompt: AnswerablePrompt) => { + const lines = + prompt.kind === 'approval' + ? [ + `*${escapeSlackText(prompt.tool)}* wants approval`, + prompt.summary && escapeSlackText(prompt.summary) + ] + : [prompt.header && `*${escapeSlackText(prompt.header)}*`, escapeSlackText(prompt.question)] + return lines.filter(Boolean).join('\n').slice(0, MAX_SECTION_CHARS) + }, + // The label as its button showed it (Slack caps those at 75), so the note stays within limits. + promptAnswered: (ownerUserId: string, label: string) => + `<@${ownerUserId}> answered: ${escapeSlackText(label.slice(0, MAX_BUTTON_LABEL_CHARS))}.`, + promptClosed: 'No longer waiting for this answer.', + onlyOwner: (ownerUserId: string) => `Only <@${ownerUserId}> can do this.`, + alreadyAnswered: 'Already answered or moved on.', // Agent-facing, not posted: a mention's earlier thread (null when Slack couldn't be read). threadStart: (thread: { name: string; text: string }[] | null, request: string) => { if (thread === null) { diff --git a/cloud/apps/connector/src/slack-thread-posts.ts b/cloud/apps/connector/src/slack-thread-posts.ts index da989e7479b..29ea1cdc0dc 100644 --- a/cloud/apps/connector/src/slack-thread-posts.ts +++ b/cloud/apps/connector/src/slack-thread-posts.ts @@ -1,12 +1,23 @@ import type { CloudDatabase, SqlRow } from '@orca-cloud/cloud-database' -import { ConnectorDeliveryResultSchema } from '@orca-cloud/connector-contract' +import { + ConnectorDeliveryResultSchema, + type ConnectorHelloRequest +} from '@orca-cloud/connector-contract' +import { knownHost } from './slack-machine-choice.js' +import { + answerDeliveryId, + ownerActionState, + pendingPrompt, + type OwnerActionState +} from './owner-controls.js' +import { refreshPromptCards, slackAnswerablePrompt } from './slack-prompt-cards.js' import type { SlackPlace, SlackPosts } from './slack-posts.js' import { escapeSlackText, slackText } from './slack-text.js' // A turn's last message is posted only if the agent hasn't replied itself by then. const FALLBACK_DELAY_MS = 5_000 -type BoundThread = SlackPlace & { hostName: string } +type BoundThread = SlackPlace & { hostName: string; hello: ConnectorHelloRequest | undefined } // A Slack thread bound to this host, with its machine's name; null for anyone else's conversation. async function boundThread( @@ -15,7 +26,7 @@ async function boundThread( externalId: string ): Promise<(BoundThread & { row: SqlRow }) | null> { const [row] = await transaction.query( - `SELECT t.*, h.host_name FROM connector_threads t + `SELECT t.*, h.host_name, h.hello FROM connector_threads t JOIN connector_hosts h ON h.host_id = t.host_id WHERE t.external_id = ? AND t.host_id = ?`, [externalId, hostId] @@ -26,6 +37,7 @@ async function boundThread( channelId: String(row.channel_id), threadTs: String(row.thread_ts), hostName: String(row.host_name), + hello: knownHost(row)?.hello, row } : null @@ -50,10 +62,18 @@ export async function refreshSlackStatus( ORDER BY seq DESC LIMIT 1`, [hostId, externalId] ) + const prompt = slackAnswerablePrompt(thread.row, thread.hello) + const place = { teamId: thread.teamId, channelId: thread.channelId, threadTs: thread.threadTs } + // The owner's own answer to the shown prompt, matched by its id. + const shown = pendingPrompt(thread.row) + const answer = shown + ? await ownerActionState(transaction, hostId, answerDeliveryId(externalId, shown.promptId)) + : null + const offered = await refreshPromptCards(transaction, posts, { ...place, externalId }, prompt) await posts.setStatus(transaction, { - ...thread, + ...place, postId: `status:${externalId}`, - text: statusText(thread.row, latest, thread.hostName) + text: statusText(thread.row, latest, thread.hostName, offered, answer) }) } @@ -114,17 +134,50 @@ export async function queueAgentReply( return await posts.find(transaction, postId) } -function statusText(thread: SqlRow, latest: SqlRow | undefined, hostName: string): string { - const result = - latest?.state === 'done' - ? ConnectorDeliveryResultSchema.safeParse(JSON.parse(String(latest.result))).data - : undefined +function resultOf(latest: SqlRow | undefined) { + return latest?.state === 'done' + ? ConnectorDeliveryResultSchema.safeParse(JSON.parse(String(latest.result))).data + : undefined +} + +// The newest delivery never reached the machine. +function unreached(latest: SqlRow | undefined): boolean { + return latest?.state === 'expired' || resultOf(latest)?.outcome === 'expired' +} + +// The owner's answer to the shown prompt: working while it is on its way, else its outcome. +function answerText(state: OwnerActionState, thread: SqlRow, hostName: string): string { + switch (state.kind) { + case 'on_its_way': + return slackText.status.working(hostName) + case 'refused': + return slackText.status.refused(hostName, state.reason) + case 'unknown': + return slackText.status.unknown + case 'unreached': + return slackText.status.offline(String(thread.owner_user_id), hostName) + } +} + +function statusText( + thread: SqlRow, + latest: SqlRow | undefined, + hostName: string, + answerableHere: boolean, + answer: OwnerActionState | null +): string { + const result = resultOf(latest) // The newest message never arriving outranks whatever the last turn left showing. - if (latest?.state === 'expired' || result?.outcome === 'expired') { + if (unreached(latest)) { return slackText.status.offline(String(thread.owner_user_id), hostName) } if (thread.pending_prompt !== null) { - return slackText.status.needsAnswer(hostName) + if (answer) { + return answerText(answer, thread, hostName) + } + return answerableHere + ? slackText.status.needsAnswerHere + : slackText.status.needsAnswer(hostName) } if (thread.turn_state === 'working') { return slackText.status.working(hostName) @@ -136,7 +189,7 @@ function statusText(thread: SqlRow, latest: SqlRow | undefined, hostName: string case 'refused': return slackText.status.refused(hostName, result.reason) case 'unsupported': - return slackText.status.refused(hostName, 'unsupported') + return slackText.status.unsupported(hostName) case 'unknown': return slackText.status.unknown }