feat(analytics): threads and turns now know which client started them

apps/mobile/src/connection/platform.ts

@@ -21,6 +21,7 @@ import * as Layer from "effect/Layer";
 import * as Option from "effect/Option";
 import * as Queue from "effect/Queue";
 import * as Stream from "effect/Stream";
+import Constants from "expo-constants";
 import * as Network from "expo-network";
 import { AppState } from "react-native";
 
@@ -166,7 +167,7 @@ const capabilitiesLayer = Layer.effectContext(
       Context.add(
         ClientPresentation,
         ClientPresentation.of({
-          metadata: authClientMetadata(),
+          metadata: authClientMetadata(Constants.expoConfig?.version),
           scopes: AuthStandardClientScopes,
         }),
       ),

apps/mobile/src/lib/authClientMetadata.ts

@@ -1,10 +1,12 @@
 import type { AuthClientPresentationMetadata } from "@t3tools/contracts";
 import { Platform } from "react-native";
 
-export function authClientMetadata(): AuthClientPresentationMetadata {
+export function authClientMetadata(appVersion?: string): AuthClientPresentationMetadata {
   return {
     label: "T3 Code Mobile",
     deviceType: "mobile",
     ...(Platform.OS === "ios" ? { os: "iOS" } : Platform.OS === "android" ? { os: "Android" } : {}),
+    surface: "mobile",
+    ...(appVersion ? { appVersion } : {}),
   };
 }

apps/mobile/src/lib/connection.test.ts

@@ -26,6 +26,14 @@ describe("mobile remote connection records", () => {
       label: "T3 Code Mobile",
       deviceType: "mobile",
       os: "iOS",
+      surface: "mobile",
+    });
+  });
+
+  it("includes the mobile app version when the client provides it", () => {
+    expect(authClientMetadata("1.2.3")).toMatchObject({
+      surface: "mobile",
+      appVersion: "1.2.3",
     });
   });
 

apps/server/src/auth/SessionStore.test.ts

@@ -4,6 +4,7 @@ import * as Duration from "effect/Duration";
 import * as Effect from "effect/Effect";
 import * as Layer from "effect/Layer";
 import * as TestClock from "effect/testing/TestClock";
+import * as SqlClient from "effect/unstable/sql/SqlClient";
 
 import * as ServerConfig from "../config.ts";
 import { PersistenceSqlError } from "../persistence/Errors.ts";
@@ -47,6 +48,7 @@ const failingSessionLookupRepositoryLayer = Layer.succeed(AuthSessions.AuthSessi
   revoke: () => Effect.fail(repositoryFailure),
   revokeAllExcept: () => Effect.fail(repositoryFailure),
   setLastConnectedAt: () => Effect.void,
+  setClientConnection: () => Effect.void,
 });
 
 const failingSessionLookupCredentialLayer = Layer.effect(
@@ -315,4 +317,35 @@ it.layer(NodeServices.layer)("SessionStore.layer", (it) => {
       expect(afterReconnect[0]?.lastConnectedAt?.toString()).not.toBe(firstConnectedAt?.toString());
     }).pipe(Effect.provide(Layer.merge(makeSessionStoreLayer(), TestClock.layer()))),
   );
+  it.effect("records client connection metadata without clearing prior values", () =>
+    Effect.gen(function* () {
+      const sessions = yield* SessionStore.SessionStore;
+      const sql = yield* SqlClient.SqlClient;
+      const issued = yield* sessions.issue({
+        subject: "client-connection-test",
+        method: "bearer-access-token",
+      });
+      const readRow = sql<{
+        readonly surface: string | null;
+        readonly appVersion: string | null;
+      }>`
+        SELECT client_surface AS "surface", client_app_version AS "appVersion"
+        FROM auth_sessions
+        WHERE session_id = ${issued.sessionId}
+      `;
+
+      yield* sessions.recordClientConnection(issued.sessionId, {
+        surface: "mobile",
+        appVersion: "1.2.0",
+      });
+      expect((yield* readRow)[0]).toEqual({ surface: "mobile", appVersion: "1.2.0" });
+
+      // A partial report (old or minimal client) must not null out stored data.
+      yield* sessions.recordClientConnection(issued.sessionId, { appVersion: "1.3.0" });
+      expect((yield* readRow)[0]).toEqual({ surface: "mobile", appVersion: "1.3.0" });
+
+      yield* sessions.recordClientConnection(issued.sessionId, {});
+      expect((yield* readRow)[0]).toEqual({ surface: "mobile", appVersion: "1.3.0" });
+    }).pipe(Effect.provide(Layer.mergeAll(makeSessionStoreLayer(), SqlitePersistenceMemory))),
+  );
 });

