mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-05 12:29:37 +01:00
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 <noreply@paperclip.ing>
This commit is contained in:
1 parent
021833a602
commit
aaf55692c4
2 files changed
+27
-9
No files matched your search
@@ -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",
|
||||
|
||||
@@ -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))
|
||||
|
||||
Reference in new issue
Block a user