diff --git a/packages/junior/src/chat/automations/edit.ts b/packages/junior/src/chat/automations/edit.ts index 4f122dc8ee..62b7315c2b 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,31 @@ 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 task ? { ...current, task } : undefined; - } - const task = await deleteScheduledAutomation(getDb(), current.task.id); - return task ? { ...current, task } : undefined; + return await getDb().transaction(async (tx) => { + // 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); + if (!task) return undefined; + await dismissItems(); + return { ...current, task }; + } + const task = await deleteScheduledAutomation(tx, current.task.id); + 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 655a23e8c1..8f3e6e4826 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,12 @@ 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' + and review_items.automation_id is null + ) or ( ${juniorDestinations.visibility} is distinct from 'public' and ( @@ -297,6 +304,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..71e6d3d37e 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,15 +612,17 @@ 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 + // 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; 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 3e8209b772..54b2432671 100644 --- a/packages/junior/src/chat/review-items/README.md +++ b/packages/junior/src/chat/review-items/README.md @@ -10,8 +10,22 @@ 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`. 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`. + + 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 +117,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/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 2045f25417..901fc770ac 100644 --- a/packages/junior/src/chat/review-items/store.ts +++ b/packages/junior/src/chat/review-items/store.ts @@ -400,6 +400,45 @@ 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. 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, + 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) + : and( + inArray(juniorReviewItems.addedByConversationId, [ + ...input.addedByConversationIds, + ]), + isNull(juniorReviewItems.automationId), + ), + ), + ); +} + /** * 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..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 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, 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 efb4d4424f..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,112 +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, @@ -178,6 +95,55 @@ describe("retention purge job", () => { destinationId: privateDest, lastActivityAtMs: BASE_MS, }); + 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}:${automationId ?? "chat"}`, + 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"); + // 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, { @@ -189,317 +155,40 @@ 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); - }); - - 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, + expect(await getReviewItem(fixture.sql.db(), pubItem)).toMatchObject({ + closeReason: "Conversation purged", + status: "dismissed", }); - 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"]); + // A later run of the Automation can still change its item. expect( - await fixture.sql.db().select().from(juniorAttachments), - ).toMatchObject([{ id: "current-attachment" }]); - }); + (await getReviewItem(fixture.sql.db(), automationItem))?.status, + ).toBe("open"); - 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, + // 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, }); - 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, + 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("shortens the window when visibility flips public to private", async () => { 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/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..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(); @@ -200,15 +203,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, @@ -217,6 +225,7 @@ describe("Slack webhook: Review item cards", () => { }), waitUntil: waitUntil.fn, services: { + ...options, getSlackAdapter: () => slackAdapter, queue, runtime: createNoopSlackWebhookRuntime(), @@ -549,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"); @@ -568,7 +592,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: {} }, @@ -611,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 () => {