diff --git a/packages/shared/src/telemetry/events.ts b/packages/shared/src/telemetry/events.ts index 5b38b75d784..b652e7cb8f8 100644 --- a/packages/shared/src/telemetry/events.ts +++ b/packages/shared/src/telemetry/events.ts @@ -153,6 +153,76 @@ export function trackAgentTaskRun( }); } +export function trackConnectorConnectionCreated( + client: TelemetryClient, + dims: { + connector_key: RawDimension<"custom">; + transport: RawDimension<"mcp_remote" | "rest_api" | "local_stdio">; + auth_kind: RawDimension<"oauth" | "api_key" | "none">; + setup_flow: RawDimension<"gallery" | "api" | "example" | "composio_sync">; + status: RawDimension<"draft" | "active" | "disabled" | "archived">; + enabled: boolean; + }, +): void { + client.track( + // @ts-expect-error -- proposed-telemetry(https://github.com/paperclipai/paperclip/issues/13578): measure which catalog connectors installations create connections for + "connector.connection_created", + dims, + ); +} + +export function trackConnectorConnectionUpdated( + client: TelemetryClient, + dims: { + connector_key: RawDimension<"custom">; + transport: RawDimension<"mcp_remote" | "rest_api" | "local_stdio">; + auth_kind: RawDimension<"oauth" | "api_key" | "none">; + change_source: RawDimension< + | "update_api" + | "gallery_setup" + | "oauth_callback" + | "credential_refresh" + | "archive" + | "example" + | "composio_sync" + >; + previous_status: RawDimension<"draft" | "active" | "disabled" | "archived">; + status: RawDimension<"draft" | "active" | "disabled" | "archived">; + previous_enabled: boolean; + enabled: boolean; + }, +): void { + client.track( + // @ts-expect-error -- proposed-telemetry(https://github.com/paperclipai/paperclip/issues/13578): measure connector lifecycle transitions (configured, paused, archived) after creation + "connector.connection_updated", + dims, + ); +} + +export function trackConnectorInvocationCompleted( + client: TelemetryClient, + dims: { + connector_key: RawDimension<"custom">; + transport: RawDimension<"mcp_remote" | "rest_api" | "local_stdio">; + status: RawDimension< + | "succeeded" + | "failed" + | "denied" + | "cancelled" + | "timed_out" + | "rate_limited" + >; + origin: RawDimension<"setup_test" | "agent" | "user" | "system" | "plugin">; + duration_seconds?: number; + }, +): void { + client.track( + // @ts-expect-error -- proposed-telemetry(https://github.com/paperclipai/paperclip/issues/13578): measure whether connected connectors are successfully used and where invocations fail + "connector.invocation_completed", + dims, + ); +} + export function trackErrorHandlerCrash( client: TelemetryClient, dims: { errorCode: string }, diff --git a/packages/shared/src/telemetry/index.ts b/packages/shared/src/telemetry/index.ts index 7628cac5927..c09c72ab81b 100644 --- a/packages/shared/src/telemetry/index.ts +++ b/packages/shared/src/telemetry/index.ts @@ -18,6 +18,9 @@ export { trackErrorHandlerCrash, trackInteractionCreated, trackInteractionResolved, + trackConnectorConnectionCreated, + trackConnectorConnectionUpdated, + trackConnectorInvocationCompleted, } from "./events.js"; export type { TelemetryConfig, diff --git a/server/src/services/connector-telemetry.test.ts b/server/src/services/connector-telemetry.test.ts new file mode 100644 index 00000000000..1f10f02f6fa --- /dev/null +++ b/server/src/services/connector-telemetry.test.ts @@ -0,0 +1,240 @@ +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 }; + +vi.mock("../telemetry.js", () => ({ + getTelemetryClient: () => telemetryClient, +})); + +const { + connectorKeyForConnection, + emitConnectorConnectionCreated, + emitConnectorConnectionUpdated, + emitConnectorInvocationCompleted, + invocationOrigin, +} = await import("./connector-telemetry.js"); + +type ToolConnectionRow = typeof toolConnections.$inferSelect; +type ToolInvocationRow = typeof toolInvocations.$inferSelect; + +function connectionRow(overrides: Record = {}): ToolConnectionRow { + return { + id: "conn-1", + connectionPurpose: "tool", + transport: "mcp_remote", + authKind: "oauth", + status: "active", + enabled: true, + config: { sourceTemplateKey: "github" }, + ...overrides, + } as unknown as ToolConnectionRow; +} + +function invocationRow(overrides: Record = {}): ToolInvocationRow { + return { + id: "inv-1", + connectionId: "conn-1", + status: "succeeded", + actorType: "agent", + runId: "run-1", + issueId: "issue-1", + gatewayId: "gw-1", + startedAt: new Date("2026-09-17T00:00:00.000Z"), + completedAt: new Date("2026-09-17T00:00:07.400Z"), + ...overrides, + } as unknown as ToolInvocationRow; +} + +function fakeDb( + invocation: ToolInvocationRow | null, + connection: ToolConnectionRow | null, +): Db { + return { + select: () => ({ + from: (table: unknown) => ({ + where: () => + Promise.resolve( + table === toolInvocations + ? invocation + ? [invocation] + : [] + : connection + ? [connection] + : [], + ), + }), + }), + } as unknown as Db; +} + +beforeEach(() => { + track.mockReset(); + telemetryClient = { track }; +}); + +describe("connectorKeyForConnection", () => { + it("keeps a sourceTemplateKey that resolves in the shared catalog", () => { + expect(connectorKeyForConnection(connectionRow())).toBe("github"); + }); + + it("reports custom for keys outside the catalog", () => { + expect( + connectorKeyForConnection(connectionRow({ config: { sourceTemplateKey: "my-internal-server" } })), + ).toBe("custom"); + }); + + it("reports custom when the key is missing or not a string", () => { + expect(connectorKeyForConnection(connectionRow({ config: {} }))).toBe("custom"); + expect( + connectorKeyForConnection(connectionRow({ config: { sourceTemplateKey: 42 } })), + ).toBe("custom"); + }); +}); + +describe("invocationOrigin", () => { + it("labels user-driven connection tests without run context as setup_test", () => { + expect( + invocationOrigin( + invocationRow({ actorType: "user", runId: null, issueId: null, gatewayId: null }), + ), + ).toBe("setup_test"); + }); + + it("keeps the actor type for run-attached invocations", () => { + expect(invocationOrigin(invocationRow())).toBe("agent"); + expect(invocationOrigin(invocationRow({ actorType: "user" }))).toBe("user"); + }); +}); + +describe("emitConnectorConnectionCreated", () => { + it("emits catalog identity and lifecycle state for tool connections", () => { + emitConnectorConnectionCreated(connectionRow(), "gallery"); + expect(track).toHaveBeenCalledWith("connector.connection_created", { + connector_key: "github", + transport: "mcp_remote", + auth_kind: "oauth", + setup_flow: "gallery", + status: "active", + enabled: true, + }); + }); + + it("never emits for channel or AI connections", () => { + emitConnectorConnectionCreated(connectionRow({ connectionPurpose: "channel" }), "api"); + expect(track).not.toHaveBeenCalled(); + }); + + it("is a no-op without an initialized telemetry client", () => { + telemetryClient = null; + expect(() => emitConnectorConnectionCreated(connectionRow(), "gallery")).not.toThrow(); + }); + + it("swallows client failures instead of failing the setup path", () => { + track.mockImplementation(() => { + throw new Error("sink offline"); + }); + expect(() => emitConnectorConnectionCreated(connectionRow(), "gallery")).not.toThrow(); + }); +}); + +describe("emitConnectorConnectionUpdated", () => { + it("stays silent when the committed write left status and enabled unchanged", () => { + emitConnectorConnectionUpdated( + connectionRow(), + { status: "active", enabled: true }, + "update_api", + ); + expect(track).not.toHaveBeenCalled(); + }); + + it("emits the transition when lifecycle status changes", () => { + emitConnectorConnectionUpdated( + connectionRow(), + { status: "draft", enabled: true }, + "oauth_callback", + ); + expect(track).toHaveBeenCalledWith("connector.connection_updated", { + connector_key: "github", + transport: "mcp_remote", + auth_kind: "oauth", + change_source: "oauth_callback", + previous_status: "draft", + status: "active", + previous_enabled: true, + enabled: true, + }); + }); + + it("emits when only the enabled flag flips", () => { + emitConnectorConnectionUpdated( + connectionRow({ enabled: false }), + { status: "active", enabled: true }, + "update_api", + ); + expect(track).toHaveBeenCalledTimes(1); + expect(track.mock.calls[0]?.[1]).toMatchObject({ + previous_enabled: true, + enabled: false, + }); + }); + + it("never emits for non-tool connections even on lifecycle changes", () => { + emitConnectorConnectionUpdated( + connectionRow({ connectionPurpose: "channel" }), + { status: "draft", enabled: true }, + "update_api", + ); + expect(track).not.toHaveBeenCalled(); + }); +}); + +describe("emitConnectorInvocationCompleted", () => { + it("emits terminal outcome, origin, and duration from the committed row", async () => { + await emitConnectorInvocationCompleted(fakeDb(invocationRow(), connectionRow()), "inv-1"); + expect(track).toHaveBeenCalledWith("connector.invocation_completed", { + connector_key: "github", + transport: "mcp_remote", + status: "succeeded", + origin: "agent", + duration_seconds: 7, + }); + }); + + it("omits duration when the invocation never recorded a start time", async () => { + await emitConnectorInvocationCompleted( + fakeDb(invocationRow({ startedAt: null, status: "denied" }), connectionRow()), + "inv-1", + ); + expect(track.mock.calls[0]?.[1]).not.toHaveProperty("duration_seconds"); + expect(track.mock.calls[0]?.[1]).toMatchObject({ status: "denied" }); + }); + + it("ignores in-flight statuses so approval waits never count as failures", async () => { + await emitConnectorInvocationCompleted( + fakeDb(invocationRow({ status: "awaiting_approval" }), connectionRow()), + "inv-1", + ); + expect(track).not.toHaveBeenCalled(); + }); + + it("ignores invocations without a connection or with a non-tool connection", async () => { + await emitConnectorInvocationCompleted( + fakeDb(invocationRow({ connectionId: null }), connectionRow()), + "inv-1", + ); + await emitConnectorInvocationCompleted( + fakeDb(invocationRow(), connectionRow({ connectionPurpose: "channel" })), + "inv-1", + ); + expect(track).not.toHaveBeenCalled(); + }); + + it("resolves without emitting when the invocation row is missing", async () => { + await expect( + emitConnectorInvocationCompleted(fakeDb(null, null), "inv-missing"), + ).resolves.toBeUndefined(); + expect(track).not.toHaveBeenCalled(); + }); +}); diff --git a/server/src/services/connector-telemetry.ts b/server/src/services/connector-telemetry.ts new file mode 100644 index 00000000000..b6088b27563 --- /dev/null +++ b/server/src/services/connector-telemetry.ts @@ -0,0 +1,209 @@ +import { eq } from "drizzle-orm"; +import { toolConnections, toolInvocations, type Db } from "@paperclipai/db"; +import { getConnectableAppDefinition } from "@paperclipai/shared"; +import { + trackConnectorConnectionCreated, + trackConnectorConnectionUpdated, + trackConnectorInvocationCompleted, +} from "@paperclipai/shared/telemetry"; +import { logger } from "../middleware/logger.js"; +import { getTelemetryClient } from "../telemetry.js"; + +type ToolConnectionRow = typeof toolConnections.$inferSelect; +type ToolInvocationRow = typeof toolInvocations.$inferSelect; + +export type ConnectorSetupFlow = "gallery" | "api" | "example" | "composio_sync"; +export type ConnectorChangeSource = + | "update_api" + | "gallery_setup" + | "oauth_callback" + | "credential_refresh" + | "archive" + | "example" + | "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. + */ +const TERMINAL_INVOCATION_STATUSES: ReadonlySet = new Set([ + "succeeded", + "failed", + "denied", + "cancelled", + "timed_out", + "rate_limited", +]); + +/** + * Connector identity for telemetry is the reviewed first-party catalog slug + * only: `config.sourceTemplateKey` validated against the shared app-definitions + * catalog. Anything else — custom MCP servers, user-named connections, keys + * that no longer resolve to a catalog entry — reports the literal `custom` so + * no user- or provider-controlled string leaves the process. + */ +export function connectorKeyForConnection( + connection: Pick, +): string { + const raw = connection.config?.sourceTemplateKey; + const key = typeof raw === "string" ? raw : null; + return key && getConnectableAppDefinition(key) ? key : "custom"; +} + +/** + * Setup-test invocations (Apps → Test tab) are separated from agent usage on + * durable invocation columns, mirroring the gateway's own test-origin + * predicate, so the split survives reloads and approval-driven completion. + */ +export function invocationOrigin( + invocation: Pick< + ToolInvocationRow, + "actorType" | "runId" | "issueId" | "gatewayId" | "connectionId" + >, +): string { + const isTestOrigin = + invocation.actorType === "user" && + invocation.runId === null && + invocation.issueId === null && + invocation.gatewayId === null && + invocation.connectionId !== null; + return isTestOrigin ? "setup_test" : invocation.actorType; +} + +function isToolPurpose(connection: Pick): boolean { + return connection.connectionPurpose === "tool"; +} + +/** + * Emits one proposed `connector.connection_created` event for a tool-purpose + * connection row that a caller already committed. Channel and AI connections + * never emit. This function never throws; a telemetry failure must never fail + * a connection setup path. + */ +export function emitConnectorConnectionCreated( + connection: ToolConnectionRow, + setupFlow: ConnectorSetupFlow, +): void { + try { + const client = getTelemetryClient(); + if (!client) return; + if (!isToolPurpose(connection)) return; + trackConnectorConnectionCreated(client, { + connector_key: connectorKeyForConnection(connection), + transport: connection.transport, + auth_kind: connection.authKind, + setup_flow: setupFlow, + status: connection.status, + enabled: connection.enabled, + }); + } catch (err) { + logger.warn( + { err, connectionId: connection.id }, + "failed to emit connector.connection_created telemetry", + ); + } +} + +/** + * Emits one proposed `connector.connection_updated` event when a committed + * write changed a tool-purpose connection's persisted lifecycle state + * (`status` or `enabled`). Metadata-only saves, health polls, credential + * rotation, and catalog refreshes do not change either field and therefore + * never emit. This function never throws. + */ +export function emitConnectorConnectionUpdated( + connection: ToolConnectionRow, + previous: Pick, + changeSource: ConnectorChangeSource, +): void { + try { + const client = getTelemetryClient(); + if (!client) return; + if (!isToolPurpose(connection)) return; + if ( + previous.status === connection.status && + previous.enabled === connection.enabled + ) { + return; + } + trackConnectorConnectionUpdated(client, { + connector_key: connectorKeyForConnection(connection), + transport: connection.transport, + auth_kind: connection.authKind, + change_source: changeSource, + previous_status: previous.status, + status: connection.status, + previous_enabled: previous.enabled, + enabled: connection.enabled, + }); + } catch (err) { + logger.warn( + { err, connectionId: connection.id }, + "failed to emit connector.connection_updated telemetry", + ); + } +} + +/** + * 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. + * + * 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. + */ +export async function emitConnectorInvocationCompleted( + db: Db, + invocationId: string, +): Promise { + try { + const client = getTelemetryClient(); + if (!client) return; + + const invocation = await db + .select() + .from(toolInvocations) + .where(eq(toolInvocations.id, invocationId)) + .then((rows) => rows[0] ?? null); + if (!invocation) return; + 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; + + const startedAtMs = invocation.startedAt + ? new Date(invocation.startedAt).getTime() + : null; + const completedAtMs = invocation.completedAt + ? new Date(invocation.completedAt).getTime() + : null; + const durationSeconds = + startedAtMs !== null && completedAtMs !== null + ? Math.max(0, Math.round((completedAtMs - startedAtMs) / 1000)) + : undefined; + + trackConnectorInvocationCompleted(client, { + connector_key: connectorKeyForConnection(connection), + transport: connection.transport, + status: invocation.status, + origin: invocationOrigin(invocation), + ...(durationSeconds === undefined ? {} : { duration_seconds: durationSeconds }), + }); + } catch (err) { + logger.warn( + { err, invocationId }, + "failed to emit connector.invocation_completed telemetry", + ); + } +} diff --git a/server/src/services/tool-access.ts b/server/src/services/tool-access.ts index 535f63689f7..f8a00882a00 100644 --- a/server/src/services/tool-access.ts +++ b/server/src/services/tool-access.ts @@ -2,6 +2,10 @@ import { isRemoteMcpConnectorMethod, connectionPurposeTransportSchema } from "@p import { instanceSettingsService } from "./instance-settings.js"; import { githubBotRequest } from "./chat-github-client.js"; import { syncConnectionCredentialBindings } from "./connection-credential-bindings.js"; +import { + emitConnectorConnectionCreated, + emitConnectorConnectionUpdated, +} from "./connector-telemetry.js"; import { canBrowseProjectRepositoryGrant, mergeProjectRepository } from "./project-repositories.js"; import { captureRunIdentity } from "./run-identity.js"; import { createHash, randomBytes, randomUUID } from "node:crypto"; @@ -6410,6 +6414,7 @@ export function toolAccessService( return { connection: updatedConnection, applicationArchived }; }); + emitConnectorConnectionUpdated(archived.connection, connection, "archive"); // Only now, with every access path closed, revoke the credentials. Each // `secrets.remove` marks the row deleted before it calls the provider, so a @@ -8012,6 +8017,7 @@ export function toolAccessService( await ensureDefaultOrganizationGrant(updated); await syncCredentialBindings(updated); await ensureRuntimeSlot(updated); + emitConnectorConnectionUpdated(updated, existing, "example"); return { row: updated, created: false }; } const connectionId = randomUUID(); @@ -8040,6 +8046,7 @@ export function toolAccessService( await ensureDefaultOrganizationGrant(created); await syncCredentialBindings(created); await ensureRuntimeSlot(created); + emitConnectorConnectionCreated(created, "example"); return { row: created, created: true }; } @@ -10382,7 +10389,10 @@ export function toolAccessService( ), ) .returning(); - if (updated) await syncCredentialBindings(updated); + if (updated) { + await syncCredentialBindings(updated); + emitConnectorConnectionUpdated(updated, connection, "credential_refresh"); + } return updated ?? null; } @@ -11270,6 +11280,11 @@ export function toolAccessService( ) .returning(); await syncCredentialBindings(reauthorizationRequired); + emitConnectorConnectionUpdated( + reauthorizationRequired, + latestConnection, + "credential_refresh", + ); } else { const activeGrantRefs = await db .select({ @@ -12719,6 +12734,15 @@ export function toolAccessService( }) .returning(); } + if (revivedConnectionPrevious) { + emitConnectorConnectionUpdated( + connectionRow, + revivedConnectionPrevious, + "gallery_setup", + ); + } else { + emitConnectorConnectionCreated(connectionRow, "gallery"); + } if (personalIdentityUserId) { // "Just me" (PAP-17835 seam #4). The credential is committed straight to // the caller's own grant; the connection row keeps only the header @@ -13744,6 +13768,11 @@ export function toolAccessService( return { profileId, profileBindings, policies, updatedConnection }; }); + emitConnectorConnectionUpdated( + transactionResult.updatedConnection, + connection, + "gallery_setup", + ); const details = await profileDetails( transactionResult.profileId, @@ -15447,6 +15476,10 @@ export function toolAccessService( : null, }); } + const preActivationLifecycle = { + status: connection.status, + enabled: connection.enabled, + }; [connection] = await db .update(toolConnections) .set({ @@ -15463,6 +15496,11 @@ export function toolAccessService( ), ) .returning(); + emitConnectorConnectionUpdated( + connection, + preActivationLifecycle, + "oauth_callback", + ); await db .update(toolApplications) .set({ status: "active", updatedAt: now() }) @@ -15558,6 +15596,10 @@ export function toolAccessService( stateRow.connectionId, stateRow.companyId, ); + const preCallbackLifecycle = { + status: connection.status, + enabled: connection.enabled, + }; const sourceTemplateKey = typeof connection.config.sourceTemplateKey === "string" ? connection.config.sourceTemplateKey @@ -15833,6 +15875,11 @@ export function toolAccessService( tx, ); }); + emitConnectorConnectionUpdated( + connection, + preCallbackLifecycle, + "oauth_callback", + ); // Personal OAuth used to return immediately after saving the grant. That // left the connection draft/paused and its catalog empty, so the person @@ -16049,6 +16096,11 @@ export function toolAccessService( await ensureDefaultOrganizationGrant(connection, tx); await syncCredentialBindings(connection, [], tx); }); + emitConnectorConnectionUpdated( + connection, + preCallbackLifecycle, + "oauth_callback", + ); await checkConnectionHealth(connection.id, input.actor); const refresh = await refreshCatalog(connection.id, input.actor, { @@ -17337,6 +17389,7 @@ export function toolAccessService( await ensureDefaultOrganizationGrant(row); await syncCredentialBindings(row); await ensureRuntimeSlot(row); + emitConnectorConnectionCreated(row, "api"); return toConnection(row); }, @@ -18227,6 +18280,7 @@ export function toolAccessService( .returning(); await syncCredentialBindings(row); await ensureRuntimeSlot(row); + emitConnectorConnectionUpdated(row, existing, "update_api"); return toConnection(row); }, diff --git a/server/src/services/tool-action-review.ts b/server/src/services/tool-action-review.ts index 01450174b22..bafa0485055 100644 --- a/server/src/services/tool-action-review.ts +++ b/server/src/services/tool-action-review.ts @@ -9,6 +9,7 @@ import { } from "@paperclipai/db"; import { conflict, forbidden, notFound } from "../errors.js"; import { assertIssueThreadInteractionResolverAudience } from "./issue-thread-interaction-resolution.js"; +import { emitConnectorInvocationCompleted } from "./connector-telemetry.js"; import { toolAccessPolicyService } from "./tool-access-policy.js"; import { logActivity, @@ -237,6 +238,11 @@ export async function commitToolActionReview( } return updated; }); + // A rejection writes the invocation's terminal `denied` status inside the + // transaction above; emit only after that commit. Approvals stay in-flight + // and reach their terminal status in the gateway execution paths. + if (input.decision === "rejected") + void emitConnectorInvocationCompleted(db, source.invocationId); for (const publication of publications) publishActivity(publication); return result; } diff --git a/server/src/services/tool-gateway.ts b/server/src/services/tool-gateway.ts index 56ab25c088a..e851c9bc703 100644 --- a/server/src/services/tool-gateway.ts +++ b/server/src/services/tool-gateway.ts @@ -8,6 +8,7 @@ import { githubGuestBotConnectionForSession, githubBotToolsForSession } from "./ import { githubChatReviewService } from "./chat-github-reviews.js"; import { runIdentityContexts } from "@paperclipai/db"; import { captureRunIdentity } from "./run-identity.js"; +import { emitConnectorInvocationCompleted } from "./connector-telemetry.js"; import { resolveManagedGitHubIdentitySelection } from "./git-credentials.js"; import { extractRemoteMcpPending } from "./remote-mcp-pending.js"; import { logger } from "../middleware/logger.js"; @@ -2223,6 +2224,7 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, input.invocation.id)); + void emitConnectorInvocationCompleted(db, input.invocation.id); await writeToolCallEvent({ invocationId: input.invocation.id, actionRequestId: input.actionRequest?.id ?? null, @@ -2260,6 +2262,7 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, input.invocation.id)); + void emitConnectorInvocationCompleted(db, input.invocation.id); throw new ToolGatewayHttpError( 500, "Approval request was not created", @@ -2309,6 +2312,7 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, input.invocation.id)); + void emitConnectorInvocationCompleted(db, input.invocation.id); throw new ToolGatewayHttpError( 500, error.message, @@ -2475,6 +2479,7 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, input.invocation.id)); + void emitConnectorInvocationCompleted(db, input.invocation.id); throw new ToolGatewayHttpError( 409, "The approval request was resolved before it could be signed", @@ -7136,6 +7141,7 @@ 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, @@ -7206,6 +7212,7 @@ 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, @@ -7320,6 +7327,7 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, invocation.id)); + void emitConnectorInvocationCompleted(db, invocation.id); await reflectToolActionInteractionLifecycle({ actionRequestId, status: "failed", @@ -7363,6 +7371,7 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, invocation.id)); + void emitConnectorInvocationCompleted(db, invocation.id); await reflectToolActionInteractionLifecycle({ actionRequestId, status: "failed", @@ -7528,6 +7537,7 @@ export function createToolGatewayService( return true; }); if (!settled) return { reasonCode, message, settled: false }; + void emitConnectorInvocationCompleted(db, input.invocationId); await reflectToolActionInteractionLifecycle({ actionRequestId: input.actionRequestId, status: "failed", @@ -7580,6 +7590,7 @@ export function createToolGatewayService( updatedAt: now, }) .where(eq(toolInvocations.id, input.invocationId)); + void emitConnectorInvocationCompleted(db, input.invocationId); await reflectToolActionInteractionLifecycle({ actionRequestId: expired.id, status: "expired", @@ -7976,6 +7987,7 @@ export function createToolGatewayService( updatedAt: now, }) .where(eq(toolInvocations.id, invocation.id)); + void emitConnectorInvocationCompleted(db, invocation.id); await db .update(toolActionRequests) .set({ status: "executed", resolvedAt: now, updatedAt: now }) @@ -9045,6 +9057,7 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, invocationId)); + void emitConnectorInvocationCompleted(db, invocationId); throw new ToolGatewayHttpError( 500, error.message, @@ -9316,6 +9329,7 @@ export function createToolGatewayService( updatedAt: now, }) .where(eq(toolInvocations.id, row.invocationId)); + void emitConnectorInvocationCompleted(db, row.invocationId); await reflectToolActionInteractionLifecycle({ actionRequestId: row.id, status, @@ -10415,6 +10429,7 @@ export function createToolGatewayService( updatedAt: completedAt, }) .where(eq(toolInvocations.id, invocationId)); + void emitConnectorInvocationCompleted(db, invocationId); if (input.approvedActionRequestId) { const [executedRequest] = await db .update(toolActionRequests) @@ -10541,6 +10556,7 @@ export function createToolGatewayService( updatedAt: completedAt, }) .where(eq(toolInvocations.id, invocationId)); + void emitConnectorInvocationCompleted(db, invocationId); if (input.approvedActionRequestId) { const [failedRequest] = await db .update(toolActionRequests) @@ -10804,6 +10820,7 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, invocationId)); + void emitConnectorInvocationCompleted(db, invocationId); await writeToolCallEvent({ invocationId, session: sessionLike, @@ -10856,6 +10873,7 @@ export function createToolGatewayService( updatedAt: new Date(), }) .where(eq(toolInvocations.id, invocationId)); + void emitConnectorInvocationCompleted(db, invocationId); await writeToolCallEvent({ invocationId, session: sessionLike,