From 354f5d37a88809ed33e3aa6d6f464238bbaebc3c Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Sun, 1 Nov 2026 20:01:03 +0000 Subject: [PATCH 1/2] fix(review-desk): Close Review items that no agent can act on Co-Authored-By: David Cramer --- packages/junior/src/chat/automations/edit.ts | 22 +++++++--- .../src/chat/conversations/sql/purge.ts | 12 +++++ .../junior/src/chat/ingress/slack-webhook.ts | 20 ++++----- .../junior/src/chat/review-items/README.md | 27 +++++++++--- .../junior/src/chat/review-items/store.ts | 33 ++++++++++++++ .../junior/src/chat/tools/review-items.ts | 2 +- .../component/conversations/retention.test.ts | 44 ++++++++++++++++++- .../api/automation-lifecycle.test.ts | 24 ++++++++++ .../slack/review-item-card-webhook.test.ts | 18 +++++--- 9 files changed, 173 insertions(+), 29 deletions(-) diff --git a/packages/junior/src/chat/automations/edit.ts b/packages/junior/src/chat/automations/edit.ts index 4f122dc8ee..994da40b6a 100644 --- a/packages/junior/src/chat/automations/edit.ts +++ b/packages/junior/src/chat/automations/edit.ts @@ -8,6 +8,7 @@ import type { AutomationEditFields } from "./edit-schema"; import type { TaskOutcome, User } from "@sentry/junior-plugin-api"; import type { AutomationEdit, AutomationUpdate } from "@/api/schema/automation"; import { getDb } from "@/chat/db"; +import { dismissOpenReviewItems } from "@/chat/review-items/store"; import { resolveTaskOutcomes } from "@/chat/task-outcomes"; import { deleteEventAutomation, @@ -393,15 +394,24 @@ export async function deleteViewerTask(user: User, id: string): Promise { /** * Delete one Automation that the caller may manage. The row stays for - * history. Returns undefined when it was already deleted. + * history. No run can act on its open Review items after the delete, so the + * same transaction dismisses them. Returns undefined when it was already + * deleted. */ export async function deleteAutomation( current: TaskCandidate, ): Promise { - if (current.kind === "event") { - const task = await deleteEventAutomation(getDb(), current.task.id); + return await getDb().transaction(async (tx) => { + await dismissOpenReviewItems(tx, { + automationId: current.task.id, + nowMs: Date.now(), + reason: "Automation deleted", + }); + if (current.kind === "event") { + const task = await deleteEventAutomation(tx, current.task.id); + return task ? { ...current, task } : undefined; + } + const task = await deleteScheduledAutomation(tx, current.task.id); return task ? { ...current, task } : undefined; - } - const task = await deleteScheduledAutomation(getDb(), current.task.id); - return task ? { ...current, task } : undefined; + }); } diff --git a/packages/junior/src/chat/conversations/sql/purge.ts b/packages/junior/src/chat/conversations/sql/purge.ts index 655a23e8c1..ad1dadd93e 100644 --- a/packages/junior/src/chat/conversations/sql/purge.ts +++ b/packages/junior/src/chat/conversations/sql/purge.ts @@ -2,6 +2,7 @@ import { and, asc, eq, inArray, isNull, or, sql } from "drizzle-orm"; import type { JuniorSqlDatabase } from "@/db/db"; import type { JuniorDestinationVisibility } from "@/db/schema/destinations"; import { requestAttachmentDeletion } from "@/chat/attachments/store"; +import { dismissOpenReviewItems } from "@/chat/review-items/store"; import { juniorConversationEvents, juniorConversationBriefs, @@ -119,6 +120,11 @@ export async function selectExpiredRoots( where bindings.parent_conversation_id = tree.conversation_id or bindings.child_conversation_id = tree.conversation_id ) + or exists ( + select 1 from junior_review_items review_items + where review_items.added_by_conversation_id = tree.conversation_id + and review_items.status = 'open' + ) or ( ${juniorDestinations.visibility} is distinct from 'public' and ( @@ -297,6 +303,12 @@ export async function purgeConversationTree( inArray(juniorAgentBindings.childConversationId, ids), ), ); + // No agent can act on an item after its Conversation is purged. + await dismissOpenReviewItems(executor.db(), { + addedByConversationIds: ids, + nowMs: args.nowMs, + reason: "Conversation purged", + }); await requestAttachmentDeletion(executor, ids, args.nowMs); const scrubMetadata = args.retention ? !isPublic diff --git a/packages/junior/src/chat/ingress/slack-webhook.ts b/packages/junior/src/chat/ingress/slack-webhook.ts index c7b8c4a3f3..a809a104b2 100644 --- a/packages/junior/src/chat/ingress/slack-webhook.ts +++ b/packages/junior/src/chat/ingress/slack-webhook.ts @@ -561,17 +561,16 @@ async function routeParsedMessage(args: { return; } - // Passive thread messages are not checked for the allowed domains yet. - if ( - route === "mention" && - !(await isSlackAuthorDomainAllowed({ + const isAuthorDomainAllowed = () => + isSlackAuthorDomainAllowed({ allowedBotUserIds: args.allowedBotUserIds, allowedDomains: args.allowedDomains, state: args.state, teamId: args.installation.teamId, user: args.event.user, - })) - ) { + }); + // Passive thread messages are not checked for the allowed domains yet. + if (route === "mention" && !(await isAuthorDomainAllowed())) { return; } @@ -613,14 +612,13 @@ async function routeParsedMessage(args: { route, state: args.state, }); - if (route !== "mention") { - await persist(); - return; - } // A message from a person to Junior answers the open questions that the - // agent asked in this Conversation, like a click on their cards. + // agent asked in this Conversation, like a click on their cards. A reply in + // a subscribed thread starts a Turn too, so it answers in the same way. It + // answers only when its author passes the allowed domain check. const closeAnswered = async () => { if (message.author.isBot === true) return; + if (route === "subscribed" && !(await isAuthorDomainAllowed())) return; await closeReviewItemsAnsweredByReply({ conversationId, getSlackAdapter: () => args.adapter, diff --git a/packages/junior/src/chat/review-items/README.md b/packages/junior/src/chat/review-items/README.md index 3e8209b772..f9b01af03a 100644 --- a/packages/junior/src/chat/review-items/README.md +++ b/packages/junior/src/chat/review-items/README.md @@ -10,8 +10,23 @@ behavior. - A `notice` only tells a person something. An `action` asks for an answer. Its `request` (`userInputRequestSchema`) is a question and 1-4 fields. - An action item is the only way the agent asks a person for input. -- The status is `open`, `done`, or `dismissed`. Only a person or an agent - closes an item. Nothing closes it on its own. +- The status is `open`, `done`, or `dismissed`. An open item closes only in + one of these ways: + - A person answers it, marks a notice done, or dismisses it (see "Person + actions"). + - A person replies in its thread (see "Reply in the thread"). + - An agent closes it with `updateReviewItem`. + - Junior purges the Conversation that added it. The item becomes + `dismissed` with the reason `Conversation purged`. + - Someone deletes the Automation that added it. The item becomes + `dismissed` with the reason `Automation deleted`. + - The plugin that added a notice closes it. A plugin close sends no + message. + + A purge and an Automation delete keep the row. They send no message and + do not change the Slack cards. A click on an old card tells the person + that the item is closed. + - An assignee is a User or **Everyone**. Everyone is a flag on the item. It means that any person who can see the item can act on it. When the agent names no assignee, a new item goes to the User of the Actor, or else to the @@ -103,9 +118,11 @@ parent with `reportBlocked`, and the parent adds the item. A person can answer an action item with a message instead of the card (`closeReviewItemsAnsweredByReply` in `actions.ts`). The message is a Slack -mention or DM to Junior, or a dashboard message, in a Conversation. Junior -saves the message first, so it always reaches the agent. Then it closes each -item that matches all of these: +mention or DM to Junior, a Slack reply without a mention in a thread that +Junior follows (when the `passive-routing` feature is on), or a dashboard +message, in a Conversation. A Slack author must pass the allowed domain +check. Junior saves the message first, so it always reaches the agent. Then +it closes each item that matches all of these: - The item is an open `action` item. - The agent added it in this Conversation (`addedByConversationId`). Items diff --git a/packages/junior/src/chat/review-items/store.ts b/packages/junior/src/chat/review-items/store.ts index 2045f25417..74ec836344 100644 --- a/packages/junior/src/chat/review-items/store.ts +++ b/packages/junior/src/chat/review-items/store.ts @@ -400,6 +400,39 @@ export async function closeReviewItem( return updated.length > 0; } +/** + * Dismiss the open Review items that an Automation or a set of + * Conversations added, when no agent can act on them any more. No person + * closes them. It sends no message and does not change the Slack cards. + */ +export async function dismissOpenReviewItems( + db: JuniorDatabase, + input: { nowMs: number; reason: string } & ( + | { addedByConversationIds: readonly string[] } + | { automationId: string } + ), +): Promise { + const now = new Date(input.nowMs); + await db + .update(juniorReviewItems) + .set({ + closeReason: input.reason, + closedAt: now, + status: "dismissed", + updatedAt: now, + }) + .where( + and( + eq(juniorReviewItems.status, "open"), + "automationId" in input + ? eq(juniorReviewItems.automationId, input.automationId) + : inArray(juniorReviewItems.addedByConversationId, [ + ...input.addedByConversationIds, + ]), + ), + ); +} + /** * Make one User the only assignee of an open Review item. Return false when * the item is closed. diff --git a/packages/junior/src/chat/tools/review-items.ts b/packages/junior/src/chat/tools/review-items.ts index 3ead20aeeb..dab26d57d7 100644 --- a/packages/junior/src/chat/tools/review-items.ts +++ b/packages/junior/src/chat/tools/review-items.ts @@ -339,7 +339,7 @@ export function createUpdateReviewItemTool( ) { return zodTool({ description: - "Change an open review desk item, or close it. Set `status` to `done` or `dismissed` with a `reason`, for example when its pull request merged. An item never closes on its own. You cannot submit the answer to an action item: only an assignee can.", + "Change an open review desk item, or close it. Set `status` to `done` or `dismissed` with a `reason`, for example when its pull request merged. An item also closes when a person answers, dismisses, or replies in its thread, when its Conversation is purged, when its Automation is deleted, or when the plugin that added it closes it. You cannot submit the answer to an action item: only an assignee can.", exposure: "deferred", source: REVIEW_DESK_TOOL_SOURCE, annotations: { diff --git a/packages/junior/tests/component/conversations/retention.test.ts b/packages/junior/tests/component/conversations/retention.test.ts index efb4d4424f..aa4f00d8e5 100644 --- a/packages/junior/tests/component/conversations/retention.test.ts +++ b/packages/junior/tests/component/conversations/retention.test.ts @@ -31,6 +31,7 @@ import { type MemoryAttachmentStorage, } from "@/chat/attachments/memory"; import type { AttachmentWrite } from "@/chat/attachments/storage"; +import { getReviewItem, insertReviewItem } from "@/chat/review-items/store"; import { createJuniorSqlFixture, type LocalJuniorSqlFixture, @@ -178,6 +179,24 @@ describe("retention purge job", () => { destinationId: privateDest, lastActivityAtMs: BASE_MS, }); + const addItem = async (conversationId: string) => + (await insertReviewItem(fixture.sql.db(), { + addedByConversationId: conversationId, + assignedToEveryone: false, + assigneeUserIds: [], + body: "The rollout waits on an answer.", + conversationIds: [conversationId], + dedupeKey: `rollout:${conversationId}`, + kind: "action", + nowMs: BASE_MS, + request: { + question: "Continue the rollout?", + fields: [{ id: "note", kind: "text", label: "Note" }], + }, + title: "Continue the rollout?", + visibility: "public", + }))!.id; + const pubItem = await addItem("pub"); // At 15 days: private is expired (14d window), public is retained (90d). const first = await runRetentionPurge(fixture.sql, { @@ -189,13 +208,36 @@ describe("retention purge job", () => { expect( (await readConversation(fixture.sql, "pub")).transcriptPurgedAt, ).toBe(null); + expect((await getReviewItem(fixture.sql.db(), pubItem))?.status).toBe( + "open", + ); - // At 91 days: the public conversation crosses its 90-day window. + // At 91 days: the public conversation crosses its 90-day window. No + // agent can act on its open item after the purge, so the item closes. const second = await runRetentionPurge(fixture.sql, { nowMs: BASE_MS + 91 * DAY_MS, }); expect(second.purged).toBe(1); expect(await eventCount(fixture.sql, "pub")).toBe(0); + expect(await getReviewItem(fixture.sql.db(), pubItem)).toMatchObject({ + closeReason: "Conversation purged", + status: "dismissed", + }); + + // An older purge left an item open. The next run closes it, and then + // the tree has no more work. + const privItem = await addItem("priv"); + const third = await runRetentionPurge(fixture.sql, { + nowMs: BASE_MS + 91 * DAY_MS, + }); + expect(third.purged).toBe(1); + expect((await getReviewItem(fixture.sql.db(), privItem))?.status).toBe( + "dismissed", + ); + const fourth = await runRetentionPurge(fixture.sql, { + nowMs: BASE_MS + 91 * DAY_MS, + }); + expect(fourth.purged).toBe(0); }); it("deletes attachment blobs after conversation content expires", async () => { diff --git a/packages/junior/tests/integration/api/automation-lifecycle.test.ts b/packages/junior/tests/integration/api/automation-lifecycle.test.ts index 85a30d84b0..0435b03e47 100644 --- a/packages/junior/tests/integration/api/automation-lifecycle.test.ts +++ b/packages/junior/tests/integration/api/automation-lifecycle.test.ts @@ -10,6 +10,7 @@ import { saveScheduledAutomation, } from "@/chat/scheduled-automations/tasks"; import { recordAutomationExecution } from "@/chat/automations/execution-stats"; +import { getReviewItem, insertReviewItem } from "@/chat/review-items/store"; import { setupAutomationEditApi as setup } from "../../fixtures/automation-edit-api"; describe("Automation lifecycle API", () => { @@ -93,6 +94,24 @@ describe("Automation lifecycle API", () => { expect( await (await app.request(`${url}/versions`)).json(), ).toMatchObject({ versions: [{ version: 1 }] }); + // A run of the Automation asked a person something. + const item = await insertReviewItem(getDb(), { + addedByConversationId: "slack:T123:U123:edit", + assignedToEveryone: false, + assigneeUserIds: [], + automationId: id, + body: "The digest is empty this week.", + conversationIds: [], + dedupeKey: "digest:empty", + kind: "action", + nowMs: Date.now(), + request: { + question: "Post an empty digest?", + fields: [{ id: "note", kind: "text", label: "Note" }], + }, + title: "Post an empty digest?", + visibility: "public", + }); // A public reader can also delete it. expect( ( @@ -103,6 +122,11 @@ describe("Automation lifecycle API", () => { ).status, ).toBe(204); expect((await app.request(`${url}/edit`)).status).toBe(404); + // No run can act on the question now, so it closes. The row stays. + expect(await getReviewItem(getDb(), item!.id)).toMatchObject({ + closeReason: "Automation deleted", + status: "dismissed", + }); } finally { await fixture.close(); } 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..01b91c6cb5 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 @@ -200,15 +200,20 @@ describe("Slack webhook: Review item cards", () => { }, ], }); - /** A message from a person to Junior in the card thread. */ - async function reply(userId: string, ts: string) { + /** + * A message from a person in the card thread. Without a mention, it is a + * reply in a thread that Junior follows. + */ + async function reply(userId: string, ts: string, { mention = true } = {}) { const waitUntil = client.waitUntil(); const response = await handleSlackWebhook({ request: client.event({ ...slackEventsApiEnvelope({ - eventType: "app_mention", + eventType: mention ? "app_mention" : "message", channel: "C123", - text: `<@${SLACK_BOT_USER_ID}> Hold, the tests are red`, + text: mention + ? `<@${SLACK_BOT_USER_ID}> Hold, the tests are red` + : "Hold, the tests are red", threadTs: "1712345.0001", ts, user: userId, @@ -568,7 +573,10 @@ describe("Slack webhook: Review item cards", () => { await reply("U456", "1712345.0003"); expect((await getReviewItem(getDb(), item.id))?.status).toBe("open"); - await reply("U123", "1712345.0004"); + // Junior follows the thread, so Ana's reply needs no mention to start a + // Turn. It answers like a mention. + await state.subscribe(THREAD); + await reply("U123", "1712345.0004", { mention: false }); expect(await getReviewItem(getDb(), item.id)).toMatchObject({ // The item saves the message text without the mention of Junior. answer: { other: "Hold, the tests are red", values: {} }, From d648164f9071ee5d083a6bb114b14bf64bb2243c Mon Sep 17 00:00:00 2001 From: "sentry-junior[bot]" <264270552+sentry-junior[bot]@users.noreply.github.com> Date: Sun, 1 Nov 2026 20:27:23 +0000 Subject: [PATCH 2/2] fix(review-items): Keep Automation items open on Conversation purge A recurring Automation refreshes its Review item by dedupe key, but the item keeps the Conversation of its first run. Purge of that Conversation closed the item while the Automation was still live. Purge now skips Automation items, and an Automation delete closes them. - Dismiss Automation items only after the delete returns a row, under the Automation lock. - Check the allowed domain of a subscribed reply only when an open item can match, so most replies need no Slack profile lookup. - Remove the plugin close path from the tool text and README. It does not exist yet. - Split the attachment retention tests into their own file, with shared seed helpers, to keep retention.test.ts under the length limit. Co-Authored-By: David Cramer --- packages/junior/src/chat/automations/edit.ts | 21 +- .../src/chat/conversations/sql/purge.ts | 1 + .../junior/src/chat/ingress/slack-webhook.ts | 7 +- .../junior/src/chat/review-items/README.md | 5 +- .../junior/src/chat/review-items/actions.ts | 6 + .../junior/src/chat/review-items/store.ts | 14 +- .../junior/src/chat/tools/review-items.ts | 2 +- .../retention-attachments.test.ts | 353 ++++++++++++++ .../component/conversations/retention.test.ts | 445 ++---------------- .../tests/fixtures/conversation-retention.ts | 79 ++++ .../slack/review-item-card-webhook.test.ts | 47 +- 11 files changed, 562 insertions(+), 418 deletions(-) create mode 100644 packages/junior/tests/component/conversations/retention-attachments.test.ts create mode 100644 packages/junior/tests/fixtures/conversation-retention.ts diff --git a/packages/junior/src/chat/automations/edit.ts b/packages/junior/src/chat/automations/edit.ts index 994da40b6a..62b7315c2b 100644 --- a/packages/junior/src/chat/automations/edit.ts +++ b/packages/junior/src/chat/automations/edit.ts @@ -402,16 +402,23 @@ export async function deleteAutomation( current: TaskCandidate, ): Promise { return await getDb().transaction(async (tx) => { - await dismissOpenReviewItems(tx, { - automationId: current.task.id, - nowMs: Date.now(), - reason: "Automation deleted", - }); + // The delete takes the lock of the Automation. It holds until this + // transaction ends, so the dismiss runs under it. + const dismissItems = () => + dismissOpenReviewItems(tx, { + automationId: current.task.id, + nowMs: Date.now(), + reason: "Automation deleted", + }); if (current.kind === "event") { const task = await deleteEventAutomation(tx, current.task.id); - return task ? { ...current, task } : undefined; + if (!task) return undefined; + await dismissItems(); + return { ...current, task }; } const task = await deleteScheduledAutomation(tx, current.task.id); - return task ? { ...current, task } : undefined; + if (!task) return undefined; + await dismissItems(); + return { ...current, task }; }); } diff --git a/packages/junior/src/chat/conversations/sql/purge.ts b/packages/junior/src/chat/conversations/sql/purge.ts index ad1dadd93e..8f3e6e4826 100644 --- a/packages/junior/src/chat/conversations/sql/purge.ts +++ b/packages/junior/src/chat/conversations/sql/purge.ts @@ -124,6 +124,7 @@ export async function selectExpiredRoots( select 1 from junior_review_items review_items where review_items.added_by_conversation_id = tree.conversation_id and review_items.status = 'open' + and review_items.automation_id is null ) or ( ${juniorDestinations.visibility} is distinct from 'public' diff --git a/packages/junior/src/chat/ingress/slack-webhook.ts b/packages/junior/src/chat/ingress/slack-webhook.ts index a809a104b2..71e6d3d37e 100644 --- a/packages/junior/src/chat/ingress/slack-webhook.ts +++ b/packages/junior/src/chat/ingress/slack-webhook.ts @@ -615,11 +615,14 @@ async function routeParsedMessage(args: { // A message from a person to Junior answers the open questions that the // agent asked in this Conversation, like a click on their cards. A reply in // a subscribed thread starts a Turn too, so it answers in the same way. It - // answers only when its author passes the allowed domain check. + // skips the allowed domain check above. The close checks the domain only + // when an open item can match, so most replies need no profile lookup. const closeAnswered = async () => { if (message.author.isBot === true) return; - if (route === "subscribed" && !(await isAuthorDomainAllowed())) return; await closeReviewItemsAnsweredByReply({ + ...(route === "subscribed" + ? { canAnswer: isAuthorDomainAllowed } + : undefined), conversationId, getSlackAdapter: () => args.adapter, queue: args.queue, diff --git a/packages/junior/src/chat/review-items/README.md b/packages/junior/src/chat/review-items/README.md index f9b01af03a..54b2432671 100644 --- a/packages/junior/src/chat/review-items/README.md +++ b/packages/junior/src/chat/review-items/README.md @@ -17,11 +17,10 @@ behavior. - A person replies in its thread (see "Reply in the thread"). - An agent closes it with `updateReviewItem`. - Junior purges the Conversation that added it. The item becomes - `dismissed` with the reason `Conversation purged`. + `dismissed` with the reason `Conversation purged`. An item from an + Automation run stays open, because a later run can still change it. - Someone deletes the Automation that added it. The item becomes `dismissed` with the reason `Automation deleted`. - - The plugin that added a notice closes it. A plugin close sends no - message. A purge and an Automation delete keep the row. They send no message and do not change the Slack cards. A click on an old card tells the person diff --git a/packages/junior/src/chat/review-items/actions.ts b/packages/junior/src/chat/review-items/actions.ts index 8539bf3bae..1e31532c06 100644 --- a/packages/junior/src/chat/review-items/actions.ts +++ b/packages/junior/src/chat/review-items/actions.ts @@ -257,6 +257,11 @@ function messageAnswer( * a message that was sent does not go again. */ export async function closeReviewItemsAnsweredByReply(args: { + /** + * Return false when the replier may not answer, such as a Slack author + * outside the allowed domains. It runs only when an item can match. + */ + canAnswer?: () => Promise; conversationId: string; getSlackAdapter?: () => SlackAdapter; nowMs?: number; @@ -267,6 +272,7 @@ export async function closeReviewItemsAnsweredByReply(args: { const db = getDb(); const items = await listReplyActionItems(db, args.conversationId); if (items.length === 0) return; + if (args.canAnswer && !(await args.canAnswer())) return; const user = await args.resolveUser(); const nowMs = args.nowMs ?? Date.now(); for (const item of items.filter((candidate) => diff --git a/packages/junior/src/chat/review-items/store.ts b/packages/junior/src/chat/review-items/store.ts index 74ec836344..901fc770ac 100644 --- a/packages/junior/src/chat/review-items/store.ts +++ b/packages/junior/src/chat/review-items/store.ts @@ -403,7 +403,10 @@ export async function closeReviewItem( /** * Dismiss the open Review items that an Automation or a set of * Conversations added, when no agent can act on them any more. No person - * closes them. It sends no message and does not change the Slack cards. + * closes them. It sends no message and does not change the Slack cards. A + * close by Conversations skips the items of an Automation, because a later + * run of the Automation can still change them. An Automation delete closes + * those. */ export async function dismissOpenReviewItems( db: JuniorDatabase, @@ -426,9 +429,12 @@ export async function dismissOpenReviewItems( eq(juniorReviewItems.status, "open"), "automationId" in input ? eq(juniorReviewItems.automationId, input.automationId) - : inArray(juniorReviewItems.addedByConversationId, [ - ...input.addedByConversationIds, - ]), + : and( + inArray(juniorReviewItems.addedByConversationId, [ + ...input.addedByConversationIds, + ]), + isNull(juniorReviewItems.automationId), + ), ), ); } diff --git a/packages/junior/src/chat/tools/review-items.ts b/packages/junior/src/chat/tools/review-items.ts index dab26d57d7..d945c30bfc 100644 --- a/packages/junior/src/chat/tools/review-items.ts +++ b/packages/junior/src/chat/tools/review-items.ts @@ -339,7 +339,7 @@ export function createUpdateReviewItemTool( ) { return zodTool({ description: - "Change an open review desk item, or close it. Set `status` to `done` or `dismissed` with a `reason`, for example when its pull request merged. An item also closes when a person answers, dismisses, or replies in its thread, when its Conversation is purged, when its Automation is deleted, or when the plugin that added it closes it. You cannot submit the answer to an action item: only an assignee can.", + "Change an open review desk item, or close it. Set `status` to `done` or `dismissed` with a `reason`, for example when its pull request merged. An item also closes when a person answers, dismisses, or replies in its thread, when its Conversation is purged, or when its Automation is deleted. You cannot submit the answer to an action item: only an assignee can.", exposure: "deferred", source: REVIEW_DESK_TOOL_SOURCE, annotations: { diff --git a/packages/junior/tests/component/conversations/retention-attachments.test.ts b/packages/junior/tests/component/conversations/retention-attachments.test.ts new file mode 100644 index 0000000000..72ab88e51d --- /dev/null +++ b/packages/junior/tests/component/conversations/retention-attachments.test.ts @@ -0,0 +1,353 @@ +import { eq } from "drizzle-orm"; +import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { + collectAttachmentGarbage, + storeAttachment, +} from "@/chat/attachments/store"; +import { + createMemoryAttachmentStorage, + type MemoryAttachmentStorage, +} from "@/chat/attachments/memory"; +import type { AttachmentWrite } from "@/chat/attachments/storage"; +import { runRetentionPurge } from "@/chat/conversations/retention"; +import { migrateSchema } from "@/chat/conversations/sql/migrations"; +import { juniorAttachments, juniorConversations } from "@/db/schema"; +import { + BASE_MS, + DAY_MS, + seedConversation, + seedDestination, +} from "../../fixtures/conversation-retention"; +import { + createJuniorSqlFixture, + type LocalJuniorSqlFixture, +} from "../../fixtures/sql"; + +/** Write one placeholder object so tests can check that cleanup removed it. */ +async function putObject( + storage: MemoryAttachmentStorage, + key: string, +): Promise { + await storage.put({ + body: Buffer.from(key), + contentType: "text/plain", + key, + }); +} + +describe("retention purge job: attachments", () => { + let fixture: LocalJuniorSqlFixture; + + beforeEach(async () => { + fixture = await createJuniorSqlFixture(); + await migrateSchema(fixture.sql); + }); + + afterEach(async () => { + await fixture.close(); + }); + + it("deletes attachment blobs after conversation content expires", async () => { + const destinationId = await seedDestination(fixture.sql, "private"); + await seedConversation(fixture.sql, { + conversationId: "with-attachment", + destinationId, + lastActivityAtMs: BASE_MS, + }); + await fixture.sql + .db() + .insert(juniorAttachments) + .values({ + id: "attachment-1", + conversationId: "with-attachment", + storageProvider: "memory", + storageKey: "key-1", + filename: "report.txt", + contentType: "text/plain", + bytes: 6, + sha256: "digest", + createdAt: new Date(BASE_MS), + }); + const storage = createMemoryAttachmentStorage(); + await putObject(storage, "key-1"); + + const result = await runRetentionPurge(fixture.sql, { + attachmentStorage: storage, + nowMs: BASE_MS + 15 * DAY_MS, + }); + + expect(result.attachments).toBe(1); + expect([...storage.objects.keys()]).toEqual([]); + expect(await fixture.sql.db().select().from(juniorAttachments)).toEqual([]); + }); + + it("deletes attachments 30 days after they are stored", async () => { + const destinationId = await seedDestination(fixture.sql, "public"); + await seedConversation(fixture.sql, { + conversationId: "active-with-attachments", + destinationId, + lastActivityAtMs: BASE_MS + 31 * DAY_MS, + }); + await fixture.sql + .db() + .insert(juniorAttachments) + .values([ + { + id: "expired-attachment", + conversationId: "active-with-attachments", + storageProvider: "memory", + storageKey: "expired-key", + filename: "old.txt", + contentType: "text/plain", + bytes: 3, + sha256: "old", + createdAt: new Date(BASE_MS), + }, + { + id: "current-attachment", + conversationId: "active-with-attachments", + storageProvider: "memory", + storageKey: "current-key", + filename: "new.txt", + contentType: "text/plain", + bytes: 3, + sha256: "new", + createdAt: new Date(BASE_MS + 2 * DAY_MS), + }, + ]); + const storage = createMemoryAttachmentStorage(); + await putObject(storage, "expired-key"); + await putObject(storage, "current-key"); + + const result = await runRetentionPurge(fixture.sql, { + attachmentStorage: storage, + nowMs: BASE_MS + 30 * DAY_MS, + }); + + expect(result.purged).toBe(0); + expect(result.attachments).toBe(1); + expect([...storage.objects.keys()]).toEqual(["current-key"]); + expect( + await fixture.sql.db().select().from(juniorAttachments), + ).toMatchObject([{ id: "current-attachment" }]); + }); + + it("keeps marked attachment rows when blob delete fails", async () => { + const destinationId = await seedDestination(fixture.sql, "private"); + await seedConversation(fixture.sql, { + conversationId: "with-attachment", + destinationId, + lastActivityAtMs: BASE_MS, + }); + await fixture.sql + .db() + .insert(juniorAttachments) + .values({ + id: "attachment-1", + conversationId: "with-attachment", + storageProvider: "memory", + storageKey: "key-1", + filename: "report.txt", + contentType: "text/plain", + bytes: 6, + sha256: "digest", + createdAt: new Date(BASE_MS), + deleteRequestedAt: new Date(BASE_MS + DAY_MS), + }); + const storage = { + ...createMemoryAttachmentStorage(), + delete: async () => { + throw new Error("blob delete failed"); + }, + }; + + await expect( + collectAttachmentGarbage({ + db: fixture.sql, + nowMs: BASE_MS + 15 * DAY_MS, + storage, + }), + ).rejects.toThrow("blob delete failed"); + + const rows = await fixture.sql.db().select().from(juniorAttachments); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + id: "attachment-1", + storageKey: "key-1", + deleteRequestedAt: expect.any(Date), + }); + }); + + it("collects unmarked attachments owned by purged conversations", async () => { + const destinationId = await seedDestination(fixture.sql, "private"); + await seedConversation(fixture.sql, { + conversationId: "purged-owner", + destinationId, + lastActivityAtMs: BASE_MS, + }); + await fixture.sql + .db() + .update(juniorConversations) + .set({ transcriptPurgedAt: new Date(BASE_MS + DAY_MS) }) + .where(eq(juniorConversations.conversationId, "purged-owner")); + await fixture.sql + .db() + .insert(juniorAttachments) + .values({ + id: "orphan-attachment", + conversationId: "purged-owner", + storageProvider: "memory", + storageKey: "orphan-key", + filename: "late.txt", + contentType: "text/plain", + bytes: 4, + sha256: "orphan", + createdAt: new Date(BASE_MS + DAY_MS), + }); + const storage = createMemoryAttachmentStorage(); + await putObject(storage, "orphan-key"); + + const result = await collectAttachmentGarbage({ + db: fixture.sql, + nowMs: BASE_MS + 15 * DAY_MS, + storage, + }); + + expect(result.deleted).toBe(1); + expect([...storage.objects.keys()]).toEqual([]); + expect(await fixture.sql.db().select().from(juniorAttachments)).toEqual([]); + }); + + it("rejects attachment stores into purged conversations", async () => { + const destinationId = await seedDestination(fixture.sql, "private"); + await seedConversation(fixture.sql, { + conversationId: "purged-owner", + destinationId, + lastActivityAtMs: BASE_MS, + }); + await fixture.sql + .db() + .update(juniorConversations) + .set({ transcriptPurgedAt: new Date(BASE_MS + DAY_MS) }) + .where(eq(juniorConversations.conversationId, "purged-owner")); + const storage = createMemoryAttachmentStorage(); + + await expect( + storeAttachment({ + conversationId: "purged-owner", + db: fixture.sql, + file: { + bytes: 4, + data: Buffer.from("late"), + filename: "late.txt", + mimeType: "text/plain", + path: "/tmp/late.txt", + }, + storage, + }), + ).rejects.toThrow(/purged conversation/); + + expect([...storage.objects.keys()]).toEqual([]); + expect(await fixture.sql.db().select().from(juniorAttachments)).toEqual([]); + }); + + it("marks and cleans a raced insert when purge wins after put", async () => { + const destinationId = await seedDestination(fixture.sql, "private"); + await seedConversation(fixture.sql, { + conversationId: "race-owner", + destinationId, + lastActivityAtMs: BASE_MS, + }); + const memoryStorage = createMemoryAttachmentStorage(); + const puts: string[] = []; + // Purge the conversation after the object is written, before the SQL row. + const storage = { + ...memoryStorage, + put: async (input: AttachmentWrite) => { + await memoryStorage.put(input); + puts.push(input.key); + await fixture.sql + .db() + .update(juniorConversations) + .set({ transcriptPurgedAt: new Date(BASE_MS + DAY_MS) }) + .where(eq(juniorConversations.conversationId, "race-owner")); + }, + }; + + await expect( + storeAttachment({ + conversationId: "race-owner", + db: fixture.sql, + file: { + bytes: 4, + data: Buffer.from("late"), + filename: "late.txt", + mimeType: "text/plain", + path: "/tmp/late.txt", + }, + storage, + }), + ).rejects.toThrow(/purged conversation/); + + const rows = await fixture.sql.db().select().from(juniorAttachments); + expect(rows).toHaveLength(1); + expect(puts).toEqual([rows[0]?.storageKey]); + expect([...memoryStorage.objects.keys()]).toEqual([]); + expect(rows[0]?.deleteRequestedAt).toEqual(expect.any(Date)); + }); + + it("rejects live-row reuse when the conversation is already purged", async () => { + const destinationId = await seedDestination(fixture.sql, "private"); + await seedConversation(fixture.sql, { + conversationId: "purged-owner", + destinationId, + lastActivityAtMs: BASE_MS, + }); + const liveKey = "live-key"; + const file = { + bytes: 4, + data: Buffer.from("late"), + filename: "late.txt", + mimeType: "text/plain", + path: "/tmp/late.txt", + }; + const storage = createMemoryAttachmentStorage(); + const stored = await storeAttachment({ + conversationId: "purged-owner", + db: fixture.sql, + file, + storage, + }); + const storedKeys = [...storage.objects.keys()]; + await fixture.sql + .db() + .update(juniorConversations) + .set({ transcriptPurgedAt: new Date(BASE_MS + DAY_MS) }) + .where(eq(juniorConversations.conversationId, "purged-owner")); + // Leave the row unmarked so reuse hits the live-row path, not revive. + await fixture.sql + .db() + .update(juniorAttachments) + .set({ deleteRequestedAt: null, storageKey: liveKey }) + .where(eq(juniorAttachments.id, stored.id)); + + await expect( + storeAttachment({ + conversationId: "purged-owner", + db: fixture.sql, + file, + storage, + }), + ).rejects.toThrow(/purged conversation/); + + // Early live-row reuse rejects before writing or deleting an object. + expect([...storage.objects.keys()]).toEqual(storedKeys); + const rows = await fixture.sql.db().select().from(juniorAttachments); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + id: stored.id, + storageKey: liveKey, + deleteRequestedAt: null, + }); + }); +}); diff --git a/packages/junior/tests/component/conversations/retention.test.ts b/packages/junior/tests/component/conversations/retention.test.ts index aa4f00d8e5..17a945bae8 100644 --- a/packages/junior/tests/component/conversations/retention.test.ts +++ b/packages/junior/tests/component/conversations/retention.test.ts @@ -1,4 +1,3 @@ -import { randomUUID } from "node:crypto"; import { eq, inArray } from "drizzle-orm"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { @@ -11,113 +10,30 @@ import { selectExpiredRoots, } from "@/chat/conversations/sql/purge"; import { - juniorAttachments, juniorConversationEvents, juniorConversationBriefs, juniorConversations, juniorDestinations, juniorAgentBindings, juniorAgentInvocations, + juniorAutomations, + juniorIdentities, } from "@/db/schema"; import type { JuniorDestinationVisibility } from "@/db/schema/destinations"; import type { JuniorSqlDatabase } from "@/db/db"; import { conversationBriefFixture } from "../../fixtures/conversation-brief"; import { - collectAttachmentGarbage, - storeAttachment, -} from "@/chat/attachments/store"; -import { - createMemoryAttachmentStorage, - type MemoryAttachmentStorage, -} from "@/chat/attachments/memory"; -import type { AttachmentWrite } from "@/chat/attachments/storage"; + BASE_MS, + DAY_MS, + seedConversation, + seedDestination, +} from "../../fixtures/conversation-retention"; import { getReviewItem, insertReviewItem } from "@/chat/review-items/store"; import { createJuniorSqlFixture, type LocalJuniorSqlFixture, } from "../../fixtures/sql"; -const DAY_MS = 24 * 60 * 60 * 1000; -const BASE_MS = Date.UTC(2026, 0, 1); - -/** Write one placeholder object so tests can check that cleanup removed it. */ -async function putObject( - storage: MemoryAttachmentStorage, - key: string, -): Promise { - await storage.put({ - body: Buffer.from(key), - contentType: "text/plain", - key, - }); -} - -async function seedDestination( - executor: JuniorSqlDatabase, - visibility: JuniorDestinationVisibility, -): Promise { - const id = randomUUID(); - await executor - .db() - .insert(juniorDestinations) - .values({ - id, - provider: "slack", - providerTenantId: "T1", - providerDestinationId: `C-${id.slice(0, 8)}`, - kind: "channel", - visibility, - createdAt: new Date(BASE_MS), - updatedAt: new Date(BASE_MS), - }); - return id; -} - -async function seedConversation( - executor: JuniorSqlDatabase, - args: { - conversationId: string; - destinationId?: string; - lastActivityAtMs: number; - parentConversationId?: string; - title?: string | null; - channelName?: string | null; - withContent?: boolean; - }, -): Promise { - const at = new Date(args.lastActivityAtMs); - await executor - .db() - .insert(juniorConversations) - .values({ - conversationId: args.conversationId, - schemaVersion: 1, - destinationId: args.destinationId ?? null, - parentConversationId: args.parentConversationId ?? null, - title: args.title === undefined ? "A title" : args.title, - channelName: args.channelName === undefined ? "eng" : args.channelName, - actor: { platform: "slack", slackUserId: "U1", teamId: "T1" }, - createdAt: new Date(BASE_MS), - lastActivityAt: at, - updatedAt: at, - executionStatus: "idle", - }); - if (args.withContent !== false) { - await executor - .db() - .insert(juniorConversationEvents) - .values({ - conversationId: args.conversationId, - seq: 0, - historyVersion: 0, - schemaVersion: 1, - type: "message", - payload: { messageId: "m1", role: "user", text: "hi" }, - createdAt: at, - }); - } -} - async function eventCount( executor: JuniorSqlDatabase, conversationId: string, @@ -179,14 +95,15 @@ describe("retention purge job", () => { destinationId: privateDest, lastActivityAtMs: BASE_MS, }); - const addItem = async (conversationId: string) => + const addItem = async (conversationId: string, automationId?: string) => (await insertReviewItem(fixture.sql.db(), { addedByConversationId: conversationId, assignedToEveryone: false, assigneeUserIds: [], + ...(automationId ? { automationId } : undefined), body: "The rollout waits on an answer.", conversationIds: [conversationId], - dedupeKey: `rollout:${conversationId}`, + dedupeKey: `rollout:${conversationId}:${automationId ?? "chat"}`, kind: "action", nowMs: BASE_MS, request: { @@ -197,6 +114,36 @@ describe("retention purge job", () => { visibility: "public", }))!.id; const pubItem = await addItem("pub"); + // A live Automation keeps its item up to date in later runs, but the item + // still names the Conversation of the first run. + await fixture.sql + .db() + .insert(juniorIdentities) + .values({ + createdAt: new Date(BASE_MS), + id: "identity:creator", + kind: "user", + provider: "slack", + providerSubjectId: "U1", + providerTenantId: "T1", + updatedAt: new Date(BASE_MS), + }); + await fixture.sql + .db() + .insert(juniorAutomations) + .values({ + channelId: "C1", + createdAtMs: BASE_MS, + creatorIdentityId: "identity:creator", + creatorSlackUserId: "U1", + id: "auto-rollout", + instruction: "Check the rollout.", + kind: "scheduled", + record: {} as any, + status: "active", + teamId: "T1", + }); + const automationItem = await addItem("pub", "auto-rollout"); // At 15 days: private is expired (14d window), public is retained (90d). const first = await runRetentionPurge(fixture.sql, { @@ -223,9 +170,13 @@ describe("retention purge job", () => { closeReason: "Conversation purged", status: "dismissed", }); + // A later run of the Automation can still change its item. + expect( + (await getReviewItem(fixture.sql.db(), automationItem))?.status, + ).toBe("open"); - // An older purge left an item open. The next run closes it, and then - // the tree has no more work. + // An older purge left an item open. The next run closes it. Then the + // tree has no more work, because the Automation item is not work. const privItem = await addItem("priv"); const third = await runRetentionPurge(fixture.sql, { nowMs: BASE_MS + 91 * DAY_MS, @@ -240,310 +191,6 @@ describe("retention purge job", () => { expect(fourth.purged).toBe(0); }); - it("deletes attachment blobs after conversation content expires", async () => { - const destinationId = await seedDestination(fixture.sql, "private"); - await seedConversation(fixture.sql, { - conversationId: "with-attachment", - destinationId, - lastActivityAtMs: BASE_MS, - }); - await fixture.sql - .db() - .insert(juniorAttachments) - .values({ - id: "attachment-1", - conversationId: "with-attachment", - storageProvider: "memory", - storageKey: "key-1", - filename: "report.txt", - contentType: "text/plain", - bytes: 6, - sha256: "digest", - createdAt: new Date(BASE_MS), - }); - const storage = createMemoryAttachmentStorage(); - await putObject(storage, "key-1"); - - const result = await runRetentionPurge(fixture.sql, { - attachmentStorage: storage, - nowMs: BASE_MS + 15 * DAY_MS, - }); - - expect(result.attachments).toBe(1); - expect([...storage.objects.keys()]).toEqual([]); - expect(await fixture.sql.db().select().from(juniorAttachments)).toEqual([]); - }); - - it("deletes attachments 30 days after they are stored", async () => { - const destinationId = await seedDestination(fixture.sql, "public"); - await seedConversation(fixture.sql, { - conversationId: "active-with-attachments", - destinationId, - lastActivityAtMs: BASE_MS + 31 * DAY_MS, - }); - await fixture.sql - .db() - .insert(juniorAttachments) - .values([ - { - id: "expired-attachment", - conversationId: "active-with-attachments", - storageProvider: "memory", - storageKey: "expired-key", - filename: "old.txt", - contentType: "text/plain", - bytes: 3, - sha256: "old", - createdAt: new Date(BASE_MS), - }, - { - id: "current-attachment", - conversationId: "active-with-attachments", - storageProvider: "memory", - storageKey: "current-key", - filename: "new.txt", - contentType: "text/plain", - bytes: 3, - sha256: "new", - createdAt: new Date(BASE_MS + 2 * DAY_MS), - }, - ]); - const storage = createMemoryAttachmentStorage(); - await putObject(storage, "expired-key"); - await putObject(storage, "current-key"); - - const result = await runRetentionPurge(fixture.sql, { - attachmentStorage: storage, - nowMs: BASE_MS + 30 * DAY_MS, - }); - - expect(result.purged).toBe(0); - expect(result.attachments).toBe(1); - expect([...storage.objects.keys()]).toEqual(["current-key"]); - expect( - await fixture.sql.db().select().from(juniorAttachments), - ).toMatchObject([{ id: "current-attachment" }]); - }); - - it("keeps marked attachment rows when blob delete fails", async () => { - const destinationId = await seedDestination(fixture.sql, "private"); - await seedConversation(fixture.sql, { - conversationId: "with-attachment", - destinationId, - lastActivityAtMs: BASE_MS, - }); - await fixture.sql - .db() - .insert(juniorAttachments) - .values({ - id: "attachment-1", - conversationId: "with-attachment", - storageProvider: "memory", - storageKey: "key-1", - filename: "report.txt", - contentType: "text/plain", - bytes: 6, - sha256: "digest", - createdAt: new Date(BASE_MS), - deleteRequestedAt: new Date(BASE_MS + DAY_MS), - }); - const storage = { - ...createMemoryAttachmentStorage(), - delete: async () => { - throw new Error("blob delete failed"); - }, - }; - - await expect( - collectAttachmentGarbage({ - db: fixture.sql, - nowMs: BASE_MS + 15 * DAY_MS, - storage, - }), - ).rejects.toThrow("blob delete failed"); - - const rows = await fixture.sql.db().select().from(juniorAttachments); - expect(rows).toHaveLength(1); - expect(rows[0]).toMatchObject({ - id: "attachment-1", - storageKey: "key-1", - deleteRequestedAt: expect.any(Date), - }); - }); - - it("collects unmarked attachments owned by purged conversations", async () => { - const destinationId = await seedDestination(fixture.sql, "private"); - await seedConversation(fixture.sql, { - conversationId: "purged-owner", - destinationId, - lastActivityAtMs: BASE_MS, - }); - await fixture.sql - .db() - .update(juniorConversations) - .set({ transcriptPurgedAt: new Date(BASE_MS + DAY_MS) }) - .where(eq(juniorConversations.conversationId, "purged-owner")); - await fixture.sql - .db() - .insert(juniorAttachments) - .values({ - id: "orphan-attachment", - conversationId: "purged-owner", - storageProvider: "memory", - storageKey: "orphan-key", - filename: "late.txt", - contentType: "text/plain", - bytes: 4, - sha256: "orphan", - createdAt: new Date(BASE_MS + DAY_MS), - }); - const storage = createMemoryAttachmentStorage(); - await putObject(storage, "orphan-key"); - - const result = await collectAttachmentGarbage({ - db: fixture.sql, - nowMs: BASE_MS + 15 * DAY_MS, - storage, - }); - - expect(result.deleted).toBe(1); - expect([...storage.objects.keys()]).toEqual([]); - expect(await fixture.sql.db().select().from(juniorAttachments)).toEqual([]); - }); - - it("rejects attachment stores into purged conversations", async () => { - const destinationId = await seedDestination(fixture.sql, "private"); - await seedConversation(fixture.sql, { - conversationId: "purged-owner", - destinationId, - lastActivityAtMs: BASE_MS, - }); - await fixture.sql - .db() - .update(juniorConversations) - .set({ transcriptPurgedAt: new Date(BASE_MS + DAY_MS) }) - .where(eq(juniorConversations.conversationId, "purged-owner")); - const storage = createMemoryAttachmentStorage(); - - await expect( - storeAttachment({ - conversationId: "purged-owner", - db: fixture.sql, - file: { - bytes: 4, - data: Buffer.from("late"), - filename: "late.txt", - mimeType: "text/plain", - path: "/tmp/late.txt", - }, - storage, - }), - ).rejects.toThrow(/purged conversation/); - - expect([...storage.objects.keys()]).toEqual([]); - expect(await fixture.sql.db().select().from(juniorAttachments)).toEqual([]); - }); - - it("marks and cleans a raced insert when purge wins after put", async () => { - const destinationId = await seedDestination(fixture.sql, "private"); - await seedConversation(fixture.sql, { - conversationId: "race-owner", - destinationId, - lastActivityAtMs: BASE_MS, - }); - const memoryStorage = createMemoryAttachmentStorage(); - const puts: string[] = []; - // Purge the conversation after the object is written, before the SQL row. - const storage = { - ...memoryStorage, - put: async (input: AttachmentWrite) => { - await memoryStorage.put(input); - puts.push(input.key); - await fixture.sql - .db() - .update(juniorConversations) - .set({ transcriptPurgedAt: new Date(BASE_MS + DAY_MS) }) - .where(eq(juniorConversations.conversationId, "race-owner")); - }, - }; - - await expect( - storeAttachment({ - conversationId: "race-owner", - db: fixture.sql, - file: { - bytes: 4, - data: Buffer.from("late"), - filename: "late.txt", - mimeType: "text/plain", - path: "/tmp/late.txt", - }, - storage, - }), - ).rejects.toThrow(/purged conversation/); - - const rows = await fixture.sql.db().select().from(juniorAttachments); - expect(rows).toHaveLength(1); - expect(puts).toEqual([rows[0]?.storageKey]); - expect([...memoryStorage.objects.keys()]).toEqual([]); - expect(rows[0]?.deleteRequestedAt).toEqual(expect.any(Date)); - }); - - it("rejects live-row reuse when the conversation is already purged", async () => { - const destinationId = await seedDestination(fixture.sql, "private"); - await seedConversation(fixture.sql, { - conversationId: "purged-owner", - destinationId, - lastActivityAtMs: BASE_MS, - }); - const liveKey = "live-key"; - const file = { - bytes: 4, - data: Buffer.from("late"), - filename: "late.txt", - mimeType: "text/plain", - path: "/tmp/late.txt", - }; - const storage = createMemoryAttachmentStorage(); - const stored = await storeAttachment({ - conversationId: "purged-owner", - db: fixture.sql, - file, - storage, - }); - const storedKeys = [...storage.objects.keys()]; - await fixture.sql - .db() - .update(juniorConversations) - .set({ transcriptPurgedAt: new Date(BASE_MS + DAY_MS) }) - .where(eq(juniorConversations.conversationId, "purged-owner")); - // Leave the row unmarked so reuse hits the live-row path, not revive. - await fixture.sql - .db() - .update(juniorAttachments) - .set({ deleteRequestedAt: null, storageKey: liveKey }) - .where(eq(juniorAttachments.id, stored.id)); - - await expect( - storeAttachment({ - conversationId: "purged-owner", - db: fixture.sql, - file, - storage, - }), - ).rejects.toThrow(/purged conversation/); - - // Early live-row reuse rejects before writing or deleting an object. - expect([...storage.objects.keys()]).toEqual(storedKeys); - const rows = await fixture.sql.db().select().from(juniorAttachments); - expect(rows).toHaveLength(1); - expect(rows[0]).toMatchObject({ - id: stored.id, - storageKey: liveKey, - deleteRequestedAt: null, - }); - }); - it("shortens the window when visibility flips public to private", async () => { const dest = await seedDestination(fixture.sql, "public"); await seedConversation(fixture.sql, { diff --git a/packages/junior/tests/fixtures/conversation-retention.ts b/packages/junior/tests/fixtures/conversation-retention.ts new file mode 100644 index 0000000000..dd3d595aa9 --- /dev/null +++ b/packages/junior/tests/fixtures/conversation-retention.ts @@ -0,0 +1,79 @@ +import { randomUUID } from "node:crypto"; +import { + juniorConversationEvents, + juniorConversations, + juniorDestinations, +} from "@/db/schema"; +import type { JuniorDestinationVisibility } from "@/db/schema/destinations"; +import type { JuniorSqlDatabase } from "@/db/db"; + +export const DAY_MS = 24 * 60 * 60 * 1000; +export const BASE_MS = Date.UTC(2026, 0, 1); + +/** Add one Slack channel Destination with the given visibility. */ +export async function seedDestination( + executor: JuniorSqlDatabase, + visibility: JuniorDestinationVisibility, +): Promise { + const id = randomUUID(); + await executor + .db() + .insert(juniorDestinations) + .values({ + id, + provider: "slack", + providerTenantId: "T1", + providerDestinationId: `C-${id.slice(0, 8)}`, + kind: "channel", + visibility, + createdAt: new Date(BASE_MS), + updatedAt: new Date(BASE_MS), + }); + return id; +} + +/** Add one Conversation, with one message event unless `withContent` is false. */ +export async function seedConversation( + executor: JuniorSqlDatabase, + args: { + conversationId: string; + destinationId?: string; + lastActivityAtMs: number; + parentConversationId?: string; + title?: string | null; + channelName?: string | null; + withContent?: boolean; + }, +): Promise { + const at = new Date(args.lastActivityAtMs); + await executor + .db() + .insert(juniorConversations) + .values({ + conversationId: args.conversationId, + schemaVersion: 1, + destinationId: args.destinationId ?? null, + parentConversationId: args.parentConversationId ?? null, + title: args.title === undefined ? "A title" : args.title, + channelName: args.channelName === undefined ? "eng" : args.channelName, + actor: { platform: "slack", slackUserId: "U1", teamId: "T1" }, + createdAt: new Date(BASE_MS), + lastActivityAt: at, + updatedAt: at, + executionStatus: "idle", + }); + if (args.withContent !== false) { + await executor + .db() + .insert(juniorConversationEvents) + .values({ + conversationId: args.conversationId, + seq: 0, + historyVersion: 0, + schemaVersion: 1, + type: "message", + payload: { messageId: "m1", role: "user", text: "hi" }, + createdAt: at, + }); + } +} 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 01b91c6cb5..b1a309db62 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 @@ -1,5 +1,6 @@ import type { Identity, User } from "@sentry/junior-plugin-api"; import { Hono } from "hono"; +import { http, HttpResponse } from "msw"; import { afterEach, beforeEach, describe, expect, it } from "vitest"; import { createConversationRoutes } from "@/api/conversations/routes"; import type { JuniorApiEnv } from "@/api/route"; @@ -15,6 +16,7 @@ import { getConversationWorkState } from "@/chat/task-execution/store"; import { createAddReviewItemTool } from "@/chat/tools/review-items"; import { juniorIdentities } from "@/db/schema"; import { resetSlackApiMockState } from "../../msw/handlers/slack-api"; +import { mswServer } from "../../msw/server"; import { SLACK_BOT_USER_ID, SLACK_SIGNING_SECRET, @@ -23,6 +25,7 @@ import { createSlackAdapterFixture, } from "../../fixtures/conversation-work"; import { slackApiOutbox } from "../../fixtures/slack-api-outbox"; +import { usersInfoOk } from "../../fixtures/slack/factories/api"; import { slackEventsApiEnvelope } from "../../fixtures/slack/factories/events"; import { createSlackWebhookTestClient } from "../../fixtures/slack/webhook-client"; @@ -153,7 +156,7 @@ describe("Slack webhook: Review item cards", () => { await closeDb(); }); - function harness() { + function harness(options: { allowedDomains?: string[] } = {}) { const state = getStateAdapter(); const queue = createConversationWorkQueueTestAdapter(); const slackAdapter = createSlackAdapterFixture(); @@ -222,6 +225,7 @@ describe("Slack webhook: Review item cards", () => { }), waitUntil: waitUntil.fn, services: { + ...options, getSlackAdapter: () => slackAdapter, queue, runtime: createNoopSlackWebhookRuntime(), @@ -554,7 +558,22 @@ describe("Slack webhook: Review item cards", () => { }); it("closes a question when its assignee replies in the thread", async () => { - const { click, queue, reply, state } = harness(); + const { click, queue, reply, state } = harness({ + allowedDomains: ["example.com"], + }); + const emails: Record = { + U123: "ana@example.com", + U456: "ben@example.com", + UOUT: "someone@contractor.dev", + }; + mswServer.use( + http.post("https://slack.com/api/users.info", async ({ request }) => { + const user = new URLSearchParams(await request.text()).get("user")!; + return HttpResponse.json( + usersInfoOk({ userId: user, email: emails[user] }), + ); + }), + ); await slackThread(THREAD, "public"); const ana = await slackUser("ana@example.com", "U123"); await slackUser("ben@example.com", "U456"); @@ -619,6 +638,30 @@ describe("Slack webhook: Review item cards", () => { "U456", "U123", ]); + + // A question with no assignees takes an answer from anyone, but not + // from a Slack author outside the allowed domains. + const unassigned = await ( + createAddReviewItemTool({ + conversationId: THREAD, + resolveActorIdentity: async () => undefined, + source: { kind: "slack" } as any, + }) as any + ).execute( + { + kind: "action", + title: "Ship release 1.2.4", + body: "The release checks passed.", + dedupe_key: "release:1.2.4", + request: SHIP_REQUEST, + }, + {}, + ); + expect(unassigned.item.assignees).toEqual([]); + await reply("UOUT", "1712345.0006", { mention: false }); + expect((await getReviewItem(getDb(), unassigned.item.id))?.status).toBe( + "open", + ); }); it("closes a question when its assignee replies on the dashboard, also when only the queue wake fails", async () => {