feat(analytics): record provider flows as duration spans

- collapse provider turn, tool call, and pr/mr create events into span-based analytics
- export duration metadata and span events through otlp
- tighten codex turn-completed notifications around active turn ids

Author: tyulyukov

apps/server/src/git/Layers/GitManager.ts

@@ -502,8 +502,11 @@ export const makeGitManager = Effect.fn("makeGitManager")(function* () {
     flush: Effect.void,
   }));
 
-  const recordGitAnalytics = (event: string, properties: Record<string, unknown>) =>
-    analytics.record(event, properties);
+  const recordGitAnalytics = (
+    event: string,
+    properties: Record<string, unknown>,
+    options?: Parameters<typeof analytics.record>[2],
+  ) => analytics.record(event, properties, options);
 
   const resolveJiraTickets = (
     threadId: ThreadId | undefined,
@@ -1310,12 +1313,41 @@ export const makeGitManager = Effect.fn("makeGitManager")(function* () {
       label: `Creating ${prOrMr}...`,
     });
     const originRepo = headContext.originRepositoryNameWithOwner;
-    yield* recordGitAnalytics("marcode.git.pr_mr.create_requested", {
+    const createStartedAtMs = Date.now();
+    const baseAnalyticsProperties = {
       "git.host.provider": detectedHostProvider ?? "unknown",
       "git.change_request.kind": changeRequestKind,
       "git.branch": headContext.headBranch,
       ...(repositoryName ? { "repository.name": repositoryName } : {}),
-    });
+    };
+    const recordCreateResult = (outcome: "created" | "error", hasUrl: boolean) => {
+      const completedAtMs = Date.now();
+      const durationMs = Math.max(1, completedAtMs - createStartedAtMs);
+      return recordGitAnalytics(
+        "marcode.git.pr_mr.create",
+        {
+          ...baseAnalyticsProperties,
+          outcome,
+          has_url: hasUrl,
+          "duration.ms": durationMs,
+        },
+        {
+          durationMs,
+          startedAt: createStartedAtMs,
+          spanEvents: [
+            {
+              name: "marcode.git.pr_mr.create.requested",
+              at: createStartedAtMs,
+            },
+            {
+              name: "marcode.git.pr_mr.create.completed",
+              at: completedAtMs,
+              attributes: { outcome, has_url: hasUrl },
+            },
+          ],
+        },
+      );
+    };
     yield* gitHostCli
       .createPullRequest({
         cwd,
@@ -1331,27 +1363,11 @@ export const makeGitManager = Effect.fn("makeGitManager")(function* () {
           schedule: Schedule.exponential(PR_CREATE_RETRY_BASE_DELAY, 2),
           while: isBranchNotReadyError,
         }),
-        Effect.tapError(() =>
-          recordGitAnalytics("marcode.git.pr_mr.create_completed", {
-            "git.host.provider": detectedHostProvider ?? "unknown",
-            "git.change_request.kind": changeRequestKind,
-            "git.branch": headContext.headBranch,
-            outcome: "error",
-            has_url: false,
-            ...(repositoryName ? { "repository.name": repositoryName } : {}),
-          }),
-        ),
+        Effect.tapError(() => recordCreateResult("error", false)),
       );
 
     const created = yield* findOpenPr(cwd, headContext);
-    yield* recordGitAnalytics("marcode.git.pr_mr.create_completed", {
-      "git.host.provider": detectedHostProvider ?? "unknown",
-      "git.change_request.kind": changeRequestKind,
-      "git.branch": headContext.headBranch,
-      outcome: "created",
-      has_url: created?.url ? true : false,
-      ...(repositoryName ? { "repository.name": repositoryName } : {}),
-    });
+    yield* recordCreateResult("created", Boolean(created?.url));
     if (!created) {
       return {
         status: "created" as const,

apps/server/src/provider/Layers/ProviderService.test.ts

@@ -43,6 +43,10 @@ import {
 } from "../../persistence/Layers/Sqlite.ts";
 import { ServerSettingsService } from "../../serverSettings.ts";
 import { AnalyticsServiceNoopLive } from "../../telemetry/Layers/AnalyticsService.ts";
+import {
+  AnalyticsService,
+  type AnalyticsServiceShape,
+} from "../../telemetry/Services/AnalyticsService.ts";
 
 const defaultServerSettingsLayer = ServerSettingsService.layerTest();
 
@@ -242,6 +246,12 @@ const hasMetricSnapshot = (
       Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value),
   );
 
+interface AnalyticsRecord {
+  readonly event: string;
+  readonly properties?: Record<string, unknown> | undefined;
+  readonly options?: Parameters<AnalyticsServiceShape["record"]>[2] | undefined;
+}
+
 function makeProviderServiceLayer() {
   const codex = makeFakeCodexAdapter();
   const claude = makeFakeCodexAdapter("claudeAgent");
@@ -382,7 +392,7 @@ it.effect("ProviderServiceLive writes canonical events to the emitting thread se
           state: "completed",
         },
       });
-      yield* sleep(20);
+      yield* sleep(100);
     }).pipe(Effect.provide(providerLayer));
 
     assert.equal(canonicalEvents.length, 1);