apps/server/src/auth/SessionStore.ts

@@ -5,6 +5,7 @@ import {
   type AuthClientMetadata,
   type AuthClientSession,
   type AuthEnvironmentScope,
+  type ClientSurface,
   type ServerAuthSessionMethod,
 } from "@t3tools/contracts";
 import * as Context from "effect/Context";
@@ -396,6 +397,13 @@ export class SessionStore extends Context.Service<
     ) => Effect.Effect<number, SessionCredentialInternalError>;
     readonly markConnected: (sessionId: AuthSessionId) => Effect.Effect<void, never>;
     readonly markDisconnected: (sessionId: AuthSessionId) => Effect.Effect<void, never>;
+    readonly recordClientConnection: (
+      sessionId: AuthSessionId,
+      client: {
+        readonly surface?: ClientSurface | undefined;
+        readonly appVersion?: string | undefined;
+      },
+    ) => Effect.Effect<void, never>;
   }
 >()("t3/auth/SessionStore") {}
 
@@ -544,6 +552,28 @@ export const make = Effect.gen(function* () {
       Effect.withSpan("SessionStore.markConnected"),
     );
 
+  // Best-effort: connection metadata must never block or fail a connect.
+  const recordClientConnection: SessionStore["Service"]["recordClientConnection"] = (
+    sessionId,
+    client,
+  ) =>
+    client.surface === undefined && client.appVersion === undefined
+      ? Effect.void
+      : authSessions
+          .setClientConnection({
+            sessionId,
+            surface: client.surface ?? null,
+            appVersion: client.appVersion ?? null,
+          })
+          .pipe(
+            Effect.catchCause((cause) =>
+              Effect.logWarning("Failed to record session client connection metadata.").pipe(
+                Effect.annotateLogs({ sessionId, cause }),
+              ),
+            ),
+            Effect.withSpan("SessionStore.recordClientConnection"),
+          );
+
   const markDisconnected: SessionStore["Service"]["markDisconnected"] = (sessionId) =>
     Ref.update(connectedSessionsRef, (current) => {
       const next = new Map(current);
@@ -912,6 +942,7 @@ export const make = Effect.gen(function* () {
     revokeAllExcept,
     markConnected,
     markDisconnected,
+    recordClientConnection,
   });
 });
 

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts

@@ -1379,4 +1379,55 @@ describe("OrchestrationEngine", () => {
 
     await system.dispose();
   });
+
+  it("stamps the dispatching client's origin onto persisted event metadata", async () => {
+    const createdAt = now();
+    const system = await createOrchestrationSystem();
+    const { engine } = system;
+
+    await system.run(
+      engine.dispatch(
+        {
+          type: "project.create",
+          commandId: CommandId.make("cmd-origin-project-create"),
+          projectId: asProjectId("project-origin"),
+          title: "Origin Project",
+          workspaceRoot: "/tmp/project-origin",
+          defaultModelSelection: {
+            instanceId: ProviderInstanceId.make("codex"),
+            model: "gpt-5-codex",
+          },
+          createdAt,
+        },
+        { origin: { surface: "mobile", appVersion: "1.2.3" } },
+      ),
+    );
+    await system.run(
+      engine.dispatch({
+        type: "project.create",
+        commandId: CommandId.make("cmd-no-origin-project-create"),
+        projectId: asProjectId("project-no-origin"),
+        title: "No Origin Project",
+        workspaceRoot: "/tmp/project-no-origin",
+        defaultModelSelection: {
+          instanceId: ProviderInstanceId.make("codex"),
+          model: "gpt-5-codex",
+        },
+        createdAt,
+      }),
+    );
+
+    const events = await system.run(
+      Stream.runCollect(engine.readEvents(0)).pipe(Effect.map((chunk) => Array.from(chunk))),
+    );
+    const withOrigin = events.find((event) => event.commandId === "cmd-origin-project-create");
+    const withoutOrigin = events.find(
+      (event) => event.commandId === "cmd-no-origin-project-create",
+    );
+
+    expect(withOrigin?.metadata.origin).toEqual({ surface: "mobile", appVersion: "1.2.3" });
+    expect(withoutOrigin?.metadata.origin).toBeUndefined();
+
+    await system.dispose();
+  });
 });

