feat(server): export how long runs take and how they end

apps/server/src/observability/Metrics.ts

@@ -63,6 +63,18 @@ export const serverEventLoopDelayMax = Metric.gauge("t3_server_event_loop_delay_
   attributes: { unit: "ms" },
 });
 
+export const runsTotal = Metric.counter("t3_runs_total", {
+  description: "Runs that finished, by outcome. A later rollback does not count again.",
+});
+
+export const runDuration = Metric.timer("t3_run_duration", {
+  description: "Wall time of a run, from when it started to its terminal status.",
+});
+
+export const runQueueWait = Metric.timer("t3_run_queue_wait", {
+  description: "Time a run waited between being requested and starting.",
+});
+
 export const gitCommandsTotal = Metric.counter("t3_git_commands_total", {
   description: "Total git commands executed by the server runtime.",
 });

apps/server/src/orchestration-v2/EventSink.ts

@@ -26,6 +26,7 @@ import * as SqlClient from "effect/sql/SqlClient";
 import { replayAndBufferProjectedLiveEvents } from "./LiveStreamBudget.ts";
 import type { UnsequencedProjectEvent } from "../persistence/OrchestrationEventStore.ts";
 import { projectDomainEventForWire } from "./WireProjection.ts";
+import { makeRunOutcomeRecorder } from "./RunMetrics.ts";
 
 import * as CommandReceiptStore from "./CommandReceiptStore.ts";
 import * as EffectOutbox from "./EffectOutbox.ts";
@@ -238,8 +239,13 @@ const layerBase: Layer.Layer<
           );
         }
       });
+    const recordRunOutcomes = makeRunOutcomeRecorder();
+    // Recorded before publication, so an interrupted publish cannot lose a committed outcome.
     const publishStoredEvents = (events: ReadonlyArray<OrchestrationV2StoredEvent>) =>
-      eventStore.publishCommitted(events).pipe(Effect.andThen(publishLiveEvents(events)));
+      recordRunOutcomes(events).pipe(
+        Effect.andThen(eventStore.publishCommitted(events)),
+        Effect.andThen(publishLiveEvents(events)),
+      );
 
     // Transactions commit one at a time, but each writer publishes after its
     // commit. If a writer is descheduled in between, a later commit reaches

apps/server/src/orchestration-v2/RunMetrics.test.ts

@@ -0,0 +1,128 @@
+import { assert, describe, it } from "@effect/vitest";
+import {
+  EventId,
+  ProviderInstanceId,
+  RunId,
+  ThreadId,
+  type OrchestrationV2Run,
+  type OrchestrationV2StoredEvent,
+} from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
+import * as Effect from "effect/Effect";
+import * as Metric from "effect/Metric";
+
+import { makeRunOutcomeRecorder } from "./RunMetrics.ts";
+
+const at = (iso: string) => DateTime.makeUnsafe(iso);
+
+const runUpdated = (
+  sequence: number,
+  run: Pick<OrchestrationV2Run, "id" | "status" | "startedAt" | "completedAt">,
+  provider: string,
+): OrchestrationV2StoredEvent => ({
+  sequence,
+  commandId: null,
+  event: {
+    type: "run.updated",
+    id: EventId.make(`event:${sequence}`),
+    threadId: ThreadId.make("thread:metrics"),
+    occurredAt: at("2026-10-19T12:00:00.000Z"),
+    payload: {
+      threadId: ThreadId.make("thread:metrics"),
+      ordinal: 1,
+      providerInstanceId: ProviderInstanceId.make(provider),
+      modelSelection: { instanceId: ProviderInstanceId.make(provider), model: "claude-opus-5-5" },
+      requestedAt: at("2026-10-19T12:00:00.000Z"),
+      ...run,
+    } as OrchestrationV2Run,
+  },
+});
+
+const runSnapshots = (provider: string) =>
+  Metric.snapshot.pipe(
+    Effect.map((snapshots) =>
+      snapshots.filter(
+        (snapshot) =>
+          snapshot.id.startsWith("t3_run") && snapshot.attributes?.["provider"] === provider,
+      ),
+    ),
+  );
+
+describe("makeRunOutcomeRecorder", () => {
+  it.effect("records a finished run once, with how long it waited and ran", () =>
+    Effect.gen(function* () {
+      const record = makeRunOutcomeRecorder();
+      const id = RunId.make("run:metrics:once");
+      const started = { startedAt: at("2026-10-19T12:00:30.000Z") };
+
+      yield* record([
+        runUpdated(1, { id, status: "running", ...started, completedAt: null }, "p1"),
+      ]);
+      const finished = { id, status: "completed" as const, ...started };
+      yield* record([
+        runUpdated(2, { ...finished, completedAt: at("2026-10-19T12:10:30.000Z") }, "p1"),
+      ]);
+      // A later update to the finished run, such as its checkpoint, is not a second run.
+      yield* record([
+        runUpdated(3, { ...finished, completedAt: at("2026-10-19T12:10:30.000Z") }, "p1"),
+      ]);
+
+      const snapshots = yield* runSnapshots("p1");
+      const counter = snapshots.find((snapshot) => snapshot.id === "t3_runs_total");
+      assert.strictEqual(counter?.type === "Counter" ? counter.state.count : undefined, 1);
+      assert.strictEqual(counter?.attributes?.["outcome"], "completed");
+      assert.strictEqual(counter?.attributes?.["modelFamily"], "claude");
+      const histogram = (id: string) => {
+        const snapshot = snapshots.find((candidate) => candidate.id === id);
+        return snapshot?.type === "Histogram" ? snapshot.state : undefined;
+      };
+      assert.strictEqual(histogram("t3_run_duration")?.sum, 600_000);
+      assert.strictEqual(histogram("t3_run_queue_wait")?.sum, 30_000);
+    }),
+  );
+
+  it.effect("does not count a rollback of a run that already finished", () =>
+    Effect.gen(function* () {
+      // A fresh recorder stands in for a restart, which forgets the runs already counted.
+      const record = makeRunOutcomeRecorder();
+      yield* record([
+        runUpdated(
+          1,
+          {
+            id: RunId.make("run:metrics:rolled-back"),
+            status: "rolled_back",
+            startedAt: at("2026-10-19T12:00:30.000Z"),
+            completedAt: at("2026-10-20T09:00:00.000Z"),
+          },
+          "p3",
+        ),
+      ]);
+
+      assert.deepStrictEqual(yield* runSnapshots("p3"), []);
+    }),
+  );
+
+  it.effect("counts a run cancelled before it started without timing it", () =>
+    Effect.gen(function* () {
+      const record = makeRunOutcomeRecorder();
+      yield* record([
+        runUpdated(
+          1,
+          {
+            id: RunId.make("run:metrics:cancelled"),
+            status: "cancelled",
+            startedAt: null,
+            completedAt: at("2026-10-19T12:01:00.000Z"),
+          },
+          "p2",
+        ),
+      ]);
+
+      const snapshots = yield* runSnapshots("p2");
+      assert.deepStrictEqual(
+        snapshots.map((snapshot) => snapshot.id),
+        ["t3_runs_total"],
+      );
+    }),
+  );
+});

