diff --git a/apps/server/src/auth/RpcAuthorization.test.ts b/apps/server/src/auth/RpcAuthorization.test.ts index 54eee4ebda4d..8b3953feb9ca 100644 --- a/apps/server/src/auth/RpcAuthorization.test.ts +++ b/apps/server/src/auth/RpcAuthorization.test.ts @@ -258,6 +258,27 @@ describe("RPC scope middleware", () => { expect(handled).toEqual([]); }).pipe(Effect.scoped), ); + + it.effect("reports only authorized requests that succeed", () => + Effect.gen(function* () { + const observed: Array = []; + const client = yield* RpcTest.makeClient(group).pipe( + Effect.provide( + Layer.mergeAll( + group.toLayerHandler(WS_METHODS.serverProbe, () => Effect.succeed({})), + group.toLayerHandler(WS_METHODS.serverRetryResourceTelemetry, () => Effect.never), + RpcAuthorization.layer([AuthOrchestrationReadScope], (method) => + Effect.sync(() => void observed.push(method)), + ), + ), + ), + ); + + yield* client[WS_METHODS.serverProbe]({}); + yield* client[WS_METHODS.serverRetryResourceTelemetry]({}).pipe(Effect.flip); + expect(observed).toEqual([WS_METHODS.serverProbe]); + }).pipe(Effect.scoped), + ); }); describe("settings mutation authorization", () => { diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index b558c08bca30..30e0ca31a594 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -267,11 +267,22 @@ const requiredScopesForRpcCall = ( return [requiredScopeForRpcMethod(method)]; }; -export const layer = (scopes: ReadonlyArray) => +/** + * `onAuthorized` runs once after an authorized request succeeds; for a stream + * RPC, that is when the stream completes. Every RPC passes through here, so it + * is the one place to observe requests without wrapping each handler. + */ +export const layer = ( + scopes: ReadonlyArray, + onAuthorized?: (method: string, payload: unknown) => Effect.Effect, +) => Layer.succeed(RpcScopeAuthorization)((effect, { rpc, payload }) => { const requiredScopes = requiredScopesForRpcCall(rpc._tag, payload); const requiredScope = requiredScopes.find((scope) => !scopes.includes(scope)); - return requiredScope === undefined ? effect : Effect.fail(rpcAuthorizationError(requiredScope)); + if (requiredScope !== undefined) return Effect.fail(rpcAuthorizationError(requiredScope)); + return onAuthorized === undefined + ? effect + : Effect.tap(effect, () => onAuthorized(rpc._tag, payload)); }); /** Retrying can install or restart tools even though ordinary listing is readable. */ diff --git a/apps/server/src/telemetry/FeatureUsage.test.ts b/apps/server/src/telemetry/FeatureUsage.test.ts new file mode 100644 index 000000000000..0be829cbb891 --- /dev/null +++ b/apps/server/src/telemetry/FeatureUsage.test.ts @@ -0,0 +1,52 @@ +import { CommandId, ORCHESTRATION_V2_WS_METHODS, ThreadId, WS_METHODS } from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; + +import { featureUsage } from "./FeatureUsage.ts"; + +const dispatch = ORCHESTRATION_V2_WS_METHODS.dispatchCommand; +const threadId = ThreadId.make("thread-private"); +const commandId = CommandId.make("command-private"); + +it("names user commands with their closed-enum variant and nothing else", () => { + expect( + featureUsage(dispatch, { + type: "checkpoint.rollback", + commandId, + threadId, + scopeId: "scope-private", + checkpointId: "checkpoint-private", + restoreFiles: false, + }), + ).toEqual({ feature: "checkpoint.rollback", variant: "conversation_only" }); + expect( + featureUsage(dispatch, { + type: "thread.runtime-mode.set", + commandId, + threadId, + runtimeMode: "full-access", + }), + ).toEqual({ feature: "thread.runtime_mode.set", variant: "full-access" }); +}); + +it("ignores messages, housekeeping commands, and untracked RPCs", () => { + expect(featureUsage(dispatch, { type: "message.dispatch", text: "secret prompt" })).toBe( + undefined, + ); + expect(featureUsage(dispatch, { type: "thread.visit", threadId })).toBe(undefined); + expect(featureUsage(WS_METHODS.terminalWrite, { data: "rm -rf" })).toBe(undefined); +}); + +it("reads RPC variants from enums, never from paths or text", () => { + expect( + featureUsage(WS_METHODS.shellOpenInEditor, { cwd: "/Users/someone/secret", editor: "cursor" }), + ).toEqual({ feature: "editor.open", variant: "cursor" }); + expect( + featureUsage(ORCHESTRATION_V2_WS_METHODS.launchThread, { + title: "Private title", + workspaceStrategy: { type: "worktree", baseRef: "main", branch: "secret-branch" }, + }), + ).toEqual({ feature: "thread.launch", variant: "worktree" }); + expect(featureUsage(WS_METHODS.terminalOpen, { cwd: "/Users/someone/secret" })).toEqual({ + feature: "terminal.open", + }); +}); diff --git a/apps/server/src/telemetry/FeatureUsage.ts b/apps/server/src/telemetry/FeatureUsage.ts new file mode 100644 index 000000000000..c5f4903c35a1 --- /dev/null +++ b/apps/server/src/telemetry/FeatureUsage.ts @@ -0,0 +1,93 @@ +import { + ORCHESTRATION_V2_WS_METHODS, + type OrchestrationV2Command, + WS_METHODS, +} from "@t3tools/contracts"; + +export interface FeatureUsage { + readonly feature: string; + readonly variant?: string; +} + +type Payload = Readonly>; + +const literal = (payload: Payload, key: string) => + typeof payload[key] === "string" ? (payload[key] as string) : undefined; + +const use = (feature: string, variant?: string): FeatureUsage => + variant === undefined ? { feature } : { feature, variant }; + +/** + * User commands sent through `orchestration.dispatchCommand` that represent a + * feature choice. Housekeeping commands (visits, reorders, acknowledgements) + * and `message.dispatch`, which `client.turn.requested` already counts, are + * left out. + */ +const COMMAND_FEATURES: Partial< + Record FeatureUsage> +> = { + "checkpoint.rollback": (command) => ({ + feature: "checkpoint.rollback", + variant: command.restoreFiles === false ? "conversation_only" : "with_files", + }), + "thread.fork": () => ({ feature: "thread.fork" }), + "thread.merge_back": () => ({ feature: "thread.merge_back" }), + "provider.switch": () => ({ feature: "provider.switch" }), + "thread.archive": () => ({ feature: "thread.archive" }), + "thread.snooze": () => ({ feature: "thread.snooze" }), + "thread.pin": () => ({ feature: "thread.pin" }), + "thread.settle": () => ({ feature: "thread.settle" }), + "thread.pull-request.link": () => ({ feature: "thread.pull_request.link" }), + "thread.pull-request.watch": () => ({ feature: "thread.pull_request.watch" }), + "thread.runtime-mode.set": (command) => + use("thread.runtime_mode.set", literal(command, "runtimeMode")), + "thread.interaction-mode.set": (command) => + use("thread.interaction_mode.set", literal(command, "interactionMode")), + "run.interrupt": () => ({ feature: "run.interrupt" }), + "queued-message.promote-to-steer": () => ({ feature: "queue.steer" }), + "queued-run.edit": () => ({ feature: "queue.edit" }), +}; + +/** + * RPCs that represent a user starting a feature. Variants come only from + * closed enums in the contracts, so no path, id, or free text can reach analytics. + */ +const METHOD_FEATURES: Readonly FeatureUsage>> = { + [ORCHESTRATION_V2_WS_METHODS.launchThread]: (payload) => { + const strategy = payload.workspaceStrategy; + return use( + "thread.launch", + typeof strategy === "object" && strategy !== null + ? literal(strategy as Payload, "type") + : undefined, + ); + }, + [WS_METHODS.terminalOpen]: () => ({ feature: "terminal.open" }), + [WS_METHODS.previewOpen]: () => ({ feature: "preview.open" }), + [WS_METHODS.deviceOpen]: () => ({ feature: "device.open" }), + [WS_METHODS.shellOpenInEditor]: (payload) => use("editor.open", literal(payload, "editor")), + [WS_METHODS.vcsCreateWorktree]: () => ({ feature: "worktree.create" }), + [WS_METHODS.gitRunStackedAction]: (payload) => + use("git.stacked_action", literal(payload, "action")), + [WS_METHODS.pullRequestsRunAction]: (payload) => + use("pull_request.action", literal(payload, "action")), + [WS_METHODS.pullRequestsSubmitReview]: () => ({ feature: "pull_request.review" }), + [WS_METHODS.pullRequestsComment]: () => ({ feature: "pull_request.comment" }), + [WS_METHODS.scheduledTasksUpsert]: () => ({ feature: "scheduled_task.save" }), + [WS_METHODS.scheduledTasksRunNow]: () => ({ feature: "scheduled_task.run_now" }), + [WS_METHODS.agentSessionsImport]: () => ({ feature: "agent_session.import" }), + [WS_METHODS.projectCloneStart]: () => ({ feature: "project.clone" }), + [WS_METHODS.sourceControlPublishRepository]: () => ({ feature: "repository.publish" }), + [WS_METHODS.serverUpsertKeybinding]: () => ({ feature: "keybinding.customize" }), +}; + +/** The feature a WebSocket request uses, or undefined when it is not a tracked feature. */ +export function featureUsage(method: string, payload: unknown): FeatureUsage | undefined { + if (typeof payload !== "object" || payload === null) return undefined; + const input = payload as Payload; + if (method === ORCHESTRATION_V2_WS_METHODS.dispatchCommand) { + const type = literal(input, "type") as OrchestrationV2Command["type"] | undefined; + return type === undefined ? undefined : COMMAND_FEATURES[type]?.(input); + } + return METHOD_FEATURES[method]?.(input); +} diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index b886499bfdfa..ade826e53f38 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -226,6 +226,7 @@ import * as ProcessResourceMonitor from "./diagnostics/ProcessResourceMonitor.ts import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; import * as HostResources from "./resourceTelemetry/HostResources.ts"; import * as AnalyticsService from "./telemetry/AnalyticsService.ts"; +import { featureUsage } from "./telemetry/FeatureUsage.ts"; import * as UsageService from "./usage/UsageService.ts"; import * as TraceDiagnostics from "./diagnostics/TraceDiagnostics.ts"; import * as PullRequestService from "./pullRequest/PullRequestService.ts"; @@ -3130,7 +3131,24 @@ export const layer = Layer.unwrap( yield* RpcServer.make(ServerWsRpcGroup, WS_RPC_SERVER_OPTIONS).pipe( Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)), Effect.provide( - Layer.merge(RpcAuthorization.layer(session.scopes), rpcInstrumentationLayer), + Layer.merge( + RpcAuthorization.layer(session.scopes, (method, payload) => { + // Anonymous product signal: the feature and its enum variant, + // credited to the connecting client's surface only. + const usage = featureUsage(method, payload); + return usage === undefined + ? Effect.void + : analytics + .record("feature.used", { + ...usage, + ...(clientAnalyticsProps.surface === undefined + ? {} + : { surface: clientAnalyticsProps.surface }), + }) + .pipe(Effect.ignore); + }), + rpcInstrumentationLayer, + ), ), Effect.forkScoped, ); diff --git a/docs/internals/product-analytics.md b/docs/internals/product-analytics.md index b07f99bdb2d2..e92218a8bdc2 100644 --- a/docs/internals/product-analytics.md +++ b/docs/internals/product-analytics.md @@ -41,6 +41,12 @@ Unknown counts stay absent. Partial usage contains valid observed counts but cannot establish a whole-turn total. Keep these distinctions when changing token normalization or building reports. +`feature.used` counts WebSocket requests that start a feature, observed in the +RPC authorization middleware after the request succeeds. +[Feature mapping](../../apps/server/src/telemetry/FeatureUsage.ts) allowlists +the methods and commands, and takes variants only from closed contract enums, +so paths, ids, and text never reach analytics. + ## Delivery The server sends batches to the ingest endpoint at `ingest.t3.codes`, which archives