Invocation log: count repeated failures on one row, short payloads past an hourly budget

packages/backend/prisma/migrations/20261006090000_tool_invocation_repeat_count/migration.sql

@@ -0,0 +1,7 @@
+-- How many identical calls this row stands for. A failing call that repeats
+-- within a minute (same connector, tool and error) is counted on the first
+-- row instead of being stored again: one workspace's own rate limit put
+-- 21,000 identical 429 rows a day into this table. Existing rows stand for one
+-- call each. Adding a column with a constant default is a catalog-only change
+-- in Postgres 11+, so this does not rewrite the table.
+ALTER TABLE "tool_invocations" ADD COLUMN "repeat_count" INTEGER NOT NULL DEFAULT 1;

packages/backend/prisma/schema.prisma

@@ -711,6 +711,10 @@ model ToolInvocation {
 
   clientInfo String? @map("client_info") // JSON: {authMethod, apiKeyName, userEmail, mcpServerName}
 
+  // How many identical calls this row stands for: a failure repeating within
+  // a minute is counted here instead of stored again (see AuditService).
+  repeatCount Int @default(1) @map("repeat_count")
+
   createdAt DateTime @default(now()) @map("created_at")
 
   @@index([toolId, createdAt])

packages/backend/src/audit/audit.breakdowns.spec.ts

@@ -20,13 +20,14 @@ describe('AuditService.getBreakdowns', () => {
       .mockResolvedValueOnce([{ userId: 'u1', _count: { _all: 10 } }])
       // byUserErr
       .mockResolvedValueOnce([]);
-    const count = jest
+    // Calls are summed over repeat_count, so totals come from aggregate().
+    const aggregate = jest
       .fn()
-      .mockResolvedValueOnce(10) // total
-      .mockResolvedValueOnce(3) // errors
-      .mockResolvedValueOnce(4); // proxyCalls
+      .mockResolvedValueOnce({ _sum: { repeatCount: 10 } }) // total
+      .mockResolvedValueOnce({ _sum: { repeatCount: 3 } }) // errors
+      .mockResolvedValueOnce({ _sum: { repeatCount: 4 } }); // proxyCalls
     return {
-      toolInvocation: { groupBy, count },
+      toolInvocation: { groupBy, aggregate },
       connector: { findMany: jest.fn().mockResolvedValue([{ id: 'c1', name: 'Koch ERP' }]) },
       mcpServerConfig: { findMany: jest.fn().mockResolvedValue([{ id: 's1', name: 'KOCH Superpowers' }]) },
       user: { findMany: jest.fn().mockResolvedValue([{ id: 'u1', email: 'a@b.c', name: 'Alice' }]) },

packages/backend/src/audit/audit.service.spec.ts

@@ -10,6 +10,8 @@ describe('AuditService', () => {
         create: jest.fn().mockResolvedValue({}),
         findMany: jest.fn().mockResolvedValue([]),
         count: jest.fn().mockResolvedValue(0),
+        aggregate: jest.fn().mockResolvedValue({ _sum: { repeatCount: 0 } }),
+        update: jest.fn().mockResolvedValue({}),
       },
       // resolveUserId consults users to satisfy the FK before insert.
       // Default: the test user exists. Individual tests override
@@ -21,6 +23,10 @@ describe('AuditService', () => {
     service = new AuditService(mockPrisma);
   });
 
+  afterEach(async () => {
+    await service.onModuleDestroy();
+  });
+
   describe('logInvocation', () => {
     it('should persist an invocation record', async () => {
       await service.logInvocation({
@@ -32,14 +38,16 @@ describe('AuditService', () => {
         durationMs: 150,
       });
 
-      expect(mockPrisma.toolInvocation.create).toHaveBeenCalledWith({
-        data: expect.objectContaining({
-          toolId: 'tool-1',
-          userId: 'user-1',
-          status: 'SUCCESS',
-          durationMs: 150,
+      expect(mockPrisma.toolInvocation.create).toHaveBeenCalledWith(
+        expect.objectContaining({
+          data: expect.objectContaining({
+            toolId: 'tool-1',
+            userId: 'user-1',
+            status: 'SUCCESS',
+            durationMs: 150,
+          }),
         }),
-      });
+      );
     });
 
     it('should not throw if persistence fails', async () => {
@@ -87,11 +95,11 @@ describe('AuditService', () => {
 
   describe('getStats', () => {
     it('should return aggregated stats', async () => {
-      mockPrisma.toolInvocation.count
-        .mockResolvedValueOnce(10)  // 24h invocations
-        .mockResolvedValueOnce(2)   // 24h errors
-        .mockResolvedValueOnce(50)  // 7d invocations
-        .mockResolvedValueOnce(100); // total
+      mockPrisma.toolInvocation.aggregate
+        .mockResolvedValueOnce({ _sum: { repeatCount: 10 } }) // 24h invocations
+        .mockResolvedValueOnce({ _sum: { repeatCount: 2 } }) // 24h errors
+        .mockResolvedValueOnce({ _sum: { repeatCount: 50 } }) // 7d invocations
+        .mockResolvedValueOnce({ _sum: { repeatCount: 100 } }); // total
 
       const stats = await service.getStats();
 
@@ -101,7 +109,80 @@ describe('AuditService', () => {
         invocations7d: 50,
         totalInvocations: 100,
       });
-      expect(mockPrisma.toolInvocation.count).toHaveBeenCalledTimes(4);
+      expect(mockPrisma.toolInvocation.aggregate).toHaveBeenCalledTimes(4);
+    });
+  });
+
+  describe('repeated failures', () => {
+    const fail = (error = '429 Too Many Requests: body={"code":429,"msg":"Request rate limit reached"}') =>
+      service.logInvocation({
+        toolId: 'tool-1',
+        connectorId: 'conn-1',
+        organizationId: 'org-1',
+        input: { view: 'api.pgb_v1' },
+        status: 'ERROR',
+        error,
+      });
+
+    beforeEach(() => {
+      let n = 0;
+      mockPrisma.toolInvocation.create.mockImplementation(async () => ({ id: `row-${++n}` }));
+    });
+
+    it('stores the first failure and counts identical ones on it', async () => {
+      await fail();
+      await fail();
+      await fail();
+      expect(mockPrisma.toolInvocation.create).toHaveBeenCalledTimes(1);
+
+      await service.flushRepeats(true);
+      expect(mockPrisma.toolInvocation.update).toHaveBeenCalledWith({
+        where: { id: 'row-1' },
+        data: { repeatCount: { increment: 2 } },
+      });
+    });
+
+    it('treats errors that differ only in numbers as the same', async () => {
+      await fail('upstream timeout after 3012 ms (request 8812)');
+      await fail('upstream timeout after 2977 ms (request 9001)');
+      expect(mockPrisma.toolInvocation.create).toHaveBeenCalledTimes(1);
+    });
+
+    it('stores a different error, or a success, as its own row', async () => {
+      await fail();
+      await fail('403 Forbidden: Access denied');
+      await service.logInvocation({ toolId: 'tool-1', connectorId: 'conn-1', input: {}, status: 'SUCCESS' });
+      expect(mockPrisma.toolInvocation.create).toHaveBeenCalledTimes(3);
+    });
+
+    it('stores the failure again once the window has passed, carrying over the count', async () => {
+      const now = jest.spyOn(Date, 'now');
+      now.mockReturnValue(1_000_000);
+      await fail();
+      await fail();
+      now.mockReturnValue(1_000_000 + 61_000);
+      await fail();
+      expect(mockPrisma.toolInvocation.create).toHaveBeenCalledTimes(2);
+      expect(mockPrisma.toolInvocation.update).toHaveBeenCalledWith({
+        where: { id: 'row-1' },
+        data: { repeatCount: { increment: 1 } },
+      });
+      now.mockRestore();
+    });
+  });
+
+  describe('payload budget', () => {
+    it('keeps only a short excerpt once an organisation has logged its hourly budget', async () => {
+      const big = { rows: 'x'.repeat(20_000) };
+      for (let i = 0; i < 1001; i++) {
+        await service.logInvocation({ toolId: 't', organizationId: 'busy-org', input: {}, output: big, status: 'SUCCESS' });
+      }
+      const calls = mockPrisma.toolInvocation.create.mock.calls;
+      const size = (call: any) => JSON.stringify(call[0].data.output).length;
+      // Within the budget the normal excerpt rules apply; past it, 512 bytes.
+      expect(size(calls[999])).toBeGreaterThan(1_500);
+      expect(size(calls[1000])).toBeLessThanOrEqual(512);
+      expect(calls[1000][0].data.status).toBe('SUCCESS');
     });
   });
 });