apps/server/src/orchestration-v2/RunMetrics.ts

@@ -0,0 +1,70 @@
+import type { OrchestrationV2Run, OrchestrationV2StoredEvent } from "@t3tools/contracts";
+import * as DateTime from "effect/DateTime";
+import * as Duration from "effect/Duration";
+import * as Effect from "effect/Effect";
+import * as Metric from "effect/Metric";
+
+import { normalizeModelMetricLabel } from "../observability/Attributes.ts";
+import {
+  metricAttributes,
+  runDuration,
+  runQueueWait,
+  runsTotal,
+} from "../observability/Metrics.ts";
+
+const REMEMBERED_RUNS = 4096;
+// How a run can end. `rolled_back` is left out (see below). Declared here rather than taken from
+// ThreadManagementService, which imports back into EventSink.
+const OUTCOMES: ReadonlySet<OrchestrationV2Run["status"]> = new Set([
+  "completed",
+  "failed",
+  "cancelled",
+  "interrupted",
+]);
+
+/**
+ * Records each run once, when a committed event first shows it finished. A finished run's row is
+ * updated again later (its checkpoint, a settlement), so the runs already counted are remembered.
+ * A rollback is not how a run ended: it rewrites runs that already finished, stamping the rollback
+ * time as their completion, so it would count them again with a duration that runs to the
+ * rollback.
+ */
+export const makeRunOutcomeRecorder = () => {
+  const counted = new Set<string>();
+  // Untraced: it runs for every committed batch inside the EventSink publish step, whose
+  // operation already owns the span.
+  return Effect.fnUntraced(function* (events: ReadonlyArray<OrchestrationV2StoredEvent>) {
+    for (const { event } of events) {
+      if (event.type !== "run.updated") continue;
+      const run = event.payload;
+      if (!OUTCOMES.has(run.status) || counted.has(run.id)) continue;
+      counted.add(run.id);
+      if (counted.size > REMEMBERED_RUNS) {
+        const oldest = counted.values().next();
+        if (!oldest.done) counted.delete(oldest.value);
+      }
+      const attributes = metricAttributes({
+        provider: run.providerInstanceId,
+        modelFamily: normalizeModelMetricLabel(run.modelSelection.model),
+        outcome: run.status,
+        delegated: run.delegatedTaskId !== undefined,
+      });
+      yield* Metric.update(Metric.withAttributes(runsTotal, attributes), 1);
+      if (run.startedAt === null) continue;
+      yield* Metric.update(
+        Metric.withAttributes(runQueueWait, attributes),
+        Duration.millis(
+          DateTime.toEpochMillis(run.startedAt) - DateTime.toEpochMillis(run.requestedAt),
+        ),
+      );
+      if (run.completedAt !== null) {
+        yield* Metric.update(
+          Metric.withAttributes(runDuration, attributes),
+          Duration.millis(
+            DateTime.toEpochMillis(run.completedAt) - DateTime.toEpochMillis(run.startedAt),
+          ),
+        );
+      }
+    }
+  });
+};