From 6ec22216598d074af25a4daefa08bdd91b1379fd Mon Sep 17 00:00:00 2001 From: "Bender (Paperclip)" Date: Thu, 17 Sep 2026 19:55:25 +0000 Subject: [PATCH] Harden proposed connector telemetry emission accuracy Review revisions to the connector telemetry proposal (#13578): - One completion event per gateway execution: success paths emit only after post-save bookkeeping settles, so a bookkeeping failure that overwrites the row to failed cannot produce a contradictory succeeded+failed pair for the same execution. - Policy denies and rate limits (rows born terminal in recordInvocation) now emit at their only completion boundary. - Composio child disable/restore emit from committed .returning() rows only; a concurrently deleted child or failed write emits nothing. Adds a test seam to reproduce that race deterministically. - The invocation emitter skips its database read entirely while the proposed event name is unregistered (TelemetryClient. isRegisteredEventName), and loads invocation+connection in a single joined query once registered. - Integration tests against the real gateway and tool-access services (embedded Postgres): success, post-success audit failure, execution failure, deny/rate-limit, approval pending/decline/approve, setup-test origin, idempotent replay, telemetry failure isolation, composio cascade/no-op/concurrent-delete/failed-write/restore. - Shared tests prove the proposal wrappers cannot queue or send through the real TelemetryClient and cover disabled/CI suppression. Co-Authored-By: Claude Fable 5 --- packages/shared/src/telemetry/client.ts | 12 +- .../proposed-connector-events.test.ts | 124 ++++ .../tool-access-connector-telemetry.test.ts | 381 ++++++++++++ .../tool-gateway-connector-telemetry.test.ts | 571 ++++++++++++++++++ .../src/services/connector-telemetry.test.ts | 65 +- server/src/services/connector-telemetry.ts | 60 +- server/src/services/tool-gateway.ts | 24 +- 7 files changed, 1193 insertions(+), 44 deletions(-) create mode 100644 packages/shared/src/telemetry/proposed-connector-events.test.ts create mode 100644 server/src/__tests__/tool-access-connector-telemetry.test.ts create mode 100644 server/src/__tests__/tool-gateway-connector-telemetry.test.ts diff --git a/packages/shared/src/telemetry/client.ts b/packages/shared/src/telemetry/client.ts index 4a131d20294..03a10a5550b 100644 --- a/packages/shared/src/telemetry/client.ts +++ b/packages/shared/src/telemetry/client.ts @@ -116,11 +116,21 @@ export class TelemetryClient { * backend event schema. */ track(eventName: K, ...args: TrackArgs): void { - if (!Object.hasOwn(PAPERCLIP_EVENTS, eventName)) return; + if (!this.isRegisteredEventName(eventName)) return; const [dimensions] = args; this.enqueue(eventName, dimensions); } + /** + * Whether `track` would enqueue this first-party event name rather than drop + * it as unregistered. Lets a caller with an expensive payload (e.g. one that + * needs database reads) skip building dimensions for a proposed event the + * client would discard anyway. + */ + isRegisteredEventName(eventName: string): boolean { + return Object.hasOwn(PAPERCLIP_EVENTS, eventName); + } + /** * Tracks plugin telemetry bridge events whose names are built dynamically * from third-party plugin input. The backend accepts only explicitly diff --git a/packages/shared/src/telemetry/proposed-connector-events.test.ts b/packages/shared/src/telemetry/proposed-connector-events.test.ts new file mode 100644 index 00000000000..38d1e84ed67 --- /dev/null +++ b/packages/shared/src/telemetry/proposed-connector-events.test.ts @@ -0,0 +1,124 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { TelemetryClient } from "./client.js"; +import { resolveTelemetryConfig } from "./config.js"; +import { + trackConnectorConnectionCreated, + trackConnectorConnectionUpdated, + trackConnectorInvocationCompleted, +} from "./events.js"; +import type { TelemetryState } from "./types.js"; + +const TEST_STATE: TelemetryState = { + installId: "test-install", + salt: "test-salt", + createdAt: "2026-01-01T00:00:00Z", + firstSeenVersion: "0.0.0", +}; + +function makeClient(config?: { enabled?: boolean }) { + const stateFactory = vi.fn(() => TEST_STATE); + return { + client: new TelemetryClient( + { enabled: config?.enabled ?? true, endpoint: "http://localhost:9999/ingest" }, + stateFactory, + "0.0.0-test", + () => 0.5, + ), + stateFactory, + }; +} + +function trackAllProposedConnectorEvents(client: TelemetryClient) { + trackConnectorConnectionCreated(client, { + connector_key: "github", + transport: "mcp_remote", + auth_kind: "oauth", + setup_flow: "gallery", + status: "active", + enabled: true, + }); + trackConnectorConnectionUpdated(client, { + connector_key: "github", + transport: "mcp_remote", + auth_kind: "oauth", + change_source: "update_api", + previous_status: "draft", + status: "active", + previous_enabled: false, + enabled: true, + }); + trackConnectorInvocationCompleted(client, { + connector_key: "github", + transport: "mcp_remote", + status: "succeeded", + origin: "agent", + duration_seconds: 3, + }); +} + +describe("proposed connector events against the real TelemetryClient", () => { + afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); + }); + + it("cannot queue or send: state stays untouched and nothing hits the wire", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const { client, stateFactory } = makeClient(); + + trackAllProposedConnectorEvents(client); + await client.flush(); + + expect(stateFactory).not.toHaveBeenCalled(); + expect(fetch).not.toHaveBeenCalled(); + }); + + it("differential control: a registered event through the same client does send", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const { client } = makeClient(); + + trackAllProposedConnectorEvents(client); + client.track("project.created", {}); + await client.flush(); + + expect(fetch).toHaveBeenCalledTimes(1); + const body = JSON.parse( + String((vi.mocked(fetch).mock.calls[0]?.[1] as RequestInit).body), + ); + expect(body.events).toEqual([ + expect.objectContaining({ name: "project.created" }), + ]); + }); + + it("reports the three connector event names as unregistered", () => { + const { client } = makeClient(); + expect(client.isRegisteredEventName("connector.connection_created")).toBe(false); + expect(client.isRegisteredEventName("connector.connection_updated")).toBe(false); + expect(client.isRegisteredEventName("connector.invocation_completed")).toBe(false); + expect(client.isRegisteredEventName("project.created")).toBe(true); + }); + + it("a disabled client drops even registered events", async () => { + vi.stubGlobal("fetch", vi.fn().mockResolvedValue({ ok: true })); + const { client, stateFactory } = makeClient({ enabled: false }); + + client.track("project.created", {}); + await client.flush(); + + expect(stateFactory).not.toHaveBeenCalled(); + expect(fetch).not.toHaveBeenCalled(); + }); + + it("environment suppression resolves telemetry off under CI and opt-out flags", () => { + vi.stubEnv("PAPERCLIP_TELEMETRY_DISABLED", "1"); + expect(resolveTelemetryConfig().enabled).toBe(false); + vi.unstubAllEnvs(); + + vi.stubEnv("DO_NOT_TRACK", "1"); + expect(resolveTelemetryConfig().enabled).toBe(false); + vi.unstubAllEnvs(); + + vi.stubEnv("GITHUB_ACTIONS", "true"); + expect(resolveTelemetryConfig().enabled).toBe(false); + }); +}); diff --git a/server/src/__tests__/tool-access-connector-telemetry.test.ts b/server/src/__tests__/tool-access-connector-telemetry.test.ts new file mode 100644 index 00000000000..45fffaa2a96 --- /dev/null +++ b/server/src/__tests__/tool-access-connector-telemetry.test.ts @@ -0,0 +1,381 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + activityLog, + companies, + companySecretBindings, + companySecretVersions, + companySecrets, + connectionGrants, + createDb, + secretAccessEvents, + toolApplications, + toolCatalogEntries, + toolConnections, + toolProfileBindings, + toolProfileEntries, + toolProfiles, +} from "@paperclipai/db"; +import type { ComposioClient } from "../services/composio.js"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; + +const track = vi.fn(); +const isRegisteredEventName = vi.fn(() => true); +vi.mock("../telemetry.js", () => ({ + getTelemetryClient: () => ({ track, isRegisteredEventName }), +})); + +const { toolAccessService } = await import("../services/tool-access.js"); +const { secretService } = await import("../services/secrets.js"); + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported + ? describe + : describe.skip; + +type Db = ReturnType; + +function createdEvents() { + return track.mock.calls.filter(([name]) => name === "connector.connection_created"); +} + +function updatedEvents() { + return track.mock.calls.filter(([name]) => name === "connector.connection_updated"); +} + +async function createCompany(db: Db) { + return db + .insert(companies) + .values({ + name: `Lifecycle telemetry ${randomUUID()}`, + issuePrefix: `LT${randomUUID().slice(0, 6).toUpperCase()}`, + }) + .returning() + .then((rows) => rows[0]!); +} + +async function createComposioParentAndChild(db: Db, companyId: string) { + const secrets = secretService(db); + const apiKey = await secrets.create(companyId, { + name: `Composio test key ${randomUUID().slice(0, 8)}`, + key: `tool_app.${randomUUID()}.credentials_apiKey`, + provider: "local_encrypted", + value: "composio-test-key", + }); + const [application] = await db + .insert(toolApplications) + .values({ companyId, name: "Composio", type: "rest_api", status: "active" }) + .returning(); + const [parent] = await db + .insert(toolConnections) + .values({ + companyId, + applicationId: application!.id, + name: "Composio", + uid: `composio/${randomUUID()}`, + transport: "rest_api", + authKind: "api_key", + status: "active", + enabled: true, + config: { sourceTemplateKey: "composio" }, + transportConfig: { sourceTemplateKey: "composio" }, + credentialRefs: [ + { + name: "credentials.apiKey", + secretId: apiKey.id, + version: "latest", + placement: "header", + key: "x-api-key", + prefix: null, + }, + ], + credentialSecretRefs: [ + { + secretId: apiKey.id, + versionSelector: "latest", + configPath: "credentials.apiKey", + required: true, + label: "Composio API key", + }, + ], + }) + .returning(); + await db.insert(companySecretBindings).values({ + companyId, + secretId: apiKey.id, + targetType: "tool_connection", + targetId: parent!.id, + configPath: "credentials.apiKey", + }); + const [child] = await db + .insert(toolConnections) + .values({ + companyId, + applicationId: application!.id, + name: "GitHub (via Composio)", + uid: `composio/github/${randomUUID()}`, + transport: "mcp_remote", + authKind: "none", + status: "active", + enabled: true, + config: { + provider: "composio", + parentConnectionId: parent!.id, + toolkitSlug: "github", + connectedAccountId: "account-github", + }, + transportConfig: {}, + }) + .returning(); + return { parent: parent!, child: child! }; +} + +function fakeComposioClient(accountStatus: () => string): ComposioClient { + return { + validateApiKey: vi.fn(async () => undefined), + listToolkits: vi.fn(async () => ({ + items: [{ slug: "github", name: "GitHub" }], + })), + listAuthConfigs: vi.fn(async () => ({ items: [] })), + createConnectLink: vi.fn(async () => ({ + link_token: "link", + redirect_url: "https://composio.test/link", + expires_at: new Date().toISOString(), + })), + listConnectedAccounts: vi.fn(async () => ({ + items: [ + { + id: "account-github", + user_id: "paperclip:test", + status: accountStatus(), + toolkit: { slug: "github" }, + auth_config: { + id: "auth-github", + auth_scheme: "OAUTH2", + is_composio_managed: true, + }, + }, + ], + })), + deleteConnectedAccount: vi.fn(async () => undefined), + createSession: vi.fn(async () => ({ + session_id: "session", + mcp: { url: "https://composio.test/mcp" }, + })), + resumeSession: vi.fn(async () => ({ + session_id: "session", + mcp: { url: "https://composio.test/mcp" }, + })), + } as unknown as ComposioClient; +} + +describeEmbeddedPostgres("connector lifecycle telemetry (tool-access)", () => { + let db!: Db; + let tempDb: Awaited> | null = + null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase( + "paperclip-lifecycle-telemetry-", + ); + db = createDb(tempDb.connectionString); + }, 30_000); + + afterEach(async () => { + track.mockClear(); + await db.delete(activityLog); + await db.delete(toolProfileEntries); + await db.delete(toolProfileBindings); + await db.delete(toolProfiles); + await db.delete(toolCatalogEntries); + await db.delete(connectionGrants); + await db.delete(toolConnections); + await db.delete(toolApplications); + await db.delete(secretAccessEvents); + await db.delete(companySecretBindings); + await db.delete(companySecretVersions); + await db.delete(companySecrets); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + function service(options: Parameters[1] = {}) { + return toolAccessService(db, { + remoteHttpEndpointLookup: async () => [{ address: "8.8.8.8", family: 4 }], + ...options, + }); + } + + it("createConnection emits one created event with the catalog key from the committed row", async () => { + const company = await createCompany(db); + const svc = service(); + await svc.createConnection(company.id, { + name: "GitHub fixture", + transport: "mcp_remote", + config: { url: "https://fixture.example/mcp", sourceTemplateKey: "github" }, + enabled: true, + status: "active", + }); + const events = createdEvents(); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ + connector_key: "github", + transport: "mcp_remote", + setup_flow: "api", + status: "active", + enabled: true, + }); + }, 30_000); + + it("updateConnection emits committed transitions only; metadata saves stay silent", async () => { + const company = await createCompany(db); + const svc = service(); + const connection = await svc.createConnection(company.id, { + name: "Custom fixture", + transport: "mcp_remote", + config: { url: "https://fixture.example/mcp" }, + enabled: true, + status: "active", + }); + track.mockClear(); + + await svc.updateConnection(connection.id, { enabled: false }); + let events = updatedEvents(); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ + connector_key: "custom", + change_source: "update_api", + previous_enabled: true, + enabled: false, + previous_status: "active", + status: "active", + }); + + track.mockClear(); + await svc.updateConnection(connection.id, { name: "Renamed fixture" }); + expect(updatedEvents()).toHaveLength(0); + }, 30_000); + + it("composio parent pause emits child transitions from committed rows and stays silent on the no-op repeat", async () => { + const company = await createCompany(db); + const { parent, child } = await createComposioParentAndChild(db, company.id); + const svc = service({ + composioClientFactory: () => fakeComposioClient(() => "ACTIVE"), + }); + + await svc.updateConnection(parent.id, { enabled: false }); + let events = updatedEvents(); + // Parent transition (update_api) + child cascade (composio_sync). + expect(events).toHaveLength(2); + const childEvent = events.find( + ([, dims]) => (dims as { change_source: string }).change_source === "composio_sync", + ); + expect(childEvent?.[1]).toMatchObject({ + connector_key: "custom", + previous_enabled: true, + enabled: false, + }); + await expect( + db.select().from(toolConnections).then((rows) => + rows.find((row) => row.id === child.id), + ), + ).resolves.toMatchObject({ enabled: false }); + + track.mockClear(); + await svc.updateConnection(parent.id, { enabled: false }); + // Parent and child both already disabled: nothing changed, nothing emits. + expect(updatedEvents()).toHaveLength(0); + }, 30_000); + + it("a child deleted between listing and write produces no event", async () => { + const company = await createCompany(db); + const { parent, child } = await createComposioParentAndChild(db, company.id); + const svc = service({ + composioClientFactory: () => fakeComposioClient(() => "ACTIVE"), + beforeComposioChildLifecycleWrite: async (childId) => { + await db.delete(toolConnections).where(eq(toolConnections.id, childId)); + }, + }); + + await svc.updateConnection(parent.id, { enabled: false }); + const events = updatedEvents(); + // Only the parent's own committed transition may report. + expect(events).toHaveLength(1); + expect( + events.every( + ([, dims]) => + (dims as { change_source: string }).change_source !== "composio_sync", + ), + ).toBe(true); + const remaining = await db.select().from(toolConnections); + expect(remaining.find((row) => row.id === child.id)).toBeUndefined(); + }, 30_000); + + it("a failed child lifecycle write emits nothing for that child", async () => { + const company = await createCompany(db); + const { parent } = await createComposioParentAndChild(db, company.id); + const svc = service({ + composioClientFactory: () => fakeComposioClient(() => "ACTIVE"), + beforeComposioChildLifecycleWrite: async () => { + throw new Error("forced child write failure"); + }, + }); + + await expect( + svc.updateConnection(parent.id, { enabled: false }), + ).rejects.toThrow(/forced child write failure/); + expect( + updatedEvents().filter( + ([, dims]) => + (dims as { change_source: string }).change_source === "composio_sync", + ), + ).toHaveLength(0); + }, 30_000); + + it("restore emits only for children whose committed enabled state changed", async () => { + const company = await createCompany(db); + const { parent, child } = await createComposioParentAndChild(db, company.id); + let accountStatus = "ACTIVE"; + const svc = service({ + composioClientFactory: () => fakeComposioClient(() => accountStatus), + }); + + await svc.updateConnection(parent.id, { enabled: false }); + track.mockClear(); + + // Inactive account: the child stays disabled, so no transition may report. + accountStatus = "INACTIVE"; + await svc.updateConnection(parent.id, { enabled: true }); + expect( + updatedEvents().filter( + ([, dims]) => + (dims as { change_source: string }).change_source === "composio_sync", + ), + ).toHaveLength(0); + + track.mockClear(); + accountStatus = "ACTIVE"; + await svc.updateConnection(parent.id, { enabled: true }); + const events = updatedEvents().filter( + ([, dims]) => + (dims as { change_source: string }).change_source === "composio_sync", + ); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ + previous_enabled: false, + enabled: true, + }); + await expect( + db.select().from(toolConnections).then((rows) => + rows.find((row) => row.id === child.id), + ), + ).resolves.toMatchObject({ enabled: true }); + }, 30_000); +}); diff --git a/server/src/__tests__/tool-gateway-connector-telemetry.test.ts b/server/src/__tests__/tool-gateway-connector-telemetry.test.ts new file mode 100644 index 00000000000..c8afd9624a0 --- /dev/null +++ b/server/src/__tests__/tool-gateway-connector-telemetry.test.ts @@ -0,0 +1,571 @@ +import { randomUUID } from "node:crypto"; +import { eq } from "drizzle-orm"; +import { afterAll, afterEach, beforeAll, describe, expect, it, vi } from "vitest"; +import { + activityLog, + agents, + companies, + companyMemberships, + connectionGrants, + createDb, + heartbeatRuns, + issues, + projects, + toolAccessAuditEvents, + toolActionRequests, + toolApplications, + toolCallEvents, + toolCatalogEntries, + toolConnections, + toolGatewayRateLimitCounters, + toolGatewaySessions, + toolInvocations, + toolPolicies, + toolProfileBindings, + toolProfiles, + toolRuntimeSlots, +} from "@paperclipai/db"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; + +// Proposal wrappers drop unregistered names inside the real client, so the +// integration assertions use a client double that reports the proposed event +// as registered. `track` observes exactly what the emitter would hand the +// real client after schema adoption. +const track = vi.fn(); +const isRegisteredEventName = vi.fn(() => true); +vi.mock("../telemetry.js", () => ({ + getTelemetryClient: () => ({ track, isRegisteredEventName }), +})); + +// Deterministic reproduction of "post-execution bookkeeping failure": the +// gateway's success path awaits writeAudit -> logActivity after the succeeded +// save; failing that call lands in the catch that overwrites the row to +// failed. +const failLogActivityForAction = { current: null as string | null }; +vi.mock("../services/activity-log.js", async (importOriginal) => { + const actual = + await importOriginal(); + return { + ...actual, + logActivity: vi.fn( + async (...args: Parameters) => { + const input = args[1] as { action?: string }; + if ( + failLogActivityForAction.current && + input?.action === failLogActivityForAction.current + ) { + throw new Error(`forced activity write failure: ${input.action}`); + } + return actual.logActivity(...args); + }, + ), + }; +}); + +const { createToolGatewayService } = await import( + "../services/tool-gateway.js" +); + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported + ? describe + : describe.skip; + +type Db = ReturnType; + +const SIGNING_SECRET = "connector-telemetry-signing-secret"; + +function mcpSuccessResponse(init: RequestInit): Response { + const body = JSON.parse(String(init.body ?? "{}")); + return new Response( + JSON.stringify({ + jsonrpc: "2.0", + id: body?.id ?? "fixture", + result: { content: [{ type: "text", text: "ok" }] }, + }), + { status: 200, headers: { "content-type": "application/json" } }, + ); +} + +async function createCompany(db: Db) { + return db + .insert(companies) + .values({ + name: `Connector telemetry ${randomUUID()}`, + issuePrefix: `CT${randomUUID().slice(0, 6).toUpperCase()}`, + }) + .returning() + .then((rows) => rows[0]!); +} + +async function createAgent(db: Db, companyId: string) { + return db + .insert(agents) + .values({ + companyId, + name: `Agent ${randomUUID()}`, + role: "engineer", + adapterType: "process", + adapterConfig: {}, + runtimeConfig: {}, + permissions: {}, + }) + .returning() + .then((rows) => rows[0]!); +} + +async function createIssueAndRun(db: Db, companyId: string, agentId: string) { + const project = await db + .insert(projects) + .values({ companyId, name: `Project ${randomUUID()}` }) + .returning() + .then((rows) => rows[0]!); + const issue = await db + .insert(issues) + .values({ + companyId, + projectId: project.id, + title: `Telemetry issue ${randomUUID()}`, + status: "in_progress", + assigneeAgentId: agentId, + }) + .returning() + .then((rows) => rows[0]!); + const run = await db + .insert(heartbeatRuns) + .values({ + companyId, + agentId, + invocationSource: "assignment", + status: "running", + contextSnapshot: { issueId: issue.id, projectId: project.id }, + }) + .returning() + .then((rows) => rows[0]!); + return { issue, run }; +} + +async function allowAllToolsForAgent(db: Db, companyId: string, agentId: string) { + const profile = await db + .insert(toolProfiles) + .values({ + companyId, + profileKey: `telemetry-all-${randomUUID()}`, + name: `Telemetry profile ${randomUUID()}`, + defaultAction: "allow", + }) + .returning() + .then((rows) => rows[0]!); + await db.insert(toolProfileBindings).values({ + companyId, + profileId: profile.id, + targetType: "agent", + targetId: agentId, + }); +} + +async function createRemoteMcpTool(db: Db, companyId: string) { + const [application] = await db + .insert(toolApplications) + .values({ + companyId, + applicationKey: `telemetry-app-${randomUUID().slice(0, 8)}`, + name: `Remote app ${randomUUID()}`, + type: "mcp_http", + status: "active", + }) + .returning(); + const config = { + url: "https://mcp.telemetry.test/mcp", + sourceTemplateKey: "github", + }; + const [connection] = await db + .insert(toolConnections) + .values({ + companyId, + applicationId: application!.id, + name: `Remote connection ${randomUUID()}`, + uid: `test/${randomUUID()}`, + transport: "mcp_remote", + status: "active", + enabled: true, + healthStatus: "ok", + config, + transportConfig: config, + credentialRefs: [], + credentialSecretRefs: [], + }) + .returning(); + await db.insert(connectionGrants).values({ + companyId, + connectionId: connection!.id, + kind: "organization", + credentialSecretRefs: [], + status: "active", + isDefault: true, + }); + const [catalogEntry] = await db + .insert(toolCatalogEntries) + .values({ + companyId, + applicationId: application!.id, + connectionId: connection!.id, + entryKind: "tool", + name: `kv_set-${randomUUID()}`, + toolName: "kv_set", + title: "KV Set", + description: "Set a value", + inputSchema: { + type: "object", + properties: { key: { type: "string" }, value: { type: "string" } }, + required: ["key", "value"], + additionalProperties: false, + }, + annotations: { readOnlyHint: false }, + riskLevel: "write", + isReadOnly: false, + isWrite: true, + isDestructive: false, + status: "active", + versionHash: randomUUID(), + }) + .returning(); + return { application: application!, connection: connection!, catalogEntry: catalogEntry! }; +} + +async function settledEvents(expectedCount: number) { + await vi.waitFor(() => { + expect( + track.mock.calls.filter( + ([name]) => name === "connector.invocation_completed", + ).length, + ).toBeGreaterThanOrEqual(expectedCount); + }); + // The emitter runs as detached background work; give any surplus emission a + // chance to land before asserting the exact count. + await new Promise((resolve) => setTimeout(resolve, 150)); + return track.mock.calls.filter( + ([name]) => name === "connector.invocation_completed", + ); +} + +describeEmbeddedPostgres("gateway connector invocation telemetry", () => { + let db!: Db; + let tempDb: Awaited> | null = + null; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase( + "paperclip-connector-telemetry-", + ); + db = createDb(tempDb.connectionString); + }, 30_000); + + afterEach(async () => { + track.mockClear(); + failLogActivityForAction.current = null; + await db.delete(activityLog); + await db.delete(toolCallEvents); + await db.delete(toolRuntimeSlots); + await db.delete(toolGatewaySessions); + await db.delete(toolGatewayRateLimitCounters); + await db.delete(toolActionRequests); + await db.delete(toolInvocations); + await db.delete(toolAccessAuditEvents); + await db.delete(toolPolicies); + await db.delete(toolProfileBindings); + await db.delete(toolProfiles); + await db.delete(toolCatalogEntries); + await db.delete(connectionGrants); + await db.delete(toolConnections); + await db.delete(toolApplications); + await db.delete(heartbeatRuns); + await db.delete(issues); + await db.delete(projects); + await db.delete(agents); + await db.delete(companyMemberships); + await db.delete(companies); + }); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + async function fixture(options: Parameters[1] = {}) { + const company = await createCompany(db); + const agent = await createAgent(db, company.id); + const { run } = await createIssueAndRun(db, company.id, agent.id); + const remoteTool = await createRemoteMcpTool(db, company.id); + await allowAllToolsForAgent(db, company.id, agent.id); + const gateway = createToolGatewayService(db, { + toolActionSigningSecret: SIGNING_SECRET, + remoteHttpRequest: async (_url, init) => mcpSuccessResponse(init), + ...options, + }); + const session = await gateway.createSession({ + companyId: company.id, + agentId: agent.id, + runId: run.id, + }); + const tool = (await gateway.listToolsForSession(session.token)).find( + (candidate) => candidate.connectionId === remoteTool.connection.id, + ); + expect(tool).toBeDefined(); + return { company, agent, run, remoteTool, gateway, session, tool: tool! }; + } + + it("emits exactly one succeeded event for a successful agent execution", async () => { + const f = await fixture(); + await expect( + f.gateway.executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "a", value: "1" }, + }), + ).resolves.toMatchObject({ status: "completed" }); + + const events = await settledEvents(1); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ + connector_key: "github", + transport: "mcp_remote", + status: "succeeded", + origin: "agent", + }); + }, 30_000); + + it("emits a single failed event when post-success audit work throws (no contradictory pair)", async () => { + const f = await fixture(); + failLogActivityForAction.current = "tool_gateway.call_completed"; + + await expect( + f.gateway.executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "a", value: "1" }, + }), + ).rejects.toThrow(/forced activity write failure/); + + const [invocation] = await db.select().from(toolInvocations); + expect(invocation?.status).toBe("failed"); + + const events = await settledEvents(1); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ status: "failed", origin: "agent" }); + expect( + events.some(([, dims]) => (dims as { status: string }).status === "succeeded"), + ).toBe(false); + }, 30_000); + + it("emits one failed event when the remote execution itself fails", async () => { + const f = await fixture({ + remoteHttpRequest: async () => + new Response("upstream exploded", { status: 500 }), + }); + await expect( + f.gateway.executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "a", value: "1" }, + }), + ).rejects.toThrow(); + + const events = await settledEvents(1); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ status: "failed" }); + }, 30_000); + + it("emits one denied event for a policy block and one rate_limited event past the limit", async () => { + const f = await fixture(); + await db.insert(toolPolicies).values({ + companyId: f.company.id, + name: "Block telemetry tool", + policyType: "block", + selectors: { connectionId: f.remoteTool.connection.id }, + priority: 1, + }); + await expect( + f.gateway.executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "a", value: "1" }, + }), + ).rejects.toMatchObject({ reasonCode: expect.any(String) }); + + let events = await settledEvents(1); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ status: "denied", origin: "agent" }); + + track.mockClear(); + await db.delete(toolPolicies); + await db.insert(toolPolicies).values({ + companyId: f.company.id, + name: "One call only", + policyType: "rate_limit", + selectors: { connectionId: f.remoteTool.connection.id }, + config: { limit: 1, windowSeconds: 60 }, + priority: 1, + }); + await expect( + f.gateway.executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "rate", value: "first" }, + }), + ).resolves.toMatchObject({ status: "completed" }); + await expect( + f.gateway.executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "rate", value: "second" }, + }), + ).rejects.toMatchObject({ status: 429 }); + + events = await settledEvents(2); + expect(events).toHaveLength(2); + expect(events.map(([, dims]) => (dims as { status: string }).status).sort()).toEqual( + ["rate_limited", "succeeded"], + ); + }, 30_000); + + it("approval flow: pending emits nothing, decline emits denied, approve emits succeeded", async () => { + const f = await fixture(); + await db.insert(toolPolicies).values({ + companyId: f.company.id, + name: "Review telemetry tool", + policyType: "require_approval", + selectors: { connectionId: f.remoteTool.connection.id }, + priority: 1, + }); + + await f.gateway + .executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "a", value: "declined" }, + }) + .then( + () => { + throw new Error("expected approval_required"); + }, + (error) => expect(error).toMatchObject({ reasonCode: "approval_required" }), + ); + // Awaiting approval is in-flight: no completion event may exist yet. The + // approval card itself emits the registered interaction.created event, so + // filter to the proposed connector event. + await new Promise((resolve) => setTimeout(resolve, 150)); + expect( + track.mock.calls.filter(([name]) => name === "connector.invocation_completed"), + ).toHaveLength(0); + + const [pendingRequest] = await db.select().from(toolActionRequests); + await f.gateway.declineActionRequest({ + companyId: f.company.id, + actionRequestId: pendingRequest!.id, + actor: { userId: "board" }, + }); + let events = await settledEvents(1); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ status: "denied", origin: "agent" }); + + track.mockClear(); + await f.gateway + .executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "a", value: "approved" }, + }) + .then( + () => { + throw new Error("expected approval_required"); + }, + (error) => expect(error).toMatchObject({ reasonCode: "approval_required" }), + ); + const approvable = await db + .select() + .from(toolActionRequests) + .where(eq(toolActionRequests.status, "pending")) + .then((rows) => rows[0]!); + await f.gateway.approveActionRequest({ + companyId: f.company.id, + actionRequestId: approvable.id, + actor: { userId: "board" }, + }); + events = await settledEvents(1); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ status: "succeeded", origin: "agent" }); + }, 40_000); + + it("labels connection-test executions as setup_test", async () => { + const f = await fixture(); + await db.insert(companyMemberships).values({ + companyId: f.company.id, + principalType: "user", + principalId: "board-user", + status: "active", + membershipRole: "member", + }); + const testCall = await f.gateway.executeTestCall({ + companyId: f.company.id, + connectionId: f.remoteTool.connection.id, + agentId: f.agent.id, + userId: "board-user", + toolName: f.tool.name, + parameters: { key: "t", value: "1" }, + }); + expect(testCall).toMatchObject({ decision: "allowed" }); + expect(testCall).not.toHaveProperty("error"); + + const events = await settledEvents(1); + expect(events).toHaveLength(1); + expect(events[0]?.[1]).toMatchObject({ + status: "succeeded", + origin: "setup_test", + }); + }, 30_000); + + it("an idempotent replay returns the stored result without a second event", async () => { + const f = await fixture(); + const idempotencyKey = `telemetry-replay-${randomUUID()}`; + await expect( + f.gateway.executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "a", value: "1" }, + idempotencyKey, + }), + ).resolves.toMatchObject({ status: "completed" }); + await settledEvents(1); + await expect( + f.gateway.executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "a", value: "1" }, + idempotencyKey, + }), + ).resolves.toMatchObject({ status: "replayed" }); + + const events = await settledEvents(1); + expect(events).toHaveLength(1); + }, 30_000); + + it("telemetry sink failures never fail the tool call", async () => { + const f = await fixture(); + track.mockImplementation(() => { + throw new Error("sink offline"); + }); + await expect( + f.gateway.executeTool({ + sessionToken: f.session.token, + tool: f.tool.name, + parameters: { key: "a", value: "1" }, + }), + ).resolves.toMatchObject({ status: "completed" }); + const [invocation] = await db.select().from(toolInvocations); + expect(invocation?.status).toBe("succeeded"); + track.mockReset(); + }, 30_000); +}); diff --git a/server/src/services/connector-telemetry.test.ts b/server/src/services/connector-telemetry.test.ts index 1f10f02f6fa..42bc937056f 100644 --- a/server/src/services/connector-telemetry.test.ts +++ b/server/src/services/connector-telemetry.test.ts @@ -2,7 +2,11 @@ import { toolConnections, toolInvocations, type Db } from "@paperclipai/db"; import { beforeEach, describe, expect, it, vi } from "vitest"; const track = vi.fn(); -let telemetryClient: { track: typeof track } | null = { track }; +const isRegisteredEventName = vi.fn(() => true); +let telemetryClient: { + track: typeof track; + isRegisteredEventName: typeof isRegisteredEventName; +} | null = { track, isRegisteredEventName }; vi.mock("../telemetry.js", () => ({ getTelemetryClient: () => telemetryClient, @@ -50,28 +54,35 @@ function invocationRow(overrides: Record = {}): ToolInvocationR function fakeDb( invocation: ToolInvocationRow | null, connection: ToolConnectionRow | null, -): Db { +): Db & { selectCalls: () => number } { + let selectCalls = 0; return { - select: () => ({ - from: (table: unknown) => ({ - where: () => - Promise.resolve( - table === toolInvocations - ? invocation - ? [invocation] - : [] - : connection - ? [connection] - : [], - ), - }), - }), - } as unknown as Db; + selectCalls: () => selectCalls, + select: () => { + selectCalls += 1; + return { + from: () => ({ + innerJoin: () => ({ + where: () => + Promise.resolve( + // Mirror the emitter's single joined read: an invocation with + // no (or a deleted) connection joins to zero rows. + invocation && connection && invocation.connectionId + ? [{ invocation, connection }] + : [], + ), + }), + }), + }; + }, + } as unknown as Db & { selectCalls: () => number }; } beforeEach(() => { track.mockReset(); - telemetryClient = { track }; + isRegisteredEventName.mockReset(); + isRegisteredEventName.mockReturnValue(true); + telemetryClient = { track, isRegisteredEventName }; }); describe("connectorKeyForConnection", () => { @@ -237,4 +248,22 @@ describe("emitConnectorInvocationCompleted", () => { ).resolves.toBeUndefined(); expect(track).not.toHaveBeenCalled(); }); + + it("reads nothing from the database while the event name is unregistered", async () => { + isRegisteredEventName.mockReturnValue(false); + const db = fakeDb(invocationRow(), connectionRow()); + await emitConnectorInvocationCompleted(db, "inv-1"); + expect(db.selectCalls()).toBe(0); + expect(track).not.toHaveBeenCalled(); + expect(isRegisteredEventName).toHaveBeenCalledWith( + "connector.invocation_completed", + ); + }); + + it("loads the invocation and connection with a single read once registered", async () => { + const db = fakeDb(invocationRow(), connectionRow()); + await emitConnectorInvocationCompleted(db, "inv-1"); + expect(db.selectCalls()).toBe(1); + expect(track).toHaveBeenCalledTimes(1); + }); }); diff --git a/server/src/services/connector-telemetry.ts b/server/src/services/connector-telemetry.ts index b6088b27563..495263f42b6 100644 --- a/server/src/services/connector-telemetry.ts +++ b/server/src/services/connector-telemetry.ts @@ -23,9 +23,12 @@ export type ConnectorChangeSource = | "composio_sync"; /** - * Terminal `tool_invocations.status` values actually written by the gateway - * and approval-review paths. `pending`, `authorized`, `awaiting_approval`, and - * `executing` are in-flight and never emit: pending approval is not a failure. + * Terminal `tool_invocations.status` values. `pending`, `authorized`, + * `awaiting_approval`, and `executing` are in-flight and never emit: pending + * approval is not a failure. `cancelled` is declared in the schema enum but no + * current code path writes it to an invocation (signing-path cancellations + * land on the action request and fail the invocation); it stays here so a + * future writer is counted without a telemetry change. */ const TERMINAL_INVOCATION_STATUSES: ReadonlySet = new Set([ "succeeded", @@ -147,16 +150,30 @@ export function emitConnectorConnectionUpdated( /** * Emits one proposed `connector.invocation_completed` event for an invocation - * a caller already wrote to a terminal status. Loads the committed row, so - * call it beside each terminal status write with - * `void emitConnectorInvocationCompleted(db, invocationId)` and never await - * it: like `agent.task_run`, this is best-effort background work and must not - * delay the caller's own response or lifecycle writes. + * a caller already wrote to a terminal status. Never await it: like + * `agent.task_run`, this is best-effort background work and must not delay + * the caller's own response or lifecycle writes. + * + * Call placement inside the gateway is deliberate and asymmetric. Failure + * paths call it right after their terminal save. Success paths call it only + * after the post-save bookkeeping (action-request settlement, tool-call + * event, audit) has completed, because a bookkeeping failure lands in a catch + * that overwrites the row to `failed` and emits there — emitting the success + * beforehand would let one execution report both outcomes. * * Self-guards, so callers do not have to re-check anything: non-terminal - * statuses, invocations without a connection, and non-tool-purpose - * connections all return without emitting. Emission is per terminal write, - * not exactly-once — a replayed terminal write on the same row emits again. + * statuses, invocations without a tool-purpose connection, and an + * unregistered (still-proposed) event name all return without emitting. The + * registration gate keeps the completion path free of telemetry reads until + * schema adoption: the runtime client drops unregistered names in `track`, + * so there is no point loading rows for an event that cannot be queued. + * + * Delivery is best-effort per execution, not exactly-once. Known residual + * duplications, accepted for the proposal stage: an outer safety-net catch + * can re-save `failed` and emit again when a failure path's own bookkeeping + * throws after its emit (test-invocation approval flow), and a crash-side + * replay that re-writes a terminal status emits again. Deduplication state + * is out of scope for this proposal. */ export async function emitConnectorInvocationCompleted( db: Db, @@ -165,22 +182,21 @@ export async function emitConnectorInvocationCompleted( try { const client = getTelemetryClient(); if (!client) return; + if (!client.isRegisteredEventName("connector.invocation_completed")) return; - const invocation = await db - .select() + const joined = await db + .select({ invocation: toolInvocations, connection: toolConnections }) .from(toolInvocations) + .innerJoin( + toolConnections, + eq(toolInvocations.connectionId, toolConnections.id), + ) .where(eq(toolInvocations.id, invocationId)) .then((rows) => rows[0] ?? null); - if (!invocation) return; + if (!joined) return; + const { invocation, connection } = joined; if (!TERMINAL_INVOCATION_STATUSES.has(invocation.status)) return; - if (!invocation.connectionId) return; - - const connection = await db - .select() - .from(toolConnections) - .where(eq(toolConnections.id, invocation.connectionId)) - .then((rows) => rows[0] ?? null); - if (!connection || !isToolPurpose(connection)) return; + if (!isToolPurpose(connection)) return; const startedAtMs = invocation.startedAt ? new Date(invocation.startedAt).getTime() diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index e851c9bc703..1f097386e36 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -7141,7 +7141,6 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, args.invocationId)); - void emitConnectorInvocationCompleted(db, args.invocationId); await writeToolCallEvent({ invocationId: args.invocationId, session: args.session, @@ -7183,6 +7182,10 @@ export function createToolGatewayService( execution: connectedMcpExecution.execution, }, }); + // After the bookkeeping above: a throw there is caught below, overwrites + // the row to failed, and emits — an earlier success emit would make one + // execution report both outcomes. + void emitConnectorInvocationCompleted(db, args.invocationId); return { decision: "allowed" as const, invocationId: args.invocationId, @@ -9130,6 +9133,9 @@ export function createToolGatewayService( } if (!accessDecision.allowed) { + // recordInvocation inserted this row already terminal (denied or + // rate_limited), so this is its only completion boundary. + void emitConnectorInvocationCompleted(db, invocationId); await writeAudit({ session, companyId: input.companyId, @@ -10295,6 +10301,9 @@ export function createToolGatewayService( }); } if (!accessDecision.allowed) { + // recordInvocation inserted this row already terminal (denied or + // rate_limited), so this is its only completion boundary. + void emitConnectorInvocationCompleted(db, invocationId); await writeAudit({ session, companyId: session.companyId, @@ -10429,7 +10438,6 @@ export function createToolGatewayService( updatedAt: completedAt, }) .where(eq(toolInvocations.id, invocationId)); - void emitConnectorInvocationCompleted(db, invocationId); if (input.approvedActionRequestId) { const [executedRequest] = await db .update(toolActionRequests) @@ -10500,6 +10508,10 @@ export function createToolGatewayService( execution: connectedMcpExecution?.execution ?? undefined, }, }); + // After the bookkeeping above: a throw there is caught below, overwrites + // the row to failed, and emits — an earlier success emit would make one + // execution report both outcomes. + void emitConnectorInvocationCompleted(db, invocationId); return { invocationId, status: "completed" as const, @@ -10738,6 +10750,9 @@ export function createToolGatewayService( } if (!accessDecision.allowed) { + // recordInvocation inserted this row already terminal (denied or + // rate_limited), so this is its only completion boundary. + void emitConnectorInvocationCompleted(db, invocationId); await writeAudit({ session: sessionLike, companyId: input.runContext.companyId, @@ -10820,7 +10835,6 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, invocationId)); - void emitConnectorInvocationCompleted(db, invocationId); await writeToolCallEvent({ invocationId, session: sessionLike, @@ -10853,6 +10867,10 @@ export function createToolGatewayService( resultSummary: resultValidation.summary, }, }); + // After the bookkeeping above: a throw there is caught below, overwrites + // the row to failed, and emits — an earlier success emit would make one + // execution report both outcomes. + void emitConnectorInvocationCompleted(db, invocationId); return resultValidation.value as typeof result; } catch (err) { const status = err instanceof ToolGatewayHttpError ? err.status : 502;