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 <noreply@anthropic.com>

packages/shared/src/telemetry/client.ts

@@ -116,11 +116,21 @@ export class TelemetryClient {
    * backend event schema.
    */
   track<K extends TelemetryEventName>(eventName: K, ...args: TrackArgs<K>): 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

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);
+  });
+});

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<typeof createDb>;
+
+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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<typeof toolAccessService>[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);
+});

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<typeof import("../services/activity-log.js")>();
+  return {
+    ...actual,
+    logActivity: vi.fn(
+      async (...args: Parameters<typeof actual.logActivity>) => {
+        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<typeof createDb>;
+
+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<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | 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<typeof createToolGatewayService>[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);
+});

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<string, unknown> = {}): 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);
+  });
 });

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<string> = 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()

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;