@@ -391,6 +401,83 @@ it.effect("ProviderServiceLive writes canonical events to the emitting thread se
   }).pipe(Effect.provide(NodeServices.layer)),
 );
 
+it.effect("ProviderServiceLive records provider turns as one duration product span", () =>
+  Effect.gen(function* () {
+    const codex = makeFakeCodexAdapter();
+    const analyticsRecords: AnalyticsRecord[] = [];
+    const registry: typeof ProviderAdapterRegistry.Service = {
+      getByProvider: (provider) =>
+        provider === "codex"
+          ? Effect.succeed(codex.adapter)
+          : Effect.fail(new ProviderUnsupportedError({ provider })),
+      listProviders: () => Effect.succeed(["codex"]),
+    };
+    const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe(
+      Layer.provide(SqlitePersistenceMemory),
+    );
+    const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer));
+    const analyticsLayer = Layer.succeed(AnalyticsService, {
+      record: (event, properties, options) =>
+        Effect.sync(() => {
+          analyticsRecords.push({ event, properties, options });
+        }),
+      flush: Effect.void,
+    });
+    const providerLayer = makeProviderServiceLive().pipe(
+      Layer.provide(Layer.succeed(ProviderAdapterRegistry, registry)),
+      Layer.provide(directoryLayer),
+      Layer.provide(defaultServerSettingsLayer),
+      Layer.provide(analyticsLayer),
+    );
+
+    yield* Effect.gen(function* () {
+      const provider = yield* ProviderService;
+      yield* sleep(10);
+      const threadId = asThreadId("thread-turn-analytics");
+      yield* provider.startSession(threadId, {
+        provider: "codex",
+        threadId,
+        runtimeMode: "full-access",
+      });
+      const turn = yield* provider.sendTurn({
+        threadId,
+        input: "hello",
+        interactionMode: "default",
+      });
+      codex.emit({
+        eventId: asEventId("evt-turn-analytics-completed"),
+        provider: "codex",
+        threadId,
+        turnId: turn.turnId,
+        createdAt: new Date(Date.now() + 250).toISOString(),
+        type: "turn.completed",
+        payload: {
+          state: "completed",
+        },
+      });
+      yield* sleep(100);
+    }).pipe(Effect.provide(providerLayer));
+
+    const turnRecord = analyticsRecords.find((record) => record.event === "marcode.provider.turn");
+
+    assert.isDefined(turnRecord);
+    assert.isUndefined(
+      analyticsRecords.find((record) => record.event === "marcode.provider.turn.sent"),
+    );
+    assert.isUndefined(
+      analyticsRecords.find((record) => record.event === "marcode.provider.turn.completed"),
+    );
+    assert.equal(turnRecord?.properties?.provider, "codex");
+    assert.equal(turnRecord?.properties?.outcome, "success");
+    assert.equal(turnRecord?.properties?.interaction_mode, "default");
+    assert.isAtLeast(Number(turnRecord?.options?.durationMs ?? 0), 1);
+    assert.deepEqual(
+      turnRecord?.options?.spanEvents?.map((event) => event.name),
+      ["marcode.provider.turn.sent", "marcode.provider.turn.completed"],
+    );
+  }).pipe(Effect.provide(NodeServices.layer)),
+);
+
 it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", () =>
   Effect.gen(function* () {
     const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "marcode-provider-service-"));