From aaf55692c4def3ce572465fa8df3ff2961d9a687 Mon Sep 17 00:00:00 2001 From: Dotta Date: Mon, 2 Nov 2026 12:25:21 -0600 Subject: [PATCH] refactor(server): inspect chat action receipts once per report Aggregate exact action completion evidence per delivery to avoid repeated unindexed action scans. Keep pending siblings and missing consumed wakes blocked. Bound the real upload integration and preserve slow-call diagnostics without extra SQL. Co-Authored-By: Paperclip --- .../__tests__/idle-completed-history.test.ts | 20 ++++++++++++++++--- .../chat-channels/idle-history-proof.ts | 16 +++++++++------ 2 files changed, 27 insertions(+), 9 deletions(-) diff --git a/server/src/__tests__/idle-completed-history.test.ts b/server/src/__tests__/idle-completed-history.test.ts index 11c45896ef9..c4fe2b56843 100644 --- a/server/src/__tests__/idle-completed-history.test.ts +++ b/server/src/__tests__/idle-completed-history.test.ts @@ -27,7 +27,10 @@ const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPos const blocked = async (reason: string) => { const check = completedHistoryChecks.find(check => check.reason === reason); if (!check) throw new Error(`Missing family ${reason}`); + const started = performance.now(); const [row] = await db.execute<{ blocked: boolean }>(sql`SELECT EXISTS (${sql.raw(check.query)}) AS blocked`); + const elapsed = performance.now() - started; + if (elapsed > 1000) console.info(`History predicate ${reason}: ${Math.round(elapsed)}ms`); return row.blocked; }; beforeAll(async () => { @@ -160,7 +163,7 @@ const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPos expect(await blocked("mcp_attachment_uploads")).toBe(true); const app = express(); app.use(transfers.router); const url = new URL(String(upload.url)); - const result = await request(app).put(url.pathname + url.search).set("Content-Type", "text/plain").send(bytes); + const result = await request(app).put(url.pathname + url.search).set("Content-Type", "text/plain").send(bytes).timeout({ response: 5000, deadline: 10_000 }); expect(result.status).toBe(200); expect(await blocked("mcp_attachment_uploads")).toBe(false); const abandoned = await transfers.getUploadUrl(principal, { ...args, requestId: randomUUID() }); @@ -171,7 +174,7 @@ const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPos await db.update(tables.mcpAttachmentUploads).set({ attachmentId: null }).where(eq(tables.mcpAttachmentUploads.id, String(upload.uploadId))); expect(await blocked("mcp_attachment_uploads")).toBe(true); } finally { await p.cleanup(); } - }); + }, 60_000); it("retains the actual rejected tool review and only its acknowledged delivery", async () => { const { commitToolActionReview } = await import("../services/tool-action-review.js"); @@ -304,7 +307,18 @@ const support = externalDatabaseUrl ? { supported: true } : await getEmbeddedPos await lease(f.endpoint, guard => resources.applyProviderAvailability({ currentEndpoint: f.endpoint, effect, credentialLease: guard, staleResourceEffect: false, claimed: delivery, candidate: { attempts: 0 } })); expect(await blocked("chat_deliveries")).toBe(false); - await db.update(tables.chatDeliveries).set({ normalizedEvent: {} }).where(eq(tables.chatDeliveries.id, delivery.id)); + const firstAction = await f.insertAction({ deliveryId: delivery.id }); + expect(await blocked("chat_deliveries")).toBe(true); + await f.service.processGitHubWebhookIngress(firstAction.id); + expect(await blocked("chat_deliveries")).toBe(false); + const secondAction = await f.insertAction({ deliveryId: delivery.id }); + expect(await blocked("chat_deliveries")).toBe(true); // one completed sibling cannot hide another pending action + await f.service.processGitHubWebhookIngress(secondAction.id); + expect(await blocked("chat_deliveries")).toBe(false); + await db.update(tables.chatDeliveries).set({ eventKind: "message", + normalizedEvent: { providerEventId: delivery.providerEventId, kind: "message" } }).where(eq(tables.chatDeliveries.id, delivery.id)); + expect(await blocked("chat_deliveries")).toBe(true); // ordinary input requires its own consumed wake, not arbitrary completed actions + await db.update(tables.chatDeliveries).set({ eventKind: "uninstallation", normalizedEvent: {} }).where(eq(tables.chatDeliveries.id, delivery.id)); expect(await blocked("chat_deliveries")).toBe(true); const [publication] = await db.insert(tables.chatPublications).values({ companyId: f.companyId, endpointId: f.endpoint.id, conversationId: conversation.id, issueId: issue.id, idempotencyKey: randomUUID(), state: "delivery_unknown", diff --git a/server/src/services/chat-channels/idle-history-proof.ts b/server/src/services/chat-channels/idle-history-proof.ts index 8f9642b3ac5..434087d5e38 100644 --- a/server/src/services/chat-channels/idle-history-proof.ts +++ b/server/src/services/chat-channels/idle-history-proof.ts @@ -42,7 +42,14 @@ function completedAction(a: string): string { } export const idleChatHistoryChecks = [ - { reason: "chat_deliveries", query: `SELECT 1 FROM chat_deliveries d WHERE ( + { reason: "chat_deliveries", query: `WITH action_outcomes AS MATERIALIZED ( + SELECT a.delivery_id, a.kind, (${completedAction('a')}) IS TRUE AS completed + FROM chat_actions a WHERE a.delivery_id IS NOT NULL + ), action_receipts AS ( + SELECT delivery_id, bool_and(completed) AS completed, + bool_or(kind = 'inbound_wakeup' AND completed) AS consumed + FROM action_outcomes GROUP BY delivery_id + ) SELECT 1 FROM chat_deliveries d LEFT JOIN action_receipts receipt ON receipt.delivery_id = d.id WHERE ( EXISTS (SELECT 1 FROM chat_endpoints e WHERE e.id = d.endpoint_id AND e.company_id = d.company_id) AND jsonb_typeof(d.normalized_event) = 'object' AND d.normalized_event->>'providerEventId' = d.provider_event_id @@ -50,14 +57,11 @@ export const idleChatHistoryChecks = [ AND d.event_kind IN ('mention', 'message', 'direct_message', 'message_updated', 'message_deleted', 'message_restored', 'reaction_added', 'reaction_removed', 'action', 'modal_submitted', 'modal_closed', 'slash_command', 'file_shared', 'installation', 'uninstallation', 'unknown') AND (d.state <> 'processed' OR d.event_kind NOT IN ('mention', 'message', 'direct_message') - OR EXISTS (SELECT 1 FROM chat_actions consumed WHERE consumed.delivery_id = d.id - AND consumed.company_id = d.company_id AND consumed.endpoint_id = d.endpoint_id - AND consumed.kind = 'inbound_wakeup' AND (${completedAction('consumed')}))) + OR receipt.consumed IS TRUE) AND d.next_attempt_at IS NULL AND (d.state = 'failed' OR (d.state IN ('processed', 'filtered') AND d.processed_at IS NOT NULL)) AND (d.conversation_id IS NULL OR EXISTS (SELECT 1 FROM chat_conversations c WHERE c.id = d.conversation_id AND c.company_id = d.company_id AND c.endpoint_id = d.endpoint_id)) - AND NOT EXISTS (SELECT 1 FROM chat_actions a WHERE a.delivery_id = d.id - AND (a.company_id <> d.company_id OR a.endpoint_id <> d.endpoint_id OR (${completedAction('a')}) IS NOT TRUE))) IS NOT TRUE` }, + AND COALESCE(receipt.completed, true)) IS NOT TRUE` }, { reason: "chat_publications", query: `SELECT 1 FROM chat_publications p WHERE ( (p.state = 'published' AND p.published_at IS NOT NULL AND p.redacted_error IS NULL OR (p.state IN ('failed', 'cancelled') AND p.next_attempt_at IS NULL))