authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-10-05 21:41:28-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-05 21:41:28-07:00
logc299308a832db382fce61233bd10a4bd13bbc60c
treed5377177dd50900d675345424161d26ce107a1cf
parent24e4b62c883eb67251bdfb07d2bf22cfd3987820
signature Signed by PGP key B5690EEEBB952194

fix(server): metrics count interrupted work on the monotonic clock (#16207)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

12 files changed, 300 insertions(+), 86 deletions(-)

apps/server/src/keybindings.test.ts+22
...@@ -601,6 +601,28 @@ it.layer(NodeServices.layer)("keybindings", (it) => {...@@ -601,6 +601,28 @@ it.layer(NodeServices.layer)("keybindings", (it) => {
601 }).pipe(Effect.provide(layerKeybindings())),601 }).pipe(Effect.provide(layerKeybindings())),
602 );602 );
603603
604 it.effect("retries a failed config read instead of keeping the failure", () =>
605 Effect.gen(function* () {
606 const fileSystem = yield* FileSystem.FileSystem;
607 const { keybindingsConfigPath } = yield* ServerConfig.ServerConfig;
608 // A directory where the file should be makes the read itself fail.
609 yield* fileSystem.makeDirectory(keybindingsConfigPath, { recursive: true });
610
611 const keybindings = yield* Keybindings.Keybindings;
612 const failed = yield* toDetailResult(keybindings.loadConfigState);
613 assertFailure(failed, "failed to read keybindings config");
614
615 yield* fileSystem.remove(keybindingsConfigPath, { recursive: true });
616 yield* writeKeybindingsConfig(keybindingsConfigPath, [
617 { key: "mod+j", command: "terminal.toggle" },
618 ]);
619
620 const configState = yield* keybindings.loadConfigState;
621 assert.deepEqual(configState.issues, []);
622 assert.isTrue(configState.keybindings.some((entry) => entry.command === "terminal.toggle"));
623 }).pipe(Effect.provide(makeKeybindingsLayer())),
624 );
625
604 it.effect("updates cached resolved config after upsert", () =>626 it.effect("updates cached resolved config after upsert", () =>
605 Effect.gen(function* () {627 Effect.gen(function* () {
606 const { keybindingsConfigPath } = yield* ServerConfig.ServerConfig;628 const { keybindingsConfigPath } = yield* ServerConfig.ServerConfig;
apps/server/src/keybindings.ts+4-3
...@@ -480,13 +480,14 @@ const make = Effect.gen(function* () {...@@ -480,13 +480,14 @@ const make = Effect.gen(function* () {
480 })),480 })),
481 );481 );
482482
483 const resolvedConfigCache = yield* Cache.make<483 // A failed read is not kept: the next read retries instead of replaying the failure.
484 const resolvedConfigCache = yield* Cache.makeWith<
484 typeof resolvedConfigCacheKey,485 typeof resolvedConfigCacheKey,
485 KeybindingsConfigState,486 KeybindingsConfigState,
486 KeybindingsConfigError487 KeybindingsConfigError
487 >({488 >(() => loadConfigStateFromDisk, {
488 capacity: 1,489 capacity: 1,
489 lookup: () => loadConfigStateFromDisk,490 timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero),
490 });491 });
491492
492 const loadConfigStateFromCacheOrDisk = Cache.get(resolvedConfigCache, resolvedConfigCacheKey);493 const loadConfigStateFromCacheOrDisk = Cache.get(resolvedConfigCache, resolvedConfigCacheKey);
apps/server/src/observability/Metrics.test.ts+59
...@@ -1,5 +1,6 @@...@@ -1,5 +1,6 @@
1import { assert, describe, it } from "@effect/vitest";1import { assert, describe, it } from "@effect/vitest";
2import { ProviderDriverKind } from "@t3tools/contracts";2import { ProviderDriverKind } from "@t3tools/contracts";
3import * as Deferred from "effect/Deferred";
3import * as Duration from "effect/Duration";4import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";5import * as Effect from "effect/Effect";
5import * as Fiber from "effect/Fiber";6import * as Fiber from "effect/Fiber";
...@@ -127,6 +128,64 @@ describe("withMetrics", () => {...@@ -127,6 +128,64 @@ describe("withMetrics", () => {
127 }),128 }),
128 );129 );
129130
131 it.effect("counts interrupted work with an interrupt outcome and its duration", () =>
132 Effect.gen(function* () {
133 const counter = Metric.counter("with_metrics_interrupt_total");
134 const timer = Metric.timer("with_metrics_interrupt_duration");
135 const started = yield* Deferred.make<void>();
136
137 const fiber = yield* Deferred.succeed(started, undefined).pipe(
138 Effect.andThen(Effect.never),
139 withMetrics({ counter, timer, attributes: { operation: "interrupt" } }),
140 Effect.forkChild,
141 );
142 yield* Deferred.await(started);
143 yield* TestClock.adjust(Duration.millis(5));
144 yield* Fiber.interrupt(fiber);
145
146 const snapshots = yield* Metric.snapshot;
147 assert.equal(
148 hasMetricSnapshot(snapshots, "with_metrics_interrupt_total", {
149 operation: "interrupt",
150 outcome: "interrupt",
151 }),
152 true,
153 );
154 const duration = findHistogramSnapshot(snapshots, "with_metrics_interrupt_duration", {
155 operation: "interrupt",
156 });
157 assert.equal(duration?.state.count, 1);
158 assert.equal(duration?.state.sum, 5);
159 }),
160 );
161
162 it.effect("measures durations on the monotonic clock, not the wall clock", () =>
163 Effect.gen(function* () {
164 const timer = Metric.timer("with_metrics_monotonic_duration");
165 const started = yield* Deferred.make<void>();
166 const finish = yield* Deferred.make<void>();
167
168 const fiber = yield* Deferred.succeed(started, undefined).pipe(
169 Effect.andThen(Deferred.await(finish)),
170 withMetrics({ timer, attributes: { operation: "monotonic" } }),
171 Effect.forkChild,
172 );
173 yield* Deferred.await(started);
174 yield* TestClock.adjust(Duration.millis(10));
175 // A backward wall-clock correction must not shorten the measured duration.
176 yield* TestClock.setTime(0);
177 yield* Deferred.succeed(finish, undefined);
178 yield* Fiber.join(fiber);
179
180 const snapshots = yield* Metric.snapshot;
181 const duration = findHistogramSnapshot(snapshots, "with_metrics_monotonic_duration", {
182 operation: "monotonic",
183 });
184 assert.equal(duration?.state.count, 1);
185 assert.equal(duration?.state.sum, 10);
186 }),
187 );
188
130 it.effect("records timer durations from nanosecond clock readings", () =>189 it.effect("records timer durations from nanosecond clock readings", () =>
131 Effect.gen(function* () {190 Effect.gen(function* () {
132 const duration = Duration.nanos(1_500_000n);191 const duration = Duration.nanos(1_500_000n);
apps/server/src/observability/Metrics.ts+22-54
...@@ -5,11 +5,7 @@ import * as Exit from "effect/Exit";...@@ -5,11 +5,7 @@ import * as Exit from "effect/Exit";
5import * as Metric from "effect/Metric";5import * as Metric from "effect/Metric";
6import { dual } from "effect/Function";6import { dual } from "effect/Function";
77
8import {8import { compactMetricAttributes, outcomeFromExit } from "./Attributes.ts";
9 compactMetricAttributes,
10 normalizeModelMetricLabel,
11 outcomeFromExit,
12} from "./Attributes.ts";
139
14export const rpcRequestsTotal = Metric.counter("t3_rpc_requests_total", {10export const rpcRequestsTotal = Metric.counter("t3_rpc_requests_total", {
15 description: "Total RPC requests handled by the websocket RPC server.",11 description: "Total RPC requests handled by the websocket RPC server.",
...@@ -19,13 +15,6 @@ export const rpcRequestDuration = Metric.timer("t3_rpc_request_duration", {...@@ -19,13 +15,6 @@ export const rpcRequestDuration = Metric.timer("t3_rpc_request_duration", {
19 description: "RPC request handling duration.",15 description: "RPC request handling duration.",
20});16});
2117
22const orchestrationEventsProcessedTotal = Metric.counter(
23 "t3_orchestration_events_processed_total",
24 {
25 description: "Total orchestration intent events processed by runtime reactors.",
26 },
27);
28
29export const orchestrationEffectClaimsTotal = Metric.counter(18export const orchestrationEffectClaimsTotal = Metric.counter(
30 "t3_orchestration_effect_claims_total",19 "t3_orchestration_effect_claims_total",
31 {20 {
...@@ -38,20 +27,16 @@ export const orchestrationEffectQueueWait = Metric.timer("t3_orchestration_effec...@@ -38,20 +27,16 @@ export const orchestrationEffectQueueWait = Metric.timer("t3_orchestration_effec
38 "Time from an orchestration effect's temporal availability until claim, including same-thread blocking.",27 "Time from an orchestration effect's temporal availability until claim, including same-thread blocking.",
39});28});
4029
41const providerSessionsTotal = Metric.counter("t3_provider_sessions_total", {30export const providerSessionsTotal = Metric.counter("t3_provider_sessions_total", {
42 description: "Total provider session lifecycle operations.",31 description: "Total provider session lifecycle operations.",
43});32});
4433
45const providerTurnsTotal = Metric.counter("t3_provider_turns_total", {34export const providerTurnsTotal = Metric.counter("t3_provider_turns_total", {
46 description: "Total provider turn lifecycle operations.",35 description: "Total provider turn lifecycle operations.",
47});36});
4837
49const providerTurnDuration = Metric.timer("t3_provider_turn_duration", {38export const providerTurnDuration = Metric.timer("t3_provider_turn_duration", {
50 description: "Provider turn request duration.",39 description: "Time for the provider adapter to start a turn, not how long the turn runs.",
51});
52
53const providerRuntimeEventsTotal = Metric.counter("t3_provider_runtime_events_total", {
54 description: "Total canonical provider runtime events processed.",
55});40});
5641
57export const gitCommandsTotal = Metric.counter("t3_git_commands_total", {42export const gitCommandsTotal = Metric.counter("t3_git_commands_total", {
...@@ -125,16 +110,13 @@ export interface WithMetricsOptions {...@@ -125,16 +110,13 @@ export interface WithMetricsOptions {
125 ) => Readonly<Record<string, unknown>>;110 ) => Readonly<Record<string, unknown>>;
126}111}
127112
128const withMetricsImpl = <A, E, R>(113const recordMetrics = (
129 effect: Effect.Effect<A, E, R>,
130 options: WithMetricsOptions,114 options: WithMetricsOptions,
131): Effect.Effect<A, E, R> =>115 startedAt: bigint,
116 exit: Exit.Exit<unknown, unknown>,
117) =>
132 Effect.gen(function* () {118 Effect.gen(function* () {
133 const startedAt = yield* Clock.currentTimeNanos;119 const duration = Duration.nanos((yield* Clock.monotonicTimeNanos) - startedAt);
134 const exit = yield* Effect.exit(effect);
135 const endedAt = yield* Clock.currentTimeNanos;
136 const elapsedNanos = endedAt > startedAt ? endedAt - startedAt : 0n;
137 const duration = Duration.nanos(elapsedNanos);
138 const baseAttributes =120 const baseAttributes =
139 typeof options.attributes === "function" ? options.attributes() : (options.attributes ?? {});121 typeof options.attributes === "function" ? options.attributes() : (options.attributes ?? {});
140122
...@@ -159,35 +141,21 @@ const withMetricsImpl = <A, E, R>(...@@ -159,35 +141,21 @@ const withMetricsImpl = <A, E, R>(
159 1,141 1,
160 );142 );
161 }143 }
162
163 if (Exit.isSuccess(exit)) {
164 return exit.value;
165 }
166 return yield* Effect.failCause(exit.cause);
167 });144 });
168145
146// Durations come from the monotonic clock, so wall-clock corrections cannot skew them, and
147// metrics are recorded in an exit finalizer, so interrupted work is counted as "interrupt".
148const withMetricsImpl = <A, E, R>(
149 effect: Effect.Effect<A, E, R>,
150 options: WithMetricsOptions,
151): Effect.Effect<A, E, R> =>
152 Effect.flatMap(Clock.monotonicTimeNanos, (startedAt) =>
153 Effect.onExit(effect, (exit) => recordMetrics(options, startedAt, exit)),
154 );
155
169export const withMetrics: {156export const withMetrics: {
170 <A, E, R>(157 (
171 options: WithMetricsOptions,158 options: WithMetricsOptions,
172 ): (effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;159 ): <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>;
173 <A, E, R>(effect: Effect.Effect<A, E, R>, options: WithMetricsOptions): Effect.Effect<A, E, R>;160 <A, E, R>(effect: Effect.Effect<A, E, R>, options: WithMetricsOptions): Effect.Effect<A, E, R>;
174} = dual(2, withMetricsImpl);161} = dual(2, withMetricsImpl);
175
176const providerMetricAttributes = (provider: string, extra?: Readonly<Record<string, unknown>>) =>
177 compactMetricAttributes({
178 provider,
179 ...extra,
180 });
181
182const providerTurnMetricAttributes = (input: {
183 readonly provider: string;
184 readonly model: string | null | undefined;
185 readonly extra?: Readonly<Record<string, unknown>>;
186}) => {
187 const modelFamily = normalizeModelMetricLabel(input.model);
188 return compactMetricAttributes({
189 provider: input.provider,
190 ...(modelFamily ? { modelFamily } : {}),
191 ...input.extra,
192 });
193};
apps/server/src/observability/RpcInstrumentation.ts+8-14
...@@ -67,12 +67,9 @@ const recordRpcStreamMetrics = <E>(...@@ -67,12 +67,9 @@ const recordRpcStreamMetrics = <E>(
67 exit: Exit.Exit<unknown, E>,67 exit: Exit.Exit<unknown, E>,
68): Effect.Effect<void, never, never> =>68): Effect.Effect<void, never, never> =>
69 Effect.gen(function* () {69 Effect.gen(function* () {
70 const endedAt = yield* Clock.currentTimeNanos;
71 const elapsedNanos = endedAt > startedAt ? endedAt - startedAt : 0n;
72
73 yield* Metric.update(70 yield* Metric.update(
74 Metric.withAttributes(rpcRequestDuration, metricAttributes({ method })),71 Metric.withAttributes(rpcRequestDuration, metricAttributes({ method })),
75 Duration.nanos(elapsedNanos),72 Duration.nanos((yield* Clock.monotonicTimeNanos) - startedAt),
76 );73 );
77 yield* Metric.update(74 yield* Metric.update(
78 Metric.withAttributes(75 Metric.withAttributes(
...@@ -111,7 +108,7 @@ export const observeRpcStream = <A, E, R>(...@@ -111,7 +108,7 @@ export const observeRpcStream = <A, E, R>(
111): Stream.Stream<A, E, R> => {108): Stream.Stream<A, E, R> => {
112 const instrumented = Stream.unwrap(109 const instrumented = Stream.unwrap(
113 Effect.gen(function* () {110 Effect.gen(function* () {
114 const startedAt = yield* Clock.currentTimeNanos;111 const startedAt = yield* Clock.monotonicTimeNanos;
115 return stream.pipe(Stream.onExit((exit) => recordRpcStreamMetrics(method, startedAt, exit)));112 return stream.pipe(Stream.onExit((exit) => recordRpcStreamMetrics(method, startedAt, exit)));
116 }),113 }),
117 );114 );
...@@ -126,15 +123,12 @@ export const observeRpcStreamEffect = <A, StreamError, StreamContext, EffectErro...@@ -126,15 +123,12 @@ export const observeRpcStreamEffect = <A, StreamError, StreamContext, EffectErro
126): Stream.Stream<A, StreamError | EffectError, StreamContext | EffectContext> => {123): Stream.Stream<A, StreamError | EffectError, StreamContext | EffectContext> => {
127 const instrumented = Stream.unwrap(124 const instrumented = Stream.unwrap(
128 Effect.gen(function* () {125 Effect.gen(function* () {
129 const startedAt = yield* Clock.currentTimeNanos;126 const startedAt = yield* Clock.monotonicTimeNanos;
130 const exit = yield* Effect.exit(effect);127 // onError also runs when the stream is interrupted before it is produced.
131128 const stream = yield* effect.pipe(
132 if (Exit.isFailure(exit)) {129 Effect.onError((cause) => recordRpcStreamMetrics(method, startedAt, Exit.failCause(cause))),
133 yield* recordRpcStreamMetrics(method, startedAt, exit);130 );
134 return yield* Effect.failCause(exit.cause);131 return stream.pipe(
135 }
136
137 return exit.value.pipe(
138 Stream.onExit((streamExit) => recordRpcStreamMetrics(method, startedAt, streamExit)),132 Stream.onExit((streamExit) => recordRpcStreamMetrics(method, startedAt, streamExit)),
139 );133 );
140 }),134 }),
apps/server/src/orchestration-v2/ProviderSessionManager.test.ts+105
...@@ -24,6 +24,7 @@ import * as Exit from "effect/Exit";...@@ -24,6 +24,7 @@ import * as Exit from "effect/Exit";
24import * as Fiber from "effect/Fiber";24import * as Fiber from "effect/Fiber";
25import * as FileSystem from "effect/FileSystem";25import * as FileSystem from "effect/FileSystem";
26import * as Layer from "effect/Layer";26import * as Layer from "effect/Layer";
27import * as Metric from "effect/Metric";
27import * as Option from "effect/Option";28import * as Option from "effect/Option";
28import * as Queue from "effect/Queue";29import * as Queue from "effect/Queue";
29import * as Ref from "effect/Ref";30import * as Ref from "effect/Ref";
...@@ -832,6 +833,110 @@ it.effect("ProviderSessionManagerV2 closes every live session for a provider ins...@@ -832,6 +833,110 @@ it.effect("ProviderSessionManagerV2 closes every live session for a provider ins
832 }),833 }),
833);834);
834835
836it.effect("ProviderSessionManagerV2 records provider session and turn metrics", () =>
837 Effect.gen(function* () {
838 const state = yield* Ref.make(emptyState);
839 const effect = Effect.gen(function* () {
840 const eventSink = yield* EventSink.EventSinkV2;
841 const idAllocator = yield* IdAllocator.IdAllocatorV2;
842 const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
843 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
844 const now = yield* DateTime.now;
845 const threadId = ThreadId.make("thread-provider-session-manager-metrics");
846 const providerSessionId = yield* idAllocator.allocate.providerSession({
847 providerInstanceId: modelSelection.instanceId,
848 threadId,
849 });
850 const providerThread = makeProviderThread({ idAllocator, threadId, providerSessionId, now });
851 const runId = idAllocator.derive.run({ threadId, ordinal: 1 });
852 yield* eventSink.write({
853 events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
854 });
855
856 const runtime = yield* manager.open({
857 threadId,
858 providerSessionId,
859 modelSelection,
860 runtimePolicy,
861 });
862 yield* runtime.startTurn({
863 appThread: (yield* projectionStore.getThreadProjection(threadId)).thread,
864 threadId,
865 runId,
866 runOrdinal: 1,
867 providerTurnOrdinal: 1,
868 attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }),
869 rootNodeId: idAllocator.derive.rootNode({ runId }),
870 providerThread,
871 message: {
872 createdBy: "user",
873 creationSource: "web",
874 messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }),
875 text: "hello",
876 attachments: [],
877 },
878 modelSelection,
879 runtimePolicy,
880 });
881 yield* runtime.interruptTurn({
882 providerThread,
883 providerTurnId: idAllocator.derive.providerTurn({
884 driver: CODEX_DRIVER,
885 nativeTurnId: "native-turn",
886 }),
887 });
888 yield* manager.close(providerSessionId);
889
890 const snapshots = yield* Metric.snapshot;
891 const has = (id: string, attributes: Readonly<Record<string, string>>) =>
892 snapshots.some(
893 (snapshot) =>
894 snapshot.id === id &&
895 Object.entries(attributes).every(
896 ([key, value]) => snapshot.attributes?.[key] === value,
897 ),
898 );
899 assert.isTrue(
900 has("t3_provider_sessions_total", {
901 provider: "codex",
902 operation: "open",
903 outcome: "success",
904 }),
905 );
906 assert.isTrue(
907 has("t3_provider_sessions_total", {
908 provider: "codex",
909 operation: "release",
910 reason: "manual_shutdown",
911 outcome: "success",
912 }),
913 );
914 assert.isTrue(
915 has("t3_provider_turns_total", {
916 provider: "codex",
917 operation: "send",
918 modelFamily: "gpt",
919 outcome: "success",
920 }),
921 );
922 assert.isTrue(has("t3_provider_turn_duration", { provider: "codex", operation: "send" }));
923 assert.isTrue(
924 has("t3_provider_turns_total", {
925 provider: "codex",
926 operation: "interrupt",
927 outcome: "success",
928 }),
929 );
930 });
931
932 yield* effect.pipe(
933 Effect.provide(makeTestLayer({ state, idleTimeoutMs: 60_000 })),
934 // A private registry keeps other tests' provider metrics out of the assertions.
935 Effect.provideService(Metric.MetricRegistry, new Map()),
936 );
937 }),
938);
939
835it.effect("ProviderSessionManagerV2 opens a duplicate session only once", () =>940it.effect("ProviderSessionManagerV2 opens a duplicate session only once", () =>
836 Effect.gen(function* () {941 Effect.gen(function* () {
837 const state = yield* Ref.make(emptyState);942 const state = yield* Ref.make(emptyState);
apps/server/src/orchestration-v2/ProviderSessionManager.ts+43-5
...@@ -30,6 +30,13 @@ import * as Scope from "effect/Scope";...@@ -30,6 +30,13 @@ import * as Scope from "effect/Scope";
30import * as Semaphore from "effect/Semaphore";30import * as Semaphore from "effect/Semaphore";
31import * as Stream from "effect/Stream";31import * as Stream from "effect/Stream";
3232
33import { normalizeModelMetricLabel } from "../observability/Attributes.ts";
34import {
35 providerSessionsTotal,
36 providerTurnDuration,
37 providerTurnsTotal,
38 withMetrics,
39} from "../observability/Metrics.ts";
33import { ProviderWorkspaceMissingError } from "../provider/Errors.ts";40import { ProviderWorkspaceMissingError } from "../provider/Errors.ts";
34import * as ProjectService from "../project/ProjectService.ts";41import * as ProjectService from "../project/ProjectService.ts";
35import * as McpProviderSession from "../mcp/McpProviderSession.ts";42import * as McpProviderSession from "../mcp/McpProviderSession.ts";
...@@ -941,7 +948,16 @@ export const layerWithOptions = (...@@ -941,7 +948,16 @@ export const layerWithOptions = (
941 if (Option.isSome(closeExit) && Exit.isFailure(closeExit.value)) {948 if (Option.isSome(closeExit) && Exit.isFailure(closeExit.value)) {
942 return yield* Effect.failCause(closeExit.value.cause);949 return yield* Effect.failCause(closeExit.value.cause);
943 }950 }
944 }),951 }).pipe(
952 withMetrics({
953 counter: providerSessionsTotal,
954 attributes: {
955 provider: entry.runtime.driver,
956 operation: "release",
957 reason: input.reason,
958 },
959 }),
960 ),
945 }),961 }),
946 ([entry]) =>962 ([entry]) =>
947 Option.match(entry, {963 Option.match(entry, {
...@@ -1402,6 +1418,18 @@ export const layerWithOptions = (...@@ -1402,6 +1418,18 @@ export const layerWithOptions = (
1402 ): ProviderAdapterV2SessionRuntime => {1418 ): ProviderAdapterV2SessionRuntime => {
1403 const providerSessionId = runtime.providerSessionId;1419 const providerSessionId = runtime.providerSessionId;
1404 const subscribeEvents = makeEventSubscription(eventSubscribers);1420 const subscribeEvents = makeEventSubscription(eventSubscribers);
1421 // Every provider's turn operations pass through here, so this is where they are
1422 // counted. Only turn starts are timed: until the provider accepts the turn.
1423 const turnMetrics = (operation: string, model?: string) =>
1424 withMetrics({
1425 counter: providerTurnsTotal,
1426 ...(operation === "send" ? { timer: providerTurnDuration } : {}),
1427 attributes: {
1428 provider: runtime.driver,
1429 operation,
1430 modelFamily: normalizeModelMetricLabel(model),
1431 },
1432 });
1405 return {1433 return {
1406 ...runtime,1434 ...runtime,
1407 subscribeEvents,1435 subscribeEvents,
...@@ -1509,7 +1537,9 @@ export const layerWithOptions = (...@@ -1509,7 +1537,9 @@ export const layerWithOptions = (
1509 }),1537 }),
1510 ).pipe(1538 ).pipe(
1511 Effect.andThen(observeActivity(providerSessionId, markBusy(providerSessionId))),1539 Effect.andThen(observeActivity(providerSessionId, markBusy(providerSessionId))),
1512 Effect.andThen(runtime.startTurn(input)),1540 Effect.andThen(
1541 runtime.startTurn(input).pipe(turnMetrics("send", input.modelSelection.model)),
1542 ),
1513 Effect.catch((error) =>1543 Effect.catch((error) =>
1514 observeActivity(providerSessionId, markIdle(providerSessionId)).pipe(1544 observeActivity(providerSessionId, markIdle(providerSessionId)).pipe(
1515 Effect.andThen(Effect.fail(error)),1545 Effect.andThen(Effect.fail(error)),
...@@ -1518,15 +1548,19 @@ export const layerWithOptions = (...@@ -1518,15 +1548,19 @@ export const layerWithOptions = (
1518 ),1548 ),
1519 steerTurn: (input) =>1549 steerTurn: (input) =>
1520 observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe(1550 observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe(
1521 Effect.andThen(runtime.steerTurn(input)),1551 Effect.andThen(runtime.steerTurn(input).pipe(turnMetrics("steer"))),
1522 ),1552 ),
1523 interruptTurn: (input) =>1553 interruptTurn: (input) =>
1524 observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe(1554 observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe(
1525 Effect.andThen(runtime.interruptTurn(input)),1555 Effect.andThen(runtime.interruptTurn(input).pipe(turnMetrics("interrupt"))),
1526 ),1556 ),
1527 respondToRuntimeRequest: (input) =>1557 respondToRuntimeRequest: (input) =>
1528 observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe(1558 observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe(
1529 Effect.andThen(runtime.respondToRuntimeRequest(input)),1559 Effect.andThen(
1560 runtime
1561 .respondToRuntimeRequest(input)
1562 .pipe(turnMetrics("runtime-request-response")),
1563 ),
1530 ),1564 ),
1531 };1565 };
1532 };1566 };
...@@ -1794,6 +1828,10 @@ export const layerWithOptions = (...@@ -1794,6 +1828,10 @@ export const layerWithOptions = (
1794 cause,1828 cause,
1795 }),1829 }),
1796 ),1830 ),
1831 withMetrics({
1832 counter: providerSessionsTotal,
1833 attributes: { provider: adapter.driver, operation: "open" },
1834 }),
1797 );1835 );
1798 const eventSubscribers = yield* Ref.make<1836 const eventSubscribers = yield* Ref.make<
1799 ReadonlyMap<number, Queue.Queue<ProviderSessionEventSignal, Cause.Done>>1837 ReadonlyMap<number, Queue.Queue<ProviderSessionEventSignal, Cause.Done>>
apps/server/src/process/externalLauncher.ts+4-3
...@@ -471,7 +471,7 @@ const resolveFileManagerRevealKind = Effect.fn("externalLauncher.resolveFileMana...@@ -471,7 +471,7 @@ const resolveFileManagerRevealKind = Effect.fn("externalLauncher.resolveFileMana
471// waiting on, or throw away work a slow host (a busy server at startup, a471// waiting on, or throw away work a slow host (a busy server at startup, a
472// long PATH) needs more than one connect to finish. A failed scan clears the472// long PATH) needs more than one connect to finish. A failed scan clears the
473// entry so the next caller starts over rather than replaying the failure.473// entry so the next caller starts over rather than replaying the failure.
474// Expiry uses the monotonic clock (Clock.currentTimeNanos), matching the474// Expiry uses the monotonic clock (Clock.monotonicTimeNanos), matching the
475// command-resolution cache in @t3tools/shared/shell, so a backward wall-clock475// command-resolution cache in @t3tools/shared/shell, so a backward wall-clock
476// adjustment cannot keep an expired entry alive.476// adjustment cannot keep an expired entry alive.
477const EDITOR_DISCOVERY_CACHE_TTL_NANOS = 60_000_000_000n;477const EDITOR_DISCOVERY_CACHE_TTL_NANOS = 60_000_000_000n;
...@@ -772,7 +772,8 @@ export const make = Effect.gen(function* () {...@@ -772,7 +772,8 @@ export const make = Effect.gen(function* () {
772 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),772 Effect.provideService(ChildProcessSpawner.ChildProcessSpawner, spawner),
773 Effect.onExit((exit) =>773 Effect.onExit((exit) =>
774 Effect.gen(function* () {774 Effect.gen(function* () {
775 const expiresAtNanos = (yield* Clock.currentTimeNanos) + EDITOR_DISCOVERY_CACHE_TTL_NANOS;775 const expiresAtNanos =
776 (yield* Clock.monotonicTimeNanos) + EDITOR_DISCOVERY_CACHE_TTL_NANOS;
776 yield* Ref.update(editorDiscoveryCache, (current) =>777 yield* Ref.update(editorDiscoveryCache, (current) =>
777 Option.isNone(current) || current.value.scan !== scan778 Option.isNone(current) || current.value.scan !== scan
778 ? current779 ? current
...@@ -789,7 +790,7 @@ export const make = Effect.gen(function* () {...@@ -789,7 +790,7 @@ export const make = Effect.gen(function* () {
789 // Claiming the cache entry and starting its scan must not be split by an790 // Claiming the cache entry and starting its scan must not be split by an
790 // interrupt, or the entry would wait on a scan that never runs.791 // interrupt, or the entry would wait on a scan that never runs.
791 const acquireEditorDiscovery = Effect.gen(function* () {792 const acquireEditorDiscovery = Effect.gen(function* () {
792 const nowNanos = yield* Clock.currentTimeNanos;793 const nowNanos = yield* Clock.monotonicTimeNanos;
793 const [scan, isNewScan] = yield* Ref.modify(794 const [scan, isNewScan] = yield* Ref.modify(
794 editorDiscoveryCache,795 editorDiscoveryCache,
795 (796 (
apps/server/src/serverSettings.test.ts+22
...@@ -303,6 +303,28 @@ it.layer(NodeServices.layer)("server settings", (it) => {...@@ -303,6 +303,28 @@ it.layer(NodeServices.layer)("server settings", (it) => {
303 }).pipe(Effect.provide(layerServerSettings())),303 }).pipe(Effect.provide(layerServerSettings())),
304 );304 );
305305
306 it.effect("retries a failed settings read instead of keeping the failure", () =>
307 Effect.gen(function* () {
308 const serverConfig = yield* ServerConfig.ServerConfig;
309 const fileSystem = yield* FileSystem.FileSystem;
310 const serverSettings = yield* ServerSettingsModule.ServerSettingsService;
311 // A directory where the file should be makes the read itself fail.
312 yield* fileSystem.makeDirectory(serverConfig.settingsPath);
313
314 const error = yield* Effect.flip(serverSettings.getSettings);
315 assert.deepInclude(error, { _tag: "ServerSettingsError", operation: "read-file" });
316
317 yield* fileSystem.remove(serverConfig.settingsPath, { recursive: true });
318 yield* fileSystem.writeFileString(
319 serverConfig.settingsPath,
320 `{ "responseStreamingMode": "turn" }`,
321 );
322
323 const settings = yield* serverSettings.getSettings;
324 assert.equal(settings.responseStreamingMode, "turn");
325 }).pipe(Effect.provide(makeServerSettingsLayer())),
326 );
327
306 it.effect("decodes nested settings patches", () =>328 it.effect("decodes nested settings patches", () =>
307 Effect.gen(function* () {329 Effect.gen(function* () {
308 assert.deepEqual(330 assert.deepEqual(
apps/server/src/serverSettings.ts+8-4
...@@ -802,10 +802,14 @@ const make = Effect.gen(function* () {...@@ -802,10 +802,14 @@ const make = Effect.gen(function* () {
802 return migrated;802 return migrated;
803 });803 });
804804
805 const settingsCache = yield* Cache.make<typeof cacheKey, ServerSettings, ServerSettingsError>({805 // A failed read is not kept: the next read retries instead of replaying the failure.
806 capacity: 1,806 const settingsCache = yield* Cache.makeWith<typeof cacheKey, ServerSettings, ServerSettingsError>(
807 lookup: () => loadSettingsFromDisk,807 () => loadSettingsFromDisk,
808 });808 {
809 capacity: 1,
810 timeToLive: (exit) => (Exit.isSuccess(exit) ? Duration.infinity : Duration.zero),
811 },
812 );
809813
810 const getSettingsFromCache = Cache.get(settingsCache, cacheKey);814 const getSettingsFromCache = Cache.get(settingsCache, cacheKey);
811815
docs/operations/observability.md+1-1
...@@ -377,7 +377,7 @@ Traces are best for one request. Metrics are best for trends....@@ -377,7 +377,7 @@ Traces are best for one request. Metrics are best for trends.
377Good metric families to watch:377Good metric families to watch:
378378
379- `t3_rpc_request_duration`379- `t3_rpc_request_duration`
380- `t3_provider_turn_duration`380- `t3_provider_turn_duration` (how long the provider adapter takes to start a turn, not the turn's run time)
381- `t3_git_command_duration`381- `t3_git_command_duration`
382382
383Counters tell you volume and failure rate:383Counters tell you volume and failure rate:
packages/shared/src/shell.ts+2-2
...@@ -492,7 +492,7 @@ function resolveCommandCandidates(...@@ -492,7 +492,7 @@ function resolveCommandCandidates(
492// just written (e.g. managed binary installs). A "not-found" outcome is also492// just written (e.g. managed binary installs). A "not-found" outcome is also
493// cached for the TTL, so a just-installed binary can stay invisible for up to493// cached for the TTL, so a just-installed binary can stay invisible for up to
494// 30s unless resolved by explicit path.494// 30s unless resolved by explicit path.
495// TTL expiry uses the monotonic clock (Clock.currentTimeNanos) so backward495// TTL expiry uses the monotonic clock (Clock.monotonicTimeNanos) so backward
496// wall-clock adjustments cannot keep expired entries alive.496// wall-clock adjustments cannot keep expired entries alive.
497const COMMAND_RESOLUTION_CACHE_TTL_NANOS = 30_000_000_000n;497const COMMAND_RESOLUTION_CACHE_TTL_NANOS = 30_000_000_000n;
498const COMMAND_RESOLUTION_CACHE_MAX_ENTRIES = 512;498const COMMAND_RESOLUTION_CACHE_MAX_ENTRIES = 512;
...@@ -642,7 +642,7 @@ const resolveCommandPathForPlatform = Effect.fn("shell.resolveCommandPathForPlat...@@ -642,7 +642,7 @@ const resolveCommandPathForPlatform = Effect.fn("shell.resolveCommandPathForPlat
642 COMMAND_RESOLUTION_CACHE_KEY_SEPARATOR,642 COMMAND_RESOLUTION_CACHE_KEY_SEPARATOR,
643 );643 );
644 const cache = yield* CommandResolutionCache;644 const cache = yield* CommandResolutionCache;
645 const nowNanos = yield* Clock.currentTimeNanos;645 const nowNanos = yield* Clock.monotonicTimeNanos;
646 const cached = cache.get(cacheKey);646 const cached = cache.get(cacheKey);
647 if (cached !== undefined && cached.expiresAtNanos > nowNanos) {647 if (cached !== undefined && cached.expiresAtNanos > nowNanos) {
648 if (cached.resolvedPath === null) {648 if (cached.resolvedPath === null) {