diff --git a/.env.example b/.env.example index 12fac543c..7713b699c 100644 --- a/.env.example +++ b/.env.example @@ -57,6 +57,25 @@ GOOGLE_CLIENT_SECRET="" # SLACK_CLIENT_ID="" # SLACK_CLIENT_SECRET="" +# Optional, and required only for inbound Slack. It is the Signing Secret on the +# Slack app's Basic Information page, not a token. Without it the events +# endpoint refuses every request, which is the safe way round: that endpoint is +# a public POST, so the signature is the only thing between a stranger and +# resuming somebody's agent run. +# +# Point the Slack app's Event Subscriptions request URL at +# API_URL + /webhooks/slack/events +# SLACK_SIGNING_SECRET="" + +# Optional, and for local development only. The hostname `bun run tunnel:slack` +# gives your machine, on a domain in your own Cloudflare account. A named tunnel +# keeps the same hostname across restarts, so Slack's request URL is set once +# and never again; the quick `cloudflared --url` tunnel invents a new hostname +# every time and Slack silently stops delivering. +# +# Nothing reads this at runtime. The script reads it, and Slack remembers it. +# SLACK_TUNNEL_HOSTNAME="crm-dev.example.com" + # Which Entra tenant may sign in. "common" (the default) accepts any work, # school or personal Microsoft account and leans on ALLOWED_SIGN_IN to decide # who actually gets in; your own tenant's GUID refuses everyone else at diff --git a/apps/agent/agent/channels/crm.ts b/apps/agent/agent/channels/crm.ts index aff3af458..98c032a5d 100644 --- a/apps/agent/agent/channels/crm.ts +++ b/apps/agent/agent/channels/crm.ts @@ -30,6 +30,7 @@ import { DISPATCH } from "../lib/dispatch-config"; import { settle } from "../lib/enrichment"; import { finishRun, runResultOf } from "../lib/run-runtime"; import { attribute } from "../lib/session-purpose"; +import { drainSlackEvents } from "../lib/slack-events"; import { createSlackChannel } from "../lib/slack-membership"; import { reconcileStaleTasks } from "../lib/stale-tasks"; import { completeTask, taskSubject } from "../lib/tasks"; @@ -146,6 +147,7 @@ export default defineChannel({ }), ); await drainAgentRuns(send); + await drainSlackEvents(send); })(), ); diff --git a/apps/agent/agent/lib/agent-actions.ts b/apps/agent/agent/lib/agent-actions.ts index 8f403b4ac..cee011441 100644 --- a/apps/agent/agent/lib/agent-actions.ts +++ b/apps/agent/agent/lib/agent-actions.ts @@ -1,12 +1,15 @@ import { AGENT_ACTION_TYPES, type AgentActionType, + SLACK_WORKSPACE_RESOURCE_ID, } from "@crm/validation/agent-manifest"; export const AGENT_ACTION_EXECUTORS = { [AGENT_ACTION_TYPES.CRM_ACTIVITY_CREATE]: "create_crm_activity", [AGENT_ACTION_TYPES.RUN_SUMMARY]: "finish_run", [AGENT_ACTION_TYPES.SLACK_MESSAGE_POST]: "post_slack_message", + [AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN]: "open_slack_channel", + [AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE]: "invite_to_slack_channel", } as const satisfies Record; export function isAgentActionType(value: string): value is AgentActionType { @@ -22,15 +25,19 @@ export type AgentActionDependency = { readonly fix: string; }; +const SLACK_DEPENDENCY = { + id: "slack", + label: "Slack", + resourceId: SLACK_WORKSPACE_RESOURCE_ID, + fix: "Connect Slack in Settings → Connections.", +} as const satisfies AgentActionDependency; + export const AGENT_ACTION_DEPENDENCIES = { [AGENT_ACTION_TYPES.CRM_ACTIVITY_CREATE]: null, [AGENT_ACTION_TYPES.RUN_SUMMARY]: null, - [AGENT_ACTION_TYPES.SLACK_MESSAGE_POST]: { - id: "slack", - label: "Slack", - resourceId: "slack:workspace", - fix: "Connect Slack in Settings → Connections.", - }, + [AGENT_ACTION_TYPES.SLACK_MESSAGE_POST]: SLACK_DEPENDENCY, + [AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN]: SLACK_DEPENDENCY, + [AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE]: SLACK_DEPENDENCY, } as const satisfies Record; export function actionDependency( diff --git a/apps/agent/agent/lib/builder-runtime.ts b/apps/agent/agent/lib/builder-runtime.ts index 4a2a2f9a8..f6449315a 100644 --- a/apps/agent/agent/lib/builder-runtime.ts +++ b/apps/agent/agent/lib/builder-runtime.ts @@ -9,6 +9,7 @@ import { readAgentModel } from "@crm/db/settings"; import { WORKSPACE_ID } from "@crm/db/workspace"; import { AGENT_ACTION_TYPES } from "@crm/validation/agent-manifest"; import { z } from "zod"; +import type { DraftAction } from "../subagents/agent_builder/lib/draft-input"; import { actionDependency } from "./agent-actions"; import { requestStaleSlackInventorySync } from "./slack-people"; @@ -43,29 +44,7 @@ export type DraftTrigger = { intervalMinutes?: number | null; }; -export type DraftAction = - | { - type: typeof AGENT_ACTION_TYPES.CRM_ACTIVITY_CREATE; - provider: "crm"; - summary: string; - activityTypes: ("NOTE" | "TASK")[]; - } - | { - type: typeof AGENT_ACTION_TYPES.RUN_SUMMARY; - provider: "crm"; - summary: string; - } - | { - type: typeof AGENT_ACTION_TYPES.SLACK_MESSAGE_POST; - provider: "slack"; - summary: string; - destination: { - kind: "channel" | "user"; - resolution: "chosen"; - id: string; - label: string; - }; - }; +export type { DraftAction }; export type DraftAgentInput = { name: string; diff --git a/apps/agent/agent/lib/run-resume.ts b/apps/agent/agent/lib/run-resume.ts new file mode 100644 index 000000000..9c21f5f9f --- /dev/null +++ b/apps/agent/agent/lib/run-resume.ts @@ -0,0 +1,130 @@ +import { db } from "@crm/db"; +import type { SendFn } from "eve/channels"; +import { APP_AUTH } from "./app-auth"; +import { runToken } from "./custom-agent-dispatch"; + +const LIVE_STATUSES = ["QUEUED", "RUNNING", "WAITING_FOR_APPROVAL"] as const; + +export type ResumeOutcome = + | { kind: "resumed"; runId: string; sessionId: string } + | { kind: "ignored"; runId: string; reason: string }; + +export type ResumeInput = { + runId: string; + message: string; + source: string; + attributes?: Readonly>; +}; + +export async function resumeAgentRun( + input: ResumeInput, + send: SendFn, +): Promise { + const { runId, message, source } = input; + + const run = await db.agentRun.findUnique({ + where: { id: runId }, + select: { + id: true, + status: true, + sessionId: true, + agentId: true, + versionId: true, + agent: { select: { status: true, name: true } }, + }, + }); + + if (!run) return { kind: "ignored", runId, reason: "no such run" }; + + if (!(LIVE_STATUSES as readonly string[]).includes(run.status)) { + return { + kind: "ignored", + runId, + reason: `the run is ${run.status.toLowerCase()}`, + }; + } + + if (run.agent.status !== "LIVE") { + return { + kind: "ignored", + runId, + reason: `the agent is ${run.agent.status.toLowerCase()}`, + }; + } + + if (!run.sessionId) { + return { + kind: "ignored", + runId, + reason: "the run has not started a session yet", + }; + } + + try { + const session = await send(message, { + auth: { + authenticator: APP_AUTH.authenticator, + principalType: APP_AUTH.principalType, + principalId: APP_AUTH.principalId, + attributes: { + purpose: "team-agent", + runId: run.id, + agentId: run.agentId, + versionId: run.versionId, + resumeSource: source, + ...input.attributes, + }, + }, + continuationToken: runToken(run.id), + title: `${run.agent.name} run`, + mode: "task", + }); + + return { kind: "resumed", runId, sessionId: session.id }; + } catch (error) { + return { + kind: "ignored", + runId, + reason: error instanceof Error ? error.message : String(error), + }; + } +} + +export async function runOnSlackChannel( + channelId: string, +): Promise { + const trimmed = channelId.trim(); + if (!trimmed) return null; + + const run = await db.agentRun.findFirst({ + where: { + slackChannelId: trimmed, + status: { in: [...LIVE_STATUSES] }, + }, + orderBy: { createdAt: "desc" }, + select: { id: true }, + }); + + return run?.id ?? null; +} + +export async function channelOfRun(runId: string): Promise { + const run = await db.agentRun.findUnique({ + where: { id: runId }, + select: { slackChannelId: true }, + }); + + return run?.slackChannelId ?? null; +} + +export async function claimSlackChannel( + runId: string, + channelId: string, +): Promise { + await db.agentRun.updateMany({ + where: { id: runId, slackChannelId: null }, + data: { slackChannelId: channelId.trim() }, + }); + + return channelOfRun(runId); +} diff --git a/apps/agent/agent/lib/run-runtime.ts b/apps/agent/agent/lib/run-runtime.ts index bc9d9c2f8..ee2840ae4 100644 --- a/apps/agent/agent/lib/run-runtime.ts +++ b/apps/agent/agent/lib/run-runtime.ts @@ -2,10 +2,16 @@ import { createHash, randomUUID } from "node:crypto"; import { ActivityType, db, type Prisma } from "@crm/db"; import type { AgentActionStatus, AgentTriggerType } from "@crm/db/enums"; import { lockIdempotencyKey } from "@crm/db/idempotency"; +import { + type AgentActionResult, + parseAgentActionResult, + readAgentActionResult, +} from "@crm/validation/agent-action"; import { AGENT_ACTION_TYPES, type AgentManifestResource, parseAgentManifest, + SLACK_WORKSPACE_RESOURCE_ID, } from "@crm/validation/agent-manifest"; import { z } from "zod"; import { readCompanyHistory, readDealHistory } from "./accounts"; @@ -13,12 +19,17 @@ import { AGENT_ACTION_EXECUTORS, isAgentActionType } from "./agent-actions"; import { readCrmHistory } from "./crm"; import { DISPATCH } from "./dispatch-config"; import { searchCrm } from "./lookup"; +import { channelOfRun, claimSlackChannel } from "./run-resume"; import { type LockedAgentRun, lockAgentRun, runTerminalEventId, } from "./run-state"; +import { toChannelName } from "./slack-channel-name"; import { slackAccessToken } from "./slack-connection"; +import { type InviteOutcome, inviteToSlackChannel } from "./slack-invite"; +import { createSlackChannel } from "./slack-membership"; +import { addDealOwner } from "./slack-owner"; const ACTION_LEASE_MS = DISPATCH.run.actionLeaseMs; const NO_ACTION_TRIGGER_TYPES = new Set( @@ -86,14 +97,21 @@ const noActionResult = z type RunActionRow = { id: string; + type: string; status: AgentActionStatus; externalId: string | null; requestHash: string | null; metadata: Prisma.JsonValue; + result: Prisma.JsonValue | null; }; type RunActionClaim = - | { claimed: false; actionId: string; externalId: string | null } + | { + claimed: false; + actionId: string; + externalId: string | null; + result: AgentActionResult | null; + } | { claimed: true; actionId: string; @@ -103,12 +121,41 @@ type RunActionClaim = const RUN_ACTION_FIELDS = { id: true, + type: true, status: true, externalId: true, requestHash: true, metadata: true, + result: true, } as const; +function storedActionResult(result: AgentActionResult): Prisma.InputJsonValue { + return parseAgentActionResult( + JSON.parse(JSON.stringify(result)), + ) as Prisma.InputJsonValue; +} + +function slackInviteResult(channelId: string, outcomes: InviteOutcome[]) { + const invited: Extract[] = []; + for (const outcome of outcomes) { + if (outcome.invited) invited.push(outcome); + } + const chosen = invited.find((outcome) => outcome.invite_id) ?? invited[0]; + if (!chosen) { + throw new Error("Slack didn't send an invite we could save."); + } + return { + externalId: chosen.invite_id ?? channelId, + result: parseAgentActionResult({ + type: AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE, + email: chosen.email, + kind: chosen.kind, + invite_id: chosen.invite_id, + url: chosen.url, + }), + }; +} + export async function approvedRunInstructions(runId: string): Promise { const run = await db.agentRun.findUnique({ where: { id: runId }, @@ -266,6 +313,7 @@ export async function createRunActivity( return { actionId: existing.id, activityId: existing.externalId, + result: readAgentActionResult(existing.type, existing.result), replayed: true, }; } @@ -302,6 +350,7 @@ export async function createRunActivity( return { actionId: claim.actionId, activityId: claim.externalId, + result: claim.result, replayed: true, }; } @@ -309,6 +358,10 @@ export async function createRunActivity( try { const activityId = `agent-action-${claim.actionId}`; const now = new Date(); + const result = parseAgentActionResult({ + type: AGENT_ACTION_TYPES.CRM_ACTIVITY_CREATE, + activityId, + }); await db.$transaction(async (tx) => { const activeRun = await lockAgentRun(tx, runId); @@ -362,12 +415,18 @@ export async function createRunActivity( data: { status: "SUCCEEDED", externalId: activityId, + result: storedActionResult(result), completedAt: now, }, }); }); - return { actionId: claim.actionId, activityId, replayed: false }; + return { + actionId: claim.actionId, + activityId, + result, + replayed: false, + }; } catch (error) { const message = error instanceof Error ? error.message : String(error); await failRunAction(claim, "ACTION_REJECTED", message); @@ -403,6 +462,7 @@ export async function postRunSlackMessage( actionId: existing.id, messageId: existing.externalId, destination: destination.label, + result: readAgentActionResult(existing.type, existing.result), replayed: true, }; } @@ -426,6 +486,7 @@ export async function postRunSlackMessage( actionId: claim.actionId, messageId: claim.externalId, destination: destination.label, + result: claim.result, replayed: true, }; } @@ -450,17 +511,23 @@ export async function postRunSlackMessage( beforePost: () => holdRunActionClaim(runId, actionId, claimedAt), }, ); + const result = parseAgentActionResult({ + type: AGENT_ACTION_TYPES.SLACK_MESSAGE_POST, + channel: posted.channel, + ts: posted.ts, + }); const messageId = `${posted.channel}:${posted.ts}`; const completed = await db.agentAction.updateMany({ where: { id: actionId, status: "RUNNING", startedAt: claimedAt }, data: { status: "SUCCEEDED", externalId: messageId, + result: storedActionResult(result), completedAt: new Date(), }, }); if (completed.count === 0) { - await recordDeliveryOutsideClaim(actionId, messageId); + await recordDeliveryOutsideClaim(actionId, messageId, result); throw new Error( "This agent run stopped while Slack was accepting the message.", ); @@ -470,6 +537,7 @@ export async function postRunSlackMessage( actionId, messageId, destination: destination.label, + result, replayed: false, }; } catch (error) { @@ -523,6 +591,7 @@ async function claimRunAction( claimed: false, actionId: action.id, externalId: action.externalId, + result: readAgentActionResult(action.type, action.result), }; } @@ -550,13 +619,14 @@ async function claimRunAction( if (claimed.count === 0) { const current = await db.agentAction.findUnique({ where: { id: action.id }, - select: { status: true, externalId: true }, + select: { status: true, externalId: true, result: true }, }); if (current?.status === "SUCCEEDED") { return { claimed: false, actionId: action.id, externalId: current.externalId, + result: readAgentActionResult(action.type, current.result), }; } throw new Error("This agent action is already in progress."); @@ -671,7 +741,8 @@ async function holdRunActionClaim( async function recordDeliveryOutsideClaim( actionId: string, - messageId: string, + externalId: string, + result: AgentActionResult, ): Promise { const delivered = "Slack accepted this message before the run stopped, and it cannot be withdrawn."; @@ -684,7 +755,8 @@ async function recordDeliveryOutsideClaim( await db.agentAction.updateMany({ where: { id: actionId, status: { not: "SUCCEEDED" }, externalId: null }, data: { - externalId: messageId, + externalId, + result: storedActionResult(result), errorMessage: current.errorMessage ? `${current.errorMessage} ${delivered}` : delivered, @@ -716,7 +788,7 @@ async function slackApiRequest( const reason = envelope.error ?? "rejected"; if (reason === "not_in_channel") { throw new Error( - "The Slack bot is not in the selected channel. Invite the app to that channel and retry the run.", + "Comp AI isn't in that channel. Invite it there, then try again.", ); } if (reason === "missing_scope") { @@ -1021,18 +1093,232 @@ function assertActivityAllowed( } } +function assertSlackActionApproved( + manifest: Prisma.JsonValue, + type: (typeof AGENT_ACTION_TYPES)[keyof typeof AGENT_ACTION_TYPES], +): void { + if (!manifestActions(manifest).some((action) => action.type === type)) { + throw new Error(`Agent version does not allow ${type}.`); + } +} + +export async function openRunSlackChannel( + runId: string, + callId: string, + input: { name: string; isPrivate: boolean }, +) { + const run = await activeRunForSlack( + runId, + AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN, + ); + + const channelName = toChannelName(input.name); + if (!channelName) { + throw new Error("That name has no letters or numbers Slack accepts."); + } + + const idempotencyKey = `${runId}:${callId}`; + const requestHash = hashRequest({ channelName, isPrivate: input.isPrivate }); + const existing = await findRunAction(idempotencyKey, requestHash); + if (existing?.status === "SUCCEEDED") { + return { + actionId: existing.id, + channelId: existing.externalId, + channelName, + watching: (await channelOfRun(runId)) === existing.externalId, + result: readAgentActionResult(existing.type, existing.result), + replayed: true, + }; + } + + const claim = await claimRunAction(existing, idempotencyKey, requestHash, { + agentId: run.agentId, + runId, + type: AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN, + provider: "slack", + targetType: "channel", + targetId: channelName, + targetLabel: `#${channelName}`, + summary: `Open #${channelName}`, + }); + if (!claim.claimed) { + return { + actionId: claim.actionId, + channelId: claim.externalId, + channelName, + watching: (await channelOfRun(runId)) === claim.externalId, + result: claim.result, + replayed: true, + }; + } + + try { + await assertRunActive(runId); + const outcome = await createSlackChannel(channelName, input.isPrivate); + if ("error" in outcome) throw new Error(outcome.error); + + const watching = await claimSlackChannel(runId, outcome.id); + const owner = await addDealOwner(runId, outcome.id).catch(() => null); + const result = parseAgentActionResult({ + type: AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN, + channelId: outcome.id, + }); + await settleRunAction(claim, outcome.id, result); + + return { + actionId: claim.actionId, + channelId: outcome.id, + channelName: outcome.name, + watching: watching === outcome.id, + owner, + result, + replayed: false, + }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + await failRunAction(claim, slackActionErrorCode(message), message); + throw error; + } +} + +export async function inviteToRunSlackChannel( + runId: string, + callId: string, + input: { emails: string[] }, +) { + const run = await activeRunForSlack( + runId, + AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE, + ); + + const channelId = await channelOfRun(runId); + if (!channelId) { + throw new Error( + "This run has no Slack channel yet. Open one with open_slack_channel first.", + ); + } + + const emails = [...new Set(input.emails.map((email) => email.trim()))].sort(); + const idempotencyKey = `${runId}:${callId}`; + const requestHash = hashRequest({ channelId, emails: emails.join(",") }); + const existing = await findRunAction(idempotencyKey, requestHash); + if (existing?.status === "SUCCEEDED") { + return { + actionId: existing.id, + channelId, + result: readAgentActionResult(existing.type, existing.result), + replayed: true, + }; + } + + const claim = await claimRunAction(existing, idempotencyKey, requestHash, { + agentId: run.agentId, + runId, + type: AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE, + provider: "slack", + targetType: "channel", + targetId: channelId, + targetLabel: channelId, + summary: + emails.length === 1 + ? "Invited 1 person" + : `Invited ${emails.length} people`, + }); + if (!claim.claimed) { + return { + actionId: claim.actionId, + channelId, + result: claim.result, + replayed: true, + }; + } + + try { + await assertRunActive(runId); + const outcomes = []; + for (const email of emails) { + outcomes.push(await inviteToSlackChannel(channelId, email)); + } + + const invited = outcomes.filter((outcome) => outcome.invited); + const refused = outcomes.filter((outcome) => !outcome.invited); + if (invited.length === 0) { + throw new Error( + refused.map((outcome) => outcome.reason).join(" ") || + "Slack refused every invitation.", + ); + } + + const stored = slackInviteResult(channelId, outcomes); + await settleRunAction(claim, stored.externalId, stored.result); + + return { + actionId: claim.actionId, + channelId, + result: stored.result, + invited, + refused, + replayed: false, + }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + await failRunAction(claim, slackActionErrorCode(message), message); + throw error; + } +} + +async function activeRunForSlack( + runId: string, + type: (typeof AGENT_ACTION_TYPES)[keyof typeof AGENT_ACTION_TYPES], +) { + const run = await db.agentRun.findUnique({ + where: { id: runId }, + select: { + id: true, + status: true, + agentId: true, + version: { select: { manifest: true } }, + }, + }); + if (!run) throw new Error("This agent run is unavailable."); + if (run.status !== "RUNNING") { + throw new Error("This agent run is not active."); + } + + assertSlackWorkspaceApproved(run.version.manifest); + assertSlackActionApproved(run.version.manifest, type); + + return run; +} + +async function settleRunAction( + claim: Extract, + externalId: string, + result: AgentActionResult, +): Promise { + const completed = await db.agentAction.updateMany({ + where: { + id: claim.actionId, + status: "RUNNING", + startedAt: claim.claimedAt, + }, + data: { + status: "SUCCEEDED", + externalId, + result: storedActionResult(result), + completedAt: new Date(), + }, + }); + if (completed.count === 0) { + await recordDeliveryOutsideClaim(claim.actionId, externalId, result); + throw new Error("This agent run stopped while Slack was still working."); + } +} + export function approvedSlackDestination( manifest: Prisma.JsonValue, ): SlackRunDestination { - const scope = manifestDataScope(manifest); - if ( - !scope.resources.some( - (resource) => - resource.kind === "integration" && resource.id === "slack:workspace", - ) - ) { - throw new Error("Agent version does not allow Slack."); - } + assertSlackWorkspaceApproved(manifest); const destinations = manifestActions(manifest).flatMap((action) => action.type === AGENT_ACTION_TYPES.SLACK_MESSAGE_POST @@ -1055,6 +1341,19 @@ export function approvedSlackDestination( return destination; } +function assertSlackWorkspaceApproved(manifest: Prisma.JsonValue): void { + const scope = manifestDataScope(manifest); + if ( + !scope.resources.some( + (resource) => + resource.kind === "integration" && + resource.id === SLACK_WORKSPACE_RESOURCE_ID, + ) + ) { + throw new Error("Agent version does not allow Slack."); + } +} + function assertResourceAllowed( mode: RunRecordScope, resources: AgentManifestResource[], diff --git a/apps/agent/agent/lib/slack-api.ts b/apps/agent/agent/lib/slack-api.ts new file mode 100644 index 000000000..3d1f2fdd2 --- /dev/null +++ b/apps/agent/agent/lib/slack-api.ts @@ -0,0 +1,95 @@ +import type { z } from "zod"; +import { SLACK } from "./slack-config"; + +type Reply = { ok: boolean; error?: string }; + +export type SlackPostBody = { + channel?: string; + users?: string; + emails?: string[]; + external_limited?: boolean; + name?: string; + is_private?: boolean; +}; + +export type SlackOutcome = + | { ok: true; data: T } + | { ok: false; error: string }; + +const UNREADABLE = "unreadable_reply"; +const RATELIMITED = "ratelimited"; +const REJECTED = "rejected"; + +async function read( + response: Response, + schema: z.ZodType, + again: () => Promise>, + attempt: number, +): Promise> { + const parsed = schema.safeParse(await response.json().catch(() => null)); + if (!parsed.success) return { ok: false, error: UNREADABLE }; + if (parsed.data.ok) return { ok: true, data: parsed.data }; + + if ( + parsed.data.error === RATELIMITED && + attempt < SLACK.request.maxAttempts + ) { + const wait = Number(response.headers.get("retry-after") ?? "1"); + await new Promise((resolve) => + setTimeout(resolve, wait * SLACK.request.retryUnitMs), + ); + return again(); + } + + return { ok: false, error: parsed.data.error ?? REJECTED }; +} + +export async function slackPost( + token: string, + method: string, + body: SlackPostBody, + schema: z.ZodType, + attempt = 1, +): Promise> { + const response = await fetch(`https://slack.com/api/${method}`, { + method: "POST", + headers: { + authorization: `Bearer ${token}`, + "content-type": "application/json; charset=utf-8", + }, + body: JSON.stringify(body), + signal: AbortSignal.timeout(SLACK.request.timeoutMs), + }); + + return read( + response, + schema, + () => slackPost(token, method, body, schema, attempt + 1), + attempt, + ); +} + +export async function slackGet( + token: string, + method: string, + query: Record, + schema: z.ZodType, + attempt = 1, +): Promise> { + const url = new URL(`https://slack.com/api/${method}`); + for (const [key, value] of Object.entries(query)) { + url.searchParams.set(key, value); + } + + const response = await fetch(url, { + headers: { authorization: `Bearer ${token}` }, + signal: AbortSignal.timeout(SLACK.request.timeoutMs), + }); + + return read( + response, + schema, + () => slackGet(token, method, query, schema, attempt + 1), + attempt, + ); +} diff --git a/apps/agent/agent/lib/slack-channel-name.ts b/apps/agent/agent/lib/slack-channel-name.ts new file mode 100644 index 000000000..e85530620 --- /dev/null +++ b/apps/agent/agent/lib/slack-channel-name.ts @@ -0,0 +1,12 @@ +import { SLACK } from "./slack-config"; + +export function toChannelName(value: string): string { + return value + .trim() + .toLowerCase() + .replace(/[^a-z0-9]+/g, "-") + .replace(/-+/g, "-") + .replace(/^-|-$/g, "") + .slice(0, SLACK.channel.maxNameChars) + .replace(/-$/, ""); +} diff --git a/apps/agent/agent/lib/slack-config.ts b/apps/agent/agent/lib/slack-config.ts index d7e687c32..0b726ee90 100644 --- a/apps/agent/agent/lib/slack-config.ts +++ b/apps/agent/agent/lib/slack-config.ts @@ -8,6 +8,10 @@ export const SLACK = { retryUnitMs: SECOND_MS, }, + channel: { + maxNameChars: 80, + }, + inventory: { pageSize: 200, channelTypes: "public_channel,private_channel", diff --git a/apps/agent/agent/lib/slack-events-config.ts b/apps/agent/agent/lib/slack-events-config.ts new file mode 100644 index 000000000..331be05fd --- /dev/null +++ b/apps/agent/agent/lib/slack-events-config.ts @@ -0,0 +1,8 @@ +const SECOND_MS = 1_000; +const MINUTE_MS = 60 * SECOND_MS; + +export const SLACK_EVENTS = { + batch: 20, + maxTextChars: 2_000, + leaseMs: MINUTE_MS, +} as const; diff --git a/apps/agent/agent/lib/slack-events.ts b/apps/agent/agent/lib/slack-events.ts new file mode 100644 index 000000000..5aa663f68 --- /dev/null +++ b/apps/agent/agent/lib/slack-events.ts @@ -0,0 +1,153 @@ +import { db, type Prisma } from "@crm/db"; +import type { SlackEvent } from "@crm/validation"; +import { schemas } from "@crm/validation"; +import { SLACK_EVENT_TYPES } from "@crm/validation/slack-events"; +import type { SendFn } from "eve/channels"; +import { resumeAgentRun, runOnSlackChannel } from "./run-resume"; +import { SLACK_EVENTS } from "./slack-events-config"; + +export type SlackEventOutcome = { + eventId: string; + resumed: boolean; + outcome: string; +}; + +type ClaimedSlackEvent = { + id: string; + eventId: string; + channelId: string | null; + payload: Prisma.JsonValue; +}; + +export async function pendingSlackEventIds(): Promise { + const now = new Date(); + const rows = await db.slackEventInbox.findMany({ + where: { + processedAt: null, + OR: [{ leasedUntil: null }, { leasedUntil: { lt: now } }], + }, + orderBy: { receivedAt: "asc" }, + take: SLACK_EVENTS.batch, + select: { id: true }, + }); + + return rows.map((row) => row.id); +} + +async function claimSlackEvent(id: string): Promise { + const now = new Date(); + const until = new Date(now.getTime() + SLACK_EVENTS.leaseMs); + + const claimed = await db.$queryRaw` + UPDATE "slackEventInbox" AS t + SET "leasedUntil" = ${until} + FROM ( + SELECT t2.id FROM "slackEventInbox" AS t2 + WHERE t2.id = ${id} + AND t2."processedAt" IS NULL + AND (t2."leasedUntil" IS NULL OR t2."leasedUntil" < ${now}) + FOR UPDATE SKIP LOCKED + ) AS due + WHERE t.id = due.id + RETURNING t.id, t."eventId", t."channelId", t.payload; + `; + + return claimed[0] ?? null; +} + +export async function dispatchSlackEvent( + id: string, + send: SendFn, +): Promise { + const row = await claimSlackEvent(id); + + if (!row) return null; + + const settle = (outcome: string, resumed = false) => + db.slackEventInbox + .updateMany({ + where: { id: row.id, processedAt: null }, + data: { processedAt: new Date(), outcome: outcome.slice(0, 300) }, + }) + .then(() => ({ eventId: row.eventId, resumed, outcome })); + + if (!row.channelId) return settle("The event names no channel."); + + const envelope = schemas.slackEvents.eventCallback.safeParse(row.payload); + if (!envelope.success) return settle("The stored payload cannot be read."); + + const runId = await runOnSlackChannel(row.channelId); + if (!runId) { + return settle(`No live agent run owns ${row.channelId}.`); + } + + const result = await resumeAgentRun( + { + runId, + message: describe(envelope.data.event), + source: `slack.${envelope.data.event.type}`, + attributes: { + slackChannelId: row.channelId, + slackEventId: row.eventId, + }, + }, + send, + ); + + if (result.kind === "resumed") { + await db.slackEventInbox.updateMany({ + where: { id: row.id }, + data: { runId }, + }); + return settle(`Resumed run ${runId}.`, true); + } + + return settle(`Run ${runId} was not resumed: ${result.reason}`); +} + +export async function drainSlackEvents(send: SendFn): Promise { + const ids = await pendingSlackEventIds(); + + const outcomes = await Promise.all( + ids.map((id) => + dispatchSlackEvent(id, send).catch((error) => { + console.error( + `[agent] Slack event ${id} could not be dispatched: ${ + error instanceof Error ? error.message : String(error) + }`, + ); + return null; + }), + ), + ); + + return outcomes.filter((outcome) => outcome?.resumed).length; +} + +export function describe(event: SlackEvent): string { + if (event.type === SLACK_EVENT_TYPES.MEMBER_JOINED) { + return [ + `Somebody joined the Slack channel this run is working in (${event.channel}).`, + event.user ? `Their Slack user id is ${event.user}.` : "", + "Carry on from where you parked.", + ] + .filter(Boolean) + .join(" "); + } + + const text = event.text?.trim() ?? ""; + + if (event.type === SLACK_EVENT_TYPES.APP_MENTION) { + return [ + `Somebody mentioned Comp AI in the Slack channel this run is working in (${event.channel})`, + event.user ? ` from ${event.user}` : "", + `: ${text.slice(0, SLACK_EVENTS.maxTextChars)}`, + ].join(""); + } + + return [ + `A message arrived in the Slack channel this run is working in (${event.channel})`, + event.user ? ` from ${event.user}` : "", + `: ${text.slice(0, SLACK_EVENTS.maxTextChars)}`, + ].join(""); +} diff --git a/apps/agent/agent/lib/slack-invite.ts b/apps/agent/agent/lib/slack-invite.ts new file mode 100644 index 000000000..7ded7d1c4 --- /dev/null +++ b/apps/agent/agent/lib/slack-invite.ts @@ -0,0 +1,115 @@ +import { schemas } from "@crm/validation"; +import { slackGet, slackPost } from "./slack-api"; +import { slackAccessToken } from "./slack-connection"; + +export type InviteOutcome = + | { + invited: true; + email: string; + kind: "member" | "connect"; + invite_id?: string; + url?: string; + } + | { invited: false; email: string; reason: string }; + +const ALREADY_IN_CHANNEL = "already_in_channel"; +const USERS_NOT_FOUND = "users_not_found"; + +export async function inviteToSlackChannel( + channelId: string, + email: string, +): Promise { + const token = await slackAccessToken(); + if (!token) { + return { invited: false, email, reason: "Slack is not connected." }; + } + + const found = await slackGet( + token, + "users.lookupByEmail", + { email }, + schemas.slack.lookupByEmail, + ); + + if (found.ok && found.data.user) { + return inviteMember(token, channelId, email, found.data.user.id); + } + + if (!found.ok && found.error !== USERS_NOT_FOUND) { + return { invited: false, email, reason: explain(found.error) }; + } + + return inviteGuest(token, channelId, email); +} + +async function inviteMember( + token: string, + channelId: string, + email: string, + userId: string, +): Promise { + const outcome = await slackPost( + token, + "conversations.invite", + { channel: channelId, users: userId }, + schemas.slack.reply, + ); + + if (outcome.ok || outcome.error === ALREADY_IN_CHANNEL) { + return { invited: true, email, kind: "member" }; + } + + return { invited: false, email, reason: explain(outcome.error) }; +} + +async function inviteGuest( + token: string, + channelId: string, + email: string, +): Promise { + const outcome = await slackPost( + token, + "conversations.inviteShared", + { channel: channelId, emails: [email], external_limited: false }, + schemas.slack.inviteShared, + ); + + if (outcome.ok) { + return { + invited: true, + email, + kind: "connect", + invite_id: outcome.data.invite_id, + url: outcome.data.url, + }; + } + + if (outcome.error === ALREADY_IN_CHANNEL) { + return { invited: true, email, kind: "connect" }; + } + + return { invited: false, email, reason: explain(outcome.error) }; +} + +function explain(error: string): string { + switch (error) { + case "not_in_channel": + return "Comp AI is not in that channel, so it cannot invite anybody."; + case "channel_not_found": + return "Slack cannot see that channel."; + case "invalid_email": + return "Slack refused that address."; + case "cannot_invite_self": + return "That address is Comp AI itself."; + case "missing_scope": + case "restricted_action": + return "This workspace doesn't let Comp AI send that invitation."; + case "org_level_email_not_allowed": + return "This workspace blocks Slack Connect invitations to that address."; + case "invalid_auth": + case "token_revoked": + return "Slack needs to be reconnected."; + default: + return `Slack refused the invitation (${error}).`; + } +} diff --git a/apps/agent/agent/lib/slack-membership.ts b/apps/agent/agent/lib/slack-membership.ts index c776d5c3d..8f174060d 100644 --- a/apps/agent/agent/lib/slack-membership.ts +++ b/apps/agent/agent/lib/slack-membership.ts @@ -1,7 +1,7 @@ import { db } from "@crm/db"; import { schemas } from "@crm/validation"; import { z } from "zod"; -import { SLACK } from "./slack-config"; +import { slackGet, slackPost } from "./slack-api"; import { slackAccessToken, slackUserToken } from "./slack-connection"; import { requestSlackInventorySync } from "./slack-people"; @@ -15,6 +15,8 @@ const ALREADY_IN_CHANNEL = "already_in_channel"; const CHANNEL_NOT_FOUND = "channel_not_found"; +const NAME_TAKEN = "name_taken"; + const channelInfo = schemas.slack.reply.extend({ channel: z .object({ @@ -28,70 +30,42 @@ async function call( token: string, method: string, body: Record, - attempt = 1, ): Promise<{ ok: boolean; error?: string }> { - const response = await fetch(`https://slack.com/api/${method}`, { - method: "POST", - headers: { - authorization: `Bearer ${token}`, - "content-type": "application/json; charset=utf-8", - }, - body: JSON.stringify(body), - signal: AbortSignal.timeout(SLACK.request.timeoutMs), - }); - - const parsed = schemas.slack.reply.safeParse(await response.json()); - if (!parsed.success) return { ok: false, error: "unreadable_reply" }; - if (parsed.data.ok) return { ok: true }; - - if ( - parsed.data.error === "ratelimited" && - attempt < SLACK.request.maxAttempts - ) { - const wait = Number(response.headers.get("retry-after") ?? "1"); - await new Promise((resolve) => - setTimeout(resolve, wait * SLACK.request.retryUnitMs), - ); - return call(token, method, body, attempt + 1); - } - - return { ok: false, error: parsed.data.error }; + const outcome = await slackPost(token, method, body, schemas.slack.reply); + return outcome.ok ? { ok: true } : { ok: false, error: outcome.error }; } async function botUserId(token: string): Promise { - const response = await fetch("https://slack.com/api/auth.test", { - headers: { authorization: `Bearer ${token}` }, - signal: AbortSignal.timeout(SLACK.request.timeoutMs), - }); - const parsed = schemas.slack.authTest.safeParse(await response.json()); - return parsed.success && parsed.data.ok - ? (parsed.data.user_id ?? null) - : null; + const outcome = await slackGet( + token, + "auth.test", + {}, + schemas.slack.authTest, + ); + + return outcome.ok ? (outcome.data.user_id ?? null) : null; } async function liveChannelState( token: string, channelId: string, ): Promise { - const url = new URL("https://slack.com/api/conversations.info"); - url.searchParams.set("channel", channelId); - try { - const response = await fetch(url, { - headers: { authorization: `Bearer ${token}` }, - signal: AbortSignal.timeout(SLACK.request.timeoutMs), - }); - const parsed = channelInfo.safeParse(await response.json()); - if (!parsed.success) return null; - - if (parsed.data.ok && parsed.data.channel) { + const outcome = await slackGet( + token, + "conversations.info", + { channel: channelId }, + channelInfo, + ); + + if (outcome.ok && outcome.data.channel) { return { - isPrivate: parsed.data.channel.is_private ?? false, - isMember: parsed.data.channel.is_member ?? false, + isPrivate: outcome.data.channel.is_private ?? false, + isMember: outcome.data.channel.is_member ?? false, }; } - return parsed.data.error === CHANNEL_NOT_FOUND + return !outcome.ok && outcome.error === CHANNEL_NOT_FOUND ? { isPrivate: true, isMember: false } : null; } catch { @@ -222,6 +196,23 @@ function explain(error: string): string { } } +async function channelNamed( + name: string, +): Promise<{ id: string; name: string } | { error: string }> { + const channel = await db.slackChannel.findFirst({ + where: { name }, + select: { id: true, name: true }, + }); + + if (channel) return channel; + + await requestSlackInventorySync(); + + return { + error: `#${name} already exists in Slack, and Comp AI cannot see it yet.`, + }; +} + export async function createSlackChannel( name: string, isPrivate: boolean, @@ -238,25 +229,23 @@ export async function createSlackChannel( }; } - const response = await fetch("https://slack.com/api/conversations.create", { - method: "POST", - headers: { - authorization: `Bearer ${token}`, - "content-type": "application/json; charset=utf-8", - }, - body: JSON.stringify({ name, is_private: isPrivate }), - signal: AbortSignal.timeout(SLACK.request.timeoutMs), - }); + const outcome = await slackPost( + token, + "conversations.create", + { name, is_private: isPrivate }, + schemas.slack.createReply, + ); - const parsed = schemas.slack.createReply.safeParse(await response.json()); - if (!parsed.success) - return { error: "Slack sent back something unreadable." }; + if (!outcome.ok) { + if (outcome.error === NAME_TAKEN) return channelNamed(name); + return { error: explain(outcome.error) }; + } - if (!parsed.data.ok || !parsed.data.channel) { - return { error: explain(parsed.data.error ?? "rejected") }; + if (!outcome.data.channel) { + return { error: "Slack sent back something unreadable." }; } - const channel = parsed.data.channel; + const channel = outcome.data.channel; await db.slackChannel.upsert({ where: { id: channel.id }, diff --git a/apps/agent/agent/lib/slack-owner.ts b/apps/agent/agent/lib/slack-owner.ts new file mode 100644 index 000000000..2bed3e9ea --- /dev/null +++ b/apps/agent/agent/lib/slack-owner.ts @@ -0,0 +1,62 @@ +import { db } from "@crm/db"; +import { schemas } from "@crm/validation"; +import { runRecord } from "@crm/validation/agent-events"; +import { slackPost } from "./slack-api"; +import { slackAccessToken } from "./slack-connection"; + +export type OwnerInvite = + | { added: true; slackUserId: string; name: string } + | { added: false; reason: string }; + +export async function addDealOwner( + runId: string, + channelId: string, +): Promise { + const run = await db.agentRun.findUnique({ + where: { id: runId }, + select: { input: true }, + }); + + const record = runRecord(run?.input); + if (record?.kind !== "deal") return null; + + const deal = await db.deal.findUnique({ + where: { id: record.id }, + select: { owner: { select: { id: true, name: true } } }, + }); + if (!deal) return null; + + const match = await db.slackMemberMatch.findUnique({ + where: { crmUserId: deal.owner.id }, + select: { slackUserId: true }, + }); + if (!match?.slackUserId) { + return { + added: false, + reason: `${deal.owner.name} has no matching Slack account.`, + }; + } + + const token = await slackAccessToken(); + if (!token) return { added: false, reason: "Slack is not connected." }; + + const outcome = await slackPost( + token, + "conversations.invite", + { channel: channelId, users: match.slackUserId }, + schemas.slack.reply, + ); + + if (outcome.ok || outcome.error === "already_in_channel") { + return { + added: true, + slackUserId: match.slackUserId, + name: deal.owner.name, + }; + } + + return { + added: false, + reason: `Slack refused to add ${deal.owner.name} (${outcome.error}).`, + }; +} diff --git a/apps/agent/agent/lib/tasks.ts b/apps/agent/agent/lib/tasks.ts index 9d8a912c7..94ce8bfcc 100644 --- a/apps/agent/agent/lib/tasks.ts +++ b/apps/agent/agent/lib/tasks.ts @@ -79,7 +79,7 @@ export async function retireExhausted( UPDATE "agentTask" AS t SET "finishedAt" = ${now}, "outcome" = ${RETIRED_OUTCOME} - WHERE t.id IN ( + FROM ( SELECT c.id FROM "agentTask" AS c WHERE c."finishedAt" IS NULL @@ -88,7 +88,8 @@ export async function retireExhausted( ORDER BY c."dueAt" ASC LIMIT ${limit} FOR UPDATE SKIP LOCKED - ) + ) AS exhausted + WHERE t.id = exhausted.id RETURNING t.id, t."contactId", t."companyId", t."dealId", t.kind; `; } diff --git a/apps/agent/agent/skills/customer-onboarding.md b/apps/agent/agent/skills/customer-onboarding.md new file mode 100644 index 000000000..c44ba8230 --- /dev/null +++ b/apps/agent/agent/skills/customer-onboarding.md @@ -0,0 +1,61 @@ +--- +description: Use when a deal closes won and the customer needs a shared Slack channel — the order the steps must run in, and what to do while you wait for a person. +--- + +# Onboarding a customer into Slack + +A deal closed won. The customer needs one place to talk to us. That place is a +Slack channel they can reach from their own workspace, which means Slack +Connect, which means somebody on their side has to accept an invitation before +anything else can happen. + +That wait is the whole shape of this job. You do not finish this work in one +turn, and you must not try. + +## The order is fixed + +1. `open_slack_channel` — name it after the customer, in plain words. This also + makes the run watch the channel, so anything said in it comes back to you. +2. `invite_to_slack_channel` — the buyer, and anybody on our side who owns the + account. +3. Stop and wait. + +Step 1 is first because it is what makes steps 2 and 3 possible. A channel you +did not open with this tool is a channel this run does not watch, so nobody's +reply will ever reach you. + +## Waiting is the work, not a failure + +When you have sent the invitations, say what you did and end your turn. Do not +poll, do not schedule a recheck, and do not report the onboarding as finished. + +The run parks. When the customer joins, or anybody writes in that channel, or +somebody mentions you there, the run wakes up with the message and you carry on +from where you stopped. You keep everything you already knew. + +A person can take a day to accept a Slack Connect invitation. That is normal and +it is not a problem to solve. + +## When you wake up + +You are told what happened: somebody joined, or somebody said something. Read it +and decide whether it needs you. + +- **Somebody joined** — greet them by name in the channel, say who we are and + what happens next. Then stop again. +- **Somebody wrote** — answer if you can, and stop. If it needs a person on our + side, say so in the channel and name them. +- **Nothing needs doing** — stop without posting. An agent that speaks every time + a channel moves is an agent people mute. + +## What goes wrong, and what it means + +- **`open_slack_channel` gives back a channel that already exists.** Somebody ran + this before, or the deal reopened and closed again. Use it. Do not invent a + second name to get a fresh channel. +- **`invited: false` with a reason.** Read the reason. "Slack is not connected" + and "this workspace does not allow it" both need a person, and neither is + fixed by trying again. +- **`watching: false`.** This run already watches a different channel, so replies + in the new one will never reach you. Say so plainly rather than waiting for a + message that cannot arrive. diff --git a/apps/agent/agent/subagents/agent_builder/lib/draft-input.ts b/apps/agent/agent/subagents/agent_builder/lib/draft-input.ts index 67d1d926b..d1efb24f4 100644 --- a/apps/agent/agent/subagents/agent_builder/lib/draft-input.ts +++ b/apps/agent/agent/subagents/agent_builder/lib/draft-input.ts @@ -55,8 +55,20 @@ const action = z.discriminatedUnion("type", [ label: z.string().trim().min(1).max(120), }), }), + z.object({ + type: z.literal(AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN), + provider: z.literal("slack"), + summary: z.string().trim().min(1).max(240), + }), + z.object({ + type: z.literal(AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE), + provider: z.literal("slack"), + summary: z.string().trim().min(1).max(240), + }), ]); +export type DraftAction = z.infer; + export const builderDraftToolInput = z.object({ name: z.string().trim().min(1).max(100), description: z.string().trim().min(1).max(320), @@ -77,6 +89,19 @@ const ACTIVITY_ACCESS = { const ACTIVITY_ORDER = ["NOTE", "TASK"] as const; +const SLACK_ACCESS = { + [AGENT_ACTION_TYPES.SLACK_MESSAGE_POST]: + "Post to approved Slack destinations", + [AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN]: "Open Slack channels", + [AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE]: "Invite people to Slack channels", +} as const; + +const SLACK_ORDER = [ + AGENT_ACTION_TYPES.SLACK_MESSAGE_POST, + AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN, + AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE, +] as const; + const INTEGRATIONS = { gmail: { kind: "integration", id: "google:gmail", label: "Gmail" }, calendar: { @@ -99,16 +124,20 @@ export function draftInputFromTool( : [], ), ); + const slackTypes = new Set(input.actions.map((entry) => entry.type)); const access = [ input.recordScope === "WORKSPACE" ? "Read workspace CRM records" : "Read selected CRM records", - ...integrations.map((integration) => { - if (integration === "gmail") return "Read connected Gmail messages"; + ...integrations.flatMap((integration) => { + if (integration === "gmail") return ["Read connected Gmail messages"]; if (integration === "calendar") { - return "Read connected Google Calendar events"; + return ["Read connected Google Calendar events"]; } - return "Post to approved Slack destinations"; + const lines = SLACK_ORDER.filter((type) => slackTypes.has(type)).map( + (type) => SLACK_ACCESS[type], + ); + return lines.length > 0 ? lines : ["Post to approved Slack destinations"]; }), ...ACTIVITY_ORDER.filter((type) => activityTypes.has(type)).map( (type) => ACTIVITY_ACCESS[type], diff --git a/apps/agent/agent/subagents/agent_runner/tools/invite_to_slack_channel.ts b/apps/agent/agent/subagents/agent_runner/tools/invite_to_slack_channel.ts new file mode 100644 index 000000000..94504ad6d --- /dev/null +++ b/apps/agent/agent/subagents/agent_runner/tools/invite_to_slack_channel.ts @@ -0,0 +1,25 @@ +import { defineTool } from "eve/tools"; +import { z } from "zod"; +import { inviteToRunSlackChannel } from "../../../lib/run-runtime"; +import { requireTeamAgentAttribute } from "../../../lib/session-purpose"; + +export default defineTool({ + description: + "Invite people to the Slack channel this run opened. An address inside this workspace is added straight away. An address outside it gets a Slack Connect invitation, which that person has to accept before they can read anything.", + inputSchema: z.object({ + emails: z + .array(z.email()) + .min(1) + .max(10) + .describe( + "Who to invite, by email address. Customers and colleagues both go here.", + ), + }), + async execute(input, ctx) { + return inviteToRunSlackChannel( + requireTeamAgentAttribute(ctx, "runId"), + ctx.callId, + input, + ); + }, +}); diff --git a/apps/agent/agent/subagents/agent_runner/tools/open_slack_channel.ts b/apps/agent/agent/subagents/agent_runner/tools/open_slack_channel.ts new file mode 100644 index 000000000..ffc76f658 --- /dev/null +++ b/apps/agent/agent/subagents/agent_runner/tools/open_slack_channel.ts @@ -0,0 +1,32 @@ +import { defineTool } from "eve/tools"; +import { z } from "zod"; +import { openRunSlackChannel } from "../../../lib/run-runtime"; +import { requireTeamAgentAttribute } from "../../../lib/session-purpose"; + +export default defineTool({ + description: + "Open the Slack channel this run works in, and start watching it. Every message and every join in that channel wakes this run up, so open the channel before you invite anybody. A name already in use gives you back the existing channel.", + inputSchema: z.object({ + name: z + .string() + .trim() + .min(1) + .max(120) + .describe( + "What to call it, in plain words: 'Acme onboarding'. Spaces and capitals are fine; Slack gets a tidied version.", + ), + isPrivate: z + .boolean() + .default(false) + .describe( + "True for work the whole workspace must not read. A customer channel is public.", + ), + }), + async execute(input, ctx) { + return openRunSlackChannel( + requireTeamAgentAttribute(ctx, "runId"), + ctx.callId, + input, + ); + }, +}); diff --git a/apps/agent/test/custom-agent-runtime.spec.ts b/apps/agent/test/custom-agent-runtime.spec.ts index 330d2a581..f72260bd6 100644 --- a/apps/agent/test/custom-agent-runtime.spec.ts +++ b/apps/agent/test/custom-agent-runtime.spec.ts @@ -244,8 +244,15 @@ describe("deployed Slack actions", () => { approvedSlackDestination({ ...manifest, dataScope: { ...manifest.dataScope, resources: [] }, + actions: [ + { + type: "run.summary", + provider: "crm", + summary: "Summarize the Slack delivery", + }, + ], }), - ).toThrow("does not allow Slack"); + ).toThrow("Agent version does not allow Slack."); }); it("posts with a stable Slack replay id", async () => { @@ -309,7 +316,9 @@ describe("deployed Slack actions", () => { "7d3e8854-79f9-48dd-a933-8cfb5994f99e", { fetcher }, ), - ).rejects.toThrow("Invite the app"); + ).rejects.toThrow( + "Comp AI isn't in that channel. Invite it there, then try again.", + ); }); }); diff --git a/apps/agent/test/run-resume.integration.spec.ts b/apps/agent/test/run-resume.integration.spec.ts new file mode 100644 index 000000000..38edb86de --- /dev/null +++ b/apps/agent/test/run-resume.integration.spec.ts @@ -0,0 +1,312 @@ +import { afterAll, beforeAll, describe, expect, it } from "bun:test"; +import { db } from "@crm/db"; +import type { SendFn } from "eve/channels"; +import { runToken } from "../agent/lib/custom-agent-dispatch"; +import { + claimSlackChannel, + resumeAgentRun, + runOnSlackChannel, +} from "../agent/lib/run-resume"; + +const suffix = crypto.randomUUID(); +const userId = `resume-user-${suffix}`; + +let agentId = ""; +let versionId = ""; + +type Delivery = { + message: string; + continuationToken?: string; + mode?: string; + attributes?: Record; +}; + +type EveSendOptions = { + continuationToken?: string; + mode?: string; + auth?: { attributes?: Record }; +}; + +const deliveries: Delivery[] = []; + +const send = (async (message: string, options?: EveSendOptions) => { + deliveries.push({ + message, + continuationToken: options?.continuationToken, + mode: options?.mode, + attributes: options?.auth?.attributes, + }); + return { id: `ses_${deliveries.length}` }; +}) as unknown as SendFn; + +const refusing = (async () => { + throw new Error("eve refused: session is not active"); +}) as unknown as SendFn; + +async function makeRun(overrides: { + status: + | "QUEUED" + | "RUNNING" + | "WAITING_FOR_APPROVAL" + | "SUCCEEDED" + | "FAILED" + | "CANCELLED"; + sessionId?: string | null; +}) { + const unique = crypto.randomUUID(); + const run = await db.agentRun.create({ + data: { + agentId, + versionId, + status: overrides.status, + triggerType: "MANUAL", + idempotencyKey: `resume-${unique}`, + correlationId: `resume-${unique}`, + sessionId: overrides.sessionId === null ? null : `ses_seed_${unique}`, + }, + select: { id: true }, + }); + return run.id; +} + +beforeAll(async () => { + await db.user.create({ + data: { + id: userId, + name: "Resume Spike", + email: `${userId}@example.test`, + }, + }); + + const agent = await db.agentDefinition.create({ + data: { + name: `Resume spike ${suffix}`, + status: "LIVE", + createdById: userId, + }, + select: { id: true }, + }); + agentId = agent.id; + + const version = await db.agentVersion.create({ + data: { + agentId, + number: 1, + status: "DEPLOYED", + createdById: userId, + instructions: "Resume spike.", + manifest: {}, + modelId: "zai/glm-5.2-fast", + sandboxPolicy: {}, + }, + select: { id: true }, + }); + versionId = version.id; + + await db.agentDefinition.update({ + where: { id: agentId }, + data: { currentVersionId: versionId }, + }); +}); + +afterAll(async () => { + await db.agentRun.deleteMany({ where: { agentId } }); + await db.agentDefinition.updateMany({ + where: { id: agentId }, + data: { currentVersionId: null }, + }); + await db.agentVersion.deleteMany({ where: { agentId } }); + await db.agentDefinition.deleteMany({ where: { id: agentId } }); + await db.user.deleteMany({ where: { id: userId } }); +}); + +describe("resuming a parked run from an outside event", () => { + it("delivers to the run's own continuation token, not a new session", async () => { + deliveries.length = 0; + const runId = await makeRun({ status: "WAITING_FOR_APPROVAL" }); + + const outcome = await resumeAgentRun( + { runId, message: "The customer joined the channel.", source: "slack" }, + send, + ); + + expect(outcome.kind).toBe("resumed"); + expect(deliveries).toHaveLength(1); + expect(deliveries[0]?.continuationToken).toBe(runToken(runId)); + expect(deliveries[0]?.message).toBe("The customer joined the channel."); + }); + + it("resumes in task mode, so it cannot pause for a per-action approval", async () => { + deliveries.length = 0; + const runId = await makeRun({ status: "RUNNING" }); + + await resumeAgentRun({ runId, message: "org id", source: "slack" }, send); + + expect(deliveries[0]?.mode).toBe("task"); + }); + + it("carries the run identity and the source, so the runner can revalidate", async () => { + deliveries.length = 0; + const runId = await makeRun({ status: "WAITING_FOR_APPROVAL" }); + + await resumeAgentRun( + { + runId, + message: "joined", + source: "slack.member_joined_channel", + attributes: { channelId: "C123" }, + }, + send, + ); + + expect(deliveries[0]?.attributes).toMatchObject({ + purpose: "team-agent", + runId, + agentId, + versionId, + resumeSource: "slack.member_joined_channel", + channelId: "C123", + }); + }); + + it("refuses a finished run, so a late event cannot restart it", async () => { + for (const status of ["SUCCEEDED", "FAILED", "CANCELLED"] as const) { + deliveries.length = 0; + const runId = await makeRun({ status }); + + const outcome = await resumeAgentRun( + { runId, message: "late", source: "slack" }, + send, + ); + + expect(outcome.kind).toBe("ignored"); + expect(deliveries).toHaveLength(0); + } + }); + + it("refuses a run that never started a session", async () => { + deliveries.length = 0; + const runId = await makeRun({ status: "QUEUED", sessionId: null }); + + const outcome = await resumeAgentRun( + { runId, message: "early", source: "slack" }, + send, + ); + + expect(outcome.kind).toBe("ignored"); + expect(deliveries).toHaveLength(0); + }); + + it("refuses a run whose agent is no longer live", async () => { + const runId = await makeRun({ status: "WAITING_FOR_APPROVAL" }); + await db.agentDefinition.update({ + where: { id: agentId }, + data: { status: "PAUSED" }, + }); + + const outcome = await resumeAgentRun( + { runId, message: "paused", source: "slack" }, + send, + ); + + await db.agentDefinition.update({ + where: { id: agentId }, + data: { status: "LIVE" }, + }); + + expect(outcome.kind).toBe("ignored"); + }); + + it("ignores an unknown run rather than throwing", async () => { + const outcome = await resumeAgentRun( + { runId: `missing-${suffix}`, message: "x", source: "slack" }, + send, + ); + + expect(outcome).toMatchObject({ kind: "ignored", reason: "no such run" }); + }); + + it("never throws when eve refuses the send", async () => { + const runId = await makeRun({ status: "WAITING_FOR_APPROVAL" }); + + const outcome = await resumeAgentRun( + { runId, message: "refused", source: "slack" }, + refusing, + ); + + expect(outcome.kind).toBe("ignored"); + if (outcome.kind !== "ignored") throw new Error("expected ignored"); + expect(outcome.reason).toContain("not active"); + }); + + it("leaves the run's status alone; the runner owns its own state", async () => { + const runId = await makeRun({ status: "WAITING_FOR_APPROVAL" }); + + await resumeAgentRun({ runId, message: "hello", source: "slack" }, send); + + const after = await db.agentRun.findUnique({ + where: { id: runId }, + select: { status: true }, + }); + expect(after?.status).toBe("WAITING_FOR_APPROVAL"); + }); +}); + +describe("finding the run an event belongs to", () => { + it("finds the live run that owns a channel", async () => { + const runId = await makeRun({ status: "WAITING_FOR_APPROVAL" }); + const channelId = `C-${crypto.randomUUID()}`; + + await claimSlackChannel(runId, channelId); + + expect(await runOnSlackChannel(channelId)).toBe(runId); + }); + + it("ignores a finished run, so its channel stops routing", async () => { + const runId = await makeRun({ status: "SUCCEEDED" }); + const channelId = `C-${crypto.randomUUID()}`; + + await claimSlackChannel(runId, channelId); + + expect(await runOnSlackChannel(channelId)).toBeNull(); + }); + + it("prefers the newest live run when a channel is reused", async () => { + const channelId = `C-${crypto.randomUUID()}`; + const older = await makeRun({ status: "WAITING_FOR_APPROVAL" }); + await claimSlackChannel(older, channelId); + + await new Promise((resolve) => setTimeout(resolve, 10)); + + const newer = await makeRun({ status: "WAITING_FOR_APPROVAL" }); + await claimSlackChannel(newer, channelId); + + expect(await runOnSlackChannel(channelId)).toBe(newer); + }); + + it("does not reassign a channel a run already claimed", async () => { + const runId = await makeRun({ status: "RUNNING" }); + const first = `C-${crypto.randomUUID()}`; + const second = `C-${crypto.randomUUID()}`; + + await claimSlackChannel(runId, first); + await claimSlackChannel(runId, second); + + expect(await runOnSlackChannel(first)).toBe(runId); + expect(await runOnSlackChannel(second)).toBeNull(); + }); + + it("says which channel the run watches, so a second claim is not silent", async () => { + const runId = await makeRun({ status: "RUNNING" }); + const first = `C-${crypto.randomUUID()}`; + const second = `C-${crypto.randomUUID()}`; + + expect(await claimSlackChannel(runId, first)).toBe(first); + expect(await claimSlackChannel(runId, second)).toBe(first); + }); + + it("reads an unknown or blank channel as nobody", async () => { + expect(await runOnSlackChannel(`C-${crypto.randomUUID()}`)).toBeNull(); + expect(await runOnSlackChannel(" ")).toBeNull(); + }); +}); diff --git a/apps/agent/test/slack-channel-actions.integration.spec.ts b/apps/agent/test/slack-channel-actions.integration.spec.ts new file mode 100644 index 000000000..4e3b16854 --- /dev/null +++ b/apps/agent/test/slack-channel-actions.integration.spec.ts @@ -0,0 +1,405 @@ +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + it, +} from "bun:test"; +import { db } from "@crm/db"; +import { channelOfRun } from "../agent/lib/run-resume"; +import { + inviteToRunSlackChannel, + openRunSlackChannel, +} from "../agent/lib/run-runtime"; + +const suffix = crypto.randomUUID(); +const userId = `slack-actions-user-${suffix}`; +const accountId = `slack-actions-account-${suffix}`; + +const realFetch = globalThis.fetch; +let created = 0; + +function slackReplies(reply: (url: string) => object) { + globalThis.fetch = (async (input: URL | RequestInfo) => { + const url = String(input instanceof Request ? input.url : input); + return new Response(JSON.stringify(reply(url)), { + headers: { "content-type": "application/json" }, + }); + }) as typeof fetch; +} + +const slackManifest = (actions: unknown[]) => ({ + triggers: [ + { type: "MANUAL", name: "Run now", summary: "Run on demand", config: {} }, + ], + dataScope: { + mode: "WORKSPACE", + summary: "Workspace CRM records", + resources: [{ kind: "integration", id: "slack:workspace", label: "Slack" }], + }, + actions, +}); + +const openAction = { + type: "slack.channel.open", + provider: "slack", + summary: "Open the customer channel", +}; +const inviteAction = { + type: "slack.channel.invite", + provider: "slack", + summary: "Invite the buyer", +}; +const summaryAction = { + type: "run.summary", + provider: "crm", + summary: "Say what happened", +}; + +let agentId = ""; + +async function makeRun(actions: unknown[]) { + const unique = crypto.randomUUID(); + const version = await db.agentVersion.create({ + data: { + agentId, + number: (await db.agentVersion.count({ where: { agentId } })) + 1, + status: "DEPLOYED", + createdById: userId, + instructions: "x", + manifest: slackManifest(actions), + modelId: "m", + sandboxPolicy: {}, + }, + select: { id: true }, + }); + const run = await db.agentRun.create({ + data: { + agentId, + versionId: version.id, + status: "RUNNING", + triggerType: "MANUAL", + idempotencyKey: `sca-${unique}`, + correlationId: `sca-${unique}`, + sessionId: `ses_${unique}`, + }, + select: { id: true }, + }); + return run.id; +} + +beforeAll(async () => { + await db.user.create({ + data: { + id: userId, + name: "Slack Actions", + email: `${userId}@example.test`, + }, + }); + const agent = await db.agentDefinition.create({ + data: { + name: `Slack actions ${suffix}`, + status: "LIVE", + createdById: userId, + }, + select: { id: true }, + }); + agentId = agent.id; + await db.account.create({ + data: { + id: accountId, + accountId: `T-${suffix}`, + providerId: "slack", + userId, + accessToken: "xoxb-slack-actions", + }, + }); +}); + +afterAll(async () => { + globalThis.fetch = realFetch; + await db.agentAction.deleteMany({ where: { agentId } }); + await db.agentRun.deleteMany({ where: { agentId } }); + await db.agentDefinition.updateMany({ + where: { id: agentId }, + data: { currentVersionId: null }, + }); + await db.agentVersion.deleteMany({ where: { agentId } }); + await db.agentDefinition.deleteMany({ where: { id: agentId } }); + await db.account.deleteMany({ where: { id: accountId } }); + await db.user.deleteMany({ where: { id: userId } }); +}); + +beforeEach(() => { + created = 0; + globalThis.fetch = realFetch; +}); + +describe("opening a channel as a deployed run", () => { + it("refuses when the version does not approve opening a channel", async () => { + const runId = await makeRun([summaryAction]); + + await expect( + openRunSlackChannel(runId, "call-1", { + name: "Acme onboarding", + isPrivate: false, + }), + ).rejects.toThrow("does not allow slack.channel.open"); + }); + + it("opens the channel and makes the run watch it", async () => { + const runId = await makeRun([openAction, summaryAction]); + const channelId = `C-${crypto.randomUUID()}`; + slackReplies(() => { + created += 1; + return { ok: true, channel: { id: channelId, name: "acme-onboarding" } }; + }); + + const outcome = await openRunSlackChannel(runId, "call-1", { + name: "Acme Onboarding", + isPrivate: false, + }); + + expect(outcome).toMatchObject({ + channelId, + channelName: "acme-onboarding", + watching: true, + replayed: false, + }); + expect(await channelOfRun(runId)).toBe(channelId); + await db.slackChannel.deleteMany({ where: { id: channelId } }); + }); + + it("replays the same call instead of opening a second channel", async () => { + const runId = await makeRun([openAction, summaryAction]); + const channelId = `C-${crypto.randomUUID()}`; + slackReplies(() => { + created += 1; + return { ok: true, channel: { id: channelId, name: "acme-onboarding" } }; + }); + + await openRunSlackChannel(runId, "call-1", { + name: "Acme Onboarding", + isPrivate: false, + }); + const again = await openRunSlackChannel(runId, "call-1", { + name: "Acme Onboarding", + isPrivate: false, + }); + + expect(again.replayed).toBe(true); + expect(again.channelId).toBe(channelId); + expect(created).toBe(1); + await db.slackChannel.deleteMany({ where: { id: channelId } }); + }); + + it("adds the deal owner when Slack knows them", async () => { + const dealId = `deal-${crypto.randomUUID()}`; + const company = await db.company.create({ + data: { name: `Owner Co ${suffix}`, domain: `${dealId}.test` }, + select: { id: true }, + }); + await db.deal.create({ + data: { + id: dealId, + name: "Owner deal", + stage: "CLOSED_WON", + companyId: company.id, + ownerId: userId, + amount: 1, + currency: "USD", + }, + }); + await db.slackMemberMatch.create({ + data: { crmUserId: userId, slackUserId: "U-OWNER" }, + }); + + const runId = await makeRun([openAction, summaryAction]); + await db.agentRun.update({ + where: { id: runId }, + data: { input: { record: { kind: "deal", id: dealId } } }, + }); + + const channelId = `C-${crypto.randomUUID()}`; + const invited: unknown[] = []; + globalThis.fetch = (async ( + input: URL | RequestInfo, + init?: RequestInit, + ) => { + const url = String(input instanceof Request ? input.url : input); + if (url.includes("conversations.invite")) { + invited.push(JSON.parse(String(init?.body))); + } + return new Response( + JSON.stringify( + url.includes("conversations.create") + ? { ok: true, channel: { id: channelId, name: "owner-co" } } + : { ok: true }, + ), + { headers: { "content-type": "application/json" } }, + ); + }) as typeof fetch; + + const outcome = await openRunSlackChannel(runId, "call-1", { + name: "Owner Co", + isPrivate: false, + }); + + expect(outcome.owner).toMatchObject({ + added: true, + slackUserId: "U-OWNER", + }); + expect(invited).toEqual([{ channel: channelId, users: "U-OWNER" }]); + + await db.slackChannel.deleteMany({ where: { id: channelId } }); + await db.slackMemberMatch.deleteMany({ where: { crmUserId: userId } }); + await db.deal.deleteMany({ where: { id: dealId } }); + await db.company.deleteMany({ where: { id: company.id } }); + }); + + it("refuses a name with nothing Slack accepts", async () => { + const runId = await makeRun([openAction, summaryAction]); + + await expect( + openRunSlackChannel(runId, "call-1", { + name: "!!! ---", + isPrivate: false, + }), + ).rejects.toThrow("no letters or numbers"); + }); +}); + +describe("inviting people as a deployed run", () => { + it("refuses before a channel exists, rather than inviting nobody", async () => { + const runId = await makeRun([inviteAction, summaryAction]); + + await expect( + inviteToRunSlackChannel(runId, "call-1", { emails: ["buyer@x.test"] }), + ).rejects.toThrow("no Slack channel yet"); + }); + + it("invites into the channel the run opened", async () => { + const runId = await makeRun([openAction, inviteAction, summaryAction]); + const channelId = `C-${crypto.randomUUID()}`; + const posted: { url: string; body: unknown }[] = []; + globalThis.fetch = (async ( + input: URL | RequestInfo, + init?: RequestInit, + ) => { + const url = String(input instanceof Request ? input.url : input); + posted.push({ + url, + body: init?.body ? JSON.parse(String(init.body)) : null, + }); + return new Response( + JSON.stringify( + url.includes("conversations.create") + ? { ok: true, channel: { id: channelId, name: "acme-onboarding" } } + : url.includes("users.lookupByEmail") + ? { ok: false, error: "users_not_found" } + : { + ok: true, + invite_id: "I1", + url: "https://slack.com/invite/x", + }, + ), + { headers: { "content-type": "application/json" } }, + ); + }) as typeof fetch; + + await openRunSlackChannel(runId, "open-1", { + name: "Acme Onboarding", + isPrivate: false, + }); + const outcome = await inviteToRunSlackChannel(runId, "invite-1", { + emails: ["buyer@customer.test"], + }); + + expect(outcome).toMatchObject({ + channelId, + replayed: false, + result: { + type: "slack.channel.invite", + invite_id: "I1", + url: "https://slack.com/invite/x", + email: "buyer@customer.test", + kind: "connect", + }, + }); + expect(outcome.invited).toHaveLength(1); + expect(outcome.invited?.[0]).toMatchObject({ + invite_id: "I1", + url: "https://slack.com/invite/x", + }); + expect( + posted.find((call) => call.url.includes("conversations.inviteShared")) + ?.body, + ).toMatchObject({ + channel: channelId, + emails: ["buyer@customer.test"], + external_limited: false, + }); + + const row = await db.agentAction.findFirst({ + where: { runId, type: "slack.channel.invite" }, + select: { externalId: true, result: true }, + }); + expect(row?.externalId).toBe("I1"); + expect(row?.result).toMatchObject({ + type: "slack.channel.invite", + invite_id: "I1", + url: "https://slack.com/invite/x", + email: "buyer@customer.test", + kind: "connect", + }); + + const replay = await inviteToRunSlackChannel(runId, "invite-1", { + emails: ["buyer@customer.test"], + }); + expect(replay).toMatchObject({ + channelId, + replayed: true, + result: { + type: "slack.channel.invite", + invite_id: "I1", + url: "https://slack.com/invite/x", + email: "buyer@customer.test", + kind: "connect", + }, + }); + await db.slackChannel.deleteMany({ where: { id: channelId } }); + }); + + it("fails the action when Slack refuses every address", async () => { + const runId = await makeRun([openAction, inviteAction, summaryAction]); + const channelId = `C-${crypto.randomUUID()}`; + slackReplies((url) => + url.includes("conversations.create") + ? { ok: true, channel: { id: channelId, name: "acme-onboarding" } } + : url.includes("users.lookupByEmail") + ? { ok: false, error: "users_not_found" } + : { ok: false, error: "restricted_action" }, + ); + + await openRunSlackChannel(runId, "open-1", { + name: "Acme Onboarding", + isPrivate: false, + }); + + await expect( + inviteToRunSlackChannel(runId, "invite-1", { + emails: ["buyer@customer.test"], + }), + ).rejects.toThrow( + "This workspace doesn't let Comp AI send that invitation.", + ); + + const action = await db.agentAction.findFirst({ + where: { runId, type: "slack.channel.invite" }, + select: { status: true }, + }); + expect(action?.status).toBe("FAILED"); + await db.slackChannel.deleteMany({ where: { id: channelId } }); + }); +}); diff --git a/apps/agent/test/slack-channel-open.spec.ts b/apps/agent/test/slack-channel-open.spec.ts new file mode 100644 index 000000000..278c9ad08 --- /dev/null +++ b/apps/agent/test/slack-channel-open.spec.ts @@ -0,0 +1,33 @@ +import { describe, expect, it } from "bun:test"; +import { toChannelName } from "../agent/lib/slack-channel-name"; + +describe("tidying a channel name for Slack", () => { + it("lowercases and joins words with a dash", () => { + expect(toChannelName("Acme Onboarding")).toBe("acme-onboarding"); + }); + + it("drops punctuation Slack refuses", () => { + expect(toChannelName("Acme, Inc. — onboarding!")).toBe( + "acme-inc-onboarding", + ); + }); + + it("collapses runs of dashes and trims the ends", () => { + expect(toChannelName(" --acme---onboarding-- ")).toBe("acme-onboarding"); + }); + + it("keeps a name Slack already accepts", () => { + expect(toChannelName("acme-onboarding")).toBe("acme-onboarding"); + }); + + it("cuts a long name to the Slack limit and leaves no trailing dash", () => { + const name = toChannelName(`${"acme ".repeat(40)}onboarding`); + + expect(name.length).toBeLessThanOrEqual(80); + expect(name.endsWith("-")).toBe(false); + }); + + it("gives back nothing when the name has no letters or numbers", () => { + expect(toChannelName("!!! ---")).toBe(""); + }); +}); diff --git a/apps/agent/test/slack-events.integration.spec.ts b/apps/agent/test/slack-events.integration.spec.ts new file mode 100644 index 000000000..b60e5bb78 --- /dev/null +++ b/apps/agent/test/slack-events.integration.spec.ts @@ -0,0 +1,289 @@ +import { + afterAll, + beforeAll, + beforeEach, + describe, + expect, + it, +} from "bun:test"; +import { db } from "@crm/db"; +import type { SlackEvent } from "@crm/validation"; +import type { SendFn } from "eve/channels"; +import { runToken } from "../agent/lib/custom-agent-dispatch"; +import { claimSlackChannel } from "../agent/lib/run-resume"; +import { + describe as describeEvent, + dispatchSlackEvent, + drainSlackEvents, +} from "../agent/lib/slack-events"; + +const suffix = crypto.randomUUID(); +const userId = `slack-events-user-${suffix}`; + +let agentId = ""; +let versionId = ""; + +const deliveries: { message: string; continuationToken?: string }[] = []; + +type EveSendOptions = { + continuationToken?: string; +}; + +const send = (async (message: string, options?: EveSendOptions) => { + deliveries.push({ + message, + continuationToken: options?.continuationToken, + }); + return { id: `ses_${deliveries.length}` }; +}) as unknown as SendFn; + +async function makeRun( + channelId: string | null, + status = "WAITING_FOR_APPROVAL", +) { + const unique = crypto.randomUUID(); + const run = await db.agentRun.create({ + data: { + agentId, + versionId, + status: status as "WAITING_FOR_APPROVAL", + triggerType: "MANUAL", + idempotencyKey: `se-${unique}`, + correlationId: `se-${unique}`, + sessionId: `ses_seed_${unique}`, + }, + select: { id: true }, + }); + if (channelId) await claimSlackChannel(run.id, channelId); + return run.id; +} + +async function inbox(event: SlackEvent, channelId: string | null) { + const eventId = `Ev-${crypto.randomUUID()}`; + const row = await db.slackEventInbox.create({ + data: { + eventId, + type: String(event.type), + channelId, + teamId: "T1", + payload: { + type: "event_callback", + event_id: eventId, + team_id: "T1", + event, + }, + }, + select: { id: true }, + }); + return row.id; +} + +beforeAll(async () => { + await db.user.create({ + data: { id: userId, name: "Slack Events", email: `${userId}@example.test` }, + }); + const agent = await db.agentDefinition.create({ + data: { + name: `Slack events ${suffix}`, + status: "LIVE", + createdById: userId, + }, + select: { id: true }, + }); + agentId = agent.id; + const version = await db.agentVersion.create({ + data: { + agentId, + number: 1, + status: "DEPLOYED", + createdById: userId, + instructions: "x", + manifest: {}, + modelId: "m", + sandboxPolicy: {}, + }, + select: { id: true }, + }); + versionId = version.id; + await db.agentDefinition.update({ + where: { id: agentId }, + data: { currentVersionId: versionId }, + }); +}); + +afterAll(async () => { + await db.slackEventInbox.deleteMany({ where: { teamId: "T1" } }); + await db.agentAction.deleteMany({ where: { agentId } }); + await db.agentRun.deleteMany({ where: { agentId } }); + await db.agentDefinition.updateMany({ + where: { id: agentId }, + data: { currentVersionId: null }, + }); + await db.agentVersion.deleteMany({ where: { agentId } }); + await db.agentDefinition.deleteMany({ where: { id: agentId } }); + await db.user.deleteMany({ where: { id: userId } }); +}); + +beforeEach(() => { + deliveries.length = 0; +}); + +describe("turning a stored Slack event into a resume", () => { + it("resumes the run that owns the channel", async () => { + const channelId = `C-${crypto.randomUUID()}`; + const runId = await makeRun(channelId); + const id = await inbox( + { type: "message", channel: channelId, user: "U1", text: "org_abc" }, + channelId, + ); + + const outcome = await dispatchSlackEvent(id, send); + + expect(outcome?.resumed).toBe(true); + expect(deliveries).toHaveLength(1); + expect(deliveries[0]?.continuationToken).toBe(runToken(runId)); + expect(deliveries[0]?.message).toContain("org_abc"); + }); + + it("marks the row processed, so a second drain does nothing", async () => { + const channelId = `C-${crypto.randomUUID()}`; + await makeRun(channelId); + const id = await inbox( + { type: "message", channel: channelId, text: "hello" }, + channelId, + ); + + await dispatchSlackEvent(id, send); + const again = await dispatchSlackEvent(id, send); + + expect(again).toBeNull(); + expect(deliveries).toHaveLength(1); + }); + + it("claims the row so two drains resume the event once", async () => { + const channelId = `C-${crypto.randomUUID()}`; + await makeRun(channelId); + const id = await inbox( + { type: "message", channel: channelId, text: "once" }, + channelId, + ); + + const outcomes = await Promise.all([ + dispatchSlackEvent(id, send), + dispatchSlackEvent(id, send), + ]); + + expect(outcomes.filter((outcome) => outcome?.resumed)).toHaveLength(1); + expect(deliveries).toHaveLength(1); + }); + + it("settles an event whose channel owns no run, rather than retrying forever", async () => { + const channelId = `C-${crypto.randomUUID()}`; + const id = await inbox( + { type: "message", channel: channelId, text: "nobody home" }, + channelId, + ); + + const outcome = await dispatchSlackEvent(id, send); + + expect(outcome?.resumed).toBe(false); + expect(outcome?.outcome).toContain("No live agent run"); + expect(deliveries).toHaveLength(0); + + const row = await db.slackEventInbox.findUnique({ + where: { id }, + select: { processedAt: true }, + }); + expect(row?.processedAt).not.toBeNull(); + }); + + it("settles an event that names no channel", async () => { + const id = await inbox({ type: "message", text: "no channel" }, null); + + const outcome = await dispatchSlackEvent(id, send); + + expect(outcome?.outcome).toContain("names no channel"); + }); + + it("records the run it resumed against the event", async () => { + const channelId = `C-${crypto.randomUUID()}`; + const runId = await makeRun(channelId); + const id = await inbox( + { type: "member_joined_channel", channel: channelId, user: "U9" }, + channelId, + ); + + await dispatchSlackEvent(id, send); + + const row = await db.slackEventInbox.findUnique({ + where: { id }, + select: { runId: true }, + }); + expect(row?.runId).toBe(runId); + }); + + it("drains every pending event and counts the resumes", async () => { + const channelId = `C-${crypto.randomUUID()}`; + await makeRun(channelId); + await inbox({ type: "message", channel: channelId, text: "a" }, channelId); + await inbox({ type: "message", channel: channelId, text: "b" }, channelId); + await inbox( + { type: "message", channel: "C-nobody", text: "c" }, + "C-nobody", + ); + + const resumed = await drainSlackEvents(send); + + expect(resumed).toBeGreaterThanOrEqual(2); + }); +}); + +describe("what the agent is told", () => { + it("names the channel and the text for a message", () => { + const message = describeEvent({ + type: "message", + channel: "C1", + user: "U1", + text: "org_abc123", + }); + + expect(message).toContain("C1"); + expect(message).toContain("U1"); + expect(message).toContain("org_abc123"); + }); + + it("says somebody joined, and tells the run to carry on", () => { + const message = describeEvent({ + type: "member_joined_channel", + channel: "C1", + user: "U7", + }); + + expect(message).toContain("joined"); + expect(message).toContain("U7"); + expect(message).toContain("Carry on"); + }); + + it("says the agent was mentioned, not that a message arrived", () => { + const message = describeEvent({ + type: "app_mention", + channel: "C1", + user: "U3", + text: "<@U1> where are we", + }); + + expect(message).toContain("mentioned"); + expect(message).toContain("U3"); + expect(message).toContain("where are we"); + }); + + it("truncates a very long message rather than sending the lot", () => { + const message = describeEvent({ + type: "message", + channel: "C1", + text: "x".repeat(9000), + }); + + expect(message.length).toBeLessThan(3000); + }); +}); diff --git a/apps/agent/test/slack-invite.integration.spec.ts b/apps/agent/test/slack-invite.integration.spec.ts new file mode 100644 index 000000000..b79d8de1a --- /dev/null +++ b/apps/agent/test/slack-invite.integration.spec.ts @@ -0,0 +1,185 @@ +import { afterEach, beforeEach, describe, expect, it } from "bun:test"; +import { db } from "@crm/db"; +import { inviteToSlackChannel } from "../agent/lib/slack-invite"; + +const USER_ID = "slack-invite-spec-user"; +const ACCOUNT_ID = "slack-invite-spec-account"; +const CHANNEL_ID = "CINVITESPEC1"; + +const realFetch = globalThis.fetch; +const requested: { url: string; body: unknown }[] = []; + +function replies(reply: (url: string) => object) { + globalThis.fetch = (async (input: URL | RequestInfo, init?: RequestInit) => { + const url = String(input instanceof Request ? input.url : input); + requested.push({ + url, + body: init?.body ? JSON.parse(String(init.body)) : null, + }); + return new Response(JSON.stringify(reply(url)), { + headers: { "content-type": "application/json" }, + }); + }) as typeof fetch; +} + +const sent = (fragment: string) => + requested.find((call) => call.url.includes(fragment)); + +beforeEach(async () => { + requested.length = 0; + await db.user.upsert({ + where: { id: USER_ID }, + create: { + id: USER_ID, + name: "Slack Invite Spec", + email: `${USER_ID}@example.com`, + }, + update: {}, + }); + await db.account.upsert({ + where: { id: ACCOUNT_ID }, + create: { + id: ACCOUNT_ID, + accountId: "T-INVITE-SPEC", + providerId: "slack", + userId: USER_ID, + accessToken: "xoxb-invite-spec", + }, + update: { accessToken: "xoxb-invite-spec" }, + }); +}); + +afterEach(async () => { + globalThis.fetch = realFetch; + await db.account.deleteMany({ where: { id: ACCOUNT_ID } }); + await db.user.deleteMany({ where: { id: USER_ID } }); +}); + +describe("inviting somebody to a channel", () => { + it("adds a colleague straight away, because Slack already knows them", async () => { + replies((url) => + url.includes("users.lookupByEmail") + ? { ok: true, user: { id: "U7" } } + : { ok: true }, + ); + + const outcome = await inviteToSlackChannel(CHANNEL_ID, "rep@ours.test"); + + expect(outcome).toEqual({ + invited: true, + email: "rep@ours.test", + kind: "member", + }); + expect(sent("conversations.invite")?.body).toMatchObject({ + channel: CHANNEL_ID, + users: "U7", + }); + expect(sent("conversations.inviteShared")).toBeUndefined(); + }); + + it("sends a Slack Connect invitation to somebody outside the workspace", async () => { + replies((url) => + url.includes("users.lookupByEmail") + ? { ok: false, error: "users_not_found" } + : { ok: true, invite_id: "I1", url: "https://slack.com/invite/abc" }, + ); + + const outcome = await inviteToSlackChannel( + CHANNEL_ID, + "buyer@customer.test", + ); + + expect(outcome).toMatchObject({ + invited: true, + kind: "connect", + invite_id: "I1", + url: "https://slack.com/invite/abc", + }); + expect(sent("conversations.inviteShared")?.body).toMatchObject({ + channel: CHANNEL_ID, + emails: ["buyer@customer.test"], + external_limited: false, + }); + }); + + it("keeps invite_id when Slack withholds the url", async () => { + replies((url) => + url.includes("users.lookupByEmail") + ? { ok: false, error: "users_not_found" } + : { ok: true, invite_id: "I1" }, + ); + + const outcome = await inviteToSlackChannel( + CHANNEL_ID, + "buyer@customer.test", + ); + + expect(outcome).toEqual({ + invited: true, + email: "buyer@customer.test", + kind: "connect", + invite_id: "I1", + url: undefined, + }); + expect(sent("conversations.inviteShared")?.body).toMatchObject({ + channel: CHANNEL_ID, + emails: ["buyer@customer.test"], + external_limited: false, + }); + }); + + it("treats somebody already in the channel as invited, so a retry is quiet", async () => { + replies((url) => + url.includes("users.lookupByEmail") + ? { ok: true, user: { id: "U7" } } + : { ok: false, error: "already_in_channel" }, + ); + + const outcome = await inviteToSlackChannel(CHANNEL_ID, "rep@ours.test"); + + expect(outcome).toMatchObject({ invited: true, kind: "member" }); + }); + + it("says the workspace refused, rather than claiming the invitation went", async () => { + replies((url) => + url.includes("users.lookupByEmail") + ? { ok: false, error: "users_not_found" } + : { ok: false, error: "restricted_action" }, + ); + + const outcome = await inviteToSlackChannel( + CHANNEL_ID, + "buyer@customer.test", + ); + + expect(outcome.invited).toBe(false); + expect(outcome).toMatchObject({ + reason: expect.stringContaining( + "This workspace doesn't let Comp AI send that invitation.", + ), + }); + }); + + it("stops on a lookup failure that is not a missing person", async () => { + replies(() => ({ ok: false, error: "invalid_auth" })); + + const outcome = await inviteToSlackChannel(CHANNEL_ID, "rep@ours.test"); + + expect(outcome.invited).toBe(false); + expect(outcome).toMatchObject({ + reason: expect.stringContaining("reconnected"), + }); + expect(sent("conversations.inviteShared")).toBeUndefined(); + }); + + it("refuses when Slack is not connected at all", async () => { + await db.account.deleteMany({ where: { id: ACCOUNT_ID } }); + + const outcome = await inviteToSlackChannel(CHANNEL_ID, "rep@ours.test"); + + expect(outcome).toMatchObject({ + invited: false, + reason: "Slack is not connected.", + }); + }); +}); diff --git a/apps/api/package.json b/apps/api/package.json index 151aec0f8..70a6add30 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -43,6 +43,7 @@ "class-transformer": "^0.5.1", "class-validator": "^0.15.1", "context.dev": "^2.7.0", + "express": "^5.2.1", "helmet": "^8.3.0", "nestjs-trpc": "^2.13.0", "reflect-metadata": "^0.2.2", diff --git a/apps/api/src/agent/agent-runs.service.ts b/apps/api/src/agent/agent-runs.service.ts index 6cff770b4..ee86ca38a 100644 --- a/apps/api/src/agent/agent-runs.service.ts +++ b/apps/api/src/agent/agent-runs.service.ts @@ -2,6 +2,7 @@ import { randomUUID } from "node:crypto"; import { type Db, Prisma } from "@crm/db"; import type { AgentRunStatus } from "@crm/db/enums"; import { lockIdempotencyKey } from "@crm/db/idempotency"; +import { readAgentActionResult } from "@crm/validation/agent-action"; import { BadRequestException, ConflictException, @@ -85,6 +86,7 @@ export class AgentRunsService { attemptCount: true, errorCode: true, errorMessage: true, + result: true, plannedAt: true, startedAt: true, completedAt: true, @@ -110,6 +112,7 @@ export class AgentRunsService { })), actions: run.actions.map((action) => ({ ...action, + result: listedActionResult(action.type, action.result), plannedAt: action.plannedAt.toISOString(), startedAt: action.startedAt?.toISOString() ?? null, completedAt: action.completedAt?.toISOString() ?? null, @@ -443,3 +446,11 @@ export class AgentRunsService { } } } + +function listedActionResult(type: string, value: Prisma.JsonValue | null) { + try { + return readAgentActionResult(type, value); + } catch { + return null; + } +} diff --git a/apps/api/src/agent/agent-trigger.service.ts b/apps/api/src/agent/agent-trigger.service.ts index 3b2143302..731bd2e59 100644 --- a/apps/api/src/agent/agent-trigger.service.ts +++ b/apps/api/src/agent/agent-trigger.service.ts @@ -555,6 +555,50 @@ export class AgentTriggerService { void this.redeliverCancellations(); } + async slackEventReceived(input: { + eventId: string; + type: string; + teamId?: string; + channelId?: string; + messageTs?: string; + payload: Prisma.InputJsonValue; + }): Promise<{ stored: boolean }> { + const existing = await this.db.slackEventInbox.findUnique({ + where: { eventId: input.eventId }, + select: { id: true }, + }); + + if (existing) return { stored: false }; + + try { + await this.db.slackEventInbox.create({ + data: { + eventId: input.eventId, + type: input.type, + teamId: input.teamId ?? null, + channelId: input.channelId ?? null, + messageTs: input.messageTs ?? null, + payload: input.payload, + }, + }); + } catch (error) { + if ( + error instanceof Prisma.PrismaClientKnownRequestError && + error.code === "P2002" + ) { + return { stored: false }; + } + this.logger.error( + { message: "Could not store Slack event", eventId: input.eventId }, + error instanceof Error ? error.stack : String(error), + ); + throw error; + } + + this.poke(); + return { stored: true }; + } + private poke(): void { this.pokeRoute("/internal/crm/dispatch"); } diff --git a/apps/api/src/agent/agents.contracts.ts b/apps/api/src/agent/agents.contracts.ts index 844e16dc0..402bae747 100644 --- a/apps/api/src/agent/agents.contracts.ts +++ b/apps/api/src/agent/agents.contracts.ts @@ -1,4 +1,5 @@ import { schemas } from "@crm/validation"; +import { agentActionResult } from "@crm/validation/agent-action"; import { z } from "zod"; export const agentManifest = schemas.agents.capabilities.loose(); @@ -269,6 +270,7 @@ const agentRunActionOutput = z.object({ attemptCount: z.number(), errorCode: z.string().nullable(), errorMessage: z.string().nullable(), + result: agentActionResult.nullable(), plannedAt: z.string(), startedAt: z.string().nullable(), completedAt: z.string().nullable(), diff --git a/apps/api/src/config/env.validation.ts b/apps/api/src/config/env.validation.ts index 08cb676c2..47b324b80 100644 --- a/apps/api/src/config/env.validation.ts +++ b/apps/api/src/config/env.validation.ts @@ -76,6 +76,10 @@ export class EnvironmentVariables { @IsString() SLACK_CLIENT_SECRET?: string; + @IsOptional() + @IsString() + SLACK_SIGNING_SECRET?: string; + @IsOptional() @IsUrl({ require_tld: false }) API_URL?: string; diff --git a/apps/api/src/create-app.ts b/apps/api/src/create-app.ts index 4a436e71c..6885950c5 100644 --- a/apps/api/src/create-app.ts +++ b/apps/api/src/create-app.ts @@ -6,7 +6,7 @@ import { type NestExpressApplication, } from "@nestjs/platform-express"; import { DocumentBuilder, SwaggerModule } from "@nestjs/swagger"; -import type { NextFunction, Request, Response } from "express"; +import { type NextFunction, type Request, type Response, raw } from "express"; import helmet from "helmet"; import { AppRouterHost } from "nestjs-trpc"; import { @@ -15,6 +15,8 @@ import { } from "trpc-to-openapi"; import { AppModule } from "./app.module"; import { ContextLogger } from "./logging/context-logger"; +import { SLACK } from "./slack/slack-config"; +import { SLACK_EVENTS_PATH } from "./slack/slack-events.controller"; import { REST_BRIDGE_PATH } from "./trpc/openapi"; import { createBaseTrpcContext } from "./trpc/trpc.context"; @@ -26,6 +28,10 @@ export async function createApp(): Promise { ); app.use(helmet()); + app.use( + SLACK_EVENTS_PATH, + raw({ type: "*/*", limit: SLACK.events.maxBodyBytes }), + ); app.useGlobalPipes( new ValidationPipe({ whitelist: true, diff --git a/apps/api/src/slack/slack-config.ts b/apps/api/src/slack/slack-config.ts index 324087188..5009d38da 100644 --- a/apps/api/src/slack/slack-config.ts +++ b/apps/api/src/slack/slack-config.ts @@ -10,6 +10,9 @@ export const SLACK = { pageSize: 50, maxPageSize: 100, }, + events: { + maxBodyBytes: 64 * 1024, + }, } as const; export const SLACK_SYNC_STATES = ["idle", "syncing", "stalled"] as const; diff --git a/apps/api/src/slack/slack-events.controller.ts b/apps/api/src/slack/slack-events.controller.ts new file mode 100644 index 000000000..7b599be50 --- /dev/null +++ b/apps/api/src/slack/slack-events.controller.ts @@ -0,0 +1,90 @@ +import { verifySlackSignature } from "@crm/auth"; +import { schemas } from "@crm/validation"; +import { + Body, + Controller, + Headers, + HttpCode, + Logger, + Post, + UnauthorizedException, +} from "@nestjs/common"; +import { ConfigService } from "@nestjs/config"; +import { ApiExcludeEndpoint } from "@nestjs/swagger"; +import { AllowAnonymous } from "@thallesp/nestjs-better-auth"; +import { AgentTriggerService } from "../agent/agent-trigger.service"; +import type { EnvironmentVariables } from "../config/env.validation"; + +export const SLACK_EVENTS_PATH = "/webhooks/slack/events"; + +@Controller("webhooks/slack") +export class SlackEventsController { + private readonly logger = new Logger(SlackEventsController.name); + private readonly secret: string | undefined; + + constructor( + private readonly agent: AgentTriggerService, + config: ConfigService, + ) { + this.secret = config.get("SLACK_SIGNING_SECRET", { infer: true }); + } + + @Post("events") + @HttpCode(200) + @AllowAnonymous() + @ApiExcludeEndpoint() + async events( + @Body() raw: Buffer, + @Headers("x-slack-request-timestamp") timestamp: string | undefined, + @Headers("x-slack-signature") signature: string | undefined, + ): Promise<{ challenge: string } | { ok: true }> { + const body = Buffer.isBuffer(raw) + ? raw.toString("utf8") + : String(raw ?? ""); + + const verdict = verifySlackSignature({ + body, + timestamp, + signature, + secret: this.secret, + }); + + if (!verdict.ok) { + this.logger.warn({ + message: "A Slack event was refused", + reason: verdict.reason, + }); + throw new UnauthorizedException("The Slack signature did not verify."); + } + + const envelope = schemas.slackEvents.slackEnvelope.safeParse( + JSON.parse(body || "null"), + ); + + if (!envelope.success) return { ok: true }; + + if (envelope.data.type === "url_verification") { + return { challenge: envelope.data.challenge }; + } + + const { event, event_id, team_id } = envelope.data; + + if (!schemas.slackEvents.isActionable(event)) return { ok: true }; + + const { stored } = await this.agent.slackEventReceived({ + eventId: event_id, + type: event.type, + teamId: team_id, + channelId: event.channel, + messageTs: event.ts, + payload: JSON.parse(body), + }); + + this.logger.log({ + message: stored ? "Slack event stored" : "Slack event already seen", + type: event.type, + }); + + return { ok: true }; + } +} diff --git a/apps/api/src/slack/slack.module.ts b/apps/api/src/slack/slack.module.ts index 8d207e357..da17b8193 100644 --- a/apps/api/src/slack/slack.module.ts +++ b/apps/api/src/slack/slack.module.ts @@ -4,9 +4,11 @@ import { TrpcModule } from "../trpc/trpc.module"; import { SlackRouter } from "./slack.router"; import { SlackChannelsService } from "./slack-channels.service"; import { SlackConnectionService } from "./slack-connection.service"; +import { SlackEventsController } from "./slack-events.controller"; @Module({ imports: [TrpcModule, AgentModule], + controllers: [SlackEventsController], providers: [SlackChannelsService, SlackConnectionService, SlackRouter], exports: [SlackConnectionService], }) diff --git a/apps/api/test/agent-events.spec.ts b/apps/api/test/agent-events.spec.ts index 2fe38f520..745cb73b0 100644 --- a/apps/api/test/agent-events.spec.ts +++ b/apps/api/test/agent-events.spec.ts @@ -43,6 +43,9 @@ beforeAll(async () => { }); afterAll(async () => { + await db.slackEventInbox.deleteMany({ + where: { eventId: { startsWith: `Ev-${suffix}-` } }, + }); await db.agentTask.deleteMany({ where: { OR: [ @@ -255,3 +258,44 @@ describe("CRM agent events", () => { ).toBe(2); }); }); + +describe("Slack event inbox writes", () => { + it("stores a Slack event once and treats a duplicate as already seen", async () => { + const eventId = `Ev-${suffix}-dup`; + const payload = { type: "event_callback", event_id: eventId }; + + expect( + await service.slackEventReceived({ + eventId, + type: "message", + payload, + }), + ).toEqual({ stored: true }); + expect( + await service.slackEventReceived({ + eventId, + type: "message", + payload, + }), + ).toEqual({ stored: false }); + }); + + it("rethrows a Slack inbox write that is not a duplicate", async () => { + const original = db.slackEventInbox.create.bind(db.slackEventInbox); + db.slackEventInbox.create = (async () => { + throw new Error("the inbox is unreachable"); + }) as typeof db.slackEventInbox.create; + + try { + await expect( + service.slackEventReceived({ + eventId: `Ev-${suffix}-fail`, + type: "message", + payload: { type: "event_callback" }, + }), + ).rejects.toThrow("the inbox is unreachable"); + } finally { + db.slackEventInbox.create = original; + } + }); +}); diff --git a/apps/api/test/agent-runs.spec.ts b/apps/api/test/agent-runs.spec.ts index 22d4d8c40..7f533f3ec 100644 --- a/apps/api/test/agent-runs.spec.ts +++ b/apps/api/test/agent-runs.spec.ts @@ -224,6 +224,43 @@ describe("manual agent runs", () => { }); }); + it("returns null for one unreadable action result and still lists the run", async () => { + const { id: runId } = await service.runNow( + { id: agentId, clientRequestId: crypto.randomUUID() }, + userId, + ); + await db.agentAction.createMany({ + data: [ + { + agentId, + runId, + type: "run.summary", + provider: "crm", + summary: "Broken result", + status: "SUCCEEDED", + idempotencyKey: `bad-result-${crypto.randomUUID()}`, + result: { not: "a stored result" }, + }, + { + agentId, + runId, + type: "run.summary", + provider: "crm", + summary: "Good result", + status: "SUCCEEDED", + idempotencyKey: `good-result-${crypto.randomUUID()}`, + result: { type: "run.summary" }, + }, + ], + }); + + const runs = await service.list(agentId, 1, userId); + expect(runs).toHaveLength(1); + const results = runs[0]?.actions.map((action) => action.result) ?? []; + expect(results).toContain(null); + expect(results).toContainEqual({ type: "run.summary" }); + }); + it("rejects a manual run while an agent is not live", async () => { const beforePokeCount = pokeCount; const draft = await db.agentDefinition.create({ diff --git a/apps/api/test/slack-events.spec.ts b/apps/api/test/slack-events.spec.ts new file mode 100644 index 000000000..17468639c --- /dev/null +++ b/apps/api/test/slack-events.spec.ts @@ -0,0 +1,198 @@ +import { beforeEach, describe, expect, it } from "bun:test"; +import { signBody } from "@crm/auth"; +import type { SlackEnvelope, SlackEvent } from "@crm/validation"; +import { UnauthorizedException } from "@nestjs/common"; +import { SlackEventsController } from "../src/slack/slack-events.controller"; + +const secret = "test-signing-secret"; + +type Stored = { + eventId: string; + type: string; + teamId?: string; + channelId?: string; + messageTs?: string; +}; + +const stored: Stored[] = []; +let nextResult = { stored: true }; + +const agent = { + slackEventReceived: async (input: Stored) => { + stored.push(input); + return nextResult; + }, +} as never; + +const config = { + get: () => secret, +} as never; + +const controller = new SlackEventsController(agent, config); + +const post = ( + payload: SlackEnvelope | { type: string; nested?: { a: number } }, + over: { secret?: string; skew?: number } = {}, +) => { + const body = JSON.stringify(payload); + const timestamp = String(Math.floor(Date.now() / 1000) + (over.skew ?? 0)); + const signature = signBody(body, timestamp, over.secret ?? secret); + + return controller.events(Buffer.from(body), timestamp, signature); +}; + +const callback = (event: SlackEvent, eventId = "Ev1") => ({ + type: "event_callback", + event_id: eventId, + team_id: "T1", + event, +}); + +beforeEach(() => { + stored.length = 0; + nextResult = { stored: true }; +}); + +describe("the Slack events endpoint", () => { + it("answers Slack's setup handshake with the challenge", async () => { + const result = await post({ + type: "url_verification", + challenge: "abc123", + }); + + expect(result).toEqual({ challenge: "abc123" }); + expect(stored).toHaveLength(0); + }); + + it("stores a member-joined event for the agent to act on", async () => { + await post( + callback({ type: "member_joined_channel", channel: "C1", user: "U1" }), + ); + + expect(stored).toHaveLength(1); + expect(stored[0]).toMatchObject({ + eventId: "Ev1", + type: "member_joined_channel", + channelId: "C1", + teamId: "T1", + }); + }); + + it("stores a human message", async () => { + await post( + callback({ + type: "message", + channel: "C1", + user: "U1", + text: "org_abc123", + }), + ); + + expect(stored).toHaveLength(1); + }); + + it("stores a mention, which is how somebody asks the agent for help", async () => { + await post( + callback({ + type: "app_mention", + channel: "C1", + user: "U1", + text: "<@U9> where are we", + ts: "1700000000.000100", + }), + ); + + expect(stored).toHaveLength(1); + expect(stored[0]).toMatchObject({ + type: "app_mention", + messageTs: "1700000000.000100", + }); + }); + + it("refuses a body that was not signed with our secret", async () => { + const attempt = post( + callback({ type: "member_joined_channel", channel: "C1" }), + { secret: "someone-elses-secret" }, + ); + + await expect(attempt).rejects.toThrow(UnauthorizedException); + expect(stored).toHaveLength(0); + }); + + it("refuses a replayed request from outside the window", async () => { + const attempt = post( + callback({ type: "member_joined_channel", channel: "C1" }), + { skew: -3600 }, + ); + + await expect(attempt).rejects.toThrow(UnauthorizedException); + expect(stored).toHaveLength(0); + }); + + it("ignores our own bot, which would otherwise talk to itself", async () => { + await post( + callback({ + type: "message", + channel: "C1", + text: "hello", + bot_id: "B1", + }), + ); + + expect(stored).toHaveLength(0); + }); + + it("ignores an event type we do not act on", async () => { + await post(callback({ type: "reaction_added", channel: "C1" })); + + expect(stored).toHaveLength(0); + }); + + it("answers 200 for a redelivery rather than storing it twice", async () => { + nextResult = { stored: false }; + + const result = await post( + callback({ type: "member_joined_channel", channel: "C1" }), + ); + + expect(result).toEqual({ ok: true }); + }); + + it("answers 200 to a shape it cannot read, so Slack stops retrying", async () => { + const result = await post({ type: "something_new", nested: { a: 1 } }); + + expect(result).toEqual({ ok: true }); + expect(stored).toHaveLength(0); + }); + + it("refuses everything when no signing secret is configured", async () => { + const unconfigured = new SlackEventsController(agent, { + get: () => undefined, + } as never); + + const body = JSON.stringify({ type: "url_verification", challenge: "x" }); + const timestamp = String(Math.floor(Date.now() / 1000)); + + await expect( + unconfigured.events( + Buffer.from(body), + timestamp, + signBody(body, timestamp, secret), + ), + ).rejects.toThrow(UnauthorizedException); + }); +}); + +describe("the Slack events body cap", () => { + it("uses Express raw middleware with an explicit size limit, then verifies", async () => { + const source = await Bun.file( + new URL("../src/create-app.ts", import.meta.url), + ).text(); + + expect(source).toContain( + 'raw({ type: "*/*", limit: SLACK.events.maxBodyBytes })', + ); + expect(source).not.toContain("collectRawBody"); + expect(source).toContain("SLACK_EVENTS_PATH"); + }); +}); diff --git a/apps/app/components/agent-builder/agent-history.tsx b/apps/app/components/agent-builder/agent-history.tsx index 7df33a1ea..76c6f6b30 100644 --- a/apps/app/components/agent-builder/agent-history.tsx +++ b/apps/app/components/agent-builder/agent-history.tsx @@ -17,6 +17,7 @@ import { } from "@crm/ui/components/alert-dialog"; import { Button } from "@crm/ui/components/button"; import { Icon } from "@crm/ui/components/icon"; +import { Link } from "@crm/ui/components/link"; import { cn } from "@crm/ui/lib/utils"; import { useState } from "react"; import { z } from "zod"; @@ -296,7 +297,7 @@ function ExpandedRun({ run }: { run: RunRow }) { - {entry.action.externalId ?? entry.action.id.slice(0, 12)} + {actionReceipt(entry.action)} ), @@ -408,6 +409,28 @@ export function AgentActivity({ activity }: { activity: Activity }) { ); } +function actionReceipt(action: RunRow["actions"][number]) { + if (action.result?.type !== "slack.channel.invite") { + return action.externalId ?? action.id.slice(0, 12); + } + + const inviteId = action.result.invite_id ?? action.externalId; + const url = action.result.url; + if (!inviteId && !url) return action.id.slice(0, 12); + + return ( + <> + {inviteId} + {inviteId && url ? " · " : null} + {url ? ( + + Invite link + + ) : null} + + ); +} + function humanStatus(value: string): string { return value .toLowerCase() diff --git a/apps/app/lib/agent-transcript.ts b/apps/app/lib/agent-transcript.ts index 2898c35bb..1dc001ebd 100644 --- a/apps/app/lib/agent-transcript.ts +++ b/apps/app/lib/agent-transcript.ts @@ -77,6 +77,8 @@ const VERBS: ToolVerbs = { schedule_recheck: "Decided when to look again", record_job_change: "Raised a job change", list_deals: "Reviewed the deal pipeline", + open_slack_channel: "Opened a Slack channel", + invite_to_slack_channel: "Invited people to the Slack channel", list_outstanding_work: "Looked for outstanding work", set_chat_title: "Named this chat", list_fields: "Read what this workspace tracks", diff --git a/bun.lock b/bun.lock index 1c6df3e63..ce9f1476d 100644 --- a/bun.lock +++ b/bun.lock @@ -57,6 +57,7 @@ "class-transformer": "^0.5.1", "class-validator": "^0.15.1", "context.dev": "^2.7.0", + "express": "^5.2.1", "helmet": "^8.3.0", "nestjs-trpc": "^2.13.0", "reflect-metadata": "^0.2.2", diff --git a/docs/environment.md b/docs/environment.md index 22417c60e..73f0c57db 100644 --- a/docs/environment.md +++ b/docs/environment.md @@ -52,6 +52,13 @@ list, read by the sign-in guard *and* the sync's "which side is external" decisi if they drifted a colleague would be refused at the door or filed as a lead. **An empty list fails closed.** Parsed on demand. `packages/auth/src/workspace.ts`. +**`SLACK_SIGNING_SECRET`** is required only for inbound Slack, and is separate +from the OAuth pair. `/webhooks/slack/events` is a public POST, so the signature +is the whole of its authentication: unset, the endpoint refuses everything rather +than trusting anyone. It is the Signing Secret on the Slack app's Basic +Information page, not a token. Point the app's Event Subscriptions request URL at +`API_URL` + `/webhooks/slack/events`. + ## Where things are - **`API_URL`** (`:3001`) mints session cookies and serves `/api/auth/*`; diff --git a/docs/setup.md b/docs/setup.md index 03b8912f7..7dfd33a3c 100644 --- a/docs/setup.md +++ b/docs/setup.md @@ -49,6 +49,30 @@ AGENT_BRIDGE_SECRET="$(openssl rand -base64 32)" the bridge — send `-H 'Host: agent.example.com'`. `GET /eve/v1/info` is the whole inventory, including a `diagnostics` count that finds files eve silently ignored. +## Inbound Slack needs one hostname, forever + +Slack posts events to a request URL it stores once. `cloudflared tunnel --url` +invents a new hostname on every restart, so the stored URL goes stale and Slack +stops delivering **without an error anywhere** — the endpoint is simply never +called. A named tunnel keeps the hostname across restarts. + +```sh +brew install cloudflared +cloudflared tunnel login # opens a browser, once +SLACK_TUNNEL_HOSTNAME="crm-dev.example.com" bun run tunnel:slack +``` + +The hostname must be on a domain in your own Cloudflare account. `tunnel:slack` +creates the tunnel if it is missing, points the DNS record at it, prints the +request URL and then runs it. Re-running is safe. Put the hostname in `.env` and +the variable can be dropped from the command. + +Paste the printed URL into the Slack app's Event Subscriptions page and +subscribe to `message.channels`, `app_mention` and `member_joined_channel`. + +**Socket Mode swallows events.** With Socket Mode on, the Request URL still shows +"Verified" and no HTTP delivery ever happens. Turn it off. + ## Running the agent The agent package's default `dev` command is interactive `eve dev`. The root diff --git a/package.json b/package.json index 534ec9d00..8e11fda6c 100644 --- a/package.json +++ b/package.json @@ -21,7 +21,8 @@ "db:reset": "turbo run db:reset", "db:seed": "turbo run db:seed", "db:studio": "turbo run db:studio", - "db:test": "bun run --filter=@crm/db db:test" + "db:test": "bun run --filter=@crm/db db:test", + "tunnel:slack": "bash scripts/slack-tunnel.sh" }, "devDependencies": { "@biomejs/biome": "^2.4.10", diff --git a/packages/auth/src/index.ts b/packages/auth/src/index.ts index 31fbcbaad..ac0e92253 100644 --- a/packages/auth/src/index.ts +++ b/packages/auth/src/index.ts @@ -66,6 +66,12 @@ export { slackScopeDrift, summariseSlackScopes, } from "./slack-scopes"; +export { + type SignatureVerdict, + SLACK_SIGNATURE, + signBody, + verifySlackSignature, +} from "./slack-signature"; export { canConfigureSso, ssoCallbackBase, diff --git a/packages/auth/src/slack-scopes.ts b/packages/auth/src/slack-scopes.ts index 898036c98..939b7685a 100644 --- a/packages/auth/src/slack-scopes.ts +++ b/packages/auth/src/slack-scopes.ts @@ -32,6 +32,12 @@ export const SLACK_SCOPES: readonly SlackScope[] = [ grant: "See private channels it has been added to", sensitive: false, }, + { + scope: "app_mentions:read", + group: "read", + grant: "See messages that mention it, so somebody can ask it for help", + sensitive: false, + }, { scope: "channels:history", group: "read", diff --git a/packages/auth/src/slack-signature.ts b/packages/auth/src/slack-signature.ts new file mode 100644 index 000000000..a8b30c152 --- /dev/null +++ b/packages/auth/src/slack-signature.ts @@ -0,0 +1,63 @@ +import { createHmac, timingSafeEqual } from "node:crypto"; + +export const SLACK_SIGNATURE = { + version: "v0", + timestampHeader: "x-slack-request-timestamp", + signatureHeader: "x-slack-signature", + toleranceSeconds: 300, +} as const; + +export type SignatureVerdict = { ok: true } | { ok: false; reason: string }; + +export type SignatureInput = { + body: string; + timestamp: string | undefined; + signature: string | undefined; + secret: string | undefined; + now?: number; +}; + +export function verifySlackSignature(input: SignatureInput): SignatureVerdict { + const { body, timestamp, signature, secret } = input; + + if (!secret) return { ok: false, reason: "no signing secret is configured" }; + if (!timestamp) + return { ok: false, reason: "the timestamp header is missing" }; + if (!signature) + return { ok: false, reason: "the signature header is missing" }; + + const sent = Number(timestamp); + if (!Number.isFinite(sent)) { + return { ok: false, reason: "the timestamp is not a number" }; + } + + const now = Math.floor((input.now ?? Date.now()) / 1000); + if (Math.abs(now - sent) > SLACK_SIGNATURE.toleranceSeconds) { + return { ok: false, reason: "the timestamp is outside the replay window" }; + } + + const expected = signBody(body, timestamp, secret); + return sameSignature(expected, signature) + ? { ok: true } + : { ok: false, reason: "the signature does not match" }; +} + +export function signBody( + body: string, + timestamp: string, + secret: string, +): string { + const digest = createHmac("sha256", secret) + .update(`${SLACK_SIGNATURE.version}:${timestamp}:${body}`) + .digest("hex"); + + return `${SLACK_SIGNATURE.version}=${digest}`; +} + +function sameSignature(expected: string, received: string): boolean { + const a = Buffer.from(expected, "utf8"); + const b = Buffer.from(received, "utf8"); + + if (a.length !== b.length) return false; + return timingSafeEqual(a, b); +} diff --git a/packages/auth/test/slack-signature.spec.ts b/packages/auth/test/slack-signature.spec.ts new file mode 100644 index 000000000..92df363d0 --- /dev/null +++ b/packages/auth/test/slack-signature.spec.ts @@ -0,0 +1,102 @@ +import { describe, expect, it } from "bun:test"; +import { + SLACK_SIGNATURE, + signBody, + verifySlackSignature, +} from "../src/slack-signature"; + +const secret = "8f742231b10e8888abcd99yyyzzz85a5"; +const body = + '{"type":"event_callback","event":{"type":"member_joined_channel"}}'; +const now = 1_700_000_000_000; +const timestamp = String(Math.floor(now / 1000)); +const signature = signBody(body, timestamp, secret); + +const verify = ( + overrides: Partial[0]>, +) => + verifySlackSignature({ + body, + timestamp, + signature, + secret, + now, + ...overrides, + }); + +describe("verifySlackSignature", () => { + it("accepts a body Slack actually signed", () => { + expect(verify({})).toEqual({ ok: true }); + }); + + it("refuses a body that changed by one byte", () => { + const tampered = `${body.slice(0, -1)} `; + expect(verify({ body: tampered }).ok).toBe(false); + }); + + it("refuses a signature from a different secret", () => { + const other = signBody(body, timestamp, "a-different-signing-secret"); + expect(verify({ signature: other }).ok).toBe(false); + }); + + it("refuses a replay outside the window", () => { + const old = String( + Math.floor(now / 1000) - SLACK_SIGNATURE.toleranceSeconds - 1, + ); + const verdict = verify({ + timestamp: old, + signature: signBody(body, old, secret), + }); + + expect(verdict).toEqual({ + ok: false, + reason: "the timestamp is outside the replay window", + }); + }); + + it("accepts a request at the edge of the window", () => { + const edge = String( + Math.floor(now / 1000) - SLACK_SIGNATURE.toleranceSeconds, + ); + expect( + verify({ timestamp: edge, signature: signBody(body, edge, secret) }).ok, + ).toBe(true); + }); + + it("refuses a future timestamp outside the window", () => { + const ahead = String( + Math.floor(now / 1000) + SLACK_SIGNATURE.toleranceSeconds + 1, + ); + expect( + verify({ timestamp: ahead, signature: signBody(body, ahead, secret) }).ok, + ).toBe(false); + }); + + it("fails closed when no signing secret is configured", () => { + expect(verify({ secret: undefined })).toEqual({ + ok: false, + reason: "no signing secret is configured", + }); + }); + + it("refuses a request missing either header", () => { + expect(verify({ timestamp: undefined }).ok).toBe(false); + expect(verify({ signature: undefined }).ok).toBe(false); + }); + + it("refuses a timestamp that is not a number", () => { + expect(verify({ timestamp: "not-a-time" })).toEqual({ + ok: false, + reason: "the timestamp is not a number", + }); + }); + + it("refuses an empty signature rather than comparing lengths oddly", () => { + expect(verify({ signature: "" }).ok).toBe(false); + }); + + it("signs in the exact form Slack documents", () => { + expect(signature.startsWith(`${SLACK_SIGNATURE.version}=`)).toBe(true); + expect(signature).toHaveLength(SLACK_SIGNATURE.version.length + 1 + 64); + }); +}); diff --git a/packages/db/prisma/migrations/20260826183121_slack_event_resume/migration.sql b/packages/db/prisma/migrations/20260826183121_slack_event_resume/migration.sql new file mode 100644 index 000000000..31005229b --- /dev/null +++ b/packages/db/prisma/migrations/20260826183121_slack_event_resume/migration.sql @@ -0,0 +1,30 @@ +-- AlterTable +ALTER TABLE "agentRun" ADD COLUMN "slackChannelId" TEXT; + +-- CreateTable +CREATE TABLE "slackEventInbox" ( + "id" TEXT NOT NULL, + "eventId" TEXT NOT NULL, + "teamId" TEXT, + "channelId" TEXT, + "type" TEXT NOT NULL, + "payload" JSONB NOT NULL, + "runId" TEXT, + "receivedAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "processedAt" TIMESTAMP(3), + "outcome" TEXT, + + CONSTRAINT "slackEventInbox_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE UNIQUE INDEX "slackEventInbox_eventId_key" ON "slackEventInbox"("eventId"); + +-- CreateIndex +CREATE INDEX "slackEventInbox_processedAt_receivedAt_idx" ON "slackEventInbox"("processedAt", "receivedAt"); + +-- CreateIndex +CREATE INDEX "slackEventInbox_channelId_idx" ON "slackEventInbox"("channelId"); + +-- CreateIndex +CREATE INDEX "agentRun_slackChannelId_status_idx" ON "agentRun"("slackChannelId", "status"); diff --git a/packages/db/prisma/migrations/20260828151505_slack_event_message_ts/migration.sql b/packages/db/prisma/migrations/20260828151505_slack_event_message_ts/migration.sql new file mode 100644 index 000000000..b35eef674 --- /dev/null +++ b/packages/db/prisma/migrations/20260828151505_slack_event_message_ts/migration.sql @@ -0,0 +1,11 @@ +/* + Warnings: + + - A unique constraint covering the columns `[channelId,messageTs]` on the table `slackEventInbox` will be added. If there are existing duplicate values, this will fail. + +*/ +-- AlterTable +ALTER TABLE "slackEventInbox" ADD COLUMN "messageTs" TEXT; + +-- CreateIndex +CREATE UNIQUE INDEX "slackEventInbox_channelId_messageTs_key" ON "slackEventInbox"("channelId", "messageTs"); diff --git a/packages/db/prisma/migrations/20260830083000_agent_action_result/migration.sql b/packages/db/prisma/migrations/20260830083000_agent_action_result/migration.sql new file mode 100644 index 000000000..750a4a42c --- /dev/null +++ b/packages/db/prisma/migrations/20260830083000_agent_action_result/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "agentAction" ADD COLUMN "result" JSONB; diff --git a/packages/db/prisma/migrations/20260830093300_slack_event_inbox_lease/migration.sql b/packages/db/prisma/migrations/20260830093300_slack_event_inbox_lease/migration.sql new file mode 100644 index 000000000..c91b656c4 --- /dev/null +++ b/packages/db/prisma/migrations/20260830093300_slack_event_inbox_lease/migration.sql @@ -0,0 +1,2 @@ +-- AlterTable +ALTER TABLE "slackEventInbox" ADD COLUMN "leasedUntil" TIMESTAMP(3); diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index 82bc1f368..aed6d00fa 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -108,6 +108,29 @@ model SlackWorkspaceGrant { @@map("slackWorkspaceGrant") } +model SlackEventInbox { + id String @id @default(cuid()) + + eventId String @unique + teamId String? + channelId String? + messageTs String? + type String + payload Json + + runId String? + + receivedAt DateTime @default(now()) + leasedUntil DateTime? + processedAt DateTime? + outcome String? + + @@unique([channelId, messageTs]) + @@index([processedAt, receivedAt]) + @@index([channelId]) + @@map("slackEventInbox") +} + model Session { id String @id expiresAt DateTime @@ -793,6 +816,7 @@ model AgentRun { triggerType AgentTriggerType status AgentRunStatus @default(QUEUED) + slackChannelId String? principalId String? sessionId String? @unique idempotencyKey String @unique @@ -827,6 +851,7 @@ model AgentRun { @@index([versionId, createdAt]) @@index([status, createdAt]) @@index([triggerId, createdAt]) + @@index([slackChannelId, status]) @@map("agentRun") } @@ -862,6 +887,7 @@ model AgentAction { targetLabel String? summary String metadata Json? + result Json? status AgentActionStatus @default(PLANNED) idempotencyKey String @unique diff --git a/packages/validation/package.json b/packages/validation/package.json index d6fb2c36a..77fc8263c 100644 --- a/packages/validation/package.json +++ b/packages/validation/package.json @@ -7,6 +7,7 @@ ".": "./src/index.ts", "./activity-meta": "./src/activity-meta.ts", "./agent-events": "./src/agent-events.ts", + "./agent-action": "./src/agent-action.ts", "./agent-manifest": "./src/agent-manifest.ts", "./builder-question": "./src/builder-question.ts", "./enrichment-queue": "./src/enrichment-queue.ts", @@ -14,7 +15,8 @@ "./eve-tool": "./src/eve-tool.ts", "./field-backfill": "./src/field-backfill.ts", "./field-templates": "./src/field-templates.ts", - "./saved-view": "./src/saved-view.ts" + "./saved-view": "./src/saved-view.ts", + "./slack-events": "./src/slack-events.ts" }, "scripts": { "check-types": "tsc --noEmit", diff --git a/packages/validation/src/agent-action.ts b/packages/validation/src/agent-action.ts new file mode 100644 index 000000000..22eaf3448 --- /dev/null +++ b/packages/validation/src/agent-action.ts @@ -0,0 +1,71 @@ +import { z } from "zod"; +import { AGENT_ACTION_TYPES } from "./agent-manifest"; + +const present = z.string().trim().min(1); + +export const result = z.discriminatedUnion("type", [ + z.object({ + type: z.literal(AGENT_ACTION_TYPES.CRM_ACTIVITY_CREATE), + activityId: present, + }), + z.object({ + type: z.literal(AGENT_ACTION_TYPES.RUN_SUMMARY), + }), + z.object({ + type: z.literal(AGENT_ACTION_TYPES.SLACK_MESSAGE_POST), + channel: present, + ts: present, + }), + z.object({ + type: z.literal(AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN), + channelId: present, + }), + z.object({ + type: z.literal(AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE), + invite_id: present.optional(), + url: present.optional(), + email: z.email(), + kind: z.enum(["member", "connect"]), + }), +]); + +export const storedResult = result.nullable(); + +export const agentActionResult = result; + +export type AgentActionResult = z.infer; + +function issues(error: z.ZodError): string { + return error.issues + .map((issue) => + issue.path.length > 0 + ? `${issue.path.join(".")} ${issue.message}` + : issue.message, + ) + .join("; "); +} + +export function parseAgentActionResult(value: unknown): AgentActionResult { + const parsed = result.safeParse(value); + if (parsed.success) return parsed.data; + throw new Error(`This agent action result: ${issues(parsed.error)}`); +} + +export function readAgentActionResult( + type: string, + value: unknown, +): AgentActionResult | null { + if (value === null || value === undefined) return null; + + const parsed = storedResult.safeParse(value); + if (!parsed.success) { + throw new Error(`This agent action result: ${issues(parsed.error)}`); + } + if (parsed.data === null) return null; + if (parsed.data.type !== type) { + throw new Error( + `This agent action result: type ${parsed.data.type} does not match action ${type}`, + ); + } + return parsed.data; +} diff --git a/packages/validation/src/agent-events.ts b/packages/validation/src/agent-events.ts index d4ce5bb2f..401b726d2 100644 --- a/packages/validation/src/agent-events.ts +++ b/packages/validation/src/agent-events.ts @@ -20,3 +20,19 @@ export const crmEventTask = z ); export type CrmEventTask = z.infer; + +export const agentRunInput = z.object({ + event: z.object({ type: z.enum(CRM_EVENT_TYPES) }).optional(), + record: z + .object({ + kind: z.enum(["company", "contact", "deal"]), + id: z.string().trim().min(1), + }) + .optional(), +}); + +export type AgentRunInput = z.infer; + +export function runRecord(value: unknown): AgentRunInput["record"] { + return agentRunInput.safeParse(value).data?.record; +} diff --git a/packages/validation/src/agent-manifest.ts b/packages/validation/src/agent-manifest.ts index 9a496e8d1..9536ab8ae 100644 --- a/packages/validation/src/agent-manifest.ts +++ b/packages/validation/src/agent-manifest.ts @@ -5,8 +5,12 @@ export const AGENT_ACTION_TYPES = { CRM_ACTIVITY_CREATE: "crm.activity.create", RUN_SUMMARY: "run.summary", SLACK_MESSAGE_POST: "slack.message.post", + SLACK_CHANNEL_OPEN: "slack.channel.open", + SLACK_CHANNEL_INVITE: "slack.channel.invite", } as const; +export const SLACK_WORKSPACE_RESOURCE_ID = "slack:workspace"; + export type AgentActionType = (typeof AGENT_ACTION_TYPES)[keyof typeof AGENT_ACTION_TYPES]; @@ -44,6 +48,16 @@ export const agentManifestAction = z.discriminatedUnion("type", [ summary: z.string(), destination: slackDestination, }), + z.object({ + type: z.literal(AGENT_ACTION_TYPES.SLACK_CHANNEL_OPEN), + provider: z.literal("slack"), + summary: z.string(), + }), + z.object({ + type: z.literal(AGENT_ACTION_TYPES.SLACK_CHANNEL_INVITE), + provider: z.literal("slack"), + summary: z.string(), + }), ]); export const agentScheduleTriggerConfig = z.object({ @@ -105,6 +119,22 @@ export const agentManifest = z } actionTypes.add(action.type); } + + const slack = manifest.actions.findIndex( + (action) => action.provider === "slack", + ); + const workspace = manifest.dataScope.resources.some( + (resource) => + resource.kind === "integration" && + resource.id === SLACK_WORKSPACE_RESOURCE_ID, + ); + if (slack >= 0 && !workspace) { + context.addIssue({ + code: "custom", + path: ["actions", slack, "provider"], + message: `A Slack action needs the ${SLACK_WORKSPACE_RESOURCE_ID} resource, or it fails on every run`, + }); + } }); export type SlackDestination = z.infer; diff --git a/packages/validation/src/index.ts b/packages/validation/src/index.ts index 0cf43c9ca..3e6f6c71a 100644 --- a/packages/validation/src/index.ts +++ b/packages/validation/src/index.ts @@ -1,5 +1,6 @@ import type { ZodType, z } from "zod"; import * as activityMeta from "./activity-meta"; +import * as agentAction from "./agent-action"; import * as agentEvents from "./agent-events"; import * as agentManifest from "./agent-manifest"; import * as agents from "./agents"; @@ -7,9 +8,11 @@ import * as builderQuestion from "./builder-question"; import * as eveStream from "./eve-stream"; import * as eveTool from "./eve-tool"; import * as slack from "./slack"; +import * as slackEvents from "./slack-events"; export const schemas = { activityMeta, + agentAction, agentEvents, agentManifest, agents, @@ -17,9 +20,11 @@ export const schemas = { eveStream, eveTool, slack, + slackEvents, } as const; export type { ActivityMeta, ActivityMetaFields } from "./activity-meta"; +export type { AgentActionResult } from "./agent-action"; export type { CrmEventTask } from "./agent-events"; export type { AgentActionType, @@ -57,6 +62,11 @@ export type { EveToolOutput, } from "./eve-tool"; export type { AuthTest, JoinPayload, OauthAccess, Reply } from "./slack"; +export type { + EventCallback, + SlackEnvelope, + SlackEvent, +} from "./slack-events"; export class InvalidInput extends Error { override readonly name = "InvalidInput"; diff --git a/packages/validation/src/slack-events.ts b/packages/validation/src/slack-events.ts new file mode 100644 index 000000000..a02c3a30c --- /dev/null +++ b/packages/validation/src/slack-events.ts @@ -0,0 +1,57 @@ +import { z } from "zod"; + +const trimmed = z.string().trim().min(1); + +export const SLACK_EVENT_TYPES = { + MEMBER_JOINED: "member_joined_channel", + MESSAGE: "message", + APP_MENTION: "app_mention", +} as const; + +export const urlVerification = z.object({ + type: z.literal("url_verification"), + challenge: trimmed, +}); + +export const slackEvent = z.object({ + type: trimmed, + channel: z.string().trim().optional(), + user: z.string().trim().optional(), + text: z.string().optional(), + ts: z.string().trim().optional(), + thread_ts: z.string().trim().optional(), + bot_id: z.string().trim().optional(), + subtype: z.string().trim().optional(), +}); + +export const eventCallback = z.object({ + type: z.literal("event_callback"), + event_id: trimmed, + team_id: z.string().trim().optional(), + event: slackEvent, +}); + +export const slackEnvelope = z.union([urlVerification, eventCallback]); + +export type SlackEvent = z.infer; +export type EventCallback = z.infer; +export type SlackEnvelope = z.infer; + +export function isFromApp(event: SlackEvent): boolean { + return Boolean(event.bot_id) || event.subtype === "bot_message"; +} + +export function isActionable(event: SlackEvent): boolean { + if (isFromApp(event)) return false; + + if (event.type === SLACK_EVENT_TYPES.MEMBER_JOINED) return true; + if (event.type === SLACK_EVENT_TYPES.APP_MENTION) { + return Boolean(event.text?.trim()); + } + + return ( + event.type === SLACK_EVENT_TYPES.MESSAGE && + event.subtype === undefined && + Boolean(event.text?.trim()) + ); +} diff --git a/packages/validation/src/slack.ts b/packages/validation/src/slack.ts index 7d276a21b..e49e797b6 100644 --- a/packages/validation/src/slack.ts +++ b/packages/validation/src/slack.ts @@ -28,6 +28,15 @@ export const reply = z.object({ error: z.string().optional(), }); +export const lookupByEmail = reply.extend({ + user: z.object({ id: z.string().trim().min(1) }).nullish(), +}); + +export const inviteShared = reply.extend({ + invite_id: z.string().trim().min(1).optional(), + url: z.string().trim().min(1).optional(), +}); + export const authTest = reply.extend({ user_id: z.string().trim().min(1).optional(), }); diff --git a/packages/validation/test/agent-action.spec.ts b/packages/validation/test/agent-action.spec.ts new file mode 100644 index 000000000..1c4ccfece --- /dev/null +++ b/packages/validation/test/agent-action.spec.ts @@ -0,0 +1,90 @@ +import { describe, expect, it } from "bun:test"; +import { InvalidInput, parse, schemas } from "../src/index"; + +describe("AgentAction result", () => { + it("keeps invite_id and url on a Slack Connect invitation", () => { + expect( + parse( + schemas.agentAction.result, + { + type: "slack.channel.invite", + invite_id: "I1", + url: "https://slack.com/invite/x", + email: "buyer@customer.test", + kind: "connect", + }, + "This agent action result", + ), + ).toEqual({ + type: "slack.channel.invite", + invite_id: "I1", + url: "https://slack.com/invite/x", + email: "buyer@customer.test", + kind: "connect", + }); + }); + + it("keeps a member invitation that Slack never gave an id", () => { + expect( + parse( + schemas.agentAction.result, + { + type: "slack.channel.invite", + email: "rep@ours.test", + kind: "member", + }, + "This agent action result", + ), + ).toEqual({ + type: "slack.channel.invite", + email: "rep@ours.test", + kind: "member", + }); + }); + + it("stores the Slack message channel and ts", () => { + expect( + parse( + schemas.agentAction.result, + { type: "slack.message.post", channel: "C1", ts: "1.2" }, + "This agent action result", + ), + ).toEqual({ type: "slack.message.post", channel: "C1", ts: "1.2" }); + }); + + it("stores the opened channel id", () => { + expect( + parse( + schemas.agentAction.result, + { type: "slack.channel.open", channelId: "C1" }, + "This agent action result", + ), + ).toEqual({ type: "slack.channel.open", channelId: "C1" }); + }); + + it("stores the created activity id", () => { + expect( + parse( + schemas.agentAction.result, + { type: "crm.activity.create", activityId: "act-1" }, + "This agent action result", + ), + ).toEqual({ type: "crm.activity.create", activityId: "act-1" }); + }); + + it("reads a missing result as null", () => { + expect( + parse(schemas.agentAction.storedResult, null, "This agent action result"), + ).toBeNull(); + }); + + it("rejects a Connect invitation that names nobody", () => { + expect(() => + parse( + schemas.agentAction.result, + { type: "slack.channel.invite", kind: "connect" }, + "This agent action result", + ), + ).toThrow(InvalidInput); + }); +}); diff --git a/packages/validation/test/slack-events.spec.ts b/packages/validation/test/slack-events.spec.ts new file mode 100644 index 000000000..de48be2c7 --- /dev/null +++ b/packages/validation/test/slack-events.spec.ts @@ -0,0 +1,125 @@ +import { describe, expect, it } from "bun:test"; +import { + eventCallback, + isActionable, + isFromApp, + SLACK_EVENT_TYPES, + type SlackEvent, + slackEnvelope, +} from "../src/slack-events"; + +const joined = { + type: "event_callback", + event_id: "Ev123", + team_id: "T1", + event: { + type: SLACK_EVENT_TYPES.MEMBER_JOINED, + channel: "C1", + user: "U1", + }, +}; + +const message = (over: Partial = {}) => ({ + type: SLACK_EVENT_TYPES.MESSAGE, + channel: "C1", + user: "U1", + text: "org_abc123", + ts: "1700000000.000100", + ...over, +}); + +describe("the envelope Slack posts", () => { + it("reads the setup handshake", () => { + const parsed = slackEnvelope.parse({ + type: "url_verification", + challenge: "abc", + }); + + expect(parsed).toEqual({ type: "url_verification", challenge: "abc" }); + }); + + it("reads a member-joined callback", () => { + const parsed = eventCallback.parse(joined); + + expect(parsed.event.channel).toBe("C1"); + expect(parsed.event_id).toBe("Ev123"); + }); + + it("refuses a callback with no event id, so nothing is deduplicated by luck", () => { + expect(eventCallback.safeParse({ ...joined, event_id: "" }).success).toBe( + false, + ); + }); + + it("keeps an unknown event type rather than dropping the delivery", () => { + const parsed = eventCallback.parse({ + ...joined, + event: { type: "reaction_added", channel: "C1" }, + }); + + expect(parsed.event.type).toBe("reaction_added"); + }); +}); + +describe("isFromApp", () => { + it("recognises our own bot, so an agent cannot answer itself", () => { + expect(isFromApp({ type: "message", bot_id: "B1" })).toBe(true); + expect(isFromApp({ type: "message", subtype: "bot_message" })).toBe(true); + }); + + it("treats a human message as a human message", () => { + expect(isFromApp(message())).toBe(false); + }); +}); + +describe("isActionable", () => { + it("acts on a member joining", () => { + expect(isActionable(joined.event)).toBe(true); + }); + + it("acts on a plain human message with text", () => { + expect(isActionable(message())).toBe(true); + }); + + it("ignores our own bot posting, which would otherwise loop", () => { + expect(isActionable(message({ bot_id: "B1" }))).toBe(false); + expect(isActionable(message({ subtype: "bot_message" }))).toBe(false); + }); + + it("ignores edits, joins-as-message and other subtypes", () => { + expect(isActionable(message({ subtype: "message_changed" }))).toBe(false); + expect(isActionable(message({ subtype: "channel_join" }))).toBe(false); + }); + + it("ignores an empty message", () => { + expect(isActionable(message({ text: "" }))).toBe(false); + expect(isActionable(message({ text: " " }))).toBe(false); + }); + + it("acts on a mention, which is how somebody asks the agent for help", () => { + expect( + isActionable({ type: "app_mention", channel: "C1", text: "<@U1> help" }), + ).toBe(true); + }); + + it("ignores a mention with no words after it", () => { + expect( + isActionable({ type: "app_mention", channel: "C1", text: " " }), + ).toBe(false); + }); + + it("ignores a mention our own bot posted", () => { + expect( + isActionable({ + type: "app_mention", + channel: "C1", + text: "<@U1> hi", + bot_id: "B1", + }), + ).toBe(false); + }); + + it("ignores event types we do not handle", () => { + expect(isActionable({ type: "reaction_added", channel: "C1" })).toBe(false); + }); +}); diff --git a/scripts/slack-tunnel.sh b/scripts/slack-tunnel.sh new file mode 100755 index 000000000..11f1a928e --- /dev/null +++ b/scripts/slack-tunnel.sh @@ -0,0 +1,56 @@ +#!/usr/bin/env bash +set -euo pipefail + +hostname="${SLACK_TUNNEL_HOSTNAME:-}" +name="${SLACK_TUNNEL_NAME:-crm-dev}" +port="${SLACK_TUNNEL_PORT:-3001}" + +root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" + +if [ -z "$hostname" ] && [ -f "$root/.env" ]; then + hostname="$(grep -E '^SLACK_TUNNEL_HOSTNAME=' "$root/.env" | tail -1 | cut -d= -f2- | tr -d '"' | tr -d "'" || true)" +fi + +if [ -z "$hostname" ]; then + cat >&2 <<'MESSAGE' +Set SLACK_TUNNEL_HOSTNAME to the name you want, on a domain in your Cloudflare +account. It is the hostname Slack keeps, so pick one and never change it: + + SLACK_TUNNEL_HOSTNAME="crm-dev.example.com" + +MESSAGE + exit 1 +fi + +if ! command -v cloudflared >/dev/null 2>&1; then + echo "cloudflared is not installed. brew install cloudflared" >&2 + exit 1 +fi + +if [ ! -f "$HOME/.cloudflared/cert.pem" ]; then + cat >&2 <<'MESSAGE' +cloudflared is not signed in. This opens a browser once, and once only: + + cloudflared tunnel login + +MESSAGE + exit 1 +fi + +if ! cloudflared tunnel info "$name" >/dev/null 2>&1; then + echo "Creating the $name tunnel." + cloudflared tunnel create "$name" +fi + +echo "Pointing $hostname at the $name tunnel." +cloudflared tunnel route dns --overwrite-dns "$name" "$hostname" + +cat <