Bugfix/compressor bug (#2833)

tim-inkeep · web-flow · commit f4a9c69d500d · 2026-03-26T16:18:33.000Z
* added compressor settings

* added compressor settings

* updarted compressor and artifact fetching

* updated bug branch for ci and comments

* added changeset

* fixed final pullfrog comments

* updated base compressor to parallellize
diff --git a/.changeset/unconscious-brown-deer.md b/.changeset/unconscious-brown-deer.md
@@ -0,0 +1,6 @@
+---
+"@inkeep/agents-api": patch
+"@inkeep/agents-core": patch
+---
+
+Fix key_findings persistence in compressor by using proper update instead of insert-only upsert
diff --git a/agents-api/src/domains/run/agents/tools/default-tools.ts b/agents-api/src/domains/run/agents/tools/default-tools.ts
@@ -3,6 +3,7 @@ import { type Tool, type ToolSet, tool } from 'ai';
 import { getLogger } from '../../../../logger';
 import { formatOversizedRetrievalReason } from '../../artifacts/artifact-utils';
 import { getModelAwareCompressionConfig } from '../../compression/BaseCompressor';
+import { SENTINEL_KEY } from '../../constants/artifact-syntax';
 import { agentSessionManager } from '../../session/AgentSession';
 import type { AgentRunContext } from '../agent-types';
 import { wrapToolWithStreaming } from './tool-wrapper';
@@ -20,6 +21,21 @@ export function getArtifactTools(ctx: AgentRunContext): Tool<any, any> {
     execute: async ({ artifactId, toolCallId }) => {
       logger.info({ artifactId, toolCallId }, 'get_artifact_full executed');
 
+      const compressor = ctx.currentCompressor;
+      if (compressor?.hasSummarizedArtifact(artifactId)) {
+        const summarized = compressor.getSummarizedArtifact(artifactId);
+        logger.info(
+          { artifactId, toolCallId },
+          'Blocked retrieval of artifact already summarized in compression'
+        );
+        return {
+          artifactId,
+          status: 'already_summarized',
+          key_findings: summarized?.key_findings ?? [],
+          hint: `This artifact's key findings are already in your compressed context. Use them directly to answer. To pass this artifact to a tool, use { "${SENTINEL_KEY.ARTIFACT}": "${artifactId}", "${SENTINEL_KEY.TOOL}": "${summarized?.tool_call_id ?? toolCallId}" } sentinel instead of retrieving it.`,
+        };
+      }
+
       const streamRequestId = ctx.streamRequestId ?? '';
       const artifactService = agentSessionManager.getArtifactService(streamRequestId);
 
diff --git a/agents-api/src/domains/run/compression/BaseCompressor.ts b/agents-api/src/domains/run/compression/BaseCompressor.ts
@@ -1,10 +1,11 @@
 import { randomUUID } from 'node:crypto';
-import type { ModelSettings } from '@inkeep/agents-core';
+import type { ModelSettings, Part } from '@inkeep/agents-core';
 import {
   estimateTokens as estimateTokensUtil,
   GENERATION_TYPES,
   getLedgerArtifacts,
   SPAN_KEYS,
+  updateLedgerArtifactParts,
 } from '@inkeep/agents-core';
 import { type Span, SpanStatusCode } from '@opentelemetry/api';
 import runDbClient from '../../../data/db/runDbClient';
@@ -539,6 +540,26 @@ export abstract class BaseCompressor {
     );
 
     this.cumulativeSummary = summary;
+
+    if (summary.related_artifacts?.length) {
+      try {
+        const persistResult = await this.persistArtifactKeyFindings(summary.related_artifacts);
+        logger.debug(
+          { ...persistResult, conversationId: this.conversationId },
+          'Artifact key_findings persistence completed'
+        );
+      } catch (err) {
+        logger.warn(
+          {
+            conversationId: this.conversationId,
+            artifactIds: summary.related_artifacts.map((a) => a.id),
+            err: err instanceof Error ? err.message : String(err),
+          },
+          'Failed to persist key_findings to artifacts'
+        );
+      }
+    }
+
     return summary;
   }
 
@@ -615,6 +636,111 @@ export abstract class BaseCompressor {
     return this.cumulativeSummary;
   }
 
+  hasSummarizedArtifact(artifactId: string): boolean {
+    return this.cumulativeSummary?.related_artifacts?.some((a) => a.id === artifactId) ?? false;
+  }
+
+  getSummarizedArtifact(
+    artifactId: string
+  ): { key_findings: string[]; tool_call_id: string } | null {
+    const artifact = this.cumulativeSummary?.related_artifacts?.find((a) => a.id === artifactId);
+    if (!artifact) return null;
+    return {
+      key_findings: artifact.key_findings,
+      tool_call_id: artifact.tool_call_id,
+    };
+  }
+
+  protected async persistArtifactKeyFindings(
+    relatedArtifacts: NonNullable<ConversationSummary['related_artifacts']>
+  ): Promise<{ persisted: number; skipped: number; failed: number }> {
+    const result = { persisted: 0, skipped: 0, failed: 0 };
+    const scopes = { tenantId: this.tenantId, projectId: this.projectId };
+
+    const artifactsWithFindings = relatedArtifacts.filter((a) => a.key_findings?.length);
+    result.skipped += relatedArtifacts.length - artifactsWithFindings.length;
+
+    if (artifactsWithFindings.length === 0) return result;
+
+    const fetchResults = await Promise.allSettled(
+      artifactsWithFindings.map((artifact) =>
+        getLedgerArtifacts(runDbClient)({ scopes, artifactId: artifact.id }).then((existing) => ({
+          artifact,
+          existing,
+        }))
+      )
+    );
+
+    const updates: Array<{ artifact: (typeof artifactsWithFindings)[number]; parts: Part[] }> = [];
+
+    for (const settled of fetchResults) {
+      if (settled.status === 'rejected') {
+        result.failed++;
+        logger.warn(
+          { conversationId: this.conversationId, error: String(settled.reason) },
+          'Failed to fetch artifact for key_findings persistence'
+        );
+        continue;
+      }
+
+      const { artifact, existing } = settled.value;
+      if (existing.length === 0) {
+        logger.debug(
+          { artifactId: artifact.id },
+          'Artifact not found in DB, skipping key_findings persistence'
+        );
+        result.skipped++;
+        continue;
+      }
+
+      const parts = structuredClone(existing[0].parts ?? []) as Part[];
+      const firstPart = parts[0];
+      if (!firstPart || firstPart.kind !== 'data' || !firstPart.data?.summary) {
+        logger.debug(
+          { artifactId: artifact.id, hasParts: parts.length > 0 },
+          'Artifact has no data part with summary structure, skipping key_findings persistence'
+        );
+        result.skipped++;
+        continue;
+      }
+
+      firstPart.data.summary = {
+        ...firstPart.data.summary,
+        key_findings: artifact.key_findings,
+      };
+
+      updates.push({ artifact, parts });
+    }
+
+    const updateResults = await Promise.allSettled(
+      updates.map(({ artifact, parts }) =>
+        updateLedgerArtifactParts(runDbClient)({ scopes, artifactId: artifact.id, parts }).then(
+          (updated) => ({ artifact, updated })
+        )
+      )
+    );
+
+    for (const settled of updateResults) {
+      if (settled.status === 'rejected') {
+        result.failed++;
+        logger.warn(
+          { conversationId: this.conversationId, error: String(settled.reason) },
+          'Failed to persist key_findings to artifact'
+        );
+        continue;
+      }
+      const { artifact, updated } = settled.value;
+      if (updated) {
+        result.persisted++;
+      } else {
+        logger.warn({ artifactId: artifact.id }, 'updateLedgerArtifactParts matched no rows');
+        result.failed++;
+      }
+    }
+
+    return result;
+  }
+
   cleanup(options: { resetSummary?: boolean; keepRecentToolCalls?: number } = {}): void {
     const { resetSummary = false, keepRecentToolCalls = 0 } = options;
     if (keepRecentToolCalls > 0) {
diff --git a/agents-api/src/domains/run/compression/__tests__/BaseCompressor.test.ts b/agents-api/src/domains/run/compression/__tests__/BaseCompressor.test.ts
@@ -691,4 +691,140 @@ describe('BaseCompressor', () => {
       expect(compressor['processedToolCalls'].size).toBeGreaterThan(0);
     });
   });
+
+  describe('hasSummarizedArtifact', () => {
+    it('should return false when cumulativeSummary is null', () => {
+      // biome-ignore lint/complexity/useLiteralKeys: accessing private property for testing
+      compressor['cumulativeSummary'] = null;
+      expect(compressor.hasSummarizedArtifact('artifact-1')).toBe(false);
+    });
+
+    it('should return false when related_artifacts is null', () => {
+      // biome-ignore lint/complexity/useLiteralKeys: accessing private property for testing
+      compressor['cumulativeSummary'] = {
+        type: 'conversation_summary_v1',
+        session_id: null,
+        _fallback: null,
+        high_level: 'summary',
+        user_intent: 'test',
+        decisions: [],
+        open_questions: [],
+        next_steps: { for_agent: [], for_user: [] },
+        related_artifacts: null,
+      };
+      expect(compressor.hasSummarizedArtifact('artifact-1')).toBe(false);
+    });
+
+    it('should return false when artifact is not in related_artifacts', () => {
+      // biome-ignore lint/complexity/useLiteralKeys: accessing private property for testing
+      compressor['cumulativeSummary'] = {
+        type: 'conversation_summary_v1',
+        session_id: null,
+        _fallback: null,
+        high_level: 'summary',
+        user_intent: 'test',
+        decisions: [],
+        open_questions: [],
+        next_steps: { for_agent: [], for_user: [] },
+        related_artifacts: [
+          {
+            id: 'other-artifact',
+            name: 'Other',
+            tool_name: 'tool',
+            tool_call_id: 'tc-1',
+            content_type: 'text',
+            key_findings: ['finding'],
+          },
+        ],
+      };
+      expect(compressor.hasSummarizedArtifact('artifact-1')).toBe(false);
+    });
+
+    it('should return true when artifact exists in related_artifacts', () => {
+      // biome-ignore lint/complexity/useLiteralKeys: accessing private property for testing
+      compressor['cumulativeSummary'] = {
+        type: 'conversation_summary_v1',
+        session_id: null,
+        _fallback: null,
+        high_level: 'summary',
+        user_intent: 'test',
+        decisions: [],
+        open_questions: [],
+        next_steps: { for_agent: [], for_user: [] },
+        related_artifacts: [
+          {
+            id: 'artifact-1',
+            name: 'Test Artifact',
+            tool_name: 'tool',
+            tool_call_id: 'tc-1',
+            content_type: 'text',
+            key_findings: ['finding'],
+          },
+        ],
+      };
+      expect(compressor.hasSummarizedArtifact('artifact-1')).toBe(true);
+    });
+  });
+
+  describe('getSummarizedArtifact', () => {
+    it('should return null when cumulativeSummary is null', () => {
+      // biome-ignore lint/complexity/useLiteralKeys: accessing private property for testing
+      compressor['cumulativeSummary'] = null;
+      expect(compressor.getSummarizedArtifact('artifact-1')).toBeNull();
+    });
+
+    it('should return null when artifact is not found', () => {
+      // biome-ignore lint/complexity/useLiteralKeys: accessing private property for testing
+      compressor['cumulativeSummary'] = {
+        type: 'conversation_summary_v1',
+        session_id: null,
+        _fallback: null,
+        high_level: 'summary',
+        user_intent: 'test',
+        decisions: [],
+        open_questions: [],
+        next_steps: { for_agent: [], for_user: [] },
+        related_artifacts: [],
+      };
+      expect(compressor.getSummarizedArtifact('artifact-1')).toBeNull();
+    });
+
+    it('should return key_findings and tool_call_id for matching artifact', () => {
+      // biome-ignore lint/complexity/useLiteralKeys: accessing private property for testing
+      compressor['cumulativeSummary'] = {
+        type: 'conversation_summary_v1',
+        session_id: null,
+        _fallback: null,
+        high_level: 'summary',
+        user_intent: 'test',
+        decisions: [],
+        open_questions: [],
+        next_steps: { for_agent: [], for_user: [] },
+        related_artifacts: [
+          {
+            id: 'artifact-1',
+            name: 'Test Artifact',
+            tool_name: 'tool',
+            tool_call_id: 'tc-42',
+            content_type: 'text',
+            key_findings: ['finding-a', 'finding-b'],
+          },
+          {
+            id: 'artifact-2',
+            name: 'Other Artifact',
+            tool_name: 'tool',
+            tool_call_id: 'tc-99',
+            content_type: 'text',
+            key_findings: ['other-finding'],
+          },
+        ],
+      };
+
+      const result = compressor.getSummarizedArtifact('artifact-1');
+      expect(result).toEqual({
+        key_findings: ['finding-a', 'finding-b'],
+        tool_call_id: 'tc-42',
+      });
+    });
+  });
 });
