diff --git a/TELEMETRY.md b/TELEMETRY.md index 189a1147da..c4cdd32c21 100644 --- a/TELEMETRY.md +++ b/TELEMETRY.md @@ -537,6 +537,33 @@ Pivot from create confirmation or Tasks page with `app.task.id`. From a fire conversation, use `app.dispatch.id` or `agent-dispatch:` as `gen_ai.conversation.id`. +### Review items + +A Review item change did not reach a linked Conversation, or a Slack card did +not change. + +Spans: `review_item.deliver` wraps the work after each change: the surfaces +show the new state, then the linked Conversations get their messages. Parent +spans are the agent tool call (`gen_ai.execute_tool` for `addReviewItem` or +`updateReviewItem`), the review desk API request, +`chat.slack_review_item_click` / `chat.slack_review_item_submit`, or the +Slack event or dashboard message request of a reply in the thread. + +Events: `review_item.message.failed` (one linked Conversation did not get the +message; the change call fails and a retry sends only missing messages), +`review_item.slack_card.sync.failed` (cards did not change; the next +delivery of the item shows its latest state), +`review_item.slack_input.persist.failed` (the Slack thread of a card click +got a message instead of Slack input) + +Attributes: `app.review_item.id`, `app.review_item.change` (`content` for an +agent content change, else `taken`, `submitted`, `completed`, `dismissed`, +`replied`, or `closed`), `app.review_item.kind`, `app.review_item.status`, +`app.review_item.visibility`, `app.review_item.conversation_count`, +`gen_ai.conversation.id` (the Conversation that missed a message) + +Pivot from a review desk item with `app.review_item.id`. + ### Auth And Resume A turn parked for auth, resumed late, or failed after callback. diff --git a/packages/junior/.dependency-cruiser.mjs b/packages/junior/.dependency-cruiser.mjs index e68877afcb..4e4b05d141 100644 --- a/packages/junior/.dependency-cruiser.mjs +++ b/packages/junior/.dependency-cruiser.mjs @@ -49,6 +49,18 @@ export default { pathNot: "^src/chat/slack/timestamp\\.ts$", }, }, + { + name: "no-review-items-to-slack", + comment: + "Review item code is provider-neutral. App setup injects Slack cards and mentions; see src/chat/review-items/README.md.", + severity: "error", + from: { + path: "^src/chat/(review-items/|tools/review-items\\.ts$)", + }, + to: { + path: "^src/chat/slack/", + }, + }, { name: "no-slack-sdk-outside-provider-modules", comment: diff --git a/packages/junior/src/api.ts b/packages/junior/src/api.ts index 1df19b0f6a..82013b9173 100644 --- a/packages/junior/src/api.ts +++ b/packages/junior/src/api.ts @@ -25,7 +25,9 @@ import { statsReportSchema, } from "./api/schema"; import { readStatsReport } from "./api/stats"; +import { getProductionSlackAdapter } from "./chat/app/production"; import { logException } from "./chat/logging"; +import { createSlackReviewItemShow } from "./chat/slack/review-item-card-sync"; import type { ConversationWorkQueue } from "./chat/task-execution/queue"; import { getVercelConversationWorkQueue } from "./chat/task-execution/vercel-queue"; import { @@ -89,7 +91,15 @@ export function createJuniorApi( queue, }), ); - app.route("/api/review-items", createReviewItemRoutes({ queue })); + app.route( + "/api/review-items", + createReviewItemRoutes({ + delivery: { + queue, + show: createSlackReviewItemShow(getProductionSlackAdapter), + }, + }), + ); app.route("/api/code", createCodeRoutes()); app.route("/api/connections", createConnectionRoutes()); app.route("/api/personal-tokens", createPersonalTokenRoutes()); diff --git a/packages/junior/src/api/conversations/create.ts b/packages/junior/src/api/conversations/create.ts index 4c3eac0574..831edb7358 100644 --- a/packages/junior/src/api/conversations/create.ts +++ b/packages/junior/src/api/conversations/create.ts @@ -15,6 +15,7 @@ import { getConversationWorkState } from "@/chat/task-execution/store"; import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; import { getProductionSlackAdapter } from "@/chat/app/production"; import { closeReviewItemsAnsweredByReply } from "@/chat/review-items/actions"; +import { createSlackReviewItemShow } from "@/chat/slack/review-item-card-sync"; import { getStateAdapter } from "@/chat/state/adapter"; import { throwApiError } from "../http"; import type { @@ -125,8 +126,10 @@ export async function appendConversationMessageForViewer( const closeAnswered = () => closeReviewItemsAnsweredByReply({ conversationId, - getSlackAdapter: getProductionSlackAdapter, - queue: services.queue, + delivery: { + queue: services.queue, + show: createSlackReviewItemShow(getProductionSlackAdapter), + }, resolveUser: async () => viewer, text: body.message, }); diff --git a/packages/junior/src/api/review-items/routes.ts b/packages/junior/src/api/review-items/routes.ts index 89a4caa5f9..3ce21b4525 100644 --- a/packages/junior/src/api/review-items/routes.ts +++ b/packages/junior/src/api/review-items/routes.ts @@ -15,7 +15,6 @@ import { } from "@/api/schema/review-item"; import { validateRequest } from "@/api/validation"; import { requireViewer } from "@/api/viewer"; -import { getProductionSlackAdapter } from "@/chat/app/production"; import { getDb } from "@/chat/db"; import { canActOnReviewItem, @@ -33,7 +32,7 @@ import { searchReviewItems, type ReviewItem, } from "@/chat/review-items/store"; -import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import type { ReviewItemDelivery } from "@/chat/review-items/changes"; const LIST_LIMIT = 100; @@ -88,7 +87,7 @@ function actionErrorResponse(error: unknown): Response { /** Create the authenticated review desk routes. */ export function createReviewItemRoutes(services: { - queue: ConversationWorkQueue; + delivery: ReviewItemDelivery; }): Hono { const app = new Hono(); app.get( @@ -163,9 +162,8 @@ export function createReviewItemRoutes(services: { try { const item = await actOnReviewItem({ action: body.action, - getSlackAdapter: getProductionSlackAdapter, + delivery: services.delivery, id, - queue: services.queue, user: viewer, ...(body.action === "submit" ? { answer: { ...body.answer, requestId: id } } diff --git a/packages/junior/src/app.ts b/packages/junior/src/app.ts index 528a8d2632..a1d6979c01 100644 --- a/packages/junior/src/app.ts +++ b/packages/junior/src/app.ts @@ -594,6 +594,7 @@ export async function createApp( bindMessageAgent(request, { queue: conversationWorkQueue }), bindSpawnAgent: (request) => bindSpawnAgent(request, { queue: conversationWorkQueue }), + conversationWorkQueue, tracePropagation, }); const runtimeServiceOverrides = { diff --git a/packages/junior/src/chat/agent/tools.ts b/packages/junior/src/chat/agent/tools.ts index 691bad34e9..9e9214986b 100644 --- a/packages/junior/src/chat/agent/tools.ts +++ b/packages/junior/src/chat/agent/tools.ts @@ -476,6 +476,7 @@ export async function wireAgentTools( ...commonToolRuntimeContext, ...toolRoute, attachmentStorage: args.run.environment?.attachmentStorage, + conversationWorkQueue: args.run.environment?.conversationWorkQueue, workspaces: { activeWorkspaceId: () => agentSandbox.sandboxRef()?.workspaceId, switch: agentSandbox.switchWorkspace, diff --git a/packages/junior/src/chat/agent/types.ts b/packages/junior/src/chat/agent/types.ts index fbe4834b87..ee4aae3bd4 100644 --- a/packages/junior/src/chat/agent/types.ts +++ b/packages/junior/src/chat/agent/types.ts @@ -33,6 +33,7 @@ import type { } from "@/chat/tools/types"; import type { AssistantMessage } from "@earendil-works/pi-ai"; import type { AttachmentStorage } from "@/chat/attachments/storage"; +import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; /** One attachment the model may see for the current instruction. */ export type AgentAttachment = { @@ -216,6 +217,8 @@ export type AgentEvent = export type AgentEnvironment = { attachmentStorage?: AttachmentStorage; configuration?: Record; + /** Queue of the app, for tools that wake other Conversations. */ + conversationWorkQueue?: ConversationWorkQueue; locationConfiguration?: LocationConfigurationService; skillDirs?: string[]; /** Per-slice override for app-owned sandbox egress trace propagation. */ diff --git a/packages/junior/src/chat/ingress/slack-review-item.ts b/packages/junior/src/chat/ingress/slack-review-item.ts index e5b4020c48..5c821aec08 100644 --- a/packages/junior/src/chat/ingress/slack-review-item.ts +++ b/packages/junior/src/chat/ingress/slack-review-item.ts @@ -14,9 +14,10 @@ import { type ReviewItemAction, } from "@/chat/review-items/actions"; import { - renderReviewItemMessage, - sendReviewItemMessages, -} from "@/chat/review-items/messages"; + deliverReviewItemChange, + type ReviewItemDelivery, +} from "@/chat/review-items/changes"; +import { renderReviewItemMessage } from "@/chat/review-items/messages"; import type { ReviewItem } from "@/chat/review-items/store"; import { parseSlackChannelId } from "@/chat/slack/ids"; import { @@ -29,6 +30,7 @@ import { type SlackReviewItemClick, } from "@/chat/slack/review-item-card"; import { + createSlackReviewItemShow, readSlackReviewItemCardConversationId, syncReviewItemSlackCards, } from "@/chat/slack/review-item-card-sync"; @@ -151,6 +153,10 @@ async function act(args: { ...target, itemId: args.itemId, })) ?? `slack:${target.channelId}:${target.threadTs}`; + const delivery: ReviewItemDelivery = { + queue: services.queue, + show: createSlackReviewItemShow(() => services.adapter), + }; let item: ReviewItem; try { const current = await readReviewItemForUser(actor.user, args.itemId); @@ -160,9 +166,8 @@ async function act(args: { // Two clicks can both see an open item. Only the click that closes it // adds the input to this thread, so one action adds one Turn. firstCloseOnly: true, - getSlackAdapter: () => services.adapter, + delivery, id: current.id, - queue: services.queue, user: actor.user, ...(args.answer ? { answer: args.answer(current) } : undefined), }); @@ -194,11 +199,11 @@ async function act(args: { logException(error, "review_item.slack_input.persist.failed", { "app.review_item.id": item.id, }); - await sendReviewItemMessages({ + await deliverReviewItemChange({ change, + delivery, item, nowMs: Date.now(), - queue: services.queue, }); } } diff --git a/packages/junior/src/chat/ingress/slack-webhook.ts b/packages/junior/src/chat/ingress/slack-webhook.ts index c7b8c4a3f3..03a826b84a 100644 --- a/packages/junior/src/chat/ingress/slack-webhook.ts +++ b/packages/junior/src/chat/ingress/slack-webhook.ts @@ -78,6 +78,7 @@ import { bindMessageAuthorIdentityId } from "@/chat/services/message-actor-ident import { createUserTokenStore } from "@/chat/capabilities/factory"; import type { UserTokenStore } from "@/chat/credentials/user-token-store"; import { publishAppHomeView } from "@/chat/slack/app-home"; +import { createSlackReviewItemShow } from "@/chat/slack/review-item-card-sync"; import { presentSlackAutomationDetails } from "@/chat/slack/automation-details"; import { getSlackClient } from "@/chat/slack/client"; import { createSlackWebApiStatusSender } from "@/chat/slack/assistant-thread/status-send"; @@ -623,8 +624,10 @@ async function routeParsedMessage(args: { if (message.author.isBot === true) return; await closeReviewItemsAnsweredByReply({ conversationId, - getSlackAdapter: () => args.adapter, - queue: args.queue, + delivery: { + queue: args.queue, + show: createSlackReviewItemShow(() => args.adapter), + }, resolveUser: async () => { const teamId = args.installation.teamId; const userId = parseActorUserId(message.author.userId); diff --git a/packages/junior/src/chat/review-items/README.md b/packages/junior/src/chat/review-items/README.md index 3e8209b772..35e28c90b8 100644 --- a/packages/junior/src/chat/review-items/README.md +++ b/packages/junior/src/chat/review-items/README.md @@ -120,9 +120,29 @@ message. A Slack person without a Junior User closes only items with no assignees. The message is the answer in its own Conversation. Each other linked Conversation gets an Event. The first close wins, so a reply after a click changes nothing. A retry of the reply, or a later reply from the same -person, sends the Events that are missing for the items that the person -closed by reply. Each Event has a key, so an Event that was sent does not -go again. +person, delivers again each item that the person closed by reply. The cards +show the close again, and only the missing Events go. Each Event has a key, +so an Event that was sent does not go again. + +## After a change + +Every change goes through one path: `deliverReviewItemChange` in +`changes.ts`. The agent tools, the person actions, and a reply in the +thread call it after they save the change. + +- First, every surface shows the new state. App setup gives the `show` + function. Today it is the Slack card sync. Review item code does not import + Slack code (`no-review-items-to-slack` in `.dependency-cruiser.mjs`). +- Then a person action or an agent close goes to the other linked + Conversations as messages. An agent change to the content sends no + messages. +- Delivery runs inline, after the change is saved. A failed message fails the + call that made the change. The caller can retry: a retry sends only the + messages that are missing. A failed card is logged. The next delivery of + the item shows its latest state. +- App setup also gives the agent tools `findMentionedUserId`. The Slack + version (`chat/slack/review-item-tools.ts`) resolves a Slack mention to a + User through the Slack Identity in the workspace of the item. ## Messages to linked Conversations @@ -160,8 +180,8 @@ each card, so Junior can change it in place. state. A closed card has no controls and shows who closed it. - A card in a thread that the item is no longer linked to loses its content. -- Cards are optional UI. A Slack failure is logged and does not fail the - change. +- Cards are optional UI. A Slack failure is logged + (`review_item.slack_card.sync.failed`) and does not fail the change. - Syncs of one item run one at a time under a state lock (`withLock` in `chat/state/locks.ts`). So two syncs cannot post two cards in one thread. Each sync reads the latest item, so a sync that waits shows the newest diff --git a/packages/junior/src/chat/review-items/actions.ts b/packages/junior/src/chat/review-items/actions.ts index 8539bf3bae..ec6c5c551f 100644 --- a/packages/junior/src/chat/review-items/actions.ts +++ b/packages/junior/src/chat/review-items/actions.ts @@ -1,14 +1,12 @@ -import type { SlackAdapter } from "@chat-adapter/slack"; import type { User } from "@sentry/junior-plugin-api"; import { getDb } from "@/chat/db"; -import { syncReviewItemSlackCards } from "@/chat/slack/review-item-card-sync"; -import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; import { isEmptyUserInputAnswer, normalizeUserInputAnswer, type UserInputAnswer, } from "@/chat/tools/runtime/user-input-request"; -import { sendReviewItemMessages, type ReviewItemChange } from "./messages"; +import { deliverReviewItemChange, type ReviewItemDelivery } from "./changes"; +import type { ReviewItemChange } from "./messages"; import { canActOnReviewItem, canReadReviewItem, @@ -110,20 +108,20 @@ function submittedAnswer( * succeeds, so a client can retry after a failure. With `firstCloseOnly`, * only the call that closes the item succeeds, and every other call gets a * conflict. Use it when the caller adds its own input for the action, such - * as a Slack card click, so one action adds one Turn. `exceptConversationId` - * is the Conversation where the person acted. It gets the person's input - * from its own surface, so it gets no message here. Slack cards of the item - * change in place; see `syncReviewItemSlackCards` for `getSlackAdapter`. + * as a card click in a chat thread, so one action adds one Turn. + * `exceptConversationId` is the Conversation where the person acted. It gets + * the person's input from its own surface, so it gets no message here. + * `delivery` shows the change and sends the messages; see + * `deliverReviewItemChange`. */ export async function actOnReviewItem(args: { action: ReviewItemAction; answer?: UserInputAnswer; + delivery: ReviewItemDelivery; exceptConversationId?: string; firstCloseOnly?: boolean; - getSlackAdapter?: () => SlackAdapter; id: string; nowMs?: number; - queue: ConversationWorkQueue; user: User; }): Promise { const { action, user } = args; @@ -150,16 +148,12 @@ export async function actOnReviewItem(args: { throw new ReviewItemActionError("conflict", "Review item is closed."); } const taken = await readReviewItemForUser(user, item.id); - await syncReviewItemSlackCards({ - getSlackAdapter: args.getSlackAdapter, - itemId: taken.id, - }); - await sendReviewItemMessages({ + await deliverReviewItemChange({ change: reviewItemChange(action, user), + delivery: args.delivery, exceptConversationId: args.exceptConversationId, item: taken, nowMs, - queue: args.queue, }); return taken; } @@ -193,16 +187,12 @@ export async function actOnReviewItem(args: { if (args.firstCloseOnly ? !closedNow : !isSameClose(closed, action, user)) { throw new ReviewItemActionError("conflict", "Review item is closed."); } - await syncReviewItemSlackCards({ - getSlackAdapter: args.getSlackAdapter, - itemId: closed.id, - }); - await sendReviewItemMessages({ + await deliverReviewItemChange({ change: reviewItemChange(action, user), + delivery: args.delivery, exceptConversationId: args.exceptConversationId, item: closed, nowMs, - queue: args.queue, }); return closed; } @@ -252,15 +242,16 @@ function messageAnswer( * `REVIEW_ITEM_REPLIED_REASON`, and the replier closes it. The reply itself * is the answer in this Conversation, so only the other linked * Conversations get a message. A click that closes the item first wins. - * A retry of the reply, or a later reply from the same person, sends the - * missing messages of the items that it closed. Each message has a key, so - * a message that was sent does not go again. + * A retry of the reply, or a later reply from the same person, delivers + * the items that it closed again: the cards show the close, and only the + * missing messages go. Each message has a key, so a message that was sent + * does not go again. `delivery` shows the change and sends the messages; + * see `deliverReviewItemChange`. */ export async function closeReviewItemsAnsweredByReply(args: { conversationId: string; - getSlackAdapter?: () => SlackAdapter; + delivery: ReviewItemDelivery; nowMs?: number; - queue: ConversationWorkQueue; resolveUser: () => Promise; text: string; }): Promise { @@ -286,21 +277,16 @@ export async function closeReviewItemsAnsweredByReply(args: { if (!closed) continue; closedItem = await getReviewItem(db, item.id); if (!closedItem) continue; - // Card sync is best effort, so a retry only needs to send messages. - await syncReviewItemSlackCards({ - getSlackAdapter: args.getSlackAdapter, - itemId: closedItem.id, - }); } - await sendReviewItemMessages({ + await deliverReviewItemChange({ change: { kind: "replied", who: user ? (user.displayName ?? user.email) : "A person", }, + delivery: args.delivery, exceptConversationId: args.conversationId, item: closedItem, nowMs, - queue: args.queue, }); } } diff --git a/packages/junior/src/chat/review-items/assignees.ts b/packages/junior/src/chat/review-items/assignees.ts index 77d6c95655..63268efb53 100644 --- a/packages/junior/src/chat/review-items/assignees.ts +++ b/packages/junior/src/chat/review-items/assignees.ts @@ -1,7 +1,6 @@ -import { and, eq, inArray, isNotNull } from "drizzle-orm"; +import { and, eq, inArray } from "drizzle-orm"; import type { ProviderTenant } from "@/api/conversations/access"; import { findUserByEmailFromSql } from "@/chat/plugins/viewer"; -import { parseSlackUserId } from "@/chat/slack/ids"; import { ToolInputError } from "@/chat/tools/execution/tool-input-error"; import type { JuniorDatabase } from "@/db/db"; import { @@ -19,52 +18,31 @@ export interface ResolvedAssignees { userIds: string[]; } -const SLACK_MENTION = /^<@([A-Z0-9]+)(?:\|[^>]*)?>$/; - -async function userIdForSlackUser( - db: JuniorDatabase, - slackUserId: string, - teamId: string, -): Promise { - const rows = await db - .select({ userId: juniorIdentities.userId }) - .from(juniorIdentities) - .where( - and( - eq(juniorIdentities.provider, "slack"), - eq(juniorIdentities.providerTenantId, teamId), - eq(juniorIdentities.providerSubjectId, slackUserId), - isNotNull(juniorIdentities.userId), - ), - ) - .limit(1); - return rows[0]?.userId ?? undefined; -} +/** + * Find the Junior User of a chat mention, such as `<@U123>`, in one provider + * tenant. Return undefined when the text is not a mention. Throw a + * `ToolInputError` when it is a mention without a Junior User. App setup + * supplies the provider version. + */ +export type FindMentionedUserId = ( + text: string, + tenant: ProviderTenant | undefined, +) => Promise; /** - * Resolve one assignee: Everyone, an email, or a Slack user. A Slack user - * must be in the Slack workspace of the item. + * Resolve one assignee: Everyone, an email, or a mention. A mention must be + * in the tenant of the item. */ export async function resolveAssignee( db: JuniorDatabase, value: string, tenant: ProviderTenant | undefined, + findMentionedUserId: FindMentionedUserId, ): Promise<{ everyone: true } | { userId: string }> { const text = value.trim(); if (text.toLowerCase() === EVERYONE_ASSIGNEE) return { everyone: true }; - const slackUserId = parseSlackUserId(SLACK_MENTION.exec(text)?.[1] ?? text); - if (slackUserId) { - const userId = - tenant?.provider === "slack" - ? await userIdForSlackUser(db, slackUserId, tenant.providerTenantId) - : undefined; - if (!userId) { - throw new ToolInputError( - `<@${slackUserId}> has no Junior user yet. Assign them by email.`, - ); - } - return { userId }; - } + const mentionedUserId = await findMentionedUserId(text, tenant); + if (mentionedUserId) return { userId: mentionedUserId }; if (text.includes("@")) { const user = await findUserByEmailFromSql(db, text); if (!user) { @@ -73,7 +51,7 @@ export async function resolveAssignee( return { userId: user.id }; } throw new ToolInputError( - `Unknown assignee ${JSON.stringify(text)}. Use ${EVERYONE_ASSIGNEE}, an email, or a Slack mention such as <@U123>.`, + `Unknown assignee ${JSON.stringify(text)}. Use ${EVERYONE_ASSIGNEE}, an email, or a mention such as <@U123>.`, ); } @@ -82,12 +60,15 @@ export async function resolveAssignees( db: JuniorDatabase, values: readonly string[], tenant: ProviderTenant | undefined, + findMentionedUserId: FindMentionedUserId, ): Promise { if (values.length === 0) { throw new ToolInputError("A review item needs at least one assignee."); } const resolved = await Promise.all( - values.map((value) => resolveAssignee(db, value, tenant)), + values.map((value) => + resolveAssignee(db, value, tenant, findMentionedUserId), + ), ); return { everyone: resolved.some((assignee) => "everyone" in assignee), diff --git a/packages/junior/src/chat/review-items/changes.ts b/packages/junior/src/chat/review-items/changes.ts new file mode 100644 index 0000000000..0d8e271e49 --- /dev/null +++ b/packages/junior/src/chat/review-items/changes.ts @@ -0,0 +1,59 @@ +import { withSpan } from "@/chat/logging"; +import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import { sendReviewItemMessages, type ReviewItemChange } from "./messages"; +import type { ReviewItem } from "./store"; + +/** + * Show the latest state of a Review item on a surface outside Junior, such + * as cards in its linked Slack threads. App setup supplies it. It is + * optional UI, so it logs its own failures and does not throw. + */ +export type ShowReviewItem = (itemId: string) => Promise; + +/** Where a Review item change goes after it is saved. */ +export interface ReviewItemDelivery { + queue: ConversationWorkQueue; + show: ShowReviewItem; +} + +/** + * Deliver one saved Review item change. Every surface shows the new state. + * A person action or an agent close (`change`) also goes to every other open + * linked Conversation. A content change from an agent has no `change`, so + * only the surfaces change. `exceptConversationId` is the Conversation that + * made the change: it already knows. A retry of the same change is safe, + * because each message has an idempotency key. + */ +export async function deliverReviewItemChange(args: { + change?: ReviewItemChange; + delivery: ReviewItemDelivery; + exceptConversationId?: string; + item: ReviewItem; + nowMs: number; +}): Promise { + const { change, delivery, item } = args; + await withSpan( + "review_item.deliver", + "review_item.deliver", + {}, + async () => { + await delivery.show(item.id); + if (!change) return; + await sendReviewItemMessages({ + change, + exceptConversationId: args.exceptConversationId, + item, + nowMs: args.nowMs, + queue: delivery.queue, + }); + }, + { + "app.review_item.id": item.id, + "app.review_item.change": change?.kind ?? "content", + "app.review_item.kind": item.kind, + "app.review_item.status": item.status, + "app.review_item.visibility": item.visibility, + "app.review_item.conversation_count": item.conversationIds.length, + }, + ); +} diff --git a/packages/junior/src/chat/review-items/messages.ts b/packages/junior/src/chat/review-items/messages.ts index d94f362efc..191b06f5f0 100644 --- a/packages/junior/src/chat/review-items/messages.ts +++ b/packages/junior/src/chat/review-items/messages.ts @@ -6,6 +6,7 @@ import { webActorFromEmail, } from "@/chat/conversations/web-input"; import { getConversationStore, getDb } from "@/chat/db"; +import { logWarn } from "@/chat/logging"; import { EVENT_AUTHOR_ID } from "@/chat/events/actor"; import { SYSTEM_IDENTITY_ID } from "@/chat/identities/actor"; import { @@ -205,7 +206,17 @@ export async function sendReviewItemMessages(args: { ), ); // Try every Conversation first, so one failure does not hide the change - // from the others. A retry sends only the missing messages. + // from the others. A retry sends only the missing messages. The caller + // reports the error; this log names each Conversation that missed it. + results.forEach((result, index) => { + if (result.status === "rejected") { + logWarn("review_item.message.failed", { + "app.review_item.id": args.item.id, + "app.review_item.change": args.change.kind, + "gen_ai.conversation.id": conversationIds[index], + }); + } + }); const failed = results.find((result) => result.status === "rejected"); if (failed) throw failed.reason; } diff --git a/packages/junior/src/chat/runtime/agent-runner.ts b/packages/junior/src/chat/runtime/agent-runner.ts index 366043c0d2..85cd015bca 100644 --- a/packages/junior/src/chat/runtime/agent-runner.ts +++ b/packages/junior/src/chat/runtime/agent-runner.ts @@ -10,6 +10,7 @@ import { import type { AgentRunOutcome } from "@/chat/runtime/agent-run-outcome"; import type { SandboxEgressTracePropagationConfig } from "@/chat/sandbox/egress/tracing"; import type { AttachmentStorage } from "@/chat/attachments/storage"; +import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; /** Run one agent-run slice behind runtime-owned orchestration boundaries. */ export interface AgentRunner { @@ -23,11 +24,13 @@ export function createAgentRunner( attachmentStorage?: AttachmentStorage; bindMessageAgent?: (run: AgentRun) => MessageAgent | undefined; bindSpawnAgent?: (run: AgentRun) => SpawnAgent | undefined; + conversationWorkQueue?: ConversationWorkQueue; streamFn?: StreamFn; tracePropagation?: SandboxEgressTracePropagationConfig; }, ): AgentRunner { const attachmentStorage = options?.attachmentStorage; + const conversationWorkQueue = options?.conversationWorkQueue; const streamFn = options?.streamFn; const tracePropagation = options?.tracePropagation; const bindSpawnAgent = options?.bindSpawnAgent; @@ -53,6 +56,8 @@ export function createAgentRunner( ...run.environment, attachmentStorage: run.environment?.attachmentStorage ?? attachmentStorage, + conversationWorkQueue: + run.environment?.conversationWorkQueue ?? conversationWorkQueue, sandboxTracePropagation: run.environment?.sandboxTracePropagation ?? tracePropagation, }, diff --git a/packages/junior/src/chat/slack/review-item-card-sync.ts b/packages/junior/src/chat/slack/review-item-card-sync.ts index abc3afcb9b..b2fe685c15 100644 --- a/packages/junior/src/chat/slack/review-item-card-sync.ts +++ b/packages/junior/src/chat/slack/review-item-card-sync.ts @@ -4,6 +4,7 @@ import { getDb } from "@/chat/db"; import { runBestEffort } from "@/chat/logging"; import { getStateAdapter } from "@/chat/state/adapter"; import { withLock } from "@/chat/state/locks"; +import type { ShowReviewItem } from "@/chat/review-items/changes"; import { getReviewItem, type ReviewItem } from "@/chat/review-items/store"; import { resolveConversationRouting } from "@/chat/services/turn-session-routing"; import { runWithSlackInstallation } from "@/chat/slack/adapter-context"; @@ -124,6 +125,16 @@ export async function readSlackReviewItemCardConversationId(card: { // Long enough for the sync that holds the lock to post or change a few cards. const SYNC_LOCK_WAIT_MS = 10_000; +/** + * Return the Slack `ShowReviewItem` for app setup. Pass `getSlackAdapter` + * outside Slack work so each card uses the token of its workspace. + */ +export function createSlackReviewItemShow( + getSlackAdapter?: () => SlackAdapter, +): ShowReviewItem { + return (itemId) => syncReviewItemSlackCards({ getSlackAdapter, itemId }); +} + /** * Show the current state of a Review item in its linked Slack threads. An * open item gets a card in each linked thread that has none. Each existing diff --git a/packages/junior/src/chat/slack/review-item-tools.ts b/packages/junior/src/chat/slack/review-item-tools.ts new file mode 100644 index 0000000000..a7986d7b71 --- /dev/null +++ b/packages/junior/src/chat/slack/review-item-tools.ts @@ -0,0 +1,55 @@ +import type { ProviderTenant } from "@/api/conversations/access"; +import { createSlackActor } from "@/chat/actor"; +import { getDb } from "@/chat/db"; +import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; +import type { ReviewItemToolServices } from "@/chat/tools/review-items"; +import { readActorIdentityFromSql } from "@/chat/plugins/viewer"; +import { ToolInputError } from "@/chat/tools/execution/tool-input-error"; +import { parseSlackUserId } from "./ids"; +import { createSlackReviewItemShow } from "./review-item-card-sync"; + +const SLACK_MENTION = /^<@([A-Z0-9]+)(?:\|[^>]*)?>$/; + +/** + * Find the Junior User of a Slack mention, such as `<@U123>` or `U123`, in + * one Slack workspace. Return undefined when the text is not a Slack user. + * Throw a `ToolInputError` when the person has no Junior User in the + * workspace of `tenant`, so an agent assigns them by email instead. + */ +async function findSlackMentionedUserId( + text: string, + tenant: ProviderTenant | undefined, +): Promise { + const slackUserId = parseSlackUserId(SLACK_MENTION.exec(text)?.[1] ?? text); + if (!slackUserId) return undefined; + const user = + tenant?.provider === "slack" + ? ( + await readActorIdentityFromSql( + getDb(), + createSlackActor(tenant.providerTenantId, slackUserId, undefined), + ) + )?.user + : undefined; + if (!user) { + throw new ToolInputError( + `<@${slackUserId}> has no Junior user yet. Assign them by email.`, + ); + } + return user.id; +} + +/** + * The Slack services of the Review item tools: Slack mentions resolve to + * Users, and Slack cards show each change. A Slack Turn has the token of its + * workspace. Other runs use the default bot token. + */ +export function createSlackReviewItemToolServices( + queue: () => ConversationWorkQueue, +): ReviewItemToolServices { + return { + findMentionedUserId: findSlackMentionedUserId, + queue, + show: createSlackReviewItemShow(), + }; +} diff --git a/packages/junior/src/chat/tools/index.ts b/packages/junior/src/chat/tools/index.ts index bbf21d33b5..15b02d75cb 100644 --- a/packages/junior/src/chat/tools/index.ts +++ b/packages/junior/src/chat/tools/index.ts @@ -39,6 +39,7 @@ import { createSlackConversationMessageSearchTool } from "@/chat/slack/tools/con import { createSlackPublicSearchTool } from "@/chat/slack/tools/public-search"; import { getSlackToolContext } from "@/chat/slack/tool-support/context"; import { createSlackMessageAddReactionTool } from "@/chat/slack/tools/message-add-reaction"; +import { createSlackReviewItemToolServices } from "@/chat/slack/review-item-tools"; import { createSlackSendFilesTool } from "@/chat/slack/tools/send-files"; import { createSendFilesTool } from "@/chat/tools/send-files"; import { getSqlExecutor } from "@/chat/db"; @@ -353,15 +354,21 @@ export function createTools( // Review items link to root Conversations, where a person can see them. // `addReviewItem` is how the agent asks a person anything, so it is loaded // in every root Turn. A child agent asks its parent instead. + const reviewItemServices = createSlackReviewItemToolServices( + () => context.conversationWorkQueue ?? getVercelConversationWorkQueue(), + ); Object.assign( tools, disableTools( { - addReviewItem: createAddReviewItemTool(context), - searchReviewItems: createSearchReviewItemsTool(context), + addReviewItem: createAddReviewItemTool(context, reviewItemServices), + searchReviewItems: createSearchReviewItemsTool( + context, + reviewItemServices, + ), updateReviewItem: createUpdateReviewItemTool( context, - getVercelConversationWorkQueue, + reviewItemServices, ), }, context.parentConversationId diff --git a/packages/junior/src/chat/tools/review-items.ts b/packages/junior/src/chat/tools/review-items.ts index 3ead20aeeb..e23744b2b4 100644 --- a/packages/junior/src/chat/tools/review-items.ts +++ b/packages/junior/src/chat/tools/review-items.ts @@ -2,8 +2,11 @@ import { randomUUID } from "node:crypto"; import { z } from "zod"; import { getDb } from "@/chat/db"; import { NO_REPLY_MARKER } from "@/chat/no-reply"; -import { sendReviewItemMessages } from "@/chat/review-items/messages"; -import { syncReviewItemSlackCards } from "@/chat/slack/review-item-card-sync"; +import { + deliverReviewItemChange, + type ReviewItemDelivery, + type ShowReviewItem, +} from "@/chat/review-items/changes"; import { canAgentChangeReviewItem, canReadReviewItem, @@ -18,6 +21,7 @@ import { resolveAssignee, resolveAssignees, type ResolvedAssignees, + type FindMentionedUserId, } from "@/chat/review-items/assignees"; import { closeReviewItem, @@ -53,7 +57,7 @@ const assigneesSchema = z .min(1) .max(20) .describe( - "Who can act on the item. Each entry is `everyone`, an email, or a Slack mention such as `<@U123>`. Everyone means any person who can see the item.", + "Who can act on the item. Each entry is `everyone`, an email, or a mention such as `<@U123>`. Everyone means any person who can see the item.", ); const subjectSchema = z @@ -105,6 +109,18 @@ async function defaultAssignees( return { everyone: false, userIds: userId ? [userId] : [] }; } +/** What app setup gives the Review item tools. */ +export interface ReviewItemToolServices { + findMentionedUserId: FindMentionedUserId; + /** Read only after a change, so tool setup does not create a queue. */ + queue: () => ConversationWorkQueue; + show: ShowReviewItem; +} + +function delivery(services: ReviewItemToolServices): ReviewItemDelivery { + return { queue: services.queue(), show: services.show }; +} + function requireRequest(kind: ReviewItem["kind"], hasRequest: boolean): void { if (kind === "action" && !hasRequest) { throw new ToolInputError("An action item needs a `request`."); @@ -167,7 +183,10 @@ function subjectInput( } /** Create the tool that adds a Review item or updates the open one. */ -export function createAddReviewItemTool(context: ReviewItemToolContext) { +export function createAddReviewItemTool( + context: ReviewItemToolContext, + services: ReviewItemToolServices, +) { return zodTool({ description: `Ask a person for input, or tell them about work that needs them. This is how you ask a person anything: the item goes on the review desk, and in a Slack thread it also shows as a card with buttons. Ask only for a decision that is truly the person's, such as their judgment, access, or approval. Make small decisions yourself. Write the question so it stands alone: a person who sees only the card must understand it. Put the recommended option first and say why in its description. Use kind \`action\` with a \`request\` to ask: a question and 1-4 fields. Each field is \`choice\` (pick one), \`multi_choice\` (pick any), or \`text\` (a short answer). For a simple decision, use one \`choice\` field. Ask for related input in one item. Use kind \`notice\` only to tell a person something. Assignees default to the person in this Turn, or to the Automation creator in an Automation run. The current Conversation is always linked. The answer arrives later as a message from the person who answered. A reply that the assignee types in the thread also closes the item. If your next step needs the answer, end the Turn now: the card is your reply, so end with exactly ${NO_REPLY_MARKER}, or finish the Automation run. If other work does not need the answer, continue with it. If an open item has the same \`dedupe_key\`, this updates that item instead of adding another.`, annotations: { @@ -258,6 +277,7 @@ export function createAddReviewItemTool(context: ReviewItemToolContext) { db, input.assignees, existing ? existing.tenant : scope.tenant, + services.findMentionedUserId, ) : existing ? { @@ -303,7 +323,11 @@ export function createAddReviewItemTool(context: ReviewItemToolContext) { ); } const item = await readableItem(existing.id, scope); - await syncReviewItemSlackCards({ itemId: item.id }); + await deliverReviewItemChange({ + delivery: delivery(services), + item, + nowMs: Date.now(), + }); return { created: false, item: itemResult(item) }; } const item = await insertReviewItem(db, { @@ -321,9 +345,11 @@ export function createAddReviewItemTool(context: ReviewItemToolContext) { "Another open item started to use this `dedupe_key`. Add the item again.", ); } - // A Slack turn has the token of its workspace. Other runs use the - // default bot token. - await syncReviewItemSlackCards({ itemId: item.id }); + await deliverReviewItemChange({ + delivery: delivery(services), + item, + nowMs: Date.now(), + }); return { created: true, item: itemResult(item) }; }, }); @@ -331,11 +357,11 @@ export function createAddReviewItemTool(context: ReviewItemToolContext) { /** * Create the tool that changes or closes a Review item. A close tells the - * other linked Conversations through `queue`. + * other linked Conversations. */ export function createUpdateReviewItemTool( context: ReviewItemToolContext, - queue: () => ConversationWorkQueue, + services: ReviewItemToolServices, ) { return zodTool({ description: @@ -422,7 +448,12 @@ export function createUpdateReviewItemTool( visibility, }); const assignees = input.assignees - ? await resolveAssignees(db, input.assignees, item.tenant) + ? await resolveAssignees( + db, + input.assignees, + item.tenant, + services.findMentionedUserId, + ) : { everyone: item.assignedToEveryone, userIds: item.assignees.map((assignee) => assignee.userId), @@ -470,25 +501,31 @@ export function createUpdateReviewItemTool( throw new ToolInputError(`Review item ${item.id} is already closed.`); } const closedItem = await readableItem(item.id, scope); - await syncReviewItemSlackCards({ itemId: closedItem.id }); - await sendReviewItemMessages({ + await deliverReviewItemChange({ change: { kind: "closed", reason: input.reason }, + delivery: delivery(services), exceptConversationId: context.conversationId, item: closedItem, nowMs, - queue: queue(), }); return { item: itemResult(closedItem) }; } const changed = await readableItem(item.id, scope); - await syncReviewItemSlackCards({ itemId: changed.id }); + await deliverReviewItemChange({ + delivery: delivery(services), + item: changed, + nowMs: Date.now(), + }); return { item: itemResult(changed) }; }, }); } /** Create the tool that searches Review items. */ -export function createSearchReviewItemsTool(context: ReviewItemToolContext) { +export function createSearchReviewItemsTool( + context: ReviewItemToolContext, + services: Pick, +) { return zodTool({ description: "Search review desk items that you can see, newest change first. For example, find the open items for one pull request before you add or close one.", @@ -515,7 +552,7 @@ export function createSearchReviewItemsTool(context: ReviewItemToolContext) { .max(320) .nullable() .optional() - .describe("`everyone`, an email, or a Slack mention."), + .describe("`everyone`, an email, or a mention such as `<@U123>`."), automation_id: z.string().trim().min(1).nullable().optional(), conversation_id: z.string().trim().min(1).nullable().optional(), limit: z @@ -540,7 +577,12 @@ export function createSearchReviewItemsTool(context: ReviewItemToolContext) { statuses, ...(input.assignee ? { - assignee: await resolveAssignee(db, input.assignee, scope.tenant), + assignee: await resolveAssignee( + db, + input.assignee, + scope.tenant, + services.findMentionedUserId, + ), } : undefined), ...(input.automation_id diff --git a/packages/junior/src/chat/tools/types.ts b/packages/junior/src/chat/tools/types.ts index 47cf1aacd2..ba0522b2bf 100644 --- a/packages/junior/src/chat/tools/types.ts +++ b/packages/junior/src/chat/tools/types.ts @@ -24,13 +24,11 @@ import type { Skill } from "@/chat/skills"; import type { JuniorToolOutput } from "@/chat/tool-support/structured-result"; import type { WebActor, LocalActor, Actor, SlackActor } from "@/chat/actor"; import type { SlackActionToken } from "@/chat/slack/action-token"; -import type { - ModelProfile, - ModelProfileConfig, -} from "@/chat/model-profile"; +import type { ModelProfile, ModelProfileConfig } from "@/chat/model-profile"; import type { GeneratedArtifactFileRef } from "@/chat/tools/sandbox/file-uploads"; import type { MessageAgent, SpawnAgent } from "@/chat/agent/types"; import type { AttachmentStorage } from "@/chat/attachments/storage"; +import type { ConversationWorkQueue } from "@/chat/task-execution/queue"; import type { Workspace } from "@/chat/workspaces/types"; import type { ConversationPrivacy } from "@/chat/conversation-privacy"; @@ -102,6 +100,8 @@ interface BaseToolRuntimeContext { /** True when a successful run sends a message to stored Destinations. */ sendsMessage: boolean; }; + /** Queue of the app. Without it, Review item tools use the Vercel queue. */ + conversationWorkQueue?: ConversationWorkQueue; handoff?: HandoffControl; messageAgent?: MessageAgent; spawnAgent?: SpawnAgent; diff --git a/packages/junior/tests/integration/api/review-items/actions.test.ts b/packages/junior/tests/integration/api/review-items/actions.test.ts index 64de7d2e75..14491b55d6 100644 --- a/packages/junior/tests/integration/api/review-items/actions.test.ts +++ b/packages/junior/tests/integration/api/review-items/actions.test.ts @@ -5,6 +5,7 @@ import { createJuniorApi, type JuniorApiVariables } from "@/api"; import { getConversationStore, getDb } from "@/chat/db"; import { resolveViewerUserFromSql } from "@/chat/plugins/viewer"; import { takeReviewItem } from "@/chat/review-items/store"; +import { createSlackReviewItemToolServices } from "@/chat/slack/review-item-tools"; import { getConversation } from "@/chat/task-execution/store"; import { createAddReviewItemTool } from "@/chat/tools/review-items"; import { @@ -12,6 +13,7 @@ import { createConversationFixture, type ConversationFixture, } from "../../../fixtures/conversation"; +import { createConversationWorkQueueTestAdapter } from "../../../fixtures/conversation-work"; const REQUEST = { question: "Approve or reject this entry?", @@ -54,11 +56,14 @@ async function slackConversation(conversationId: string): Promise { } async function addItem(input: Record): Promise { - const tool = createAddReviewItemTool({ - conversationId: "slack:CPUB:1.000001", - resolveActorIdentity: async () => undefined, - source: { kind: "scheduled_automation" }, - }); + const tool = createAddReviewItemTool( + { + conversationId: "slack:CPUB:1.000001", + resolveActorIdentity: async () => undefined, + source: { kind: "scheduled_automation" }, + }, + createSlackReviewItemToolServices(createConversationWorkQueueTestAdapter), + ); const result = await (tool as any).execute(input, {}); return result.item.id; } diff --git a/packages/junior/tests/integration/review-item-tools.test.ts b/packages/junior/tests/integration/review-item-tools.test.ts index 8e1e591ec4..2b2c4ed50d 100644 --- a/packages/junior/tests/integration/review-item-tools.test.ts +++ b/packages/junior/tests/integration/review-item-tools.test.ts @@ -13,6 +13,7 @@ import { createSearchReviewItemsTool, createUpdateReviewItemTool, } from "@/chat/tools/review-items"; +import { createSlackReviewItemToolServices } from "@/chat/slack/review-item-tools"; import { getConversation } from "@/chat/task-execution/store"; import type { ToolRuntimeContext } from "@/chat/tools/types"; import { createConversationWorkQueueTestAdapter } from "../fixtures/conversation-work"; @@ -28,11 +29,13 @@ async function execute(tool: any, input: unknown): Promise { const queue = createConversationWorkQueueTestAdapter(); +const services = createSlackReviewItemToolServices(() => queue); + function tools(context: Context) { return { - add: createAddReviewItemTool(context), - search: createSearchReviewItemsTool(context), - update: createUpdateReviewItemTool(context, () => queue), + add: createAddReviewItemTool(context, services), + search: createSearchReviewItemsTool(context, services), + update: createUpdateReviewItemTool(context, services), }; } diff --git a/packages/junior/tests/integration/slack/review-item-card-webhook.test.ts b/packages/junior/tests/integration/slack/review-item-card-webhook.test.ts index a7d615913c..1a6419c996 100644 --- a/packages/junior/tests/integration/slack/review-item-card-webhook.test.ts +++ b/packages/junior/tests/integration/slack/review-item-card-webhook.test.ts @@ -12,6 +12,8 @@ import { actOnReviewItem } from "@/chat/review-items/actions"; import { getReviewItem } from "@/chat/review-items/store"; import { disconnectStateAdapter, getStateAdapter } from "@/chat/state/adapter"; import { getConversationWorkState } from "@/chat/task-execution/store"; +import { createSlackReviewItemShow } from "@/chat/slack/review-item-card-sync"; +import { createSlackReviewItemToolServices } from "@/chat/slack/review-item-tools"; import { createAddReviewItemTool } from "@/chat/tools/review-items"; import { juniorIdentities } from "@/db/schema"; import { resetSlackApiMockState } from "../../msw/handlers/slack-api"; @@ -107,14 +109,17 @@ async function addItem( input: Record, conversationId = THREAD, ) { - const tool = createAddReviewItemTool({ - conversationId, - resolveActorIdentity: async () => ({ - identity: { id: `identity:${user.id}` } as Identity, - user, - }), - source: { kind: "slack" } as any, - }); + const tool = createAddReviewItemTool( + { + conversationId, + resolveActorIdentity: async () => ({ + identity: { id: `identity:${user.id}` } as Identity, + user, + }), + source: { kind: "slack" } as any, + }, + createSlackReviewItemToolServices(createConversationWorkQueueTestAdapter), + ); return (await (tool as any).execute(input, {})) as { item: { assignees: string[]; id: string }; }; @@ -482,8 +487,8 @@ describe("Slack webhook: Review item cards", () => { await actOnReviewItem({ action: "submit", answer: { requestId: item.id, values: { notes: ["x".repeat(3_000)] } }, + delivery: { queue, show: createSlackReviewItemShow() }, id: item.id, - queue, user: ana, }); @@ -598,8 +603,8 @@ describe("Slack webhook: Review item cards", () => { actOnReviewItem({ action: "submit", answer: { requestId: item.id, values: { decision: ["Ship it"] } }, + delivery: { queue, show: createSlackReviewItemShow() }, id: item.id, - queue, user: ana, }), ).rejects.toThrow("Review item is closed.");