diff --git a/doc/plans/2026-10-24-recurring-server-sleep-stack.md b/doc/plans/2026-10-24-recurring-server-sleep-stack.md index 4aede3a325c..58f805f3244 100644 --- a/doc/plans/2026-10-24-recurring-server-sleep-stack.md +++ b/doc/plans/2026-10-24-recurring-server-sleep-stack.md @@ -1504,3 +1504,13 @@ fix; all 184 milestone/integration/unit interaction cases and production server typechecking pass. Remaining terminal-interaction writers are document-target expiry, deleted-addressee cancellation and startup supersession; other producer audits and production migration remain pending. + + +Slice 28 covers the remaining direct terminal-interaction writers in the +interaction service: document-target expiry, deleted addressees and startup +supersession. Seven real-producer regressions failed before the fix; all 191 +worker/integration/unit cases and production server typechecking pass. Database +fixture setup/cleanup moved to a cohesive helper, keeping the worker test file +below 1,000 lines. The broader raw-writer audit still includes external services +(secret proposals, tool reviews, conversation reset and native review retirement), +plus the previously listed publication/configuration/recovery dependencies. diff --git a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md index ec374576a57..be7a64e739a 100644 --- a/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md +++ b/doc/plans/2026-10-25-chat-run-milestone-sleep-audit.md @@ -840,3 +840,48 @@ an addressed agent, and startup supersession. Document-change callers are in `routes/issues.ts`; the owning state mutation is in the interaction service. Audit those before removing the lifecycle poll. This checkpoint still does not register the worker in production or establish final CI/review acceptance. + + +## Remaining terminal transitions in the interaction service + +The resolution-time stale-document helper, explicit document-target invalidation, +deleted-addressee cancellation and startup supersession now register milestone +intent before their first interaction write. The guards follow the existing +response-slot policy: question/confirmation for addressee cancellation, and +confirmation for document targets. No validity, permission, locking, publication, +or expiry behavior changes. Existing transactions own the notification. + +New regressions use actual document creation/update/deletion and the interaction +service's document-invalidation/accept/reject entry points. An early document +write is allowed to settle before interaction expiry, proving that the last +state transition still wakes projection. Deleted-addressee cases use another +agent, so the source run and its chat endpoint remain valid. Startup supersession +creates both pending cards before startup with ordinary sibling replacement +explicitly disabled; the actual repair selects and expires the older one. + +- `/tmp/sleep28-fixture-extraction-tests.log`: all 65 existing worker cases pass + after extracting only shared fixture creation/cleanup into + `__tests__/helpers/chat-milestone-fixtures.ts`. Worker, clock and quiet assertions + remain in the tests; cleanup still occurs after coordinator shutdown. The + helper tracks only fixture company/user IDs and now cleans fixture documents. +- `/tmp/sleep28-terminal-interactions-before.log`: all seven new cases reach + the correct terminal interaction state but fail to produce the eligible + failure notice: document change/deletion, stale accept/reject, two deleted- + addressee variants and startup supersession. There are no setup failures. +- `/tmp/sleep28-terminal-interactions-tests.log`: all 191 cases pass across + 72 worker, 106 service integration and 13 service unit cases. The two real-DB + suites use their separate existing isolated databases. +- `/tmp/sleep28-terminal-interactions-typecheck.log`: production server + typecheck passes; tests are excluded. `git diff --check` passes. + +The raw interaction-writer audit also found owners outside this service. These +must be reconciled with their outer transaction signals before removing polling: +`agent-conversations.ts`, `secret-proposals.ts`, `tool-action-review.ts`, +`tool-gateway.ts`, native automatic/obsolete review retirement, and native +external-review supersession. Result/summary-only writes, connection-intent-only +writes, and writes already covered by an outer status-decision/run transaction +need explicit disposition rather than indiscriminate new notifications. + +Provider receipts, conversation/configuration changes and alternate recovery +writers remain under audit. This checkpoint does not register the new worker +in production or remove lifecycle polling, and is not final PR acceptance. diff --git a/server/src/__tests__/chat-run-milestone-work.test.ts b/server/src/__tests__/chat-run-milestone-work.test.ts index 2e866745ea8..45aa4784f6d 100644 --- a/server/src/__tests__/chat-run-milestone-work.test.ts +++ b/server/src/__tests__/chat-run-milestone-work.test.ts @@ -1,8 +1,9 @@ +import { documentService } from "../services/documents.js"; +import { createChatMilestoneFixtures } from "./helpers/chat-milestone-fixtures.js"; import { randomUUID } from "node:crypto"; import { eq } from "drizzle-orm"; import { afterAll, afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "vitest"; -import { activityLog, agents, agentWakeupRequests, authUsers, caseHandoffs, cases, companyMemberships, chatActions, chatConversations, chatEndpoints, chatMessageLinks, chatPublications, companies, completionContracts, createDb, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, workAssessments, subscribeDatabaseWork, - heartbeatRunEvents, heartbeatRuns, issueComments, issueQuestionResponseDeliveries, issueThreadInteractions, issues, toolApplications, toolConnections } from "@paperclipai/db"; +import { agents, agentWakeupRequests, caseHandoffs, cases, chatConversations, chatEndpoints, chatPublications, completionContracts, createDb, nativeRunFinalizations, nativeRunResults, statusDecisions, statusDecisionEffects, workAssessments, subscribeDatabaseWork, heartbeatRuns, issueComments, issueQuestionResponseDeliveries, issueThreadInteractions, issues } from "@paperclipai/db"; import { getEmbeddedPostgresTestSupport, startEmbeddedPostgresTestDatabase } from "./helpers/embedded-postgres.js"; import { resolveExternalTestDatabaseUrl } from "./helpers/external-test-database.js"; import { createDeliveryWorkCoordinator } from "../services/delivery-work-coordinator.js"; @@ -37,68 +38,23 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu let coordinator: ReturnType | undefined; let enabled = true; const errors: unknown[] = []; - const fixtureCompanies: string[] = []; - const fixtureUsers: string[] = []; + let fixtures: ReturnType; type Fixture = Awaited>; let f: Fixture; beforeAll(async () => { if (!external) temporary = await startEmbeddedPostgresTestDatabase("chat-milestones-"); db = createDb(external ?? temporary!.connectionString, { maxConnections: 6, idleTimeoutSeconds: 0, maxLifetimeSeconds: 0 }); + fixtures = createChatMilestoneFixtures(db); }, 60_000); - async function fixture(companyId = randomUUID()) { - const agentId = randomUUID(), issueId = randomUUID(), endpointId = randomUUID(), conversationId = randomUUID(), commentId = randomUUID(), userId = randomUUID(); - if (!fixtureCompanies.includes(companyId)) { - fixtureCompanies.push(companyId); - await db.insert(companies).values({ id: companyId, name: "Milestones", issuePrefix: `M${companyId.slice(0, 7)}` }); - } - fixtureUsers.push(userId); - await db.insert(authUsers).values({ id: userId, name: "Owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); - await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "owner" }); - await db.insert(agents).values({ id: agentId, companyId, name: "Maya", adapterType: "paperclip_runner" }); - await db.insert(issues).values({ id: issueId, companyId, title: "Chat task", status: "in_progress", assigneeAgentId: agentId, responsibleUserId: userId }); - const applicationId = randomUUID(), connectionId = randomUUID(); - await db.insert(toolApplications).values({ id: applicationId, companyId, name: `Chat ${applicationId}`, type: "chat" }); - await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Chat", uid: connectionId, transport: "chat_sdk", connectionPurpose: "channel", enabled: true, status: "active" }); - await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, assignedAgentId: agentId, publicId: endpointId, provider: "slack", status: "active" }); - await db.insert(chatConversations).values({ id: conversationId, companyId, endpointId, issueId, externalConversationId: conversationId, externalThreadId: conversationId, externalLabel: "Chat" }); - await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorType: "user", body: "Start work" }); - await db.insert(chatMessageLinks).values({ companyId, endpointId, conversationId, commentId, providerMessageId: commentId, direction: "inbound" }); - return { companyId, agentId, issueId, endpointId, conversationId, commentId, userId }; - } + const fixture = (companyId?: string) => fixtures.create(companyId); beforeEach(async () => { enabled = true; errors.length = 0; f = await fixture(); vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true }); }); afterEach(async () => { await coordinator?.stop(); coordinator = undefined; vi.useRealTimers(); vi.restoreAllMocks(); - for (const companyId of fixtureCompanies.splice(0)) { - await db.delete(activityLog).where(eq(activityLog.companyId, companyId)); - await db.delete(caseHandoffs).where(eq(caseHandoffs.companyId, companyId)); - await db.delete(cases).where(eq(cases.companyId, companyId)); - await db.delete(chatActions).where(eq(chatActions.companyId, companyId)); - await db.delete(chatMessageLinks).where(eq(chatMessageLinks.companyId, companyId)); - await db.delete(chatPublications).where(eq(chatPublications.companyId, companyId)); - await db.delete(chatConversations).where(eq(chatConversations.companyId, companyId)); - await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)); - await db.delete(issueQuestionResponseDeliveries).where(eq(issueQuestionResponseDeliveries.companyId, companyId)); - await db.delete(issueThreadInteractions).where(eq(issueThreadInteractions.companyId, companyId)); - await db.delete(issueComments).where(eq(issueComments.companyId, companyId)); - await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.companyId, companyId)); - await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.companyId, companyId)); - await db.delete(statusDecisionEffects).where(eq(statusDecisionEffects.companyId, companyId)); - await db.delete(statusDecisions).where(eq(statusDecisions.companyId, companyId)); - await db.delete(workAssessments).where(eq(workAssessments.companyId, companyId)); - await db.delete(nativeRunResults).where(eq(nativeRunResults.companyId, companyId)); - await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); - await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); - await db.delete(completionContracts).where(eq(completionContracts.companyId, companyId)); - await db.delete(issues).where(eq(issues.companyId, companyId)); - await db.delete(agents).where(eq(agents.companyId, companyId)); - await db.delete(companyMemberships).where(eq(companyMemberships.companyId, companyId)); - await db.delete(companies).where(eq(companies.id, companyId)); - } - for (const userId of fixtureUsers.splice(0)) await db.delete(authUsers).where(eq(authUsers.id, userId)); + await fixtures.cleanup(); expect(errors).toEqual([]); }); afterAll(async () => { await db?.$client.end({ timeout: 1 }); await temporary?.cleanup(); }); @@ -640,23 +596,30 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu ["confirmation", "withdraw"], ["confirmation", "expire"], ["question", "comment"], ["question", "historical comment"], ["confirmation", "comment"], ["confirmation", "historical comment"], - ["question", "replacement"], ["confirmation", "replacement"], + ["question", "replacement"], ["confirmation", "replacement"], ["confirmation", "startup supersession"], + ["question", "addressee deletion"], ["confirmation", "addressee deletion"], ] as const)("rechecks an unsent %s after %s", async (kind, action) => { const source = await run(f, { status: "failed", errorCode: "test_run_failed" }); const interactions = issueThreadInteractionService(db); const issue = { id: f.issueId, companyId: f.companyId, status: "in_progress" }; - const createInteraction = (sourceRunId: string) => interactions.create(issue, kind === "confirmation" ? { - kind: "request_confirmation", continuationPolicy: "wake_assignee", sourceRunId, + const addressee = action === "addressee deletion" + ? (await db.insert(agents).values({ companyId: f.companyId, name: "Reviewer", adapterType: "paperclip_runner" }).returning())[0] : null; + const createInteraction = (sourceRunId: string, supersedePendingSiblingInteractions = true) => interactions.create(issue, kind === "confirmation" ? { + kind: "request_confirmation", continuationPolicy: "wake_assignee", sourceRunId, addresseeAgentId: addressee?.id, title: "Confirm", payload: { version: 1, prompt: "Continue?", supersedeOnUserComment: true }, } : { - kind: "ask_user_questions", continuationPolicy: "wake_assignee", sourceRunId, + kind: "ask_user_questions", continuationPolicy: "wake_assignee", sourceRunId, addresseeAgentId: addressee?.id, title: "Choose", payload: { version: 1, title: "Choose", supersedeOnUserComment: true, questions: [{ id: "choice", prompt: "Continue?", selectionMode: "single", required: true, options: [{ id: "yes", label: "Yes" }], }] }, - }, { agentId: f.agentId, runId: sourceRunId }); + }, { agentId: f.agentId, runId: sourceRunId }, { supersedePendingSiblingInteractions }); const interaction = await createInteraction(source.id); - const replacement = action === "replacement" ? await run() : null; + const replacement = action === "replacement" || action === "startup supersession" ? await run() : null; + if (action === "startup supersession") { + await createInteraction(replacement!.id, false); + await db.update(issueThreadInteractions).set({ createdAt: new Date(Date.now() - 1000) }).where(eq(issueThreadInteractions.id, interaction.id)); + } if (action === "historical answer") { await interactions.dismissInteraction(issue, interaction.id, { userId: f.userId }); await db.update(issues).set({ status: "done" }).where(eq(issues.id, f.issueId)); @@ -664,7 +627,11 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu await start(); await flush(); expect((await publications()).some(row => row.idempotencyKey.startsWith(`run:${source.id}:`))).toBe(false); await quiet(); - if (action === "replacement") { + if (action === "addressee deletion") { + expect((await interactions.cancelPendingForDeletedAddressee(f.companyId, addressee!.id)).map(row => row.id)).toContain(interaction.id); + } else if (action === "startup supersession") { + expect(await interactions.sweepSupersededPendingRequestConfirmations()).toMatchObject({ expired: 1 }); + } else if (action === "replacement") { await createInteraction(replacement!.id); expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0].status).toBe("expired"); } else if (action === "comment" || action === "historical comment") { @@ -698,6 +665,37 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu } await quiet(); }); + it.each(["change", "delete", "accept", "reject"] as const)("rechecks a document confirmation after %s detects a stale target", async mode => { + const source = await run(f, { status: "failed", errorCode: "test_run_failed" }); + const interactions = issueThreadInteractionService(db), docs = documentService(db); + const issue = { id: f.issueId, companyId: f.companyId, status: "in_progress", projectId: null, goalId: null }; + const input = { issueId: f.issueId, key: "plan", format: "markdown", body: "First plan", createdByUserId: f.userId }; + const { document } = await docs.upsertIssueDocument(input); + const interaction = await interactions.create(issue, { + kind: "request_confirmation", continuationPolicy: "wake_assignee", sourceRunId: source.id, + payload: { version: 1, prompt: "Apply this plan?", target: { type: "issue_document", issueId: f.issueId, + documentId: document.id, key: document.key, revisionId: document.latestRevisionId!, revisionNumber: document.latestRevisionNumber } }, + }, { agentId: f.agentId, runId: source.id }); + await start(); await flush(); + expect((await publications()).some(row => row.idempotencyKey.startsWith(`run:${source.id}:`))).toBe(false); + await quiet(); + const changed = mode === "delete" + ? { ...(await docs.deleteIssueDocument(f.issueId, document.key))!, latestRevisionId: null, latestRevisionNumber: null } + : (await docs.upsertIssueDocument({ ...input, body: "Revised plan", baseRevisionId: document.latestRevisionId })).document; + await flush(); await quiet(); + if (mode === "accept" || mode === "reject") { + await expect(mode === "accept" + ? interactions.acceptInteraction(issue, interaction.id, {}, { userId: f.userId }) + : interactions.rejectInteraction(issue, interaction.id, {}, { userId: f.userId })) + .rejects.toMatchObject({ status: 409, details: { code: "interaction_stale_target" } }); + } else { + expect((await interactions.expireStaleRequestConfirmationsForIssueDocument(issue, changed, { userId: f.userId })).map(row => row.id)).toContain(interaction.id); + } + expect((await db.select().from(issueThreadInteractions).where(eq(issueThreadInteractions.id, interaction.id)))[0].status).toBe("expired"); + await flush(); + expect((await publications()).map(row => row.idempotencyKey)).toContain(`run:${source.id}:failed:${f.endpointId}`); + await quiet(); + }); it("wakes a question continuation after its in-flight prompt receipt commits", async () => { await db.update(chatEndpoints).set({ provider: "discord" }).where(eq(chatEndpoints.id, f.endpointId)); const source = await run(f, { status: "failed", contextSnapshot: { ...runValues().contextSnapshot, source: "chat:discord" } }); diff --git a/server/src/__tests__/helpers/chat-milestone-fixtures.ts b/server/src/__tests__/helpers/chat-milestone-fixtures.ts new file mode 100644 index 00000000000..00302e480a8 --- /dev/null +++ b/server/src/__tests__/helpers/chat-milestone-fixtures.ts @@ -0,0 +1,72 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { + activityLog, agents, agentWakeupRequests, authUsers, + caseHandoffs, cases, companyMemberships, chatActions, + chatConversations, chatEndpoints, chatMessageLinks, chatPublications, + companies, completionContracts, documentRevisions, documents, + issueDocuments, type Db, nativeRunFinalizations, nativeRunResults, + statusDecisions, statusDecisionEffects, workAssessments, heartbeatRunEvents, + heartbeatRuns, issueComments, issueQuestionResponseDeliveries, issueThreadInteractions, + issues, toolApplications, toolConnections, +} from "@paperclipai/db"; + +// This owns only fixture data. Each test retains its worker, clock and quiet +// assertions, and must stop the worker before cleanup. +export function createChatMilestoneFixtures(db: Db) { + const fixtureCompanies: string[] = []; + const fixtureUsers: string[] = []; + async function create(companyId: string = randomUUID()) { + const agentId = randomUUID(), issueId = randomUUID(), endpointId = randomUUID(), conversationId = randomUUID(), commentId = randomUUID(), userId = randomUUID(); + if (!fixtureCompanies.includes(companyId)) { + fixtureCompanies.push(companyId); + await db.insert(companies).values({ id: companyId, name: "Milestones", issuePrefix: `M${companyId.slice(0, 7)}` }); + } + fixtureUsers.push(userId); + await db.insert(authUsers).values({ id: userId, name: "Owner", email: `${userId}@example.test`, createdAt: new Date(), updatedAt: new Date() }); + await db.insert(companyMemberships).values({ companyId, principalType: "user", principalId: userId, status: "active", membershipRole: "owner" }); + await db.insert(agents).values({ id: agentId, companyId, name: "Maya", adapterType: "paperclip_runner" }); + await db.insert(issues).values({ id: issueId, companyId, title: "Chat task", status: "in_progress", assigneeAgentId: agentId, responsibleUserId: userId }); + const applicationId = randomUUID(), connectionId = randomUUID(); + await db.insert(toolApplications).values({ id: applicationId, companyId, name: `Chat ${applicationId}`, type: "chat" }); + await db.insert(toolConnections).values({ id: connectionId, companyId, applicationId, name: "Chat", uid: connectionId, transport: "chat_sdk", connectionPurpose: "channel", enabled: true, status: "active" }); + await db.insert(chatEndpoints).values({ id: endpointId, companyId, connectionId, assignedAgentId: agentId, publicId: endpointId, provider: "slack", status: "active" }); + await db.insert(chatConversations).values({ id: conversationId, companyId, endpointId, issueId, externalConversationId: conversationId, externalThreadId: conversationId, externalLabel: "Chat" }); + await db.insert(issueComments).values({ id: commentId, companyId, issueId, authorType: "user", body: "Start work" }); + await db.insert(chatMessageLinks).values({ companyId, endpointId, conversationId, commentId, providerMessageId: commentId, direction: "inbound" }); + return { companyId, agentId, issueId, endpointId, conversationId, commentId, userId }; + } + async function cleanup() { + for (const companyId of fixtureCompanies.splice(0)) { + await db.delete(activityLog).where(eq(activityLog.companyId, companyId)); + await db.delete(caseHandoffs).where(eq(caseHandoffs.companyId, companyId)); + await db.delete(cases).where(eq(cases.companyId, companyId)); + await db.delete(chatActions).where(eq(chatActions.companyId, companyId)); + await db.delete(chatMessageLinks).where(eq(chatMessageLinks.companyId, companyId)); + await db.delete(chatPublications).where(eq(chatPublications.companyId, companyId)); + await db.delete(chatConversations).where(eq(chatConversations.companyId, companyId)); + await db.delete(chatEndpoints).where(eq(chatEndpoints.companyId, companyId)); + await db.delete(issueQuestionResponseDeliveries).where(eq(issueQuestionResponseDeliveries.companyId, companyId)); + await db.delete(issueThreadInteractions).where(eq(issueThreadInteractions.companyId, companyId)); + await db.delete(issueComments).where(eq(issueComments.companyId, companyId)); + await db.delete(heartbeatRunEvents).where(eq(heartbeatRunEvents.companyId, companyId)); + await db.delete(nativeRunFinalizations).where(eq(nativeRunFinalizations.companyId, companyId)); + await db.delete(statusDecisionEffects).where(eq(statusDecisionEffects.companyId, companyId)); + await db.delete(statusDecisions).where(eq(statusDecisions.companyId, companyId)); + await db.delete(workAssessments).where(eq(workAssessments.companyId, companyId)); + await db.delete(nativeRunResults).where(eq(nativeRunResults.companyId, companyId)); + await db.delete(heartbeatRuns).where(eq(heartbeatRuns.companyId, companyId)); + await db.delete(agentWakeupRequests).where(eq(agentWakeupRequests.companyId, companyId)); + await db.delete(completionContracts).where(eq(completionContracts.companyId, companyId)); + await db.delete(issueDocuments).where(eq(issueDocuments.companyId, companyId)); + await db.delete(documentRevisions).where(eq(documentRevisions.companyId, companyId)); + await db.delete(documents).where(eq(documents.companyId, companyId)); + await db.delete(issues).where(eq(issues.companyId, companyId)); + await db.delete(agents).where(eq(agents.companyId, companyId)); + await db.delete(companyMemberships).where(eq(companyMemberships.companyId, companyId)); + await db.delete(companies).where(eq(companies.id, companyId)); + } + for (const userId of fixtureUsers.splice(0)) await db.delete(authUsers).where(eq(authUsers.id, userId)); + } + return { create, cleanup }; +} diff --git a/server/src/services/issue-thread-interactions.ts b/server/src/services/issue-thread-interactions.ts index fbac12e0a6f..92d58eccbee 100644 --- a/server/src/services/issue-thread-interactions.ts +++ b/server/src/services/issue-thread-interactions.ts @@ -1898,6 +1898,7 @@ async function expireStaleRequestConfirmationTarget( snapshot, }); const expireInTransaction = async (tx: Db | any) => { + if (args.row.kind === "request_confirmation") await notifyChatRunMilestones(tx, args.row); const [updated] = await tx .update(issueThreadInteractions) .set({ @@ -3251,6 +3252,7 @@ export function issueThreadInteractionService( const cancelled: IssueThreadInteraction[] = []; for (const row of rows) { const updated = await db.transaction(async (tx) => { + if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); if (row.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const [resolved] = await tx .update(issueThreadInteractions) @@ -3336,6 +3338,7 @@ export function issueThreadInteractionService( const expired: IssueThreadInteraction[] = []; for (const { row, replacementInteractionId } of supersededRows) { const updated = await db.transaction(async (tx) => { + if (["ask_user_questions", "request_confirmation"].includes(row.kind)) await notifyChatRunMilestones(tx, row); if (row.kind === "request_confirmation") await notifyDeliveryWork(tx, DELIVERY_QUEUES.toolReviewProjection); const [updatedRow] = await tx .update(issueThreadInteractions) @@ -4724,6 +4727,7 @@ export function issueThreadInteractionService( document, }); const updated = await db.transaction(async (tx) => { + if (row.kind === "request_confirmation") await notifyChatRunMilestones(tx, row); const [resolved] = await tx .update(issueThreadInteractions) .set({