diff --git a/packages/agents-core/src/__tests__/data-access/ledgerArtifacts.test.ts b/packages/agents-core/src/__tests__/data-access/ledgerArtifacts.test.ts
@@ -6,6 +6,7 @@ import {
   deleteLedgerArtifactsByTask,
   getLedgerArtifacts,
   getLedgerArtifactsByContext,
+  updateLedgerArtifactParts,
 } from '../../data-access/runtime/ledgerArtifacts';
 import type { AgentsRunDatabaseClient } from '../../db/runtime/runtime-client';
 import { testRunDbClient } from '../setup';
@@ -639,4 +640,55 @@ describe('Ledger Artifacts Data Access', () => {
       expect(result).toBe(0);
     });
   });
+
+  describe('updateLedgerArtifactParts', () => {
+    it('should return true when artifact is successfully updated', async () => {
+      const mockReturning = vi.fn().mockResolvedValue([{ id: testArtifactId }]);
+      const mockWhere = vi.fn().mockReturnValue({ returning: mockReturning });
+      const mockSet = vi.fn().mockReturnValue({ where: mockWhere });
+      const mockUpdate = vi.fn().mockReturnValue({ set: mockSet });
+
+      const mockDb = {
+        ...db,
+        update: mockUpdate,
+      } as any;
+
+      const newParts = [{ kind: 'data' as const, data: { summary: { key_findings: ['a', 'b'] } } }];
+
+      const result = await updateLedgerArtifactParts(mockDb)({
+        scopes: { tenantId: testTenantId, projectId: testProjectId },
+        artifactId: testArtifactId,
+        parts: newParts,
+      });
+
+      expect(result).toBe(true);
+      expect(mockUpdate).toHaveBeenCalled();
+      expect(mockSet).toHaveBeenCalledWith(
+        expect.objectContaining({
+          parts: newParts,
+          updatedAt: expect.any(String),
+        })
+      );
+    });
+
+    it('should return false when no artifact matches', async () => {
+      const mockReturning = vi.fn().mockResolvedValue([]);
+      const mockWhere = vi.fn().mockReturnValue({ returning: mockReturning });
+      const mockSet = vi.fn().mockReturnValue({ where: mockWhere });
+      const mockUpdate = vi.fn().mockReturnValue({ set: mockSet });
+
+      const mockDb = {
+        ...db,
+        update: mockUpdate,
+      } as any;
+
+      const result = await updateLedgerArtifactParts(mockDb)({
+        scopes: { tenantId: testTenantId, projectId: testProjectId },
+        artifactId: 'non-existent-artifact',
+        parts: [{ kind: 'text' as const, text: 'test' }],
+      });
+
+      expect(result).toBe(false);
+    });
+  });
 });
diff --git a/packages/agents-core/src/data-access/runtime/ledgerArtifacts.ts b/packages/agents-core/src/data-access/runtime/ledgerArtifacts.ts