apps/server/src/orchestration/Layers/OrchestrationEngine.ts

@@ -1,4 +1,5 @@
 import type {
+  OrchestrationClientOrigin,
   OrchestrationEvent,
   OrchestrationReadModel,
   ProjectId,
@@ -54,6 +55,7 @@ const isOrchestrationCommandInvariantError = Schema.is(OrchestrationCommandInvar
 
 interface CommandEnvelope {
   command: OrchestrationCommand;
+  origin: OrchestrationClientOrigin | undefined;
   result: Deferred.Deferred<{ sequence: number }, OrchestrationDispatchError>;
   startedAtMs: number;
 }
@@ -182,7 +184,16 @@ const makeOrchestrationEngine = Effect.gen(function* () {
                 }),
           ),
         );
-        const eventBases = Array.isArray(eventBase) ? eventBase : [eventBase];
+        const plannedEvents = Array.isArray(eventBase) ? eventBase : [eventBase];
+        // Stamp the dispatching client's origin onto every event the command
+        // produced. The decider stays pure; attribution is an engine concern.
+        const eventBases =
+          envelope.origin === undefined
+            ? plannedEvents
+            : plannedEvents.map((planned) => ({
+                ...planned,
+                metadata: { ...planned.metadata, origin: envelope.origin },
+              }));
         const committedCommand = yield* sql
           .withTransaction(
             Effect.gen(function* () {
@@ -329,11 +340,12 @@ const makeOrchestrationEngine = Effect.gen(function* () {
   const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>
     eventStore.readFromSequence(fromSequenceExclusive, limit);
 
-  const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
+  const dispatch: OrchestrationEngineShape["dispatch"] = (command, options) =>
     Effect.gen(function* () {
       const result = yield* Deferred.make<{ sequence: number }, OrchestrationDispatchError>();
       yield* Queue.offer(commandQueue, {
         command,
+        origin: options?.origin,
         result,
         startedAtMs: yield* Clock.currentTimeMillis,
       });

apps/server/src/orchestration/Services/OrchestrationEngine.ts

@@ -10,7 +10,11 @@
  *
  * @module OrchestrationEngineService
  */
-import type { OrchestrationCommand, OrchestrationEvent } from "@t3tools/contracts";
+import type {
+  OrchestrationClientOrigin,
+  OrchestrationCommand,
+  OrchestrationEvent,
+} from "@t3tools/contracts";
 import * as Context from "effect/Context";
 import type * as Effect from "effect/Effect";
 import type * as Stream from "effect/Stream";
@@ -41,13 +45,16 @@ export interface OrchestrationEngineShape {
    * Dispatch a validated orchestration command.
    *
    * @param command - Valid orchestration command.
+   * @param options - Optional client origin (surface/app version) stamped into
+   *   the metadata of every event the command produces.
    * @returns Effect containing the sequence of the persisted event.
    *
    * Dispatch is serialized through an internal queue and deduplicated via
    * command receipts.
    */
   readonly dispatch: (
     command: OrchestrationCommand,
+    options?: { readonly origin?: OrchestrationClientOrigin },
   ) => Effect.Effect<{ sequence: number }, OrchestrationDispatchError, never>;
 
   /**

apps/server/src/persistence/AuthSessions.ts

@@ -10,6 +10,7 @@ import {
   AuthClientMetadataDeviceType,
   AuthEnvironmentScopes,
   AuthSessionId,
+  ClientSurface,
   ServerAuthSessionMethod,
 } from "@t3tools/contracts";
 
@@ -82,6 +83,13 @@ export const SetAuthSessionLastConnectedAtInput = Schema.Struct({
 });
 export type SetAuthSessionLastConnectedAtInput = typeof SetAuthSessionLastConnectedAtInput.Type;
 
+export const SetAuthSessionClientConnectionInput = Schema.Struct({
+  sessionId: AuthSessionId,
+  surface: Schema.NullOr(ClientSurface),
+  appVersion: Schema.NullOr(Schema.String),
+});
+export type SetAuthSessionClientConnectionInput = typeof SetAuthSessionClientConnectionInput.Type;
+
 export class AuthSessionRepository extends Context.Service<
   AuthSessionRepository,
   {
@@ -103,6 +111,9 @@ export class AuthSessionRepository extends Context.Service<
     readonly setLastConnectedAt: (
       input: SetAuthSessionLastConnectedAtInput,
     ) => Effect.Effect<void, AuthSessionRepositoryError>;
+    readonly setClientConnection: (
+      input: SetAuthSessionClientConnectionInput,
+    ) => Effect.Effect<void, AuthSessionRepositoryError>;
   }
 >()("t3/persistence/AuthSessions/AuthSessionRepository") {}
 
@@ -281,6 +292,20 @@ export const make = Effect.gen(function* () {
       `,
   });
 
+  // COALESCE keeps the previous value when a client reports only one field, so
+  // a partial report never nulls out data a fuller client stored earlier.
+  const setClientConnectionRow = SqlSchema.void({
+    Request: SetAuthSessionClientConnectionInput,
+    execute: ({ sessionId, surface, appVersion }) =>
+      sql`
+        UPDATE auth_sessions
+        SET client_surface = COALESCE(${surface}, client_surface),
+            client_app_version = COALESCE(${appVersion}, client_app_version)
+        WHERE session_id = ${sessionId}
+          AND revoked_at IS NULL
+      `,
+  });
+
   const revokeSessionRows = SqlSchema.findAll({
     Request: RevokeAuthSessionInput,
     Result: Schema.Struct({ sessionId: AuthSessionId }),
@@ -404,13 +429,25 @@ export const make = Effect.gen(function* () {
       ),
     );
 
+  const setClientConnection: AuthSessionRepository["Service"]["setClientConnection"] = (input) =>
+    setClientConnectionRow(input).pipe(
+      Effect.mapError(
+        toPersistenceSqlOrDecodeError(
+          "AuthSessionRepository.setClientConnection:query",
+          "AuthSessionRepository.setClientConnection:encodeRequest",
+          { sessionId: input.sessionId },
+        ),
+      ),
+    );
+
   return {
     create,
     getById,
     listActive,
     revoke,
     revokeAllExcept,
     setLastConnectedAt,
+    setClientConnection,
   } satisfies AuthSessionRepository["Service"];
 });
 

apps/server/src/persistence/Migrations.ts

@@ -53,6 +53,7 @@ import Migration0037 from "./Migrations/037_ProjectionTurnsKeysetIndex.ts";
 import Migration0038 from "./Migrations/038_ProjectionThreadsPinOrderKey.ts";
 import Migration0039 from "./Migrations/039_ProjectionProjectsDefaultThreadEnvMode.ts";
 import Migration0040 from "./Migrations/040_ProjectionProjectFaviconPath.ts";
+import Migration0041 from "./Migrations/041_AuthSessionClientConnection.ts";
 
 /**
  * Migration loader with all migrations defined inline.
@@ -105,6 +106,7 @@ export const migrationEntries = [
   [38, "ProjectionThreadsPinOrderKey", Migration0038],
   [39, "ProjectionProjectsDefaultThreadEnvMode", Migration0039],
   [40, "ProjectionProjectFaviconPath", Migration0040],
+  [41, "AuthSessionClientConnection", Migration0041],
 ] as const;
 
 export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const);