diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 80d43043df6a..dbca27b8eb9f 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -13,7 +13,12 @@ import * as Stream from "effect/Stream"; import type * as Types from "effect/Types"; import { AiError, McpProtocol, McpSchema, McpServer, Tool, Toolkit } from "effect/ai"; import { HttpRouter, HttpServerRequest, HttpServerResponse } from "effect/http"; -import { OrchestratorMcpFailure, PreviewAutomationError } from "@t3tools/contracts"; +import { + OrchestratorMcpFailure, + PreviewAutomationError, + type RunId, + ThreadId, +} from "@t3tools/contracts"; import packageJson from "../../package.json" with { type: "json" }; import * as ServerConfig from "../config.ts"; @@ -21,6 +26,15 @@ import * as DeviceService from "../device/DeviceService.ts"; import * as HtmlRender from "../htmlRender/HtmlRender.ts"; import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.ts"; import * as ThreadManagementService from "../orchestration-v2/ThreadManagementService.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as AnalyticsService from "../telemetry/AnalyticsService.ts"; +import { + type AgentThread, + agentToolProperties, + handoffSettings, + MAX_REPORTED_DEPTH, + toolOutcome, +} from "../telemetry/ProviderDimensions.ts"; import * as McpInvocationContext from "./McpInvocationContext.ts"; import * as McpToolAccess from "./McpToolAccess.ts"; import * as OrchestratorMcpService from "./OrchestratorMcpService.ts"; @@ -746,6 +760,189 @@ const registerHtmlPreview = Effect.fn("McpHttpServer.registerHtmlPreview")(funct ); }); +/** + * Tools that hand work to another thread's agent: they create it, message it, + * steer it, or move context into it. Their events name the receiving agent too. + */ +const HANDOFF_TOOLS: ReadonlySet = new Set([ + "delegate_task", + "create_threads", + "t3_thread_launch", + "t3_thread_send", + "t3_thread_send_attachments", + "t3_thread_fork", + "t3_thread_merge_back", + "t3_queue_edit", + "t3_queue_promote_to_steer", + "t3_pending_request_respond", + "schedule_task", + "run_scheduled_task_now", +]); + +/** Handoff tools whose result is only `{ sequence }`; their input names the receiving thread. */ +const INPUT_TARGET_TOOLS: ReadonlySet = new Set([ + "t3_queue_edit", + "t3_queue_promote_to_steer", + "t3_pending_request_respond", +]); + +const resultThreadKeys = ["childThreadId", "targetThreadId", "boundThreadId", "threadId"] as const; + +/** Thread ids a handoff tool result names: the thread that received the work. */ +const resultThreadIds = (content: unknown): ReadonlyArray => { + if (typeof content !== "object" || content === null) return []; + const record = content as Readonly>; + if (Array.isArray(record.threads)) return record.threads.flatMap(resultThreadIds); + const key = resultThreadKeys.find((candidate) => typeof record[candidate] === "string"); + return key === undefined ? [] : [record[key] as string]; +}; + +/** + * Thread ids that received a handoff. The input's `threadId` is used only after + * the call succeeded, so it is one the handler already resolved and authorized. + */ +const handoffTargetIds = ( + tool: string, + args: unknown, + result: McpSchema.CallToolResult | undefined, +): ReadonlyArray => { + if (!INPUT_TARGET_TOOLS.has(tool)) return resultThreadIds(result?.structuredContent); + if (result === undefined || toolOutcome(result).outcome !== "ok") return []; + const threadId = (args as { readonly threadId?: unknown } | null | undefined)?.threadId; + return typeof threadId === "string" ? [threadId] : []; +}; + +/** + * Runs a tool registration against a server that records one anonymous + * `mcp.tool.invoked` event per call: the tool, the calling agent's provider, + * the outcome with its failure code, and the duration. A handoff tool also + * reports the caller's model, origin, and delegation depth, the settings the + * agent chose, and the provider, model, and modes of each thread that received + * the work. Ids, prompts, and free text are never recorded. + */ +const withToolAnalytics = (registration: Effect.Effect) => + Effect.gen(function* () { + const server = yield* McpServer.McpServer; + const analytics = yield* Effect.serviceOption(AnalyticsService.AnalyticsService); + if (Option.isNone(analytics)) return yield* registration; + const registry = yield* Effect.serviceOption(ProviderRegistry.ProviderRegistry); + const threads = yield* Effect.serviceOption(ThreadManagementService.ThreadManagementService); + const shellOf = (threadId: string) => + Option.isNone(threads) + ? Effect.succeed(null) + : threads.value + .getThreadShell(ThreadId.make(threadId)) + .pipe(Effect.orElseSucceed(() => null)); + // Delegation depth: parents walked from the caller, capped. + const depthOf = (thread: AgentThread) => + Effect.gen(function* () { + let depth = 0; + let parentId = + thread.lineage.relationshipToParent === "subagent" ? thread.lineage.parentThreadId : null; + while (parentId !== null && depth < MAX_REPORTED_DEPTH) { + depth += 1; + const parent = yield* shellOf(parentId); + parentId = + parent?.lineage.relationshipToParent === "subagent" + ? parent.lineage.parentThreadId + : null; + } + return depth; + }); + // Whether the caller's active run was started by a scheduled task. + const scheduledRunOf = (threadId: string, runId: RunId | null) => + Option.isNone(threads) || runId === null + ? Effect.succeed(false) + : threads.value + .getThreadRecords(ThreadId.make(threadId), ["runs", "messages"], { + runIds: [runId], + messageRoles: ["user"], + }) + .pipe( + Effect.map((records) => { + const run = records.runs.find((candidate) => candidate.id === runId); + return records.messages.some( + (message) => + message.id === run?.userMessageId && message.scheduledTaskId !== undefined, + ); + }), + Effect.orElseSucceed(() => false), + ); + const record = ( + tool: string, + args: unknown, + result: McpSchema.CallToolResult | undefined, + durationMs: number, + ) => + Effect.gen(function* () { + // Client credentials (no calling thread) report an unknown caller. + const invocation = Option.getOrUndefined( + yield* Effect.serviceOption(McpInvocationContext.McpInvocationContext), + )?.thread; + const providers = Option.isSome(registry) ? yield* registry.value.getProviders : []; + // Only handoff tools pay for thread lookups; the rest report the + // caller's provider from the credential. + const handoff = HANDOFF_TOOLS.has(tool); + const caller = + handoff && invocation !== undefined + ? ((yield* shellOf(invocation.threadId)) ?? undefined) + : undefined; + const targetIds = handoff + ? [...new Set(handoffTargetIds(tool, args, result))].filter( + (threadId) => threadId !== invocation?.threadId, + ) + : []; + const targets = yield* Effect.forEach(targetIds, shellOf); + for (const properties of agentToolProperties({ + tool, + providers, + caller, + ...(invocation === undefined + ? {} + : { callerProviderInstanceId: invocation.providerInstanceId }), + ...(caller === undefined || invocation === undefined + ? {} + : { + callerDepth: yield* depthOf(caller), + callerScheduledRun: yield* scheduledRunOf(invocation.threadId, caller.activeRunId), + }), + targets: targets.filter((thread) => thread !== null), + outcome: toolOutcome(result), + ...(handoff ? { settings: handoffSettings(tool, args, result?.structuredContent) } : {}), + durationMs, + })) { + yield* analytics.value.record("mcp.tool.invoked", properties); + } + }).pipe(Effect.ignoreCause); + const recordingServer = McpServer.McpServer.of({ + ...server, + addTool: (options) => + server.addTool({ + ...options, + handle: (payload) => + Effect.gen(function* () { + const startedAt = yield* Clock.currentTimeMillis; + return yield* options.handle(payload).pipe( + Effect.onExit((exit) => { + const result = exit._tag === "Success" ? exit.value : undefined; + // A call that asks for input is retried with it, and that call is the one counted. + if (result instanceof McpSchema.InputRequired) return Effect.void; + // Handoff tools look up threads to record; the response must not wait on them. + return Clock.currentTimeMillis.pipe( + Effect.flatMap((endedAt) => + record(options.tool.name, payload, result, endedAt - startedAt), + ), + Effect.forkDetach, + Effect.asVoid, + ); + }), + ); + }), + }), + }); + return yield* registration.pipe(Effect.provideService(McpServer.McpServer, recordingServer)); + }); + /** * `McpServer.toolkit` asks for every service its tools declare when it * registers them, but the auth middleware provides `McpInvocationContext` to @@ -756,7 +953,9 @@ export const toolkitRegistration = , EX, toolkit: Toolkit.Toolkit, handlers: McpToolAccess.HandlersLayer, ) => { - const registration = McpServer.toolkit(decodeAsAdvertised(toolkit)); + const registration = Layer.effectDiscard( + withToolAnalytics(McpServer.registerToolkit(decodeAsAdvertised(toolkit))), + ).pipe(Layer.provide(McpServer.McpServer.layer)); // @effect-diagnostics-next-line unsafeEffectTypeAssertion:off - the auth middleware provides it per request. const registered = registration as Layer.Layer< never, @@ -769,7 +968,10 @@ export const toolkitRegistration = , EX, const imageToolRegistration = , A, E, R, EX, RX>( register: Effect.Effect, handlers: McpToolAccess.HandlersLayer, -) => Layer.effectDiscard(register).pipe(Layer.provide(McpToolAccess.HandlersLayer.layer(handlers))); +) => + Layer.effectDiscard(withToolAnalytics(register)).pipe( + Layer.provide(McpToolAccess.HandlersLayer.layer(handlers)), + ); export const layerHtmlToolkit = Layer.mergeAll( toolkitRegistration(HtmlRenderToolkit, HtmlHandlers.layerRender), diff --git a/apps/server/src/mcp/toolkits/core.test.ts b/apps/server/src/mcp/toolkits/core.test.ts index 45ea89082f42..aaa5b176db46 100644 --- a/apps/server/src/mcp/toolkits/core.test.ts +++ b/apps/server/src/mcp/toolkits/core.test.ts @@ -18,6 +18,7 @@ import { type OrchestrationV2ProjectedTurnItem, type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as TestClock from "effect/testing/TestClock"; import * as Layer from "effect/Layer"; @@ -49,6 +50,7 @@ import * as McpHttpServer from "../McpHttpServer.ts"; import * as McpInvocationContext from "../McpInvocationContext.ts"; import * as McpToolAccessTestkit from "../McpToolAccess.testkit.ts"; import { dispatchFailure } from "../threadAccess.ts"; +import * as AnalyticsService from "../../telemetry/AnalyticsService.ts"; import { OrchestratorToolkit } from "./orchestrator/tools.ts"; import { PreviewToolkit } from "./preview/tools.ts"; import { PreviewControlsToolkit } from "./previewControls/tools.ts"; @@ -362,6 +364,131 @@ it.effect("keeps unexpected handler defects private through the production regis ), ); +it.effect("records the calling agent and the agent a tool acted on", () => { + // Recording runs detached from the tool response, so each expected event is awaited. + const recorded = [ + Deferred.makeUnsafe<{ event: string; properties: object }>(), + Deferred.makeUnsafe<{ event: string; properties: object }>(), + ]; + let recordedCount = 0; + const shell = (id: ThreadId, instanceId: string, model: string) => + ({ + id, + projectId: "mcp-core-project", + providerInstanceId: ProviderInstanceId.make(instanceId), + modelSelection: { instanceId: ProviderInstanceId.make(instanceId), model }, + runtimeMode: "full-access", + interactionMode: "default", + archivedAt: null, + deletedAt: null, + activeRunId: "mcp-core-run", + createdBy: "agent", + creationSource: "mcp", + lineage: { + parentThreadId: id === threadId ? ThreadId.make("mcp-core-parent") : null, + relationshipToParent: id === threadId ? "subagent" : null, + rootThreadId: ThreadId.make("mcp-core-parent"), + }, + }) as never; + return Effect.gen(function* () { + const server = yield* McpServer.McpServer; + yield* server + .callTool({ name: "t3_thread_fork", arguments: { sourcePoint: { type: "latest_stable" } } }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, client), + ); + const { event, properties } = yield* Deferred.await(recorded[0]!); + expect(event).toBe("mcp.tool.invoked"); + expect(properties).toMatchObject({ + tool: "t3_thread_fork", + outcome: "ok", + callerProvider: "codex", + callerModel: "gpt-5.5", + callerOrigin: "agent", + callerDepth: 1, + targetProvider: "claudeAgent", + targetModel: "claude-opus-5-5", + targetRuntimeMode: "full-access", + targetInteractionMode: "default", + crossProvider: true, + }); + expect(properties).toHaveProperty("durationMs"); + expect(Object.values(properties).some((value) => String(value).includes("mcp-core"))).toBe( + false, + ); + + // Queue tools return only a sequence, so the receiving thread comes from the input. + yield* server + .callTool({ + name: "t3_queue_edit", + arguments: { threadId: "mcp-core-other", queuedRunId: "mcp-core-queued", text: "secret" }, + }) + .pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, scope), + Effect.provideService(McpSchema.McpServerClient, client), + ); + // The fork recorded a single event, so the next one is the queue edit. + expect((yield* Deferred.await(recorded[1]!)).properties).toMatchObject({ + tool: "t3_queue_edit", + outcome: "ok", + callerProvider: "codex", + targetProvider: "claudeAgent", + crossProvider: true, + }); + }).pipe( + Effect.provide( + layerThreadToolkit.pipe( + Layer.provideMerge(McpServer.McpServer.layer), + Layer.provide(NodeCrypto.layer), + Layer.provide( + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadShell: (id) => + Effect.succeed( + id === threadId + ? shell(threadId, "codex", "gpt-5.5") + : shell(id, "claudeAgent", "claude-opus-5-5"), + ), + getThreadRecords: () => Effect.succeed({ runs: [], messages: [] } as never), + getProjectThreadRecords: () => + Effect.succeed({ thread: shell(threadId, "codex", "gpt-5.5") } as never), + dispatch: () => Effect.succeed({ sequence: 1 } as never), + }), + ), + Layer.provide( + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([ + { + instanceId: ProviderInstanceId.make("codex"), + driver: "codex", + models: [{ slug: "gpt-5.5", isCustom: false }], + } as never, + { + instanceId: ProviderInstanceId.make("claudeAgent"), + driver: "claudeAgent", + models: [{ slug: "claude-opus-5-5", isCustom: false }], + } as never, + ]), + }), + ), + Layer.provide( + Layer.succeed( + AnalyticsService.AnalyticsService, + AnalyticsService.AnalyticsService.of({ + record: (event, properties) => + Deferred.succeed(recorded[recordedCount++]!, { + event, + properties: properties ?? {}, + }).pipe(Effect.asVoid), + flush: Effect.void, + }), + ), + ), + ), + ), + ); +}); + it("keeps MCP preference output allowlisted and Unicode-bounded", () => { const settings = { ...DEFAULT_SERVER_SETTINGS, diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 2d35647f0948..59daedc15107 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -45,6 +45,7 @@ import * as PullRequestReviewGuides from "./persistence/PullRequestReviewGuides. import * as PullRequestReviewGuideService from "./pullRequest/PullRequestReviewGuideService.ts"; import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; +import * as DelegatedTaskAnalytics from "./telemetry/DelegatedTaskAnalytics.ts"; import * as ProviderEventIngestor from "./orchestration-v2/ProviderEventIngestor.ts"; import * as ModelManifest from "./provider/ModelManifest.ts"; import * as ResetCreditCoordinator from "./provider/resetCreditCoordinator.ts"; @@ -554,6 +555,7 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( ), ), layerThreadPullRequestWorker, + DelegatedTaskAnalytics.layer, Layer.effectDiscard( Effect.gen(function* () { const service = yield* PullRequestSyncReactor.PullRequestSyncReactor; diff --git a/apps/server/src/telemetry/DelegatedTaskAnalytics.test.ts b/apps/server/src/telemetry/DelegatedTaskAnalytics.test.ts new file mode 100644 index 000000000000..aea7aec786cd --- /dev/null +++ b/apps/server/src/telemetry/DelegatedTaskAnalytics.test.ts @@ -0,0 +1,102 @@ +import { ProviderInstanceId, ThreadId } from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import * as AnalyticsService from "./AnalyticsService.ts"; +import * as DelegatedTaskAnalytics from "./DelegatedTaskAnalytics.ts"; + +const parentThreadId = ThreadId.make("parent-private"); + +const subagentEvent = (status: string, origin = "app_owned") => + ({ + type: "subagent.updated", + payload: { + id: "task-private", + threadId: parentThreadId, + origin, + status, + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + model: "claude-opus-5-5", + prompt: "secret task", + completionWake: "always", + startedAt: DateTime.makeUnsafe(0), + completedAt: DateTime.makeUnsafe(90_000), + }, + }) as never; + +it.effect("records each delegated task's final outcome once, without ids or prompts", () => { + const recorded: Array<{ event: string; properties: unknown }> = []; + let shellLookups = 0; + return Effect.gen(function* () { + const analytics = yield* DelegatedTaskAnalytics.DelegatedTaskAnalytics; + yield* analytics.onSubagent(subagentEvent("running")); + yield* analytics.onSubagent(subagentEvent("completed", "provider_native")); + // The first parent lookup fails; the next final update still reports the task. + yield* analytics.onSubagent(subagentEvent("completed")); + yield* analytics.onSubagent(subagentEvent("completed")); + yield* analytics.onSubagent(subagentEvent("completed")); + expect(recorded).toEqual([ + { + event: "mcp.delegated_task.finished", + properties: { + status: "completed", + callerProvider: "codex", + callerModel: "gpt-5.5", + targetProvider: "claudeAgent", + targetModel: "claude-opus-5-5", + crossProvider: true, + completionWake: "always", + durationSeconds: 90, + }, + }, + ]); + }).pipe( + Effect.provide( + DelegatedTaskAnalytics.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(Orchestrator.OrchestratorV2)({ + streamDomainEvents: Stream.empty, + getThreadShell: () => + ++shellLookups === 1 + ? Effect.die("thread store unavailable") + : Effect.succeed({ + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.5", + }, + } as never), + }), + Layer.mock(ProviderRegistry.ProviderRegistry)({ + getProviders: Effect.succeed([ + { + instanceId: ProviderInstanceId.make("codex"), + driver: "codex", + models: [{ slug: "gpt-5.5", isCustom: false }], + } as never, + { + instanceId: ProviderInstanceId.make("claudeAgent"), + driver: "claudeAgent", + models: [{ slug: "claude-opus-5-5", isCustom: false }], + } as never, + ]), + }), + Layer.succeed( + AnalyticsService.AnalyticsService, + AnalyticsService.AnalyticsService.of({ + record: (event, properties) => + Effect.sync(() => void recorded.push({ event, properties })), + flush: Effect.void, + }), + ), + ), + ), + ), + ), + ); +}); diff --git a/apps/server/src/telemetry/DelegatedTaskAnalytics.ts b/apps/server/src/telemetry/DelegatedTaskAnalytics.ts new file mode 100644 index 000000000000..e3f50aef6a8d --- /dev/null +++ b/apps/server/src/telemetry/DelegatedTaskAnalytics.ts @@ -0,0 +1,93 @@ +/** + * Records one anonymous `mcp.delegated_task.finished` event when an app-owned + * delegated task reaches a final status, so we can see whether delegation + * between providers actually completes. The event carries the parent and child + * provider and model, the final status, and the child's runtime in seconds. + * + * @module DelegatedTaskAnalytics + */ +import type { OrchestrationV2DomainEvent } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; + +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ProviderRegistry from "../provider/ProviderRegistry.ts"; +import { forkParked } from "../serverActivation.ts"; +import * as AnalyticsService from "./AnalyticsService.ts"; +import { crossProvider, providerDimensions } from "./ProviderDimensions.ts"; + +type SubagentEvent = Extract; + +const FINAL_STATUSES: ReadonlySet = new Set([ + "completed", + "failed", + "cancelled", + "interrupted", +]); + +export class DelegatedTaskAnalytics extends Context.Service< + DelegatedTaskAnalytics, + { readonly onSubagent: (event: SubagentEvent) => Effect.Effect } +>()("t3/telemetry/DelegatedTaskAnalytics") {} + +const make = Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const analytics = yield* AnalyticsService.AnalyticsService; + const registry = yield* ProviderRegistry.ProviderRegistry; + // A task can be re-reported with the same final status; count it once per process. + const reported = new Set(); + + const onSubagent = (event: SubagentEvent) => + Effect.gen(function* () { + const task = event.payload; + if (task.origin !== "app_owned" || !FINAL_STATUSES.has(task.status)) return; + if (reported.has(task.id)) return; + const parent = yield* orchestrator.getThreadShell(task.threadId); + const providers = yield* registry.getProviders; + const parentDimensions = + parent === null + ? { provider: "unknown" } + : providerDimensions(providers, parent.modelSelection); + const child = providerDimensions(providers, { + instanceId: task.providerInstanceId, + model: task.model ?? "", + }); + const cross = crossProvider(parentDimensions.provider, child.provider); + const durationSeconds = + task.startedAt !== null && task.completedAt !== null + ? Math.max( + 0, + Math.round( + (DateTime.toEpochMillis(task.completedAt) - + DateTime.toEpochMillis(task.startedAt)) / + 1000, + ), + ) + : undefined; + yield* analytics.record("mcp.delegated_task.finished", { + status: task.status, + callerProvider: parentDimensions.provider, + ...(parentDimensions.model === undefined ? {} : { callerModel: parentDimensions.model }), + targetProvider: child.provider, + ...(child.model === undefined ? {} : { targetModel: child.model }), + ...(cross === undefined ? {} : { crossProvider: cross }), + ...(task.completionWake === undefined ? {} : { completionWake: task.completionWake }), + ...(durationSeconds === undefined ? {} : { durationSeconds }), + }); + // Only after recording: a failed lookup leaves the next final update to report it. + reported.add(task.id); + }).pipe(Effect.ignoreCause); + + yield* forkParked( + Stream.runForEach(orchestrator.streamDomainEvents, (event) => + event.type === "subagent.updated" ? onSubagent(event) : Effect.void, + ), + ); + + return DelegatedTaskAnalytics.of({ onSubagent }); +}); + +export const layer = Layer.effect(DelegatedTaskAnalytics, make); diff --git a/apps/server/src/telemetry/ProviderDimensions.test.ts b/apps/server/src/telemetry/ProviderDimensions.test.ts new file mode 100644 index 000000000000..67b33fe6f962 --- /dev/null +++ b/apps/server/src/telemetry/ProviderDimensions.test.ts @@ -0,0 +1,226 @@ +import { ProviderInstanceId, ThreadId, type ServerProvider } from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; + +import { + type AgentThread, + agentToolProperties, + handoffSettings, + callerOrigin, + toolOutcome, +} from "./ProviderDimensions.ts"; + +const provider = ( + instanceId: string, + driver: string, + models: ReadonlyArray<{ slug: string; isCustom: boolean }>, +) => + ({ + instanceId: ProviderInstanceId.make(instanceId), + driver, + models, + }) as unknown as ServerProvider; + +const providers = [ + provider("work-codex", "codex", [{ slug: "gpt-5.5", isCustom: false }]), + provider("claudeAgent", "claudeAgent", [ + { slug: "claude-opus-5-5", isCustom: false }, + { slug: "my-private-finetune", isCustom: true }, + ]), + provider("opencode", "opencode", [{ slug: "ollama/acme-internal", isCustom: false }]), +]; + +const thread = ( + instanceId: string, + model: string, + overrides: Partial = {}, +): AgentThread => ({ + modelSelection: { instanceId: ProviderInstanceId.make(instanceId), model }, + runtimeMode: "full-access", + interactionMode: "default", + createdBy: "user", + creationSource: "web", + lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: ThreadId.make("t") }, + ...overrides, +}); + +it("relates the calling agent to each agent that received work", () => { + expect( + agentToolProperties({ + tool: "create_threads", + providers, + caller: thread("work-codex", "gpt-5.5", { createdBy: "agent", creationSource: "mcp" }), + callerDepth: 1, + targets: [ + thread("claudeAgent", "claude-opus-5-5", { interactionMode: "plan" }), + thread("work-codex", "gpt-5.5"), + ], + outcome: { outcome: "ok" }, + settings: { batchSize: 2, targetChosen: true }, + durationMs: 42, + }), + ).toEqual([ + { + tool: "create_threads", + callerProvider: "codex", + callerModel: "gpt-5.5", + callerOrigin: "agent", + callerDepth: 1, + durationMs: 42, + outcome: "ok", + batchSize: 2, + targetChosen: true, + targetProvider: "claudeAgent", + targetModel: "claude-opus-5-5", + targetRuntimeMode: "full-access", + targetInteractionMode: "plan", + crossProvider: true, + }, + { + tool: "create_threads", + callerProvider: "codex", + callerModel: "gpt-5.5", + callerOrigin: "agent", + callerDepth: 1, + durationMs: 42, + outcome: "ok", + batchSize: 2, + targetChosen: true, + targetProvider: "codex", + targetModel: "gpt-5.5", + targetRuntimeMode: "full-access", + targetInteractionMode: "default", + crossProvider: false, + }, + ]); +}); + +it("reports a non-handoff tool from the credential's provider alone", () => { + expect( + agentToolProperties({ + tool: "preview_click", + providers, + caller: undefined, + callerProviderInstanceId: ProviderInstanceId.make("claudeAgent"), + targets: [], + outcome: { outcome: "error", errorCode: "capability_denied" }, + durationMs: 3, + }), + ).toEqual([ + { + tool: "preview_click", + callerProvider: "claudeAgent", + durationMs: 3, + outcome: "error", + errorCode: "capability_denied", + }, + ]); +}); + +it("omits custom, user-configured, and unknown models and instance names", () => { + expect( + agentToolProperties({ + tool: "t3_thread_send", + providers, + caller: thread("claudeAgent", "my-private-finetune"), + targets: [thread("opencode", "ollama/acme-internal"), thread("removed-instance", "gpt-5.5")], + outcome: { outcome: "ok" }, + }).map(({ callerProvider, callerModel, targetProvider, targetModel }) => ({ + callerProvider, + callerModel, + targetProvider, + targetModel, + })), + ).toEqual([ + { + callerProvider: "claudeAgent", + callerModel: undefined, + targetProvider: "opencode", + targetModel: undefined, + }, + { + callerProvider: "claudeAgent", + callerModel: undefined, + targetProvider: "unknown", + targetModel: undefined, + }, + ]); +}); + +it("reads handoff settings from enums and flags, never from prompts or ids", () => { + expect( + handoffSettings( + "delegate_task", + { task: "secret", mode: "wait", target: { driverKind: "claudeAgent" } }, + { taskId: "t", waitTimedOut: true }, + ), + ).toEqual({ targetChosen: true, mode: "wait", waitTimedOut: true }); + expect(handoffSettings("delegate_task", { task: "secret" }, { waitTimedOut: false })).toEqual({ + targetChosen: false, + mode: "async", + }); + expect( + handoffSettings( + "t3_thread_launch", + { message: "secret", workspaceStrategy: { type: "worktree", branch: "private" } }, + {}, + ), + ).toEqual({ targetChosen: false, workspace: "worktree" }); + expect(handoffSettings("t3_thread_launch", { scratch: true }, {})).toMatchObject({ + workspace: "scratch", + }); + expect( + handoffSettings( + "create_threads", + { threads: [{ prompt: "a" }, { prompt: "b", target: {} }] }, + {}, + ), + ).toEqual({ batchSize: 2, targetChosen: true }); + expect(handoffSettings("t3_thread_send", { message: "secret" }, { delivery: "steered" })).toEqual( + { delivery: "steered" }, + ); +}); + +it("drops input strings outside the schema's literals, since rejected calls are recorded too", () => { + expect(handoffSettings("delegate_task", { mode: "/home/me/secret" }, undefined)).toEqual({ + targetChosen: false, + mode: "async", + }); + expect( + handoffSettings("t3_thread_launch", { workspaceStrategy: { type: "/home/me/secret" } }, {}), + ).toEqual({ targetChosen: false, workspace: "root" }); +}); + +it("reports crossProvider only when both providers resolve", () => { + const [event] = agentToolProperties({ + tool: "t3_thread_send", + providers, + caller: thread("work-codex", "gpt-5.5"), + targets: [thread("removed-instance", "gpt-5.5")], + outcome: { outcome: "ok" }, + }); + expect(event).toMatchObject({ callerProvider: "codex", targetProvider: "unknown" }); + expect(event).not.toHaveProperty("crossProvider"); +}); + +it("reads failure codes from declared tool errors even without isError", () => { + expect( + toolOutcome({ + structuredContent: { + _tag: "OrchestratorMcpFailure", + code: "capability_denied", + message: "private detail", + }, + }), + ).toEqual({ outcome: "error", errorCode: "capability_denied" }); + expect(toolOutcome({ isError: false, structuredContent: { threadId: "t" } })).toEqual({ + outcome: "ok", + }); + expect(toolOutcome(undefined)).toEqual({ outcome: "error", errorCode: "exception" }); +}); + +it("tells user, agent, system, and scheduled work apart", () => { + expect(callerOrigin({ thread: { createdBy: "user" }, scheduledRun: false })).toBe("user"); + expect(callerOrigin({ thread: { createdBy: "agent" }, scheduledRun: false })).toBe("agent"); + expect(callerOrigin({ thread: { createdBy: "system" }, scheduledRun: false })).toBe("system"); + expect(callerOrigin({ thread: { createdBy: "user" }, scheduledRun: true })).toBe("scheduler"); +}); diff --git a/apps/server/src/telemetry/ProviderDimensions.ts b/apps/server/src/telemetry/ProviderDimensions.ts new file mode 100644 index 000000000000..4a668ee5af78 --- /dev/null +++ b/apps/server/src/telemetry/ProviderDimensions.ts @@ -0,0 +1,210 @@ +import type { + ModelSelection, + OrchestrationV2ThreadShell, + ServerProvider, +} from "@t3tools/contracts"; + +/** + * Drivers whose model catalogs are published by the vendor. Other drivers + * (OpenCode, Pi, ACP agents) list models from the user's own configuration, + * such as local Ollama tags, so their model names stay out of analytics. + */ +const VENDOR_CATALOG_DRIVERS: ReadonlySet = new Set([ + "codex", + "claudeAgent", + "cursor", + "grok", + "antigravity", +]); + +export interface ProviderDimensions { + readonly provider: string; + readonly model?: string; +} + +/** + * Anonymous provider and model for a model selection. Instance ids are + * user-defined, so only the driver kind is reported. The model is reported + * only when it comes from a vendor catalog and is not a user-added custom model. + */ +export function providerDimensions( + providers: ReadonlyArray, + selection: Pick, +): ProviderDimensions { + const provider = providers.find((candidate) => candidate.instanceId === selection.instanceId); + if (provider === undefined) return { provider: "unknown" }; + const model = provider.models.find((candidate) => candidate.slug === selection.model); + return VENDOR_CATALOG_DRIVERS.has(provider.driver) && model !== undefined && !model.isCustom + ? { provider: provider.driver, model: model.slug } + : { provider: provider.driver }; +} + +/** The thread fields agent analytics reads. Ids are used for lookups, never reported. */ +export type AgentThread = Pick< + OrchestrationV2ThreadShell, + "modelSelection" | "runtimeMode" | "interactionMode" | "createdBy" | "creationSource" | "lineage" +>; + +type Fields = Readonly>; + +const asFields = (value: unknown): Fields => + typeof value === "object" && value !== null ? (value as Fields) : {}; + +const stringField = (fields: Fields, key: string) => + typeof fields[key] === "string" ? (fields[key] as string) : undefined; + +/** A string from the tool input only when it is one of the schema's literals. */ +const oneOf = (value: string | undefined, allowed: ReadonlyArray) => + value !== undefined && allowed.includes(value) ? value : undefined; + +/** + * Whether two resolved providers differ. Unresolved selections compare as + * unknown rather than as a cross-provider handoff. + */ +export const crossProvider = (caller: string, target: string) => + caller === "unknown" || target === "unknown" ? undefined : caller !== target; + +const defined = (entries: Readonly>) => + Object.fromEntries(Object.entries(entries).filter(([, value]) => value !== undefined)); + +/** + * How deep a thread sits in a delegation tree: 0 for a top-level thread, 1 for + * its subagent, and so on. Forks are top-level threads of their own. Depth is + * counted by walking parents, capped so a corrupt lineage cannot loop. + */ +export const MAX_REPORTED_DEPTH = 5; + +/** + * Who started the work that made this call. Scheduled runs keep the task + * creator's provenance, so the run's message marks them instead. + */ +export function callerOrigin(input: { + readonly thread: Pick; + readonly scheduledRun: boolean; +}) { + if (input.scheduledRun) return "scheduler"; + if (input.thread.createdBy === "agent") return "agent"; + return input.thread.createdBy === "system" ? "system" : "user"; +} + +/** + * Settings the agent chose for a handoff, read only from closed enums and + * booleans in the tool's arguments and result. Prompts, titles, and ids are + * never read. + */ +export function handoffSettings(tool: string, args: unknown, result: unknown): Fields { + const input = asFields(args); + const output = asFields(result); + switch (tool) { + case "delegate_task": + return defined({ + targetChosen: input.target !== undefined, + // Rejected calls are recorded too, so input strings must match a literal. + mode: oneOf(stringField(input, "mode"), ["async", "wait"]) ?? "async", + ...(output.waitTimedOut === true ? { waitTimedOut: true } : {}), + }); + case "create_threads": { + const threads = Array.isArray(input.threads) ? input.threads : []; + return { + batchSize: threads.length, + targetChosen: threads.some((thread) => asFields(thread).target !== undefined), + }; + } + case "t3_thread_launch": + return defined({ + targetChosen: input.modelSelection !== undefined, + workspace: + input.scratch === true + ? "scratch" + : (oneOf(stringField(asFields(input.workspaceStrategy), "type"), [ + "root", + "worktree", + "existing_worktree", + ]) ?? "root"), + }); + case "t3_thread_send": + return defined({ delivery: stringField(output, "delivery") }); + case "t3_thread_send_attachments": + return { delivery: "auto" }; + case "schedule_task": + return { bindToCurrentThread: input.bindToCurrentThread !== false }; + default: + return {}; + } +} + +/** Outcome of a tool call, with the stable failure code when the call failed. */ +export function toolOutcome( + result: + | { readonly isError?: boolean | undefined; readonly structuredContent?: unknown } + | undefined, +) { + if (result === undefined) return { outcome: "error", errorCode: "exception" }; + const content = asFields(result.structuredContent); + // T3 tool failures are declared errors whose structured content carries a + // closed `code`; the MCP layer does not always set isError for them. + const code = stringField(content, "code") ?? stringField(asFields(content.error), "_tag"); + if (result.isError === true || (content._tag !== undefined && code !== undefined)) { + return { outcome: "error", errorCode: code ?? "unknown" }; + } + return { outcome: "ok" }; +} + +/** + * Event properties for an agent tool call: the calling agent, and when the call + * handed work to other threads, the agents running them. Each target counts + * once, so a create_threads batch on two providers reports both. + */ +export function agentToolProperties(input: { + readonly tool: string; + readonly providers: ReadonlyArray; + readonly caller: AgentThread | undefined; + readonly callerProviderInstanceId?: ModelSelection["instanceId"]; + readonly callerDepth?: number; + readonly callerScheduledRun?: boolean; + readonly targets: ReadonlyArray; + readonly outcome: Fields; + readonly settings?: Fields; + readonly durationMs?: number; +}): ReadonlyArray>> { + const caller = + input.caller !== undefined + ? providerDimensions(input.providers, input.caller.modelSelection) + : input.callerProviderInstanceId !== undefined + ? { + provider: providerDimensions(input.providers, { + instanceId: input.callerProviderInstanceId, + model: "", + }).provider, + } + : { provider: "unknown" }; + const base = defined({ + tool: input.tool, + callerProvider: caller.provider, + callerModel: caller.model, + callerOrigin: + input.caller === undefined + ? undefined + : callerOrigin({ + thread: input.caller, + scheduledRun: input.callerScheduledRun === true, + }), + callerDepth: + input.callerDepth === undefined ? undefined : Math.min(input.callerDepth, MAX_REPORTED_DEPTH), + durationMs: input.durationMs, + ...input.outcome, + ...input.settings, + }); + if (input.targets.length === 0) return [base]; + return input.targets.map((thread) => { + const target = providerDimensions(input.providers, thread.modelSelection); + return defined({ + ...base, + targetProvider: target.provider, + targetModel: target.model, + targetRuntimeMode: thread.runtimeMode, + targetInteractionMode: thread.interactionMode, + crossProvider: crossProvider(caller.provider, target.provider), + }); + }); +} diff --git a/docs/internals/product-analytics.md b/docs/internals/product-analytics.md index e92218a8bdc2..4d754c3eff0c 100644 --- a/docs/internals/product-analytics.md +++ b/docs/internals/product-analytics.md @@ -47,6 +47,16 @@ RPC authorization middleware after the request succeeds. the methods and commands, and takes variants only from closed contract enums, so paths, ids, and text never reach analytics. +Agent tool events report driver kinds, never instance ids: users name +instances, so an instance id can identify an account or employer. +[Provider dimensions](../../apps/server/src/telemetry/ProviderDimensions.ts) +report a model only when it is a non-custom entry in a vendor catalog. OpenCode, +Pi, and ACP agents list models from user configuration, such as local Ollama +tags, so their models stay out. Handoff settings come from closed enums and +booleans in tool arguments; prompts, titles, and ids are never read. Thread ids +are used only to look up the threads on each side and never leave the server, +which is also why tool events carry no per-thread sequence. + ## Delivery The server sends batches to the ingest endpoint at `ingest.t3.codes`, which archives