From d021f57bf3a500e419e800c3eb957568bed8f713 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 6 Oct 2026 17:51:52 -0700 Subject: [PATCH] refactor(server): instrument WS RPCs in group middleware (#15548) Co-authored-by: Claude Opus 5.5 (1M context) --- .../observability/RpcInstrumentation.test.ts | 589 ++-- .../src/observability/RpcInstrumentation.ts | 335 ++- apps/server/src/ws.ts | 2513 ++++++----------- docs/internals/effect-services.md | 16 +- docs/operations/observability.md | 4 +- 5 files changed, 1443 insertions(+), 2014 deletions(-) diff --git a/apps/server/src/observability/RpcInstrumentation.test.ts b/apps/server/src/observability/RpcInstrumentation.test.ts index c7fb7d12b0958a0dbe478b22f4204d343daeac2c..32baf29d068cadfc6ce0e082f00624eb6a22d407 100644 --- a/apps/server/src/observability/RpcInstrumentation.test.ts +++ b/apps/server/src/observability/RpcInstrumentation.test.ts @@ -1,313 +1,336 @@ import { assert, describe, it } from "@effect/vitest"; -import { WS_METHODS } from "@t3tools/contracts"; +import { + AuthOrchestrationOperateScope, + AuthOrchestrationReadScope, + type AuthEnvironmentScope, + ScheduledTaskError, + ScheduledTaskId, + WS_METHODS, + WsRpcGroup, +} from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; import * as Metric from "effect/Metric"; +import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; import * as Tracer from "effect/Tracer"; +import * as RpcTest from "effect/rpc/RpcTest"; import * as TestClock from "effect/testing/TestClock"; -import { - observeRpcEffect, - observeRpcStream, - observeRpcStreamEffect, -} from "./RpcInstrumentation.ts"; - -const hasMetricSnapshot = ( - snapshots: ReadonlyArray, - id: string, - attributes: Readonly>, -) => - snapshots.some( - (snapshot) => - snapshot.id === id && - Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value), +import { RPC_REQUIRED_SCOPES } from "../auth/RpcAuthorization.ts"; +import * as RpcAuthorization from "../auth/RpcAuthorization.ts"; +import { RpcInstrumentation, rpcInstrumentationLayer } from "./RpcInstrumentation.ts"; + +type WsRpcMethod = keyof typeof RPC_REQUIRED_SCOPES; + +/** The server group narrowed to `tags`, so a test only implements the handlers it calls. */ +const groupOf = >(...tags: Tags) => + WsRpcGroup.omit( + ...[...WsRpcGroup.requests.keys()].filter( + (tag): tag is Exclude => + !(tags as ReadonlyArray).includes(tag), + ), + ).middleware(RpcInstrumentation); + +/** The middleware ws.ts installs for a connection with `scopes`. */ +const connectionMiddleware = (scopes: ReadonlyArray) => + Layer.merge(RpcAuthorization.layer(scopes), rpcInstrumentationLayer); +const readOnlyConnection = connectionMiddleware([AuthOrchestrationReadScope]); +const taskId = ScheduledTaskId.make("scheduled-task:instrumented"); +const rpcSpanDefaults = { "rpc.transport": "websocket", "rpc.system": "effect-rpc" }; + +/** Runs a test with a fresh metric registry and a tracer that keeps every span it ends. */ +const withTelemetry = ( + test: (ended: ReadonlyArray) => Effect.Effect, +) => { + const ended: Array = []; + const tracer = Tracer.make({ + span: (options) => { + const span = new Tracer.NativeSpan(options); + const end = span.end.bind(span); + span.end = (endTime, exit) => { + end(endTime, exit); + ended.push(span); + }; + return span; + }, + }); + return test(ended).pipe( + Effect.scoped, + Effect.withTracer(tracer), + Effect.provideService(Metric.MetricRegistry, new Map()), ); +}; -const findHistogramSnapshot = ( +const rpcSpans = (ended: ReadonlyArray) => + ended.filter((span) => span.name.startsWith("ws.rpc.")); + +// RpcTest keeps Effect's own RpcServer/RpcClient spans, which ws.ts turns off with +// `disableTracing: true`. Everything else comes from the middleware or the handlers. +const appSpans = (ended: ReadonlyArray) => + ended.filter((span) => !/^Rpc(Server|Client)\./.test(span.name)); + +const exitTag = (span: Tracer.NativeSpan | undefined) => + span?.status._tag === "Ended" ? span.status.exit._tag : undefined; + +const parentSpanId = (span: Tracer.NativeSpan | undefined) => + span?.parent._tag === "Some" ? span.parent.value.spanId : undefined; + +const requestCount = ( snapshots: ReadonlyArray, - id: string, - attributes: Readonly>, + method: string, + outcome: string, ) => + snapshots.find( + (snapshot): snapshot is Extract => + snapshot.type === "Counter" && + snapshot.id === "t3_rpc_requests_total" && + snapshot.attributes?.["method"] === method && + snapshot.attributes?.["outcome"] === outcome, + )?.state; + +const requestDuration = (snapshots: ReadonlyArray, method: string) => snapshots.find( (snapshot): snapshot is Extract => snapshot.type === "Histogram" && - snapshot.id === id && - Object.entries(attributes).every(([key, value]) => snapshot.attributes?.[key] === value), - ); - -const collectSpanNames = ( - effect: Effect.Effect, -): Effect.Effect, E, R> => - Effect.gen(function* () { - const spanNames: Array = []; - const tracer = Tracer.make({ - span: (options) => { - const span = new Tracer.NativeSpan(options); - const end = span.end.bind(span); - - span.end = (endTime, exit) => { - end(endTime, exit); - if (span.sampled) { - spanNames.push(span.name); - } - }; - - return span; - }, - }); - - yield* effect.pipe(Effect.withTracer(tracer)); - - return spanNames; - }); - -describe("RpcInstrumentation", () => { - it.effect("records success metrics for unary RPC handlers", () => - Effect.gen(function* () { - yield* observeRpcEffect("rpc.instrumentation.success", Effect.succeed("ok"), { - "rpc.aggregate": "test", - }).pipe(Effect.withSpan("rpc.instrumentation.success.span")); - - const snapshots = yield* Metric.snapshot; - - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { - method: "rpc.instrumentation.success", - outcome: "success", - }), - true, - ); - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { - method: "rpc.instrumentation.success", - }), - true, - ); - }), - ); - - it.effect("records failure outcomes for unary RPC handlers", () => - Effect.gen(function* () { - yield* Effect.exit( - observeRpcEffect("rpc.instrumentation.failure", Effect.fail("boom"), { - "rpc.aggregate": "test", - }).pipe(Effect.withSpan("rpc.instrumentation.failure.span")), - ); - - const snapshots = yield* Metric.snapshot; - - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { - method: "rpc.instrumentation.failure", - outcome: "failure", - }), - true, - ); - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { - method: "rpc.instrumentation.failure", - }), - true, - ); - }), - ); - - it.effect("records subscription activation metrics for stream RPC handlers", () => - Effect.gen(function* () { - const events = yield* Stream.runCollect( - observeRpcStreamEffect( - "rpc.instrumentation.stream", - Effect.succeed(Stream.make("a", "b")), - { "rpc.aggregate": "test" }, - ).pipe(Stream.withSpan("rpc.instrumentation.stream.span")), - ); - - assert.deepStrictEqual(Array.from(events), ["a", "b"]); - - const snapshots = yield* Metric.snapshot; - - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { - method: "rpc.instrumentation.stream", - outcome: "success", - }), - true, - ); - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { - method: "rpc.instrumentation.stream", - }), - true, - ); - }), - ); - - it.effect("records failure outcomes for direct stream RPC handlers during consumption", () => - Effect.gen(function* () { - const exit = yield* Stream.runCollect( - observeRpcStream( - "rpc.instrumentation.stream.failure", - Stream.make("a").pipe(Stream.concat(Stream.fail("boom"))), - { "rpc.aggregate": "test" }, - ).pipe(Stream.withSpan("rpc.instrumentation.stream.failure.span")), - ).pipe(Effect.exit); - - assert.equal(Exit.isFailure(exit), true); - - const snapshots = yield* Metric.snapshot; - - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { - method: "rpc.instrumentation.stream.failure", - outcome: "failure", - }), - true, - ); - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { - method: "rpc.instrumentation.stream.failure", - }), - true, - ); - }), + snapshot.id === "t3_rpc_request_duration" && + snapshot.attributes?.["method"] === method, + )?.state; + +describe("WS RPC instrumentation middleware", () => { + it.effect("records one span and request metric per call, including rejected calls", () => + withTelemetry((ended) => + Effect.gen(function* () { + const group = groupOf( + WS_METHODS.serverProbe, + WS_METHODS.scheduledTasksList, + WS_METHODS.serverRetryResourceTelemetry, + WS_METHODS.pullRequestsSubscribeRefreshes, + WS_METHODS.scheduledTasksSubscribe, + WS_METHODS.serverGetSettings, + ); + const client = yield* RpcTest.makeClient(group).pipe( + Effect.provide( + Layer.mergeAll( + group.toLayerHandler(WS_METHODS.serverProbe, () => + Effect.succeed({}).pipe(Effect.withSpan("serverProbe.child")), + ), + group.toLayerHandler(WS_METHODS.scheduledTasksList, () => + Effect.annotateCurrentSpan({ "scheduled_task.id": taskId }).pipe( + Effect.andThen(new ScheduledTaskError({ message: "List failed." })), + ), + ), + group.toLayerHandler(WS_METHODS.serverRetryResourceTelemetry, () => + Effect.die("authorization let a rejected call through"), + ), + group.toLayerHandler(WS_METHODS.pullRequestsSubscribeRefreshes, () => + Stream.make(1, 2), + ), + group.toLayerHandler(WS_METHODS.scheduledTasksSubscribe, () => + Stream.concat( + Stream.make({ tasks: [] }), + Stream.fail(new ScheduledTaskError({ message: "Subscription failed." })), + ), + ), + group.toLayerHandler(WS_METHODS.serverGetSettings, () => Effect.die("broken")), + readOnlyConnection, + ), + ), + ); + + assert.deepStrictEqual(yield* client[WS_METHODS.serverProbe]({}), {}); + const listError = yield* client[WS_METHODS.scheduledTasksList]({}).pipe(Effect.flip); + assert.equal(listError._tag, "ScheduledTaskError"); + // Retrying telemetry needs operate scope, so the handler must not run. + const rejection = yield* client[WS_METHODS.serverRetryResourceTelemetry]({}).pipe( + Effect.flip, + ); + assert.equal(rejection._tag, "EnvironmentAuthorizationError"); + const refreshes = yield* Stream.runCollect( + client[WS_METHODS.pullRequestsSubscribeRefreshes]({}), + ); + assert.deepStrictEqual(Array.from(refreshes), [1, 2]); + const subscribeError = yield* Stream.runDrain( + client[WS_METHODS.scheduledTasksSubscribe]({}), + ).pipe(Effect.flip); + assert.equal(subscribeError._tag, "ScheduledTaskError"); + const settingsExit = yield* Effect.exit(client[WS_METHODS.serverGetSettings]({})); + assert.isTrue(Exit.hasDies(settingsExit)); + + assert.deepStrictEqual( + rpcSpans(ended).map((span) => [span.name, Object.fromEntries(span.attributes)]), + [ + [ + "ws.rpc.server.probe", + { + ...rpcSpanDefaults, + "rpc.method": WS_METHODS.serverProbe, + "rpc.aggregate": "server", + }, + ], + [ + "ws.rpc.scheduledTasks.list", + { + ...rpcSpanDefaults, + "rpc.method": WS_METHODS.scheduledTasksList, + "rpc.aggregate": "scheduledTasks", + "scheduled_task.id": taskId, + }, + ], + [ + "ws.rpc.server.retryResourceTelemetry", + { + ...rpcSpanDefaults, + "rpc.method": WS_METHODS.serverRetryResourceTelemetry, + "rpc.aggregate": "server", + }, + ], + [ + "ws.rpc.pullRequests.subscribeRefreshes", + { + ...rpcSpanDefaults, + "rpc.method": WS_METHODS.pullRequestsSubscribeRefreshes, + "rpc.aggregate": "pull-requests", + }, + ], + [ + "ws.rpc.scheduledTasks.subscribe", + { + ...rpcSpanDefaults, + "rpc.method": WS_METHODS.scheduledTasksSubscribe, + "rpc.aggregate": "scheduledTasks", + }, + ], + [ + "ws.rpc.server.getSettings", + { + ...rpcSpanDefaults, + "rpc.method": WS_METHODS.serverGetSettings, + "rpc.aggregate": "server", + }, + ], + ], + ); + assert.deepStrictEqual(rpcSpans(ended).map(exitTag), [ + "Success", + "Failure", + "Failure", + "Success", + "Failure", + "Failure", + ]); + const probeSpan = rpcSpans(ended)[0]; + const child = ended.find((span) => span.name === "serverProbe.child"); + assert.equal(parentSpanId(child), probeSpan?.spanId); + + const snapshots = yield* Metric.snapshot; + assert.deepStrictEqual( + [ + requestCount(snapshots, WS_METHODS.serverProbe, "success"), + requestCount(snapshots, WS_METHODS.scheduledTasksList, "failure"), + requestCount(snapshots, WS_METHODS.serverRetryResourceTelemetry, "failure"), + requestCount(snapshots, WS_METHODS.pullRequestsSubscribeRefreshes, "success"), + requestCount(snapshots, WS_METHODS.scheduledTasksSubscribe, "failure"), + requestCount(snapshots, WS_METHODS.serverGetSettings, "failure"), + ].map((state) => state?.count), + [1, 1, 1, 1, 1, 1], + ); + for (const method of group.requests.keys()) { + assert.equal(requestDuration(snapshots, method)?.count, 1); + } + }), + ), ); - it.effect("records direct stream durations from nanosecond clock readings", () => - Effect.gen(function* () { - const duration = Duration.nanos(1_500_000n); - const events = yield* Effect.gen(function* () { - const fiber = yield* Stream.runCollect( - observeRpcStream( - WS_METHODS.serverGetProcessDiagnostics, - Stream.fromEffect(Effect.sleep(duration).pipe(Effect.as("ok"))), - { - "rpc.aggregate": "test", - }, + it.effect("keeps a stream's span and duration open until the subscription is interrupted", () => + withTelemetry((ended) => + Effect.gen(function* () { + const waiting = yield* Deferred.make(); + const group = groupOf(WS_METHODS.pullRequestsSubscribeRefreshes); + const client = yield* RpcTest.makeClient(group).pipe( + Effect.provide( + Layer.mergeAll( + group.toLayerHandler(WS_METHODS.pullRequestsSubscribeRefreshes, () => + Stream.concat( + Stream.make(1), + Stream.fromEffect( + Deferred.succeed(waiting, undefined).pipe( + Effect.andThen(Effect.never), + Effect.withSpan("refreshes.wait"), + ), + ), + ), + ), + readOnlyConnection, + ), ), + ); + + // Wall and monotonic time disagree before the call starts, so a duration that mixes + // the two clocks is caught. + yield* TestClock.adjust(Duration.seconds(1)); + yield* TestClock.setTime(0); + const consumer = yield* Stream.runDrain( + client[WS_METHODS.pullRequestsSubscribeRefreshes]({}), ).pipe(Effect.forkChild); - - yield* Effect.yieldNow; - yield* TestClock.adjust(duration); - return yield* Fiber.join(fiber); - }).pipe(Effect.provide(TestClock.layer())); - - assert.deepStrictEqual(Array.from(events), ["ok"]); - - const snapshots = yield* Metric.snapshot; - const snapshot = findHistogramSnapshot(snapshots, "t3_rpc_request_duration", { - method: WS_METHODS.serverGetProcessDiagnostics, - }); - - assert.equal(snapshot?.state.count, 1); - assert.equal(snapshot?.state.sum, 1.5); - }), + yield* Deferred.await(waiting); + yield* TestClock.adjust(Duration.millis(250)); + // A backward wall-clock correction must not shorten the measured duration. + yield* TestClock.setTime(0); + assert.deepStrictEqual(appSpans(ended), []); + + // The client waits for the server to stop the call, which ends the RPC span. + yield* Fiber.interrupt(consumer); + + const [rpcSpan] = rpcSpans(ended); + assert.equal(rpcSpan?.name, "ws.rpc.pullRequests.subscribeRefreshes"); + assert.equal(exitTag(rpcSpan), "Failure"); + const waitSpan = ended.find((span) => span.name === "refreshes.wait"); + assert.equal(parentSpanId(waitSpan), rpcSpan?.spanId); + + const snapshots = yield* Metric.snapshot; + assert.equal( + requestCount(snapshots, WS_METHODS.pullRequestsSubscribeRefreshes, "interrupt")?.count, + 1, + ); + const duration = requestDuration(snapshots, WS_METHODS.pullRequestsSubscribeRefreshes); + assert.equal(duration?.count, 1); + assert.equal(duration?.sum, 250); + }), + ), ); - it.effect("records failure outcomes when a stream RPC effect produces a failing stream", () => - Effect.gen(function* () { - const exit = yield* Stream.runCollect( - observeRpcStreamEffect( - "rpc.instrumentation.stream.effect.failure", - Effect.succeed(Stream.fail("boom")), - { "rpc.aggregate": "test" }, - ).pipe(Stream.withSpan("rpc.instrumentation.stream.effect.failure.span")), - ).pipe(Effect.exit); - - assert.equal(Exit.isFailure(exit), true); - - const snapshots = yield* Metric.snapshot; - - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_requests_total", { - method: "rpc.instrumentation.stream.effect.failure", - outcome: "failure", - }), - true, - ); - assert.equal( - hasMetricSnapshot(snapshots, "t3_rpc_request_duration", { - method: "rpc.instrumentation.stream.effect.failure", - }), - true, - ); - }), - ); - - it.effect("records spans for traced stream RPC handlers", () => - Effect.gen(function* () { - const spanNames = yield* collectSpanNames( - Stream.runCollect( - observeRpcStream( - "rpc.instrumentation.traced.stream", - Stream.fromEffect( - Effect.succeed("ok").pipe(Effect.withSpan("rpc.instrumentation.traced.stream.child")), - ), - { "rpc.aggregate": "test" }, - ), - ), - ); - - assert.equal(spanNames.includes("ws.rpc.rpc.instrumentation.traced.stream"), true); - assert.equal(spanNames.includes("rpc.instrumentation.traced.stream.child"), true); - }), - ); - - it.effect("does not create spans for disabled unary RPC handlers", () => - Effect.gen(function* () { - const spanNames = yield* collectSpanNames( - observeRpcEffect( - WS_METHODS.serverGetTraceDiagnostics, - Effect.succeed("ok").pipe(Effect.withSpan("rpc.instrumentation.disabled.unary.child")), - { "rpc.aggregate": "test" }, - ), - ); - - assert.deepStrictEqual(spanNames, []); - }), - ); - - it.effect("does not create spans for disabled direct stream RPC handlers", () => - Effect.gen(function* () { - const spanNames = yield* collectSpanNames( - Stream.runCollect( - observeRpcStream( - WS_METHODS.serverGetTraceDiagnostics, - Stream.fromEffect( - Effect.succeed("ok").pipe( - Effect.withSpan("rpc.instrumentation.disabled.stream.child"), + it.effect("records metrics but no spans for methods with tracing disabled", () => + withTelemetry((ended) => + Effect.gen(function* () { + const group = groupOf(WS_METHODS.serverSignalProcess); + const client = yield* RpcTest.makeClient(group).pipe( + Effect.provide( + Layer.mergeAll( + group.toLayerHandler(WS_METHODS.serverSignalProcess, (input) => + Effect.succeed({ + pid: input.pid, + signal: input.signal, + signaled: true, + message: Option.none(), + }).pipe(Effect.withSpan("signalProcess.child")), ), + connectionMiddleware([AuthOrchestrationReadScope, AuthOrchestrationOperateScope]), ), - { "rpc.aggregate": "test" }, ), - ), - ); - - assert.deepStrictEqual(spanNames, []); - }), - ); + ); - it.effect("does not create spans for disabled stream effect RPC handlers", () => - Effect.gen(function* () { - const spanNames = yield* collectSpanNames( - Stream.runCollect( - observeRpcStreamEffect( - WS_METHODS.serverGetTraceDiagnostics, - Effect.succeed( - Stream.fromEffect( - Effect.succeed("ok").pipe( - Effect.withSpan("rpc.instrumentation.disabled.stream.effect.consume"), - ), - ), - ).pipe(Effect.withSpan("rpc.instrumentation.disabled.stream.effect.create")), - { "rpc.aggregate": "test" }, - ), - ), - ); + const input = { pid: 4242, startTimeMs: 0, signal: "SIGINT" } as const; + assert.equal((yield* client[WS_METHODS.serverSignalProcess](input)).signaled, true); - assert.deepStrictEqual(spanNames, []); - }), + assert.deepStrictEqual(appSpans(ended), []); + const snapshots = yield* Metric.snapshot; + assert.equal(requestCount(snapshots, WS_METHODS.serverSignalProcess, "success")?.count, 1); + }), + ), ); }); diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index 1f3d172b5763586f8facbb7bbd4730d1f4081d3e..f38e1aeff1d9deccf6f0df76865000f3b93247f5 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -1,14 +1,198 @@ -import { WS_METHODS } from "@t3tools/contracts"; -import * as Clock from "effect/Clock"; -import * as Duration from "effect/Duration"; +import { ORCHESTRATION_V2_WS_METHODS, WS_METHODS, type WsRpcGroup } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; -import * as Exit from "effect/Exit"; -import * as Metric from "effect/Metric"; +import * as Layer from "effect/Layer"; import * as References from "effect/References"; -import * as Stream from "effect/Stream"; +import type * as RpcGroup from "effect/rpc/RpcGroup"; +import * as RpcMiddleware from "effect/rpc/RpcMiddleware"; -import { outcomeFromExit } from "./Attributes.ts"; -import { metricAttributes, rpcRequestDuration, rpcRequestsTotal, withMetrics } from "./Metrics.ts"; +import { rpcRequestDuration, rpcRequestsTotal, withMetrics } from "./Metrics.ts"; + +type WsRpcMethod = RpcGroup.Rpcs["_tag"]; + +/** + * The `rpc.aggregate` span attribute of every WebSocket RPC. Trace queries and dashboards group by + * these labels, so they keep their historical values even where they differ from the method + * prefix. Adding an RPC to `WsRpcGroup` without a label is a type error. + */ +const RPC_AGGREGATES = { + [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: "orchestrationV2", + [ORCHESTRATION_V2_WS_METHODS.getWorkflowScript]: "orchestration", + [ORCHESTRATION_V2_WS_METHODS.getTurnItem]: "orchestration", + [ORCHESTRATION_V2_WS_METHODS.getTurnDiff]: "orchestration", + [ORCHESTRATION_V2_WS_METHODS.getFullThreadDiff]: "orchestration", + [ORCHESTRATION_V2_WS_METHODS.searchThreads]: "orchestration", + [ORCHESTRATION_V2_WS_METHODS.getArchivedShellSnapshot]: "orchestration", + [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: "orchestrationV2", + [ORCHESTRATION_V2_WS_METHODS.launchThread]: "orchestration", + [ORCHESTRATION_V2_WS_METHODS.subscribeArchivedShell]: "orchestration", + [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: "orchestrationV2", + [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: "orchestrationV2", + [WS_METHODS.projectsMutate]: "orchestration", + [WS_METHODS.serverProbe]: "server", + [WS_METHODS.serverGetConfig]: "server", + [WS_METHODS.serverRefreshProviders]: "server", + [WS_METHODS.serverUpdateProvider]: "server", + [WS_METHODS.providerAuthStart]: "provider", + [WS_METHODS.providerConsumeResetCredit]: "provider", + [WS_METHODS.providerAuthComplete]: "provider", + [WS_METHODS.chatGptReconnectProfile]: "provider", + [WS_METHODS.chatGptImportProfile]: "provider", + [WS_METHODS.chatGptHandoffSubscribe]: "provider", + [WS_METHODS.codexAuthCallbackSubscribe]: "provider", + [WS_METHODS.providerAuthRespond]: "provider", + [WS_METHODS.providerAuthCancel]: "provider", + [WS_METHODS.providerAuthLogout]: "provider", + [WS_METHODS.providerAuthSubscribe]: "provider", + [WS_METHODS.providerInstallStart]: "provider", + [WS_METHODS.providerInstallCancel]: "provider", + [WS_METHODS.providerInstallSubscribe]: "provider", + [WS_METHODS.providerInstallRemove]: "provider", + [WS_METHODS.serverUpdateServer]: "server", + [WS_METHODS.serverUpdateServerWithProgress]: "server", + [WS_METHODS.serverCommitDesktopUpdate]: "server", + [WS_METHODS.serverUpsertKeybinding]: "server", + [WS_METHODS.serverRemoveKeybinding]: "server", + [WS_METHODS.serverGetSettings]: "server", + [WS_METHODS.serverUpdateSettings]: "server", + [WS_METHODS.serverSearchAcpRegistry]: "server", + [WS_METHODS.serverPrepareAcpRegistryAgent]: "server", + [WS_METHODS.serverUninstallAcpRegistryManagedBinary]: "server", + [WS_METHODS.serverAcceptAcpRegistryUrlAuth]: "server", + [WS_METHODS.serverListAcpRegistrySessions]: "server", + [WS_METHODS.serverImportAcpRegistrySession]: "server", + [WS_METHODS.serverDeleteAcpRegistrySession]: "server", + [WS_METHODS.serverListAcpRegistryProviders]: "server", + [WS_METHODS.serverSetAcpRegistryProvider]: "server", + [WS_METHODS.serverDisableAcpRegistryProvider]: "server", + [WS_METHODS.serverLogoutAcpRegistry]: "server", + [WS_METHODS.serverDiscoverSourceControl]: "server", + [WS_METHODS.serverGetTraceDiagnostics]: "server", + [WS_METHODS.serverGetProcessDiagnostics]: "server", + [WS_METHODS.serverGetHostResources]: "server", + [WS_METHODS.serverGetProcessResourceHistory]: "server", + [WS_METHODS.serverGetResourceTelemetryHistory]: "server", + [WS_METHODS.serverRetryResourceTelemetry]: "server", + [WS_METHODS.serverGetUsageSummary]: "server", + [WS_METHODS.serverRefreshUsageRates]: "server", + [WS_METHODS.serverSignalProcess]: "server", + [WS_METHODS.serverReportClientActivity]: "server", + [WS_METHODS.serverReportHostPowerState]: "server", + [WS_METHODS.serverGetBackgroundPolicy]: "server", + [WS_METHODS.scheduledTasksList]: "scheduledTasks", + [WS_METHODS.scheduledTasksSubscribe]: "scheduledTasks", + [WS_METHODS.scheduledTasksUpsert]: "scheduledTasks", + [WS_METHODS.scheduledTasksSetEnabled]: "scheduledTasks", + [WS_METHODS.scheduledTasksDelete]: "scheduledTasks", + [WS_METHODS.scheduledTasksRunNow]: "scheduledTasks", + [WS_METHODS.scheduledTasksRotateWebhookToken]: "scheduledTasks", + [WS_METHODS.scheduledTasksListWebhookDeliveries]: "scheduledTasks", + [WS_METHODS.scheduledTasksGetWebhookDelivery]: "scheduledTasks", + [WS_METHODS.secretsAnswerRequest]: "secrets", + [WS_METHODS.cloudGetRelayClientStatus]: "cloud", + [WS_METHODS.cloudInstallRelayClient]: "cloud", + [WS_METHODS.pullRequestsList]: "pull-requests", + [WS_METHODS.pullRequestsListStats]: "pull-requests", + [WS_METHODS.pullRequestsSummary]: "pull-requests", + [WS_METHODS.pullRequestsRouting]: "pull-requests", + [WS_METHODS.pullRequestsRoutingIdentity]: "pull-requests", + [WS_METHODS.pullRequestsStack]: "pull-requests", + [WS_METHODS.pullRequestsLinkedThreads]: "pull-requests", + [WS_METHODS.pullRequestsDetail]: "pull-requests", + [WS_METHODS.pullRequestsPreview]: "pull-requests", + [WS_METHODS.pullRequestsChecks]: "pull-requests", + [WS_METHODS.pullRequestsActivity]: "pull-requests", + [WS_METHODS.pullRequestsThreadComments]: "pull-requests", + [WS_METHODS.pullRequestsDiffFileContents]: "pull-requests", + [WS_METHODS.pullRequestsFilesViewed]: "pull-requests", + [WS_METHODS.pullRequestsRunAction]: "pull-requests", + [WS_METHODS.pullRequestsUpdate]: "pull-requests", + [WS_METHODS.pullRequestsComment]: "pull-requests", + [WS_METHODS.pullRequestsUpdateComment]: "pull-requests", + [WS_METHODS.pullRequestsSubmitReview]: "pull-requests", + [WS_METHODS.pullRequestsReplyToThread]: "pull-requests", + [WS_METHODS.pullRequestsSetThreadResolution]: "pull-requests", + [WS_METHODS.pullRequestsSetReaction]: "pull-requests", + [WS_METHODS.pullRequestsSetFilesViewed]: "pull-requests", + [WS_METHODS.pullRequestsInvalidate]: "pull-requests", + [WS_METHODS.pullRequestsSubscribeRefreshes]: "pull-requests", + [WS_METHODS.pullRequestsReviewerCandidates]: "pull-requests", + [WS_METHODS.pullRequestsRequestReviewers]: "pull-requests", + [WS_METHODS.pullRequestsLabelCandidates]: "pull-requests", + [WS_METHODS.pullRequestsSetLabels]: "pull-requests", + [WS_METHODS.sourceControlLookupRepository]: "source-control", + [WS_METHODS.sourceControlCloneRepository]: "source-control", + [WS_METHODS.sourceControlPublishRepository]: "source-control", + [WS_METHODS.projectCloneStart]: "source-control", + [WS_METHODS.projectCloneCancel]: "source-control", + [WS_METHODS.projectCloneRetry]: "source-control", + [WS_METHODS.subscribeProjectClones]: "source-control", + [WS_METHODS.projectsListEntries]: "workspace", + [WS_METHODS.projectsReadFile]: "workspace", + [WS_METHODS.projectsSearchContents]: "workspace", + [WS_METHODS.projectsSearchEntries]: "workspace", + [WS_METHODS.projectsWriteFile]: "workspace", + [WS_METHODS.projectsEnsureScratch]: "orchestration", + [WS_METHODS.projectsCreateNew]: "orchestration", + [WS_METHODS.shellOpenInEditor]: "workspace", + [WS_METHODS.filesystemBrowse]: "workspace", + [WS_METHODS.agentSessionsScan]: "workspace", + [WS_METHODS.agentSessionsImport]: "workspace", + [WS_METHODS.assetsCreateUrl]: "workspace", + [WS_METHODS.assetsPersistChatAttachments]: "orchestration", + [WS_METHODS.attachmentsCreateUploadUrl]: "workspace", + [WS_METHODS.attachmentsDelete]: "workspace", + [WS_METHODS.providerUploadFeedback]: "provider", + [WS_METHODS.subscribeVcsStatus]: "vcs", + [WS_METHODS.subscribeWorktreeSetup]: "vcs", + [WS_METHODS.worktreeSetupCancel]: "vcs", + [WS_METHODS.subscribeResourceTelemetry]: "server", + [WS_METHODS.vcsRefreshStatus]: "vcs", + [WS_METHODS.vcsPull]: "git", + [WS_METHODS.gitRunStackedAction]: "vcs", + [WS_METHODS.gitResolvePullRequest]: "git", + [WS_METHODS.gitPreparePullRequestThread]: "git", + [WS_METHODS.vcsListRefs]: "vcs", + [WS_METHODS.vcsCreateWorktree]: "vcs", + [WS_METHODS.vcsRemoveWorktree]: "vcs", + [WS_METHODS.vcsCreateRef]: "vcs", + [WS_METHODS.vcsSwitchRef]: "vcs", + [WS_METHODS.vcsInit]: "vcs", + [WS_METHODS.reviewGetDiffPreview]: "review", + [WS_METHODS.reviewGetDiffFileContents]: "review", + [WS_METHODS.terminalOpen]: "terminal", + [WS_METHODS.terminalAttach]: "terminal", + [WS_METHODS.terminalWrite]: "terminal", + [WS_METHODS.terminalResize]: "terminal", + [WS_METHODS.terminalClear]: "terminal", + [WS_METHODS.terminalRestart]: "terminal", + [WS_METHODS.terminalClose]: "terminal", + [WS_METHODS.subscribeTerminalEvents]: "terminal", + [WS_METHODS.subscribeTerminalMetadata]: "terminal", + [WS_METHODS.previewOpen]: "preview", + [WS_METHODS.previewNavigate]: "preview", + [WS_METHODS.previewResize]: "preview", + [WS_METHODS.previewAdjust]: "preview", + [WS_METHODS.previewRefresh]: "preview", + [WS_METHODS.previewClose]: "preview", + [WS_METHODS.previewList]: "preview", + [WS_METHODS.previewClearProfile]: "preview", + [WS_METHODS.previewReportStatus]: "preview", + [WS_METHODS.subscribePreviewEvents]: "preview", + [WS_METHODS.subscribeDiscoveredLocalServers]: "preview", + [WS_METHODS.deviceConfigure]: "device", + [WS_METHODS.deviceTestHost]: "device", + [WS_METHODS.deviceList]: "device", + [WS_METHODS.deviceOpen]: "device", + [WS_METHODS.deviceClose]: "device", + [WS_METHODS.deviceShutdown]: "device", + [WS_METHODS.deviceDetail]: "device", + [WS_METHODS.deviceAction]: "device", + [WS_METHODS.subscribeDeviceState]: "device", + [WS_METHODS.subscribeServerConfig]: "server", + [WS_METHODS.subscribeServerLifecycle]: "server", + [WS_METHODS.subscribeAuthAccess]: "auth", + [WS_METHODS.subscribeBackgroundPolicy]: "server", +} as const satisfies Readonly>; const RPC_SPAN_PREFIX = "ws.rpc"; const DEFAULT_RPC_SPAN_ATTRIBUTES = { @@ -22,117 +206,36 @@ const RPC_METHODS_WITH_TRACING_DISABLED: ReadonlySet = new Set([ WS_METHODS.serverSignalProcess, ]); -function shouldTraceRpc(method: string): boolean { - return !RPC_METHODS_WITH_TRACING_DISABLED.has(method); -} +/** + * Records each WebSocket RPC's span and request metrics. `ws.ts` adds it to the server's group + * after `RpcScopeAuthorization`, so it wraps authorization and also records rejected calls. + */ +export class RpcInstrumentation extends RpcMiddleware.Service()( + "t3/server/RpcInstrumentation", +) {} -const rpcSpanAttributes = ( - method: string, - traceAttributes?: Readonly>, -): Record => ({ - ...DEFAULT_RPC_SPAN_ATTRIBUTES, - "rpc.method": method, - ...traceAttributes, -}); +/** + * Wraps each WebSocket RPC call in its `ws.rpc.` span and records its request counter and + * duration. For a stream RPC, the middleware receives the whole stream run, so the span and the + * metrics cover the subscription until it ends, fails, or is interrupted. Methods in + * `RPC_METHODS_WITH_TRACING_DISABLED` record metrics but no spans, for the call or anything it runs. + */ +export const rpcInstrumentationLayer = Layer.succeed(RpcInstrumentation)((effect, { rpc }) => { + const method = rpc._tag; + const measured = effect.pipe( + withMetrics({ counter: rpcRequestsTotal, timer: rpcRequestDuration, attributes: { method } }), + ); -const withRpcEffectTracing = ( - method: string, - effect: Effect.Effect, - traceAttributes?: Readonly>, -): Effect.Effect => - shouldTraceRpc(method) - ? effect.pipe( - Effect.withSpan(`${RPC_SPAN_PREFIX}.${method}`, { - attributes: rpcSpanAttributes(method, traceAttributes), - }), - ) - : effect.pipe(Effect.provideService(References.TracerEnabled, false)); - -const withRpcStreamTracing = ( - method: string, - stream: Stream.Stream, - traceAttributes?: Readonly>, -): Stream.Stream => - shouldTraceRpc(method) - ? stream.pipe( - Stream.withSpan(`${RPC_SPAN_PREFIX}.${method}`, { - attributes: rpcSpanAttributes(method, traceAttributes), - }), - ) - : stream.pipe(Stream.provideService(References.TracerEnabled, false)); - -const recordRpcStreamMetrics = ( - method: string, - startedAt: bigint, - exit: Exit.Exit, -): Effect.Effect => - Effect.gen(function* () { - yield* Metric.update( - Metric.withAttributes(rpcRequestDuration, metricAttributes({ method })), - Duration.nanos((yield* Clock.monotonicTimeNanos) - startedAt), - ); - yield* Metric.update( - Metric.withAttributes( - rpcRequestsTotal, - metricAttributes({ - method, - outcome: outcomeFromExit(exit), - }), - ), - 1, - ); - }); - -export const observeRpcEffect = ( - method: string, - effect: Effect.Effect, - traceAttributes?: Readonly>, -): Effect.Effect => { - const instrumented = effect.pipe( - withMetrics({ - counter: rpcRequestsTotal, - timer: rpcRequestDuration, + if (RPC_METHODS_WITH_TRACING_DISABLED.has(method)) { + return measured.pipe(Effect.provideService(References.TracerEnabled, false)); + } + return measured.pipe( + Effect.withSpan(`${RPC_SPAN_PREFIX}.${method}`, { attributes: { - method, + ...DEFAULT_RPC_SPAN_ATTRIBUTES, + "rpc.method": method, + "rpc.aggregate": RPC_AGGREGATES[method as WsRpcMethod], }, }), ); - - return withRpcEffectTracing(method, instrumented, traceAttributes); -}; - -export const observeRpcStream = ( - method: string, - stream: Stream.Stream, - traceAttributes?: Readonly>, -): Stream.Stream => { - const instrumented = Stream.unwrap( - Effect.gen(function* () { - const startedAt = yield* Clock.monotonicTimeNanos; - return stream.pipe(Stream.onExit((exit) => recordRpcStreamMetrics(method, startedAt, exit))); - }), - ); - - return withRpcStreamTracing(method, instrumented, traceAttributes); -}; - -export const observeRpcStreamEffect = ( - method: string, - effect: Effect.Effect, EffectError, EffectContext>, - traceAttributes?: Readonly>, -): Stream.Stream => { - const instrumented = Stream.unwrap( - Effect.gen(function* () { - const startedAt = yield* Clock.monotonicTimeNanos; - // onError also runs when the stream is interrupted before it is produced. - const stream = yield* effect.pipe( - Effect.onError((cause) => recordRpcStreamMetrics(method, startedAt, Exit.failCause(cause))), - ); - return stream.pipe( - Stream.onExit((streamExit) => recordRpcStreamMetrics(method, startedAt, streamExit)), - ); - }), - ); - - return withRpcStreamTracing(method, instrumented, traceAttributes); -}; +}); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index dcf9fad568352fcbf333a6cee7e21811c33447dc..ad1b9815acc3d14e5658a9eef76d81bc3f37def6 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -159,11 +159,6 @@ import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; import { userFacingDispatchErrorMessage } from "./orchestration-v2/UserFacingErrors.ts"; -import { - observeRpcEffect, - observeRpcStream, - observeRpcStreamEffect, -} from "./observability/RpcInstrumentation.ts"; import * as ProviderRegistry from "./provider/ProviderRegistry.ts"; import * as ProviderInstanceRegistry from "./provider/ProviderInstanceRegistry.ts"; import * as AcpRegistrySupport from "./provider/acp/AcpRegistrySupport.ts"; @@ -215,6 +210,7 @@ import * as BackgroundPolicy from "./background/BackgroundPolicy.ts"; import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts"; import { requiredScopeForDeviceList, rpcAuthorizationError } from "./auth/RpcAuthorization.ts"; import * as RpcAuthorization from "./auth/RpcAuthorization.ts"; +import { RpcInstrumentation, rpcInstrumentationLayer } from "./observability/RpcInstrumentation.ts"; import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts"; import * as ProcessResourceMonitor from "./diagnostics/ProcessResourceMonitor.ts"; import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts"; @@ -541,7 +537,8 @@ function projectFileFailureContext( const PROVIDER_STATUS_DEBOUNCE_MS = 200; -const ServerWsRpcGroup = WsRpcGroup; +// Middleware added later wraps middleware added earlier, so instrumentation wraps authorization. +const ServerWsRpcGroup = WsRpcGroup.middleware(RpcInstrumentation); // When a resuming client's cursor is more than this many events behind the // current head, skip the per-event catch-up replay and send a fresh shell // snapshot instead. Replaying each intervening event costs a shell refetch; @@ -1808,851 +1805,595 @@ const layerWsRpc = ( const handlers = ServerWsRpcGroup.of({ [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command) => - observeRpcEffect( - ORCHESTRATION_V2_WS_METHODS.dispatchCommand, - startup - .enqueueCommand( - // A retry also restarts the preparation work the launch owns. - (command.type === "prepared-run.retry" - ? threadLaunch.retryPreparation(command) - : ThreadMessageIntake.dispatchCommand( - ThreadManagementService.withCreationProvenance(command, { - createdBy: "user", - creationSource: - "creationSource" in command ? command.creationSource : "web", - }), - ) - ).pipe(Effect.provide(intakeContext)), - ) - .pipe( - Effect.tap(() => recordClientCommandAnalytics(command)), - Effect.map((result) => ({ sequence: result.sequence })), - Effect.mapError((cause) => { - const detail = userFacingDispatchErrorMessage(cause); - return new OrchestrationV2DispatchCommandError({ - commandId: command.commandId, - commandType: command.type, - message: detail ?? "Failed to dispatch orchestration V2 command", - ...(detail === undefined ? {} : { detail }), - cause, - }); - }), - ), - { - "rpc.aggregate": "orchestrationV2", - "orchestration_v2.command_id": command.commandId, - "orchestration_v2.command_type": command.type, - "orchestration_v2.thread_id": - command.type === "thread.fork" || command.type === "thread.merge_back" - ? command.targetThreadId - : command.type === "delegated_task.request" || - command.type === "delegated_task.wake-policy" || - command.type === "delegated_task.completion-delivery.acknowledge" || - command.type === "delegated_task.completion-delivery.dispose" || - command.type === "thread.created.record" - ? command.parentThreadId - : command.threadId, - ...(command.type === "thread.fork" || command.type === "thread.merge_back" - ? { "orchestration_v2.source_thread_id": command.sourceThreadId } - : {}), - }, + Effect.annotateCurrentSpan({ + "orchestration_v2.command_id": command.commandId, + "orchestration_v2.command_type": command.type, + "orchestration_v2.thread_id": + command.type === "thread.fork" || command.type === "thread.merge_back" + ? command.targetThreadId + : command.type === "delegated_task.request" || + command.type === "delegated_task.wake-policy" || + command.type === "delegated_task.completion-delivery.acknowledge" || + command.type === "delegated_task.completion-delivery.dispose" || + command.type === "thread.created.record" + ? command.parentThreadId + : command.threadId, + ...(command.type === "thread.fork" || command.type === "thread.merge_back" + ? { "orchestration_v2.source_thread_id": command.sourceThreadId } + : {}), + }).pipe( + Effect.andThen( + startup + .enqueueCommand( + // A retry also restarts the preparation work the launch owns. + (command.type === "prepared-run.retry" + ? threadLaunch.retryPreparation(command) + : ThreadMessageIntake.dispatchCommand( + ThreadManagementService.withCreationProvenance(command, { + createdBy: "user", + creationSource: + "creationSource" in command ? command.creationSource : "web", + }), + ) + ).pipe(Effect.provide(intakeContext)), + ) + .pipe( + Effect.tap(() => recordClientCommandAnalytics(command)), + Effect.map((result) => ({ sequence: result.sequence })), + Effect.mapError((cause) => { + const detail = userFacingDispatchErrorMessage(cause); + return new OrchestrationV2DispatchCommandError({ + commandId: command.commandId, + commandType: command.type, + message: detail ?? "Failed to dispatch orchestration V2 command", + ...(detail === undefined ? {} : { detail }), + cause, + }); + }), + ), + ), ), [ORCHESTRATION_V2_WS_METHODS.getWorkflowScript]: (input) => - observeRpcEffect( - ORCHESTRATION_V2_WS_METHODS.getWorkflowScript, - readWorkflowScript({ scriptPath: input.scriptPath }), - { "rpc.aggregate": "orchestration" }, - ), + readWorkflowScript({ scriptPath: input.scriptPath }), [ORCHESTRATION_V2_WS_METHODS.getTurnItem]: (input) => - observeRpcEffect( - ORCHESTRATION_V2_WS_METHODS.getTurnItem, - threadManagement.getTurnItem(input).pipe( - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: "Failed to load turn item", - cause, - }), - ), + threadManagement.getTurnItem(input).pipe( + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: "Failed to load turn item", + cause, + }), ), - { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_V2_WS_METHODS.getTurnDiff]: (input) => - observeRpcEffect( - ORCHESTRATION_V2_WS_METHODS.getTurnDiff, - checkpointDiffQuery.getTurnDiff(input).pipe( - Effect.mapError( - (cause) => - new OrchestrationGetTurnDiffError({ - message: "Failed to load turn diff", - cause, - }), - ), + checkpointDiffQuery.getTurnDiff(input).pipe( + Effect.mapError( + (cause) => + new OrchestrationGetTurnDiffError({ + message: "Failed to load turn diff", + cause, + }), ), - { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_V2_WS_METHODS.getFullThreadDiff]: (input) => - observeRpcEffect( - ORCHESTRATION_V2_WS_METHODS.getFullThreadDiff, - checkpointDiffQuery.getFullThreadDiff(input).pipe( - Effect.mapError( - (cause) => - new OrchestrationGetFullThreadDiffError({ - message: "Failed to load full thread diff", - cause, - }), - ), + checkpointDiffQuery.getFullThreadDiff(input).pipe( + Effect.mapError( + (cause) => + new OrchestrationGetFullThreadDiffError({ + message: "Failed to load full thread diff", + cause, + }), ), - { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_V2_WS_METHODS.searchThreads]: (input) => - observeRpcEffect( - ORCHESTRATION_V2_WS_METHODS.searchThreads, - threadSearch.search(input).pipe( - Effect.mapError( - (cause) => - new OrchestrationSearchThreadsError({ - message: "Failed to search threads", - cause, - }), - ), + threadSearch.search(input).pipe( + Effect.mapError( + (cause) => + new OrchestrationSearchThreadsError({ + message: "Failed to search threads", + cause, + }), ), - { "rpc.aggregate": "orchestration" }, ), [ORCHESTRATION_V2_WS_METHODS.getArchivedShellSnapshot]: (_input) => - observeRpcEffect( - ORCHESTRATION_V2_WS_METHODS.getArchivedShellSnapshot, - getOrchestrationV2ArchivedShellSnapshot, - { "rpc.aggregate": "orchestration" }, - ), + getOrchestrationV2ArchivedShellSnapshot, [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: (input) => - observeRpcEffect( - ORCHESTRATION_V2_WS_METHODS.getThreadProjection, - // Pre-pagination clients still call this compatibility endpoint. - // Keep stale clients from materializing an unbounded transcript. - threadManagement - .getThreadSnapshotWindow(input.threadId, { - rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, - }) - .pipe( - Effect.map((snapshot) => - projectThreadProjectionForWire( - buildBoundedThreadProjection({ - projection: snapshot.projection, - snapshotSequence: snapshot.snapshotSequence, - }).projection, + Effect.annotateCurrentSpan({ "orchestration_v2.thread_id": input.threadId }).pipe( + Effect.andThen( + // Pre-pagination clients still call this compatibility endpoint. + // Keep stale clients from materializing an unbounded transcript. + threadManagement + .getThreadSnapshotWindow(input.threadId, { + rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT, + }) + .pipe( + Effect.map((snapshot) => + projectThreadProjectionForWire( + buildBoundedThreadProjection({ + projection: snapshot.projection, + snapshotSequence: snapshot.snapshotSequence, + }).projection, + ), + ), + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: `Failed to load orchestration V2 thread ${input.threadId}`, + cause, + }), ), ), - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: `Failed to load orchestration V2 thread ${input.threadId}`, - cause, - }), - ), - ), - { - "rpc.aggregate": "orchestrationV2", - "orchestration_v2.thread_id": input.threadId, - }, + ), ), [ORCHESTRATION_V2_WS_METHODS.launchThread]: (input) => - observeRpcEffect( - ORCHESTRATION_V2_WS_METHODS.launchThread, - startup - .enqueueCommand( - ThreadMessageIntake.launchThread({ - commandId: input.commandId, - ...(input.threadId === undefined ? {} : { threadId: input.threadId }), - ...(input.reuseExistingThread === undefined - ? {} - : { reuseExistingThread: input.reuseExistingThread }), - projectId: input.projectId, - title: input.title, - ...(input.generateTitle === undefined - ? {} - : { generateTitle: input.generateTitle }), - modelSelection: input.modelSelection, - runtimeMode: input.runtimeMode, - interactionMode: input.interactionMode, - workspaceStrategy: input.workspaceStrategy, - ...(input.initialMessage === undefined - ? {} - : { - initialMessage: { - ...(input.initialMessage.messageId === undefined - ? {} - : { messageId: input.initialMessage.messageId }), - text: input.initialMessage.text, - attachments: input.initialMessage.attachments, - ...(input.initialMessage.context === undefined - ? {} - : { context: input.initialMessage.context }), - }, - }), - createdBy: "user", - creationSource: input.creationSource ?? "web", - }).pipe(Effect.provide(intakeContext)), - ) - .pipe( - Effect.tap(() => - analytics - .record("client.thread.started", originProps) - .pipe( - Effect.andThen( - input.initialMessage === undefined - ? Effect.void - : analytics.record("client.turn.requested", originProps), + Effect.annotateCurrentSpan({ + "orchestration_v2.command_id": input.commandId, + "orchestration_v2.project_id": input.projectId, + }).pipe( + Effect.andThen( + startup + .enqueueCommand( + ThreadMessageIntake.launchThread({ + commandId: input.commandId, + ...(input.threadId === undefined ? {} : { threadId: input.threadId }), + ...(input.reuseExistingThread === undefined + ? {} + : { reuseExistingThread: input.reuseExistingThread }), + projectId: input.projectId, + title: input.title, + ...(input.generateTitle === undefined + ? {} + : { generateTitle: input.generateTitle }), + modelSelection: input.modelSelection, + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + workspaceStrategy: input.workspaceStrategy, + ...(input.initialMessage === undefined + ? {} + : { + initialMessage: { + ...(input.initialMessage.messageId === undefined + ? {} + : { messageId: input.initialMessage.messageId }), + text: input.initialMessage.text, + attachments: input.initialMessage.attachments, + ...(input.initialMessage.context === undefined + ? {} + : { context: input.initialMessage.context }), + }, + }), + createdBy: "user", + creationSource: input.creationSource ?? "web", + }).pipe(Effect.provide(intakeContext)), + ) + .pipe( + Effect.tap(() => + analytics + .record("client.thread.started", originProps) + .pipe( + Effect.andThen( + input.initialMessage === undefined + ? Effect.void + : analytics.record("client.turn.requested", originProps), + ), + Effect.ignore, ), - Effect.ignore, - ), + ), + Effect.map((result) => ({ + ...result, + projection: projectThreadProjectionForWire(result.projection), + })), + Effect.catchTags({ + AttachmentClaimError: (cause) => + new OrchestrationV2ThreadLaunchError({ + commandId: input.commandId, + projectId: input.projectId, + message: cause.message, + cause, + }), + ThreadLaunchError: (cause) => + new OrchestrationV2ThreadLaunchError({ + commandId: input.commandId, + projectId: input.projectId, + message: "Failed to launch thread", + cause, + }), + ServerRuntimeStartupError: (cause) => + new OrchestrationV2ThreadLaunchError({ + commandId: input.commandId, + projectId: input.projectId, + message: "Failed to launch thread", + cause, + }), + }), ), - Effect.map((result) => ({ - ...result, - projection: projectThreadProjectionForWire(result.projection), - })), - Effect.catchTags({ - AttachmentClaimError: (cause) => - new OrchestrationV2ThreadLaunchError({ - commandId: input.commandId, - projectId: input.projectId, - message: cause.message, - cause, - }), - ThreadLaunchError: (cause) => - new OrchestrationV2ThreadLaunchError({ - commandId: input.commandId, - projectId: input.projectId, - message: "Failed to launch thread", - cause, - }), - ServerRuntimeStartupError: (cause) => - new OrchestrationV2ThreadLaunchError({ - commandId: input.commandId, - projectId: input.projectId, - message: "Failed to launch thread", - cause, - }), - }), - ), - { - "rpc.aggregate": "orchestration", - "orchestration_v2.command_id": input.commandId, - "orchestration_v2.project_id": input.projectId, - }, + ), ), [ORCHESTRATION_V2_WS_METHODS.subscribeArchivedShell]: (_input) => - observeRpcStreamEffect( - ORCHESTRATION_V2_WS_METHODS.subscribeArchivedShell, - subscribeOrchestrationV2ArchivedShell(), - { "rpc.aggregate": "orchestration" }, - ), + Stream.unwrap(subscribeOrchestrationV2ArchivedShell()), [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input) => - observeRpcStreamEffect( - ORCHESTRATION_V2_WS_METHODS.subscribeShell, - subscribeOrchestrationV2Shell(input), - { - "rpc.aggregate": "orchestrationV2", - }, - ), + Stream.unwrap(subscribeOrchestrationV2Shell(input)), [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input) => - observeRpcStreamEffect( - ORCHESTRATION_V2_WS_METHODS.subscribeThread, - subscribeOrchestrationV2Thread(input), - { - "rpc.aggregate": "orchestrationV2", - "orchestration_v2.thread_id": input.threadId, - }, + Stream.unwrap( + Effect.annotateCurrentSpan({ "orchestration_v2.thread_id": input.threadId }).pipe( + Effect.andThen(subscribeOrchestrationV2Thread(input)), + ), ), [WS_METHODS.scheduledTasksList]: (_input) => - observeRpcEffect( - WS_METHODS.scheduledTasksList, - scheduledTasks.list().pipe(Effect.map(withVisibleWebhookUrls)), - { "rpc.aggregate": "scheduledTasks" }, - ), + scheduledTasks.list().pipe(Effect.map(withVisibleWebhookUrls)), [WS_METHODS.scheduledTasksSubscribe]: (_input) => - observeRpcStream( - WS_METHODS.scheduledTasksSubscribe, - scheduledTasks.subscribeList().pipe(Stream.map(withVisibleWebhookUrls)), - { "rpc.aggregate": "scheduledTasks" }, - ), - [WS_METHODS.scheduledTasksUpsert]: (input) => - observeRpcEffect(WS_METHODS.scheduledTasksUpsert, scheduledTasks.upsert(input), { - "rpc.aggregate": "scheduledTasks", - }), + scheduledTasks.subscribeList().pipe(Stream.map(withVisibleWebhookUrls)), + [WS_METHODS.scheduledTasksUpsert]: (input) => scheduledTasks.upsert(input), [WS_METHODS.scheduledTasksSetEnabled]: (input) => - observeRpcEffect(WS_METHODS.scheduledTasksSetEnabled, scheduledTasks.setEnabled(input), { - "rpc.aggregate": "scheduledTasks", - "scheduled_task.id": input.id, - }), + Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( + Effect.andThen(scheduledTasks.setEnabled(input)), + ), [WS_METHODS.scheduledTasksDelete]: (input) => - observeRpcEffect(WS_METHODS.scheduledTasksDelete, scheduledTasks.delete(input), { - "rpc.aggregate": "scheduledTasks", - "scheduled_task.id": input.id, - }), + Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( + Effect.andThen(scheduledTasks.delete(input)), + ), [WS_METHODS.scheduledTasksRunNow]: (input) => - observeRpcEffect(WS_METHODS.scheduledTasksRunNow, scheduledTasks.runNow(input), { - "rpc.aggregate": "scheduledTasks", - "scheduled_task.id": input.id, - }), + Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( + Effect.andThen(scheduledTasks.runNow(input)), + ), [WS_METHODS.scheduledTasksRotateWebhookToken]: (input) => - observeRpcEffect( - WS_METHODS.scheduledTasksRotateWebhookToken, - scheduledTasks.rotateWebhookToken(input), - { "rpc.aggregate": "scheduledTasks", "scheduled_task.id": input.id }, + Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( + Effect.andThen(scheduledTasks.rotateWebhookToken(input)), ), [WS_METHODS.secretsAnswerRequest]: (input) => - observeRpcEffect(WS_METHODS.secretsAnswerRequest, secretRequests.answer(input), { - "rpc.aggregate": "secrets", - "orchestration_v2.thread_id": input.threadId, - }), + Effect.annotateCurrentSpan({ "orchestration_v2.thread_id": input.threadId }).pipe( + Effect.andThen(secretRequests.answer(input)), + ), [WS_METHODS.scheduledTasksListWebhookDeliveries]: (input) => - observeRpcEffect( - WS_METHODS.scheduledTasksListWebhookDeliveries, - scheduledTasks.listWebhookDeliveries(input), - { "rpc.aggregate": "scheduledTasks", "scheduled_task.id": input.id }, + Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( + Effect.andThen(scheduledTasks.listWebhookDeliveries(input)), ), [WS_METHODS.scheduledTasksGetWebhookDelivery]: (input) => - observeRpcEffect( - WS_METHODS.scheduledTasksGetWebhookDelivery, - scheduledTasks.getWebhookDelivery(input), - { "rpc.aggregate": "scheduledTasks", "scheduled_task.id": input.id }, - ), - [WS_METHODS.serverProbe]: (_input) => - observeRpcEffect(WS_METHODS.serverProbe, Effect.succeed({}), { - "rpc.aggregate": "server", - }), - [WS_METHODS.serverGetConfig]: (_input) => - observeRpcEffect( - WS_METHODS.serverGetConfig, - loadServerConfig({ usageLimitsCommand: false }), - { - "rpc.aggregate": "server", - }, + Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( + Effect.andThen(scheduledTasks.getWebhookDelivery(input)), ), + [WS_METHODS.serverProbe]: (_input) => Effect.succeed({}), + [WS_METHODS.serverGetConfig]: (_input) => loadServerConfig({ usageLimitsCommand: false }), [WS_METHODS.serverSearchAcpRegistry]: (input) => - observeRpcEffect( - WS_METHODS.serverSearchAcpRegistry, - acpRegistryCatalog - .search(input) - .pipe(Effect.mapError(AcpRegistrySupport.toAcpRegistryOperationError)), - { "rpc.aggregate": "server" }, - ), + acpRegistryCatalog + .search(input) + .pipe(Effect.mapError(AcpRegistrySupport.toAcpRegistryOperationError)), [WS_METHODS.serverPrepareAcpRegistryAgent]: (input) => - observeRpcEffect( - WS_METHODS.serverPrepareAcpRegistryAgent, - acpRegistryCatalog - .prepare(input) - .pipe(Effect.mapError(AcpRegistrySupport.toAcpRegistryOperationError)), - { - "rpc.aggregate": "server", - "acp_registry.agent_id": input.agentId, - }, + Effect.annotateCurrentSpan({ "acp_registry.agent_id": input.agentId }).pipe( + Effect.andThen( + acpRegistryCatalog + .prepare(input) + .pipe(Effect.mapError(AcpRegistrySupport.toAcpRegistryOperationError)), + ), ), [WS_METHODS.serverUninstallAcpRegistryManagedBinary]: (input) => - observeRpcEffect( - WS_METHODS.serverUninstallAcpRegistryManagedBinary, - acpRegistryCatalog - .uninstallManagedBinary(input) - .pipe(Effect.mapError(AcpRegistrySupport.toAcpRegistryOperationError)), - { - "rpc.aggregate": "server", - "acp_registry.agent_id": input.agentId, - }, + Effect.annotateCurrentSpan({ "acp_registry.agent_id": input.agentId }).pipe( + Effect.andThen( + acpRegistryCatalog + .uninstallManagedBinary(input) + .pipe(Effect.mapError(AcpRegistrySupport.toAcpRegistryOperationError)), + ), ), [WS_METHODS.serverAcceptAcpRegistryUrlAuth]: (input) => - observeRpcEffect( - WS_METHODS.serverAcceptAcpRegistryUrlAuth, - acpRegistryRuntimeCoordinator - .acceptUrlAuthentication(input) - .pipe(Effect.map((accepted) => ({ accepted }))), - { - "rpc.aggregate": "server", - "provider.instance_id": input.instanceId, - }, + Effect.annotateCurrentSpan({ "provider.instance_id": input.instanceId }).pipe( + Effect.andThen( + acpRegistryRuntimeCoordinator + .acceptUrlAuthentication(input) + .pipe(Effect.map((accepted) => ({ accepted }))), + ), ), [WS_METHODS.serverListAcpRegistrySessions]: (input) => - observeRpcEffect( - WS_METHODS.serverListAcpRegistrySessions, - listAcpRegistrySessions(input), - { - "rpc.aggregate": "server", - "provider.instance_id": input.instanceId, - "project.id": input.projectId, - }, - ), + Effect.annotateCurrentSpan({ + "provider.instance_id": input.instanceId, + "project.id": input.projectId, + }).pipe(Effect.andThen(listAcpRegistrySessions(input))), [WS_METHODS.serverImportAcpRegistrySession]: (input) => - observeRpcEffect( - WS_METHODS.serverImportAcpRegistrySession, - importAcpRegistrySession(input), - { - "rpc.aggregate": "server", - "provider.instance_id": input.instanceId, - "project.id": input.projectId, - }, - ), + Effect.annotateCurrentSpan({ + "provider.instance_id": input.instanceId, + "project.id": input.projectId, + }).pipe(Effect.andThen(importAcpRegistrySession(input))), [WS_METHODS.serverDeleteAcpRegistrySession]: (input) => - observeRpcEffect( - WS_METHODS.serverDeleteAcpRegistrySession, - deleteAcpRegistrySession(input), - { - "rpc.aggregate": "server", - "provider.instance_id": input.instanceId, - "project.id": input.projectId, - }, - ), + Effect.annotateCurrentSpan({ + "provider.instance_id": input.instanceId, + "project.id": input.projectId, + }).pipe(Effect.andThen(deleteAcpRegistrySession(input))), [WS_METHODS.serverListAcpRegistryProviders]: (input) => - observeRpcEffect( - WS_METHODS.serverListAcpRegistryProviders, - listAcpRegistryProviders(input), - { - "rpc.aggregate": "server", - "provider.instance_id": input.instanceId, - "project.id": input.projectId, - }, - ), + Effect.annotateCurrentSpan({ + "provider.instance_id": input.instanceId, + "project.id": input.projectId, + }).pipe(Effect.andThen(listAcpRegistryProviders(input))), [WS_METHODS.serverSetAcpRegistryProvider]: (input) => - observeRpcEffect(WS_METHODS.serverSetAcpRegistryProvider, setAcpRegistryProvider(input), { - "rpc.aggregate": "server", + Effect.annotateCurrentSpan({ "provider.instance_id": input.instanceId, "project.id": input.projectId, - }), + }).pipe(Effect.andThen(setAcpRegistryProvider(input))), [WS_METHODS.serverDisableAcpRegistryProvider]: (input) => - observeRpcEffect( - WS_METHODS.serverDisableAcpRegistryProvider, - disableAcpRegistryProvider(input), - { - "rpc.aggregate": "server", - "provider.instance_id": input.instanceId, - "project.id": input.projectId, - }, - ), + Effect.annotateCurrentSpan({ + "provider.instance_id": input.instanceId, + "project.id": input.projectId, + }).pipe(Effect.andThen(disableAcpRegistryProvider(input))), [WS_METHODS.serverLogoutAcpRegistry]: (input) => - observeRpcEffect( - WS_METHODS.serverLogoutAcpRegistry, - Effect.gen(function* () { - const { instance, manager } = yield* acpSessionManager(input.instanceId); - const snapshot = yield* instance.snapshot.getSnapshot; - if (snapshot.auth.canLogout !== true) { - return yield* new AcpRegistryOperationError({ - reason: "logout_unsupported", - message: "The ACP agent does not advertise logout.", - }); - } - if (instance.auth) { - yield* providerAuth.logout(input).pipe( - Effect.mapError( - (cause) => - new AcpRegistryOperationError({ - reason: "logout_failed", - message: "Could not sign out of the ACP agent.", - cause, - }), - ), - ); - } else { - yield* providerSessionManager.closeInstance(input.instanceId).pipe( - Effect.mapError( - (cause) => - new AcpRegistryOperationError({ - reason: "logout_failed", - message: "Could not stop live sessions before ACP logout.", - cause, - }), - ), - ); - yield* manager.logout(config.cwd); - } - yield* providerRegistry.refreshInstance(input.instanceId); - return { loggedOut: true } as const; - }), - { - "rpc.aggregate": "server", - "provider.instance_id": input.instanceId, - }, + Effect.annotateCurrentSpan({ "provider.instance_id": input.instanceId }).pipe( + Effect.andThen( + Effect.gen(function* () { + const { instance, manager } = yield* acpSessionManager(input.instanceId); + const snapshot = yield* instance.snapshot.getSnapshot; + if (snapshot.auth.canLogout !== true) { + return yield* new AcpRegistryOperationError({ + reason: "logout_unsupported", + message: "The ACP agent does not advertise logout.", + }); + } + if (instance.auth) { + yield* providerAuth.logout(input).pipe( + Effect.mapError( + (cause) => + new AcpRegistryOperationError({ + reason: "logout_failed", + message: "Could not sign out of the ACP agent.", + cause, + }), + ), + ); + } else { + yield* providerSessionManager.closeInstance(input.instanceId).pipe( + Effect.mapError( + (cause) => + new AcpRegistryOperationError({ + reason: "logout_failed", + message: "Could not stop live sessions before ACP logout.", + cause, + }), + ), + ); + yield* manager.logout(config.cwd); + } + yield* providerRegistry.refreshInstance(input.instanceId); + return { loggedOut: true } as const; + }), + ), ), [WS_METHODS.serverRefreshProviders]: (input) => - observeRpcEffect( - WS_METHODS.serverRefreshProviders, - Effect.gen(function* () { - // Only explicit catalog refreshes bypass T3's caches. Workspace - // discovery and background status checks retain their timers. - if (input.refreshModels) { - yield* modelManifest.forceRefresh; - const instances = yield* providerInstances.listInstances; - yield* Effect.forEach( - instances.filter( - (instance) => - input.instanceId === undefined || input.instanceId === instance.instanceId, - ), + Effect.gen(function* () { + // Only explicit catalog refreshes bypass T3's caches. Workspace + // discovery and background status checks retain their timers. + if (input.refreshModels) { + yield* modelManifest.forceRefresh; + const instances = yield* providerInstances.listInstances; + yield* Effect.forEach( + instances.filter( (instance) => - Effect.gen(function* () { - yield* instance.invalidateCaches ?? Effect.void; - const maintenance = yield* instance.snapshot.resolveMaintenance({ - fresh: true, - }); - if (maintenance.packageName) - providerVersionCache.delete(maintenance.packageName); - }), - { concurrency: "unbounded", discard: true }, - ); - } - // An untargeted refresh is "re-read everything's status", which - // includes quota from configured usage-limit sources. Awaited, - // not forked: the RPC scope closes on return and would - // interrupt a fork before the hub answered. - if (input.instanceId === undefined) { - yield* usageLimitSources.refresh; - } - let providers = yield* input.cwd !== undefined && input.instanceId !== undefined - ? providerRegistry.refreshWorkspaceSnapshot({ - instanceId: input.instanceId, - cwd: input.cwd, - fresh: input.fresh === true, - }) - : input.instanceId !== undefined - ? providerRegistry.refreshInstance(input.instanceId) - : providerRegistry.refresh(); - if (input.refreshModels) { - const instances = yield* providerInstances.listInstances; - for (const instance of instances) { - if ( - !instance.refreshModels || - (input.instanceId !== undefined && input.instanceId !== instance.instanceId) || - !providers.some( - (provider) => - provider.instanceId === instance.instanceId && - provider.enabled && - provider.installed, - ) + input.instanceId === undefined || input.instanceId === instance.instanceId, + ), + (instance) => + Effect.gen(function* () { + yield* instance.invalidateCaches ?? Effect.void; + const maintenance = yield* instance.snapshot.resolveMaintenance({ + fresh: true, + }); + if (maintenance.packageName) + providerVersionCache.delete(maintenance.packageName); + }), + { concurrency: "unbounded", discard: true }, + ); + } + // An untargeted refresh is "re-read everything's status", which + // includes quota from configured usage-limit sources. Awaited, + // not forked: the RPC scope closes on return and would + // interrupt a fork before the hub answered. + if (input.instanceId === undefined) { + yield* usageLimitSources.refresh; + } + let providers = yield* input.cwd !== undefined && input.instanceId !== undefined + ? providerRegistry.refreshWorkspaceSnapshot({ + instanceId: input.instanceId, + cwd: input.cwd, + fresh: input.fresh === true, + }) + : input.instanceId !== undefined + ? providerRegistry.refreshInstance(input.instanceId) + : providerRegistry.refresh(); + if (input.refreshModels) { + const instances = yield* providerInstances.listInstances; + for (const instance of instances) { + if ( + !instance.refreshModels || + (input.instanceId !== undefined && input.instanceId !== instance.instanceId) || + !providers.some( + (provider) => + provider.instanceId === instance.instanceId && + provider.enabled && + provider.installed, ) - continue; - yield* instance.refreshModels().pipe( - Effect.mapError( - (error) => - new ProviderSetupError({ - instanceId: instance.instanceId, - operation: "refresh-models", - detail: error.detail, - }), - ), - ); - providers = yield* providerRegistry.refreshInstance(instance.instanceId); - } + ) + continue; + yield* instance.refreshModels().pipe( + Effect.mapError( + (error) => + new ProviderSetupError({ + instanceId: instance.instanceId, + operation: "refresh-models", + detail: error.detail, + }), + ), + ); + providers = yield* providerRegistry.refreshInstance(instance.instanceId); } - return { providers }; - }), - { "rpc.aggregate": "server" }, - ), + } + return { providers }; + }), [WS_METHODS.providerUploadFeedback]: (input) => - observeRpcEffect( - WS_METHODS.providerUploadFeedback, - Effect.gen(function* () { - const projection = yield* threadManagement.getThreadRecords(input.threadId, [ - "providerThreads", - ]); - const providerThread = - projection.providerThreads.find( - (candidate) => candidate.id === projection.thread.activeProviderThreadId, - ) ?? projection.providerThreads.at(-1); - const providerSessionId = providerThread?.providerSessionId ?? null; - if (providerThread === undefined || providerSessionId === null) { - return yield* Effect.fail( - new ProviderUploadFeedbackError({ + Effect.gen(function* () { + const projection = yield* threadManagement.getThreadRecords(input.threadId, [ + "providerThreads", + ]); + const providerThread = + projection.providerThreads.find( + (candidate) => candidate.id === projection.thread.activeProviderThreadId, + ) ?? projection.providerThreads.at(-1); + const providerSessionId = providerThread?.providerSessionId ?? null; + if (providerThread === undefined || providerSessionId === null) { + return yield* Effect.fail( + new ProviderUploadFeedbackError({ + threadId: input.threadId, + cause: "No provider session has run in this thread yet.", + }), + ); + } + const runtime = Option.getOrNull(yield* providerSessionsV2.get(providerSessionId)); + if (runtime === null) { + return yield* Effect.fail( + new ProviderUploadFeedbackError({ + threadId: input.threadId, + cause: "The provider session is no longer running. Send a message first.", + }), + ); + } + if (runtime.uploadFeedback === undefined) { + return yield* Effect.fail( + new ProviderUploadFeedbackError({ + threadId: input.threadId, + cause: `Provider '${runtime.driver}' does not support feedback uploads.`, + }), + ); + } + return yield* runtime.uploadFeedback({ + providerThread, + ...(input.reason === undefined ? {} : { reason: input.reason }), + }); + }).pipe( + Effect.mapError((cause) => + isProviderUploadFeedbackError(cause) + ? cause + : new ProviderUploadFeedbackError({ threadId: input.threadId, - cause: "No provider session has run in this thread yet.", + cause, }), - ); - } - const runtime = Option.getOrNull(yield* providerSessionsV2.get(providerSessionId)); - if (runtime === null) { - return yield* Effect.fail( - new ProviderUploadFeedbackError({ - threadId: input.threadId, - cause: "The provider session is no longer running. Send a message first.", - }), - ); - } - if (runtime.uploadFeedback === undefined) { - return yield* Effect.fail( - new ProviderUploadFeedbackError({ - threadId: input.threadId, - cause: `Provider '${runtime.driver}' does not support feedback uploads.`, - }), - ); - } - return yield* runtime.uploadFeedback({ - providerThread, - ...(input.reason === undefined ? {} : { reason: input.reason }), - }); - }).pipe( - Effect.mapError((cause) => - isProviderUploadFeedbackError(cause) - ? cause - : new ProviderUploadFeedbackError({ - threadId: input.threadId, - cause, - }), - ), ), - { "rpc.aggregate": "provider" }, ), [WS_METHODS.serverUpdateProvider]: (input) => - observeRpcEffect( - WS_METHODS.serverUpdateProvider, - providerMaintenanceRunner.updateProvider(input), - { - "rpc.aggregate": "server", - }, - ), + providerMaintenanceRunner.updateProvider(input), [WS_METHODS.providerConsumeResetCredit]: (input) => - observeRpcEffect( - WS_METHODS.providerConsumeResetCredit, - Effect.gen(function* () { - if ("sourceId" in input) return yield* usageLimitSources.consumeResetCredit(input); - const instance = yield* providerInstances.getInstance(input.instanceId); - // A disabled instance must not spend anything on its account. - if (instance === undefined || !instance.enabled) { - return yield* new ProviderSetupError({ - instanceId: input.instanceId, - operation: "consume-reset-credit", - detail: instance ? "This provider is disabled." : "Provider instance not found.", - }); - } - if (instance.consumeResetCredit === undefined) { - return yield* new ProviderSetupError({ - instanceId: input.instanceId, - operation: "consume-reset-credit", - detail: "This provider does not bank reset credits.", - }); - } - const outcome = yield* instance.consumeResetCredit().pipe( - Effect.mapError( - (error) => - new ProviderSetupError({ - instanceId: input.instanceId, - operation: "consume-reset-credit", - detail: error.detail, - cause: error, - }), - ), - ); - return { outcome }; - }), - { "rpc.aggregate": "provider" }, - ), - [WS_METHODS.providerAuthStart]: (input) => - observeRpcEffect( - WS_METHODS.providerAuthStart, - providerAuth.start(input, currentSessionId), - { "rpc.aggregate": "provider" }, - ), + Effect.gen(function* () { + if ("sourceId" in input) return yield* usageLimitSources.consumeResetCredit(input); + const instance = yield* providerInstances.getInstance(input.instanceId); + // A disabled instance must not spend anything on its account. + if (instance === undefined || !instance.enabled) { + return yield* new ProviderSetupError({ + instanceId: input.instanceId, + operation: "consume-reset-credit", + detail: instance ? "This provider is disabled." : "Provider instance not found.", + }); + } + if (instance.consumeResetCredit === undefined) { + return yield* new ProviderSetupError({ + instanceId: input.instanceId, + operation: "consume-reset-credit", + detail: "This provider does not bank reset credits.", + }); + } + const outcome = yield* instance.consumeResetCredit().pipe( + Effect.mapError( + (error) => + new ProviderSetupError({ + instanceId: input.instanceId, + operation: "consume-reset-credit", + detail: error.detail, + cause: error, + }), + ), + ); + return { outcome }; + }), + [WS_METHODS.providerAuthStart]: (input) => providerAuth.start(input, currentSessionId), [WS_METHODS.providerAuthRespond]: (input) => - observeRpcEffect( - WS_METHODS.providerAuthRespond, - providerAuth.respond(input, currentSessionId), - { - "rpc.aggregate": "provider", - instanceId: input.instanceId, - }, + Effect.annotateCurrentSpan({ instanceId: input.instanceId }).pipe( + Effect.andThen(providerAuth.respond(input, currentSessionId)), ), [WS_METHODS.providerAuthComplete]: (input) => - observeRpcEffect( - WS_METHODS.providerAuthComplete, - providerAuth.complete(input, currentSessionId), - { "rpc.aggregate": "provider" }, - ), + providerAuth.complete(input, currentSessionId), [WS_METHODS.chatGptReconnectProfile]: (input) => providerAuth.reconnectProfile(input), [WS_METHODS.chatGptImportProfile]: (input) => providerAuth.importProfile(input), [WS_METHODS.chatGptHandoffSubscribe]: (input) => subscribeChatGptHandoff(input, currentSessionId), - [WS_METHODS.codexAuthCallbackSubscribe]: (input) => - observeRpcStream( - WS_METHODS.codexAuthCallbackSubscribe, - subscribeCodexAuthCallback(input), - { - "rpc.aggregate": "provider", - }, - ), - [WS_METHODS.providerAuthCancel]: (input) => - observeRpcEffect( - WS_METHODS.providerAuthCancel, - providerAuth.cancel(input, currentSessionId), - { "rpc.aggregate": "provider" }, - ), - [WS_METHODS.providerAuthLogout]: (input) => - observeRpcEffect(WS_METHODS.providerAuthLogout, providerAuth.logout(input), { - "rpc.aggregate": "provider", - }), + [WS_METHODS.codexAuthCallbackSubscribe]: (input) => subscribeCodexAuthCallback(input), + [WS_METHODS.providerAuthCancel]: (input) => providerAuth.cancel(input, currentSessionId), + [WS_METHODS.providerAuthLogout]: (input) => providerAuth.logout(input), [WS_METHODS.providerAuthSubscribe]: (input) => - observeRpcStream( - WS_METHODS.providerAuthSubscribe, - providerAuth.subscribe(input, currentSessionId), - { "rpc.aggregate": "provider" }, - ), - [WS_METHODS.providerInstallStart]: (input) => - observeRpcEffect(WS_METHODS.providerInstallStart, providerInstallation.start(input), { - "rpc.aggregate": "provider", - }), - [WS_METHODS.providerInstallCancel]: (input) => - observeRpcEffect(WS_METHODS.providerInstallCancel, providerInstallation.cancel(input), { - "rpc.aggregate": "provider", - }), - [WS_METHODS.providerInstallSubscribe]: (input) => - observeRpcStream( - WS_METHODS.providerInstallSubscribe, - providerInstallation.subscribe(input), - { "rpc.aggregate": "provider" }, - ), - [WS_METHODS.providerInstallRemove]: (input) => - observeRpcEffect(WS_METHODS.providerInstallRemove, providerInstallation.remove(input), { - "rpc.aggregate": "provider", - }), - [WS_METHODS.serverUpdateServer]: (input) => - observeRpcEffect(WS_METHODS.serverUpdateServer, serverSelfUpdate.update(input), { - "rpc.aggregate": "server", - }), + providerAuth.subscribe(input, currentSessionId), + [WS_METHODS.providerInstallStart]: (input) => providerInstallation.start(input), + [WS_METHODS.providerInstallCancel]: (input) => providerInstallation.cancel(input), + [WS_METHODS.providerInstallSubscribe]: (input) => providerInstallation.subscribe(input), + [WS_METHODS.providerInstallRemove]: (input) => providerInstallation.remove(input), + [WS_METHODS.serverUpdateServer]: (input) => serverSelfUpdate.update(input), [WS_METHODS.serverUpdateServerWithProgress]: (input) => - observeRpcStream( - WS_METHODS.serverUpdateServerWithProgress, - Stream.callback((queue) => - serverSelfUpdate - .update(input, (stage) => + Stream.callback((queue) => + serverSelfUpdate + .update(input, (stage) => + Queue.offer(queue, { + type: "progress", + stage, + }).pipe(Effect.asVoid), + ) + .pipe( + Effect.flatMap((result) => Queue.offer(queue, { - type: "progress", - stage, - }).pipe(Effect.asVoid), - ) - .pipe( - Effect.flatMap((result) => - Queue.offer(queue, { - type: "complete", - result, - }), - ), - Effect.catchTags({ - ServerSelfUpdateError: (error) => Queue.fail(queue, error), + type: "complete", + result, }), - Effect.andThen(Queue.end(queue)), - Effect.forkScoped, ), - ), - { "rpc.aggregate": "server" }, + Effect.catchTags({ + ServerSelfUpdateError: (error) => Queue.fail(queue, error), + }), + Effect.andThen(Queue.end(queue)), + Effect.forkScoped, + ), ), [WS_METHODS.serverCommitDesktopUpdate]: (input) => - observeRpcEffect( - WS_METHODS.serverCommitDesktopUpdate, - serverSelfUpdate.commitDesktopUpdate(input.requestId), - { "rpc.aggregate": "server" }, - ), + serverSelfUpdate.commitDesktopUpdate(input.requestId), [WS_METHODS.serverUpsertKeybinding]: (rule) => - observeRpcEffect( - WS_METHODS.serverUpsertKeybinding, - Effect.gen(function* () { - const keybindingsConfig = yield* keybindings.upsertKeybindingRule(rule); - return { keybindings: keybindingsConfig, issues: [] }; - }), - { "rpc.aggregate": "server" }, - ), + Effect.gen(function* () { + const keybindingsConfig = yield* keybindings.upsertKeybindingRule(rule); + return { keybindings: keybindingsConfig, issues: [] }; + }), [WS_METHODS.serverRemoveKeybinding]: (rule) => - observeRpcEffect( - WS_METHODS.serverRemoveKeybinding, - Effect.gen(function* () { - const keybindingsConfig = yield* keybindings.removeKeybindingRule(rule); - return { keybindings: keybindingsConfig, issues: [] }; - }), - { "rpc.aggregate": "server" }, - ), + Effect.gen(function* () { + const keybindingsConfig = yield* keybindings.removeKeybindingRule(rule); + return { keybindings: keybindingsConfig, issues: [] }; + }), [WS_METHODS.serverGetSettings]: (_input) => - observeRpcEffect( - WS_METHODS.serverGetSettings, - serverSettings.getSettings.pipe( - Effect.map(ServerSettings.redactServerSettingsForClient), - ), - { - "rpc.aggregate": "server", - }, - ), + serverSettings.getSettings.pipe(Effect.map(ServerSettings.redactServerSettingsForClient)), [WS_METHODS.serverUpdateSettings]: ({ patch, providerInstanceMutation }) => - observeRpcEffect( - WS_METHODS.serverUpdateSettings, - Effect.gen(function* () { - const deviceHosts = patch.deviceHosts - ? yield* remoteSshDeviceHosts(patch.deviceHosts).pipe( - Effect.provide(deviceHostContext), - ) - : undefined; - const nextPatch = { ...patch, ...(deviceHosts ? { deviceHosts } : {}) }; - const settings = yield* providerInstanceMutation === undefined - ? serverSettings.updateSettings(nextPatch) - : serverSettings.updateProviderInstance(providerInstanceMutation, nextPatch); - return ServerSettings.redactServerSettingsForClient(settings); - }), - { - "rpc.aggregate": "server", - }, - ), - [WS_METHODS.serverDiscoverSourceControl]: (_input) => - observeRpcEffect( - WS_METHODS.serverDiscoverSourceControl, - sourceControlDiscovery.discover, - { - "rpc.aggregate": "server", - }, - ), + Effect.gen(function* () { + const deviceHosts = patch.deviceHosts + ? yield* remoteSshDeviceHosts(patch.deviceHosts).pipe( + Effect.provide(deviceHostContext), + ) + : undefined; + const nextPatch = { ...patch, ...(deviceHosts ? { deviceHosts } : {}) }; + const settings = yield* providerInstanceMutation === undefined + ? serverSettings.updateSettings(nextPatch) + : serverSettings.updateProviderInstance(providerInstanceMutation, nextPatch); + return ServerSettings.redactServerSettingsForClient(settings); + }), + [WS_METHODS.serverDiscoverSourceControl]: (_input) => sourceControlDiscovery.discover, [WS_METHODS.serverGetTraceDiagnostics]: (_input) => - observeRpcEffect( - WS_METHODS.serverGetTraceDiagnostics, - TraceDiagnostics.readTraceDiagnostics({ - traceFilePath: config.serverTracePath, - maxFiles: config.traceMaxFiles, - }), - { - "rpc.aggregate": "server", - }, - ), - [WS_METHODS.serverGetProcessDiagnostics]: (_input) => - observeRpcEffect(WS_METHODS.serverGetProcessDiagnostics, processDiagnostics.read, { - "rpc.aggregate": "server", - }), - [WS_METHODS.serverGetHostResources]: (_input) => - observeRpcEffect(WS_METHODS.serverGetHostResources, hostResources.read, { - "rpc.aggregate": "server", + TraceDiagnostics.readTraceDiagnostics({ + traceFilePath: config.serverTracePath, + maxFiles: config.traceMaxFiles, }), + [WS_METHODS.serverGetProcessDiagnostics]: (_input) => processDiagnostics.read, + [WS_METHODS.serverGetHostResources]: (_input) => hostResources.read, [WS_METHODS.serverGetProcessResourceHistory]: (input) => - observeRpcEffect( - WS_METHODS.serverGetProcessResourceHistory, - processResourceMonitor.readHistory(input), - { - "rpc.aggregate": "server", - }, - ), + processResourceMonitor.readHistory(input), [WS_METHODS.serverGetResourceTelemetryHistory]: (input) => - observeRpcEffect( - WS_METHODS.serverGetResourceTelemetryHistory, - resourceTelemetry.readHistory(input), - { - "rpc.aggregate": "server", - }, - ), - [WS_METHODS.serverGetUsageSummary]: (input) => - observeRpcEffect(WS_METHODS.serverGetUsageSummary, usage.readSummary(input), { - "rpc.aggregate": "server", - }), - [WS_METHODS.serverRefreshUsageRates]: (_input) => - observeRpcEffect(WS_METHODS.serverRefreshUsageRates, usage.refreshRates, { - "rpc.aggregate": "server", - }), - [WS_METHODS.serverRetryResourceTelemetry]: (_input) => - observeRpcEffect(WS_METHODS.serverRetryResourceTelemetry, resourceTelemetry.retry, { - "rpc.aggregate": "server", - }), - [WS_METHODS.serverSignalProcess]: (input) => - observeRpcEffect(WS_METHODS.serverSignalProcess, processDiagnostics.signal(input), { - "rpc.aggregate": "server", - }), + resourceTelemetry.readHistory(input), + [WS_METHODS.serverGetUsageSummary]: (input) => usage.readSummary(input), + [WS_METHODS.serverRefreshUsageRates]: (_input) => usage.refreshRates, + [WS_METHODS.serverRetryResourceTelemetry]: (_input) => resourceTelemetry.retry, + [WS_METHODS.serverSignalProcess]: (input) => processDiagnostics.signal(input), [WS_METHODS.serverReportClientActivity]: (input, metadata) => Ref.update(rpcClientIds, (clientIds) => { const next = new Set(clientIds); @@ -2660,583 +2401,303 @@ const layerWsRpc = ( return next; }).pipe( Effect.andThen( - observeRpcEffect( - WS_METHODS.serverReportClientActivity, - backgroundPolicy.reportClientActivity( - currentSessionId, - RpcClientId.make(metadata.client.id), - input, - ), - { "rpc.aggregate": "server" }, + backgroundPolicy.reportClientActivity( + currentSessionId, + RpcClientId.make(metadata.client.id), + input, ), ), ), [WS_METHODS.serverReportHostPowerState]: (input) => - observeRpcEffect( - WS_METHODS.serverReportHostPowerState, - backgroundPolicy.reportHostPowerState(input), - { "rpc.aggregate": "server" }, - ), - [WS_METHODS.serverGetBackgroundPolicy]: (_input) => - observeRpcEffect(WS_METHODS.serverGetBackgroundPolicy, backgroundPolicy.snapshot, { - "rpc.aggregate": "server", - }), - [WS_METHODS.cloudGetRelayClientStatus]: (_input) => - observeRpcEffect(WS_METHODS.cloudGetRelayClientStatus, relayClient.resolve, { - "rpc.aggregate": "cloud", - }), + backgroundPolicy.reportHostPowerState(input), + [WS_METHODS.serverGetBackgroundPolicy]: (_input) => backgroundPolicy.snapshot, + [WS_METHODS.cloudGetRelayClientStatus]: (_input) => relayClient.resolve, [WS_METHODS.cloudInstallRelayClient]: (_input) => - observeRpcStream( - WS_METHODS.cloudInstallRelayClient, - Stream.callback( - (queue) => - relayClient - .installWithProgress((event) => Queue.offer(queue, event).pipe(Effect.asVoid)) - .pipe( - Effect.flatMap((status) => - Queue.offer(queue, { - type: "complete", - status, + Stream.callback((queue) => + relayClient + .installWithProgress((event) => Queue.offer(queue, event).pipe(Effect.asVoid)) + .pipe( + Effect.flatMap((status) => + Queue.offer(queue, { + type: "complete", + status, + }), + ), + Effect.catchTags({ + RelayClientInstallError: (error) => + Queue.fail( + queue, + new RelayClientInstallFailedError({ + reason: error.reason, + message: error.message, }), ), - Effect.catchTags({ - RelayClientInstallError: (error) => - Queue.fail( - queue, - new RelayClientInstallFailedError({ - reason: error.reason, - message: error.message, - }), - ), - }), - Effect.andThen(Queue.end(queue)), - Effect.forkScoped, - ), - ), - { "rpc.aggregate": "cloud" }, + }), + Effect.andThen(Queue.end(queue)), + Effect.forkScoped, + ), ), - [WS_METHODS.pullRequestsList]: (input) => - observeRpcEffect(WS_METHODS.pullRequestsList, pullRequests.list(input), { - "rpc.aggregate": "pull-requests", - }), - [WS_METHODS.pullRequestsListStats]: (input) => - observeRpcEffect(WS_METHODS.pullRequestsListStats, pullRequests.listStats(input), { - "rpc.aggregate": "pull-requests", - }), - [WS_METHODS.pullRequestsRoutingIdentity]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsRoutingIdentity, - pullRequests.routingIdentity(input), - { - "rpc.aggregate": "pull-requests", - }, - ), - [WS_METHODS.pullRequestsRouting]: (input) => - observeRpcEffect(WS_METHODS.pullRequestsRouting, pullRequests.routing(input), { - "rpc.aggregate": "pull-requests", - }), + [WS_METHODS.pullRequestsList]: (input) => pullRequests.list(input), + [WS_METHODS.pullRequestsListStats]: (input) => pullRequests.listStats(input), + [WS_METHODS.pullRequestsRoutingIdentity]: (input) => pullRequests.routingIdentity(input), + [WS_METHODS.pullRequestsRouting]: (input) => pullRequests.routing(input), [WS_METHODS.pullRequestsSummary]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsSummary, - withPullRequestViewer(input, pullRequests.summary(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.summary(input)), [WS_METHODS.pullRequestsStack]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsStack, - withPullRequestViewer(input, pullRequests.stack(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.stack(input)), [WS_METHODS.pullRequestsLinkedThreads]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsLinkedThreads, - resolvePullRequestSyncKey(input).pipe( - Effect.flatMap((key) => - key === null - ? Effect.succeed({ threads: [] }) - : listLinkedPullRequestThreads(key).pipe( - Effect.provideService(SqlClient.SqlClient, sql), - ), - ), + resolvePullRequestSyncKey(input).pipe( + Effect.flatMap((key) => + key === null + ? Effect.succeed({ threads: [] }) + : listLinkedPullRequestThreads(key).pipe( + Effect.provideService(SqlClient.SqlClient, sql), + ), ), - { "rpc.aggregate": "pull-requests" }, ), [WS_METHODS.pullRequestsDetail]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsDetail, - withPullRequestViewer(input, pullRequests.detail(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.detail(input)), [WS_METHODS.pullRequestsPreview]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsPreview, - withPullRequestViewer(input, pullRequests.preview(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.preview(input)), [WS_METHODS.pullRequestsChecks]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsChecks, - withPullRequestViewer(input, pullRequests.checks(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.checks(input)), [WS_METHODS.pullRequestsActivity]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsActivity, - withPullRequestViewer(input, pullRequests.activity(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.activity(input)), [WS_METHODS.pullRequestsThreadComments]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsThreadComments, - withPullRequestViewer(input, pullRequests.threadComments(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.threadComments(input)), [WS_METHODS.pullRequestsDiffFileContents]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsDiffFileContents, - withPullRequestViewer(input, pullRequests.diffFileContents(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.diffFileContents(input)), [WS_METHODS.pullRequestsFilesViewed]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsFilesViewed, - withPullRequestViewer(input, pullRequests.filesViewed(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.filesViewed(input)), [WS_METHODS.pullRequestsSetFilesViewed]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsSetFilesViewed, - withPullRequestViewer(input, pullRequests.setFilesViewed(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.setFilesViewed(input)), [WS_METHODS.pullRequestsRunAction]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsRunAction, - withPullRequestViewer(input, pullRequests.runAction(input)).pipe( - Effect.tap(() => - resolvePullRequestSyncKey(input).pipe( - Effect.flatMap((key) => - key === null ? Effect.void : pullRequestSync.requestSync(key), - ), + withPullRequestViewer(input, pullRequests.runAction(input)).pipe( + Effect.tap(() => + resolvePullRequestSyncKey(input).pipe( + Effect.flatMap((key) => + key === null ? Effect.void : pullRequestSync.requestSync(key), ), ), ), - { "rpc.aggregate": "pull-requests" }, ), [WS_METHODS.pullRequestsUpdate]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsUpdate, - withPullRequestViewer(input, pullRequests.update(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.update(input)), [WS_METHODS.pullRequestsComment]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsComment, - withPullRequestViewer(input, pullRequests.comment(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.comment(input)), [WS_METHODS.pullRequestsUpdateComment]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsUpdateComment, - withPullRequestViewer(input, pullRequests.updateComment(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.updateComment(input)), [WS_METHODS.pullRequestsSubmitReview]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsSubmitReview, - withPullRequestViewer(input, pullRequests.submitReview(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.submitReview(input)), [WS_METHODS.pullRequestsReplyToThread]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsReplyToThread, - withPullRequestViewer(input, pullRequests.replyToThread(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.replyToThread(input)), [WS_METHODS.pullRequestsSetThreadResolution]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsSetThreadResolution, - withPullRequestViewer(input, pullRequests.setThreadResolution(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.setThreadResolution(input)), [WS_METHODS.pullRequestsSetReaction]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsSetReaction, - withPullRequestViewer(input, pullRequests.setReaction(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.setReaction(input)), [WS_METHODS.pullRequestsInvalidate]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsInvalidate, - pullRequests.invalidate(input, { notifyReaders: true }).pipe( - // A reader asking for fresh host state also wants the thread badges it feeds to - // catch up, including a merged link the sweep would otherwise never revisit. - Effect.andThen( - input.reference === undefined || input.filesViewedOnly === true - ? Effect.void - : resolvePullRequestSyncKey(input.reference).pipe( - Effect.flatMap((key) => - key === null ? Effect.void : pullRequestSync.requestSync(key), - ), + pullRequests.invalidate(input, { notifyReaders: true }).pipe( + // A reader asking for fresh host state also wants the thread badges it feeds to + // catch up, including a merged link the sweep would otherwise never revisit. + Effect.andThen( + input.reference === undefined || input.filesViewedOnly === true + ? Effect.void + : resolvePullRequestSyncKey(input.reference).pipe( + Effect.flatMap((key) => + key === null ? Effect.void : pullRequestSync.requestSync(key), ), - ), + ), ), - { "rpc.aggregate": "pull-requests" }, - ), - [WS_METHODS.pullRequestsSubscribeRefreshes]: () => - observeRpcStream( - WS_METHODS.pullRequestsSubscribeRefreshes, - pullRequests.subscribeRefreshes, - { "rpc.aggregate": "pull-requests" }, ), + [WS_METHODS.pullRequestsSubscribeRefreshes]: () => pullRequests.subscribeRefreshes, [WS_METHODS.pullRequestsReviewerCandidates]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsReviewerCandidates, - withPullRequestViewer(input, pullRequests.reviewerCandidates(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.reviewerCandidates(input)), [WS_METHODS.pullRequestsRequestReviewers]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsRequestReviewers, - withPullRequestViewer(input, pullRequests.requestReviewers(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.requestReviewers(input)), [WS_METHODS.pullRequestsLabelCandidates]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsLabelCandidates, - withPullRequestViewer(input, pullRequests.labelCandidates(input)), - { "rpc.aggregate": "pull-requests" }, - ), + withPullRequestViewer(input, pullRequests.labelCandidates(input)), [WS_METHODS.pullRequestsSetLabels]: (input) => - observeRpcEffect( - WS_METHODS.pullRequestsSetLabels, - withPullRequestViewer(input, pullRequests.setLabels(input)), - { - "rpc.aggregate": "pull-requests", - }, - ), + withPullRequestViewer(input, pullRequests.setLabels(input)), [WS_METHODS.sourceControlLookupRepository]: (input) => - observeRpcEffect( - WS_METHODS.sourceControlLookupRepository, - sourceControlRepositories.lookupRepository(input), - { - "rpc.aggregate": "source-control", - }, - ), + sourceControlRepositories.lookupRepository(input), [WS_METHODS.sourceControlCloneRepository]: (input) => - observeRpcEffect( - WS_METHODS.sourceControlCloneRepository, - sourceControlRepositories.cloneRepository(input), - { - "rpc.aggregate": "source-control", - }, - ), + sourceControlRepositories.cloneRepository(input), [WS_METHODS.projectCloneStart]: (input) => - observeRpcEffect( - WS_METHODS.projectCloneStart, - projectCloneTracker.start(input, { - createProject: (project) => - projectService - .create({ - commandId: CommandId.make(`project-clone-create:${project.projectId}`), + projectCloneTracker.start(input, { + createProject: (project) => + projectService + .create({ + commandId: CommandId.make(`project-clone-create:${project.projectId}`), + projectId: project.projectId, + title: project.title, + workspaceRoot: project.workspaceRoot, + createWorkspaceRootIfMissing: true, + }) + .pipe( + Effect.asVoid, + Effect.mapError( + (cause) => + new OrchestrationDispatchCommandError({ + message: "Failed to create clone project.", + cause, + }), + ), + ), + onCloned: (project) => + repositoryIdentityResolver.resolve(project.workspaceRoot, { refresh: true }).pipe( + Effect.andThen( + projectService.update({ + commandId: CommandId.make(`project-clone-done:${project.projectId}`), projectId: project.projectId, - title: project.title, - workspaceRoot: project.workspaceRoot, - createWorkspaceRootIfMissing: true, - }) - .pipe( - Effect.asVoid, - Effect.mapError( - (cause) => - new OrchestrationDispatchCommandError({ - message: "Failed to create clone project.", - cause, - }), - ), - ), - onCloned: (project) => - repositoryIdentityResolver.resolve(project.workspaceRoot, { refresh: true }).pipe( - Effect.andThen( - projectService.update({ - commandId: CommandId.make(`project-clone-done:${project.projectId}`), - projectId: project.projectId, - }), - ), - Effect.andThen(refreshGitStatus(project.workspaceRoot)), - Effect.ignoreCause({ log: true }), + }), ), - }), - { "rpc.aggregate": "source-control" }, - ), + Effect.andThen(refreshGitStatus(project.workspaceRoot)), + Effect.ignoreCause({ log: true }), + ), + }), [WS_METHODS.projectsEnsureScratch]: () => - observeRpcEffect( - WS_METHODS.projectsEnsureScratch, - managedFolders.ensureScratchProject.pipe( - Effect.mapError( - (cause) => new OrchestrationDispatchCommandError({ message: cause.message, cause }), - ), + managedFolders.ensureScratchProject.pipe( + Effect.mapError( + (cause) => new OrchestrationDispatchCommandError({ message: cause.message, cause }), ), - { "rpc.aggregate": "orchestration" }, ), [WS_METHODS.projectsCreateNew]: (input) => - observeRpcEffect( - WS_METHODS.projectsCreateNew, - managedFolders - .createNamedProject(input) - .pipe( - Effect.mapError( - (cause) => - new OrchestrationDispatchCommandError({ message: cause.message, cause }), - ), + managedFolders + .createNamedProject(input) + .pipe( + Effect.mapError( + (cause) => new OrchestrationDispatchCommandError({ message: cause.message, cause }), ), - { "rpc.aggregate": "orchestration" }, - ), + ), [WS_METHODS.projectCloneCancel]: (input) => - observeRpcEffect( - WS_METHODS.projectCloneCancel, - projectCloneTracker - .cancel(input.projectId) - .pipe(Effect.map((applied) => ({ applied }))), - { "rpc.aggregate": "source-control" }, - ), + projectCloneTracker.cancel(input.projectId).pipe(Effect.map((applied) => ({ applied }))), [WS_METHODS.projectCloneRetry]: (input) => - observeRpcEffect( - WS_METHODS.projectCloneRetry, - projectCloneTracker.retry(input.projectId).pipe(Effect.map((applied) => ({ applied }))), - { "rpc.aggregate": "source-control" }, - ), - [WS_METHODS.subscribeProjectClones]: () => - observeRpcStream(WS_METHODS.subscribeProjectClones, projectCloneTracker.stream, { - "rpc.aggregate": "source-control", - }), + projectCloneTracker.retry(input.projectId).pipe(Effect.map((applied) => ({ applied }))), + [WS_METHODS.subscribeProjectClones]: () => projectCloneTracker.stream, [WS_METHODS.sourceControlPublishRepository]: (input) => - observeRpcEffect( - WS_METHODS.sourceControlPublishRepository, - sourceControlRepositories.publishRepository(input).pipe( - // A new remote can change the cached identity. Only the `cwd` entry - // refreshes, so after a publish from a linked worktree the project - // root entry waits for its TTL. - Effect.tap(() => repositoryIdentityResolver.resolve(input.cwd, { refresh: true })), - Effect.tap(() => refreshGitStatus(input.cwd)), - ), - { - "rpc.aggregate": "source-control", - }, + sourceControlRepositories.publishRepository(input).pipe( + // A new remote can change the cached identity. Only the `cwd` entry + // refreshes, so after a publish from a linked worktree the project + // root entry waits for its TTL. + Effect.tap(() => repositoryIdentityResolver.resolve(input.cwd, { refresh: true })), + Effect.tap(() => refreshGitStatus(input.cwd)), ), [WS_METHODS.projectsSearchEntries]: (input) => - observeRpcEffect( - WS_METHODS.projectsSearchEntries, - workspaceEntries.search(input).pipe( - Effect.mapError( - (cause) => - new ProjectSearchEntriesError({ - cwd: input.cwd, - queryLength: input.query.length, - limit: input.limit, - ...projectEntriesFailureContext(cause), - cause, - }), - ), + workspaceEntries.search(input).pipe( + Effect.mapError( + (cause) => + new ProjectSearchEntriesError({ + cwd: input.cwd, + queryLength: input.query.length, + limit: input.limit, + ...projectEntriesFailureContext(cause), + cause, + }), ), - { "rpc.aggregate": "workspace" }, ), [WS_METHODS.projectsSearchContents]: (input) => - observeRpcEffect( - WS_METHODS.projectsSearchContents, - workspaceEntries.searchContents(input).pipe( - Effect.mapError( - (cause) => - new ProjectSearchContentsError({ - cwd: input.cwd, - queryLength: input.query.length, - limit: input.limit, - ...projectEntriesFailureContext(cause), - cause, - }), - ), + workspaceEntries.searchContents(input).pipe( + Effect.mapError( + (cause) => + new ProjectSearchContentsError({ + cwd: input.cwd, + queryLength: input.query.length, + limit: input.limit, + ...projectEntriesFailureContext(cause), + cause, + }), ), - { "rpc.aggregate": "workspace" }, ), [WS_METHODS.projectsListEntries]: (input) => - observeRpcEffect( - WS_METHODS.projectsListEntries, - workspaceEntries.list(input).pipe( - Effect.mapError( - (cause) => - new ProjectListEntriesError({ - ...input, - ...projectEntriesFailureContext(cause), - cause, - }), - ), + workspaceEntries.list(input).pipe( + Effect.mapError( + (cause) => + new ProjectListEntriesError({ + ...input, + ...projectEntriesFailureContext(cause), + cause, + }), ), - { "rpc.aggregate": "workspace" }, ), [WS_METHODS.projectsReadFile]: (input) => - observeRpcEffect( - WS_METHODS.projectsReadFile, - workspaceFileSystem.readFile(input).pipe( - Effect.mapError( - (cause) => - new ProjectReadFileError({ - ...input, - ...projectFileFailureContext(cause), - cause, - }), - ), + workspaceFileSystem.readFile(input).pipe( + Effect.mapError( + (cause) => + new ProjectReadFileError({ + ...input, + ...projectFileFailureContext(cause), + cause, + }), ), - { "rpc.aggregate": "workspace" }, ), [WS_METHODS.projectsWriteFile]: (input) => - observeRpcEffect( - WS_METHODS.projectsWriteFile, - workspaceFileSystem.writeFile(input).pipe( - Effect.mapError( - (cause) => - new ProjectWriteFileError({ - cwd: input.cwd, - relativePath: input.relativePath, - ...projectFileFailureContext(cause), - cause, - }), - ), + workspaceFileSystem.writeFile(input).pipe( + Effect.mapError( + (cause) => + new ProjectWriteFileError({ + cwd: input.cwd, + relativePath: input.relativePath, + ...projectFileFailureContext(cause), + cause, + }), ), - { "rpc.aggregate": "workspace" }, ), [WS_METHODS.projectsMutate]: (mutation) => - observeRpcEffect( - WS_METHODS.projectsMutate, - startup.enqueueCommand(mutateProject(mutation)).pipe( - Effect.mapError( - (cause) => - new ProjectMutationError({ - commandId: mutation.commandId, - message: - cause._tag === "ProjectNotEmptyError" - ? cause.message - : "Failed to mutate project.", - cause, - }), - ), + startup.enqueueCommand(mutateProject(mutation)).pipe( + Effect.mapError( + (cause) => + new ProjectMutationError({ + commandId: mutation.commandId, + message: + cause._tag === "ProjectNotEmptyError" + ? cause.message + : "Failed to mutate project.", + cause, + }), ), - { "rpc.aggregate": "orchestration" }, ), - [WS_METHODS.shellOpenInEditor]: (input) => - observeRpcEffect(WS_METHODS.shellOpenInEditor, externalLauncher.launchEditor(input), { - "rpc.aggregate": "workspace", - }), + [WS_METHODS.shellOpenInEditor]: (input) => externalLauncher.launchEditor(input), [WS_METHODS.filesystemBrowse]: (input) => - observeRpcEffect( - WS_METHODS.filesystemBrowse, - workspaceEntries.browse(input).pipe( - Effect.mapError( - (cause) => - new FilesystemBrowseError({ - ...input, - ...filesystemBrowseFailureContext(cause), - cause, - }), - ), + workspaceEntries.browse(input).pipe( + Effect.mapError( + (cause) => + new FilesystemBrowseError({ + ...input, + ...filesystemBrowseFailureContext(cause), + cause, + }), ), - { "rpc.aggregate": "workspace" }, ), - [WS_METHODS.attachmentsCreateUploadUrl]: (input) => - observeRpcEffect(WS_METHODS.attachmentsCreateUploadUrl, issueAttachmentUploadUrl(input), { - "rpc.aggregate": "workspace", - }), - [WS_METHODS.attachmentsDelete]: (input) => - observeRpcEffect( - WS_METHODS.attachmentsDelete, - deletePendingAttachment(input.attachmentId), - { "rpc.aggregate": "workspace" }, - ), - [WS_METHODS.agentSessionsScan]: () => - observeRpcEffect(WS_METHODS.agentSessionsScan, agentSessionScanner.scan, { - "rpc.aggregate": "workspace", - }), + [WS_METHODS.attachmentsCreateUploadUrl]: (input) => issueAttachmentUploadUrl(input), + [WS_METHODS.attachmentsDelete]: (input) => deletePendingAttachment(input.attachmentId), + [WS_METHODS.agentSessionsScan]: () => agentSessionScanner.scan, [WS_METHODS.agentSessionsImport]: (input) => - observeRpcEffect( - WS_METHODS.agentSessionsImport, - agentSessionImporter.importRecentAgentThreads(input), - { "rpc.aggregate": "workspace" }, - ), + agentSessionImporter.importRecentAgentThreads(input), [WS_METHODS.assetsCreateUrl]: (input) => - observeRpcEffect( - WS_METHODS.assetsCreateUrl, - Effect.gen(function* () { - const path = yield* Path.Path; - // An absolute media path can be linked from a thread on another environment. - if ( - input.resource._tag === "attachment" || - input.resource._tag === "native-app-icon" || - input.resource._tag === "tool-output-image" || - // GitHub media names the repository it authenticates through itself. - input.resource._tag === "github-media" || - (input.resource._tag === "media-file" && path.isAbsolute(input.resource.path)) - ) { - return yield* issueAssetUrl({ resource: input.resource }); - } - if (input.resource._tag === "draft-workspace-file") { - // A project draft names its workspace directly; there is no - // thread to resolve one from. - return yield* issueAssetUrl({ - resource: input.resource, - workspaceRoot: input.resource.cwd, - }); - } - if (input.resource._tag === "project-favicon") { - const project = yield* projectStore - .findActiveByWorkspaceRoot(input.resource.cwd) - .pipe( - Effect.mapError( - (cause) => - new AssetWorkspaceContextResolutionError({ - resource: input.resource, - cause, - }), - ), - ); - if (Option.isNone(project)) { - return yield* new AssetWorkspaceContextNotFoundError({ - resource: input.resource, - }); - } - // A cloned project exists before its files do. Clients ask again - // when the clone lands (see createProjectFaviconUrlAtomFamily). - const clone = yield* projectCloneTracker.get(project.value.projectId); - return yield* issueAssetUrl({ - resource: input.resource, - ...(project.value.faviconPath - ? { projectFaviconPath: project.value.faviconPath } - : {}), - projectCheckoutPending: - clone !== null && - clone.phase !== "done" && - clone.destinationPath === project.value.workspaceRoot, - }); - } - const thread = yield* threadManagement - .getThreadRecords(input.resource.threadId, []) + Effect.gen(function* () { + const path = yield* Path.Path; + // An absolute media path can be linked from a thread on another environment. + if ( + input.resource._tag === "attachment" || + input.resource._tag === "native-app-icon" || + input.resource._tag === "tool-output-image" || + // GitHub media names the repository it authenticates through itself. + input.resource._tag === "github-media" || + (input.resource._tag === "media-file" && path.isAbsolute(input.resource.path)) + ) { + return yield* issueAssetUrl({ resource: input.resource }); + } + if (input.resource._tag === "draft-workspace-file") { + // A project draft names its workspace directly; there is no + // thread to resolve one from. + return yield* issueAssetUrl({ + resource: input.resource, + workspaceRoot: input.resource.cwd, + }); + } + if (input.resource._tag === "project-favicon") { + const project = yield* projectStore + .findActiveByWorkspaceRoot(input.resource.cwd) .pipe( Effect.mapError( (cause) => @@ -3246,366 +2707,217 @@ const layerWsRpc = ( }), ), ); - const project = yield* projectService.getById(thread.thread.projectId).pipe( - Effect.mapError( - (cause) => - new AssetWorkspaceContextResolutionError({ - resource: input.resource, - cause, - }), - ), - ); if (Option.isNone(project)) { return yield* new AssetWorkspaceContextNotFoundError({ resource: input.resource, }); } + // A cloned project exists before its files do. Clients ask again + // when the clone lands (see createProjectFaviconUrlAtomFamily). + const clone = yield* projectCloneTracker.get(project.value.projectId); return yield* issueAssetUrl({ resource: input.resource, - workspaceRoot: thread.thread.worktreePath ?? project.value.workspaceRoot, + ...(project.value.faviconPath + ? { projectFaviconPath: project.value.faviconPath } + : {}), + projectCheckoutPending: + clone !== null && + clone.phase !== "done" && + clone.destinationPath === project.value.workspaceRoot, }); - }), - { "rpc.aggregate": "workspace" }, - ), + } + const thread = yield* threadManagement + .getThreadRecords(input.resource.threadId, []) + .pipe( + Effect.mapError( + (cause) => + new AssetWorkspaceContextResolutionError({ + resource: input.resource, + cause, + }), + ), + ); + const project = yield* projectService.getById(thread.thread.projectId).pipe( + Effect.mapError( + (cause) => + new AssetWorkspaceContextResolutionError({ + resource: input.resource, + cause, + }), + ), + ); + if (Option.isNone(project)) { + return yield* new AssetWorkspaceContextNotFoundError({ + resource: input.resource, + }); + } + return yield* issueAssetUrl({ + resource: input.resource, + workspaceRoot: thread.thread.worktreePath ?? project.value.workspaceRoot, + }); + }), [WS_METHODS.assetsPersistChatAttachments]: (input) => - observeRpcEffect( - WS_METHODS.assetsPersistChatAttachments, - persistChatAttachments(input).pipe(Effect.map((attachments) => ({ attachments }))), - { "rpc.aggregate": "orchestration" }, - ), + persistChatAttachments(input).pipe(Effect.map((attachments) => ({ attachments }))), [WS_METHODS.subscribeVcsStatus]: (input) => - observeRpcStream( - WS_METHODS.subscribeVcsStatus, - vcsStatusBroadcaster.streamStatus(input, { - automaticRemoteRefreshInterval: automaticGitFetchInterval, - }), - { - "rpc.aggregate": "vcs", - }, - ), - [WS_METHODS.subscribeWorktreeSetup]: (input) => - observeRpcStream( - WS_METHODS.subscribeWorktreeSetup, - worktreeSetupTracker.stream(input.threadId), - { "rpc.aggregate": "vcs" }, - ), + vcsStatusBroadcaster.streamStatus(input, { + automaticRemoteRefreshInterval: automaticGitFetchInterval, + }), + [WS_METHODS.subscribeWorktreeSetup]: (input) => worktreeSetupTracker.stream(input.threadId), [WS_METHODS.worktreeSetupCancel]: (input) => - observeRpcEffect( - WS_METHODS.worktreeSetupCancel, - worktreeSetupTracker - .cancel(input.threadId) - .pipe(Effect.map((cancelled) => ({ cancelled }))), - { "rpc.aggregate": "vcs" }, - ), - [WS_METHODS.vcsRefreshStatus]: (input) => - observeRpcEffect( - WS_METHODS.vcsRefreshStatus, - vcsStatusBroadcaster.refreshStatus(input.cwd), - { - "rpc.aggregate": "vcs", - }, - ), + worktreeSetupTracker + .cancel(input.threadId) + .pipe(Effect.map((cancelled) => ({ cancelled }))), + [WS_METHODS.vcsRefreshStatus]: (input) => vcsStatusBroadcaster.refreshStatus(input.cwd), [WS_METHODS.vcsPull]: (input) => - observeRpcEffect( - WS_METHODS.vcsPull, - gitWorkflow.pullCurrentBranch(input.cwd).pipe( - Effect.matchCauseEffect({ - onFailure: (cause) => Effect.failCause(cause), - onSuccess: (result) => - refreshGitStatus(input.cwd).pipe(Effect.ignore({ log: true }), Effect.as(result)), - }), - ), - { "rpc.aggregate": "git" }, + gitWorkflow.pullCurrentBranch(input.cwd).pipe( + Effect.matchCauseEffect({ + onFailure: (cause) => Effect.failCause(cause), + onSuccess: (result) => + refreshGitStatus(input.cwd).pipe(Effect.ignore({ log: true }), Effect.as(result)), + }), ), [WS_METHODS.gitRunStackedAction]: (input) => - observeRpcStream( - WS_METHODS.gitRunStackedAction, - Stream.callback((queue) => - gitWorkflow - .runStackedAction(input, { - actionId: input.actionId, - progressReporter: { - publish: (event) => Queue.offer(queue, event).pipe(Effect.asVoid), - }, - }) - .pipe( - Effect.matchCauseEffect({ - onFailure: (cause) => Queue.failCause(queue, cause), - onSuccess: (result) => - (input.threadId === undefined - ? Effect.void - : linkCreatedPullRequest({ - threadId: input.threadId, - result, - commandId: serverCommandId("pr-created-link"), - }).pipe( - Effect.provideService(Orchestrator.OrchestratorV2, orchestrationEngine), - Effect.provideService(ProjectService.ProjectService, projectService), - ) - ).pipe( - Effect.andThen( - refreshPushedPullRequests(input, result).pipe( - Effect.provideService(Orchestrator.OrchestratorV2, orchestrationEngine), - Effect.provideService(ProjectStore.ProjectStoreV2, projectStore), - Effect.provideService( - PullRequestService.PullRequestService, - pullRequests, - ), + Stream.callback((queue) => + gitWorkflow + .runStackedAction(input, { + actionId: input.actionId, + progressReporter: { + publish: (event) => Queue.offer(queue, event).pipe(Effect.asVoid), + }, + }) + .pipe( + Effect.matchCauseEffect({ + onFailure: (cause) => Queue.failCause(queue, cause), + onSuccess: (result) => + (input.threadId === undefined + ? Effect.void + : linkCreatedPullRequest({ + threadId: input.threadId, + result, + commandId: serverCommandId("pr-created-link"), + }).pipe( + Effect.provideService(Orchestrator.OrchestratorV2, orchestrationEngine), + Effect.provideService(ProjectService.ProjectService, projectService), + ) + ).pipe( + Effect.andThen( + refreshPushedPullRequests(input, result).pipe( + Effect.provideService(Orchestrator.OrchestratorV2, orchestrationEngine), + Effect.provideService(ProjectStore.ProjectStoreV2, projectStore), + Effect.provideService( + PullRequestService.PullRequestService, + pullRequests, ), ), - Effect.andThen(refreshGitStatus(input.cwd)), - Effect.andThen(Queue.end(queue).pipe(Effect.asVoid)), ), - }), - ), - ), - { "rpc.aggregate": "vcs" }, - ), - [WS_METHODS.gitResolvePullRequest]: (input) => - observeRpcEffect( - WS_METHODS.gitResolvePullRequest, - gitWorkflow.resolvePullRequest(input), - { - "rpc.aggregate": "git", - }, + Effect.andThen(refreshGitStatus(input.cwd)), + Effect.andThen(Queue.end(queue).pipe(Effect.asVoid)), + ), + }), + ), ), + [WS_METHODS.gitResolvePullRequest]: (input) => gitWorkflow.resolvePullRequest(input), [WS_METHODS.gitPreparePullRequestThread]: (input) => - observeRpcEffect( - WS_METHODS.gitPreparePullRequestThread, - gitWorkflow - .preparePullRequestThread(input) - .pipe(Effect.tap(() => refreshGitStatus(input.cwd))), - { "rpc.aggregate": "git" }, - ), - [WS_METHODS.vcsListRefs]: (input) => - observeRpcEffect(WS_METHODS.vcsListRefs, gitWorkflow.listRefs(input), { - "rpc.aggregate": "vcs", - }), + gitWorkflow + .preparePullRequestThread(input) + .pipe(Effect.tap(() => refreshGitStatus(input.cwd))), + [WS_METHODS.vcsListRefs]: (input) => gitWorkflow.listRefs(input), [WS_METHODS.vcsCreateWorktree]: (input) => - observeRpcEffect( - WS_METHODS.vcsCreateWorktree, - gitWorkflow.createWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), - { "rpc.aggregate": "vcs" }, - ), + gitWorkflow.createWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), [WS_METHODS.vcsRemoveWorktree]: (input) => - observeRpcEffect( - WS_METHODS.vcsRemoveWorktree, - gitWorkflow.removeWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), - { "rpc.aggregate": "vcs" }, - ), + gitWorkflow.removeWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), [WS_METHODS.vcsCreateRef]: (input) => - observeRpcEffect( - WS_METHODS.vcsCreateRef, - gitWorkflow.createRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), - { "rpc.aggregate": "vcs" }, - ), + gitWorkflow.createRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), [WS_METHODS.vcsSwitchRef]: (input) => - observeRpcEffect( - WS_METHODS.vcsSwitchRef, - gitWorkflow.switchRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), - { "rpc.aggregate": "vcs" }, - ), + gitWorkflow.switchRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), [WS_METHODS.vcsInit]: (input) => - observeRpcEffect( - WS_METHODS.vcsInit, - vcsProvisioning - .initRepository(input) - .pipe(Effect.tap(() => refreshGitStatus(input.cwd))), - { "rpc.aggregate": "vcs" }, - ), - [WS_METHODS.reviewGetDiffPreview]: (input) => - observeRpcEffect(WS_METHODS.reviewGetDiffPreview, review.getDiffPreview(input), { - "rpc.aggregate": "review", - }), - [WS_METHODS.reviewGetDiffFileContents]: (input) => - observeRpcEffect( - WS_METHODS.reviewGetDiffFileContents, - review.getDiffFileContents(input), - { "rpc.aggregate": "review" }, - ), - [WS_METHODS.terminalOpen]: (input) => - observeRpcEffect(WS_METHODS.terminalOpen, terminalManager.open(input), { - "rpc.aggregate": "terminal", - }), + vcsProvisioning.initRepository(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))), + [WS_METHODS.reviewGetDiffPreview]: (input) => review.getDiffPreview(input), + [WS_METHODS.reviewGetDiffFileContents]: (input) => review.getDiffFileContents(input), + [WS_METHODS.terminalOpen]: (input) => terminalManager.open(input), [WS_METHODS.terminalAttach]: (input) => - observeRpcStream( - WS_METHODS.terminalAttach, - Stream.callback((queue) => - Effect.acquireRelease( - terminalManager.attachStream(input, (event) => Queue.offer(queue, event)), - (unsubscribe) => Effect.sync(unsubscribe), - ), + Stream.callback((queue) => + Effect.acquireRelease( + terminalManager.attachStream(input, (event) => Queue.offer(queue, event)), + (unsubscribe) => Effect.sync(unsubscribe), ), - { "rpc.aggregate": "terminal" }, ), - [WS_METHODS.terminalWrite]: (input) => - observeRpcEffect(WS_METHODS.terminalWrite, terminalManager.write(input), { - "rpc.aggregate": "terminal", - }), - [WS_METHODS.terminalResize]: (input) => - observeRpcEffect(WS_METHODS.terminalResize, terminalManager.resize(input), { - "rpc.aggregate": "terminal", - }), - [WS_METHODS.terminalClear]: (input) => - observeRpcEffect(WS_METHODS.terminalClear, terminalManager.clear(input), { - "rpc.aggregate": "terminal", - }), - [WS_METHODS.terminalRestart]: (input) => - observeRpcEffect(WS_METHODS.terminalRestart, terminalManager.restart(input), { - "rpc.aggregate": "terminal", - }), - [WS_METHODS.terminalClose]: (input) => - observeRpcEffect(WS_METHODS.terminalClose, terminalManager.close(input), { - "rpc.aggregate": "terminal", - }), + [WS_METHODS.terminalWrite]: (input) => terminalManager.write(input), + [WS_METHODS.terminalResize]: (input) => terminalManager.resize(input), + [WS_METHODS.terminalClear]: (input) => terminalManager.clear(input), + [WS_METHODS.terminalRestart]: (input) => terminalManager.restart(input), + [WS_METHODS.terminalClose]: (input) => terminalManager.close(input), [WS_METHODS.subscribeTerminalEvents]: (_input) => - observeRpcStream( - WS_METHODS.subscribeTerminalEvents, - Stream.callback((queue) => - Effect.acquireRelease( - terminalManager.subscribe((event) => Queue.offer(queue, event)), - (unsubscribe) => Effect.sync(unsubscribe), - ), + Stream.callback((queue) => + Effect.acquireRelease( + terminalManager.subscribe((event) => Queue.offer(queue, event)), + (unsubscribe) => Effect.sync(unsubscribe), ), - { "rpc.aggregate": "terminal" }, ), [WS_METHODS.subscribeTerminalMetadata]: (_input) => - observeRpcStream( - WS_METHODS.subscribeTerminalMetadata, - Stream.callback((queue) => - Effect.acquireRelease( - terminalManager.subscribeMetadata((event) => Queue.offer(queue, event)), - (unsubscribe) => Effect.sync(unsubscribe), - ), + Stream.callback((queue) => + Effect.acquireRelease( + terminalManager.subscribeMetadata((event) => Queue.offer(queue, event)), + (unsubscribe) => Effect.sync(unsubscribe), ), - { "rpc.aggregate": "terminal" }, ), - [WS_METHODS.previewOpen]: (input) => - observeRpcEffect(WS_METHODS.previewOpen, previewManager.open(input), { - "rpc.aggregate": "preview", - }), - [WS_METHODS.previewNavigate]: (input) => - observeRpcEffect(WS_METHODS.previewNavigate, previewManager.navigate(input), { - "rpc.aggregate": "preview", - }), - [WS_METHODS.previewResize]: (input) => - observeRpcEffect(WS_METHODS.previewResize, previewManager.resize(input), { - "rpc.aggregate": "preview", - }), - [WS_METHODS.previewAdjust]: (input) => - observeRpcEffect(WS_METHODS.previewAdjust, previewManager.adjust(input), { - "rpc.aggregate": "preview", - }), - [WS_METHODS.previewRefresh]: (input) => - observeRpcEffect(WS_METHODS.previewRefresh, previewManager.refresh(input), { - "rpc.aggregate": "preview", - }), - [WS_METHODS.previewClose]: (input) => - observeRpcEffect(WS_METHODS.previewClose, previewManager.close(input), { - "rpc.aggregate": "preview", - }), - [WS_METHODS.previewList]: (input) => - observeRpcEffect(WS_METHODS.previewList, previewManager.list(input), { - "rpc.aggregate": "preview", - }), - [WS_METHODS.previewClearProfile]: (input) => - observeRpcEffect( - WS_METHODS.previewClearProfile, - serverBrowser.clearProfile(input.profileId), - { "rpc.aggregate": "preview" }, - ), - [WS_METHODS.previewReportStatus]: (input) => - observeRpcEffect(WS_METHODS.previewReportStatus, previewManager.reportStatus(input), { - "rpc.aggregate": "preview", - }), - [WS_METHODS.subscribePreviewEvents]: (_input) => - observeRpcStream(WS_METHODS.subscribePreviewEvents, previewManager.events, { - "rpc.aggregate": "preview", - }), - [WS_METHODS.deviceConfigure]: (input) => - observeRpcEffect(WS_METHODS.deviceConfigure, deviceService.configure(input), { - "rpc.aggregate": "device", - }), - [WS_METHODS.deviceTestHost]: (input) => - observeRpcEffect(WS_METHODS.deviceTestHost, deviceService.testHost(input), { - "rpc.aggregate": "device", - }), + [WS_METHODS.previewOpen]: (input) => previewManager.open(input), + [WS_METHODS.previewNavigate]: (input) => previewManager.navigate(input), + [WS_METHODS.previewResize]: (input) => previewManager.resize(input), + [WS_METHODS.previewAdjust]: (input) => previewManager.adjust(input), + [WS_METHODS.previewRefresh]: (input) => previewManager.refresh(input), + [WS_METHODS.previewClose]: (input) => previewManager.close(input), + [WS_METHODS.previewList]: (input) => previewManager.list(input), + [WS_METHODS.previewClearProfile]: (input) => serverBrowser.clearProfile(input.profileId), + [WS_METHODS.previewReportStatus]: (input) => previewManager.reportStatus(input), + [WS_METHODS.subscribePreviewEvents]: (_input) => previewManager.events, + [WS_METHODS.deviceConfigure]: (input) => deviceService.configure(input), + [WS_METHODS.deviceTestHost]: (input) => deviceService.testHost(input), [WS_METHODS.deviceList]: (input) => - observeRpcEffect( - WS_METHODS.deviceList, - input.inspectOnly && !input.updateTool - ? deviceService.inspect - : authorizeEffect( - requiredScopeForDeviceList(input), - input.updateTool - ? deviceService.updateTool(input.updateTool) - : input.retryHostId - ? deviceService.retryHost(input.retryHostId) - : deviceService.list, - ), - { - "rpc.aggregate": "device", - }, - ), - [WS_METHODS.deviceOpen]: (input) => - observeRpcEffect(WS_METHODS.deviceOpen, deviceService.open(input), { - "rpc.aggregate": "device", - }), - [WS_METHODS.deviceClose]: (input) => - observeRpcEffect(WS_METHODS.deviceClose, deviceService.close(input), { - "rpc.aggregate": "device", - }), - [WS_METHODS.deviceShutdown]: (input) => - observeRpcEffect(WS_METHODS.deviceShutdown, deviceService.shutdown(input), { - "rpc.aggregate": "device", - }), - [WS_METHODS.deviceDetail]: (input) => - observeRpcEffect(WS_METHODS.deviceDetail, deviceService.detail(input), { - "rpc.aggregate": "device", - }), - [WS_METHODS.deviceAction]: (input) => - observeRpcEffect(WS_METHODS.deviceAction, deviceService.action(input), { - "rpc.aggregate": "device", - }), - [WS_METHODS.subscribeDeviceState]: (_input) => - observeRpcStream( - WS_METHODS.subscribeDeviceState, - DeviceService.stateStream(deviceService), - { "rpc.aggregate": "device" }, - ), + input.inspectOnly && !input.updateTool + ? deviceService.inspect + : authorizeEffect( + requiredScopeForDeviceList(input), + input.updateTool + ? deviceService.updateTool(input.updateTool) + : input.retryHostId + ? deviceService.retryHost(input.retryHostId) + : deviceService.list, + ), + [WS_METHODS.deviceOpen]: (input) => deviceService.open(input), + [WS_METHODS.deviceClose]: (input) => deviceService.close(input), + [WS_METHODS.deviceShutdown]: (input) => deviceService.shutdown(input), + [WS_METHODS.deviceDetail]: (input) => deviceService.detail(input), + [WS_METHODS.deviceAction]: (input) => deviceService.action(input), + [WS_METHODS.subscribeDeviceState]: (_input) => DeviceService.stateStream(deviceService), [WS_METHODS.subscribeDiscoveredLocalServers]: (input) => - observeRpcStream( - WS_METHODS.subscribeDiscoveredLocalServers, - Stream.callback((queue) => - Effect.gen(function* () { - const configuredUrls = input.configuredUrls ?? []; - yield* portDiscovery.retain; - const initial = yield* portDiscovery.scan(configuredUrls); - const initialScannedAt = DateTime.formatIso(yield* DateTime.now); - yield* Queue.offer(queue, { - servers: initial, - scannedAt: initialScannedAt, - configuredUrlProbing: true, - }); - yield* portDiscovery.subscribe( - { configuredUrls, initialSnapshot: initial }, - (servers) => - Effect.gen(function* () { - const scannedAt = DateTime.formatIso(yield* DateTime.now); - yield* Queue.offer(queue, { - servers, - scannedAt, - configuredUrlProbing: true, - }); - }), - ); - }), - ), - { "rpc.aggregate": "preview" }, + Stream.callback((queue) => + Effect.gen(function* () { + const configuredUrls = input.configuredUrls ?? []; + yield* portDiscovery.retain; + const initial = yield* portDiscovery.scan(configuredUrls); + const initialScannedAt = DateTime.formatIso(yield* DateTime.now); + yield* Queue.offer(queue, { + servers: initial, + scannedAt: initialScannedAt, + configuredUrlProbing: true, + }); + yield* portDiscovery.subscribe( + { configuredUrls, initialSnapshot: initial }, + (servers) => + Effect.gen(function* () { + const scannedAt = DateTime.formatIso(yield* DateTime.now); + yield* Queue.offer(queue, { + servers, + scannedAt, + configuredUrlProbing: true, + }); + }), + ); + }), ), [WS_METHODS.subscribeServerConfig]: (input) => - observeRpcStreamEffect( - WS_METHODS.subscribeServerConfig, + Stream.unwrap( Effect.gen(function* () { const usageLimitsCommand = input.usageLimitsCommand === true; const config = yield* loadServerConfig({ usageLimitsCommand }); @@ -3705,11 +3017,9 @@ const layerWsRpc = ( withLateEditorConfig(config, liveUpdates, externalLauncher), ); }), - { "rpc.aggregate": "server" }, ), [WS_METHODS.subscribeServerLifecycle]: (_input) => - observeRpcStreamEffect( - WS_METHODS.subscribeServerLifecycle, + Stream.unwrap( Effect.gen(function* () { const liveBuffer = yield* Queue.unbounded(); yield* Effect.forkScoped( @@ -3727,11 +3037,9 @@ const layerWsRpc = ( ); return Stream.concat(rpcInitialItems(snapshotEvents), liveEvents); }), - { "rpc.aggregate": "server" }, ), [WS_METHODS.subscribeAuthAccess]: (_input) => - observeRpcStreamEffect( - WS_METHODS.subscribeAuthAccess, + Stream.unwrap( Effect.gen(function* () { const initialSnapshot = yield* loadAuthAccessSnapshot(); const revisionRef = yield* Ref.make(1); @@ -3761,27 +3069,18 @@ const layerWsRpc = ( liveEvents, ); }), - { "rpc.aggregate": "auth" }, ), [WS_METHODS.subscribeBackgroundPolicy]: (_input) => - observeRpcStream( - WS_METHODS.subscribeBackgroundPolicy, - Stream.unwrap( - Effect.map(backgroundPolicy.subscribe, ({ latest, changes }) => - Stream.concat(Stream.make(latest), changes), - ), + Stream.unwrap( + Effect.map(backgroundPolicy.subscribe, ({ latest, changes }) => + Stream.concat(Stream.make(latest), changes), ), - { "rpc.aggregate": "server" }, ), [WS_METHODS.subscribeResourceTelemetry]: (_input) => - observeRpcStream( - WS_METHODS.subscribeResourceTelemetry, - Stream.unwrap( - Effect.map(resourceTelemetry.subscribe, ({ latest, changes }) => - Stream.concat(Stream.make(latest), changes), - ), + Stream.unwrap( + Effect.map(resourceTelemetry.subscribe, ({ latest, changes }) => + Stream.concat(Stream.make(latest), changes), ), - { "rpc.aggregate": "server" }, ), }); return handlers; @@ -3841,7 +3140,9 @@ export const layer = Layer.unwrap( const { protocol, httpEffect } = yield* RpcServer.makeProtocolWithHttpEffectWebsocket; yield* RpcServer.make(ServerWsRpcGroup, WS_RPC_SERVER_OPTIONS).pipe( Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)), - Effect.provide(RpcAuthorization.layer(session.scopes)), + Effect.provide( + Layer.merge(RpcAuthorization.layer(session.scopes), rpcInstrumentationLayer), + ), Effect.forkScoped, ); // @effect-diagnostics-next-line returnEffectInGen:off diff --git a/docs/internals/effect-services.md b/docs/internals/effect-services.md index 49a7d2424cac72e4151a29fdc9a61d0a312d8ea0..4da6357c0a91df44efb40b127cfc71dbf254bf45 100644 --- a/docs/internals/effect-services.md +++ b/docs/internals/effect-services.md @@ -24,15 +24,17 @@ message) are fine next to the service; the capability itself is the method. ```ts // ws.ts: a thin handler [WS_METHODS.projectsCreateNew]: (input) => - observeRpcEffect( - WS_METHODS.projectsCreateNew, - projectFolders.createNamedProject(input).pipe( - Effect.mapError((cause) => new ProjectCreateNewError({ cause })), - ), - { "rpc.aggregate": "orchestration" }, - ), + projectFolders + .createNamedProject(input) + .pipe(Effect.mapError((cause) => new ProjectCreateNewError({ cause }))), ``` +Handlers don't add their own spans or request metrics. Group middleware authorizes every call +([`RpcAuthorization.ts`](../../apps/server/src/auth/RpcAuthorization.ts)), and the server's group +also instruments it +([`RpcInstrumentation.ts`](../../apps/server/src/observability/RpcInstrumentation.ts)). A handler +with per-call context, such as a thread id, adds it with `Effect.annotateCurrentSpan`. + ## Shape of a service module One module per service, in this order: imports, errors and schemas, the `Context.Service` tag with diff --git a/docs/operations/observability.md b/docs/operations/observability.md index c97a187eb08ae8bb91f617e2d7af92f7dadf28ab..c199f6aba73195b59b3cad1bdea0a6d352d80666 100644 --- a/docs/operations/observability.md +++ b/docs/operations/observability.md @@ -635,8 +635,8 @@ wins for its signal. `otlp` is the default, and any other exporter name, such as Current high-value span and metric boundaries include: -- Effect RPC websocket request spans from `effect/rpc` -- RPC request metrics in `apps/server/src/observability/RpcInstrumentation.ts` +- WebSocket RPC request spans (`ws.rpc.`) and metrics in + `apps/server/src/observability/RpcInstrumentation.ts` - startup phases - orchestration command processing - provider session and turn operations -- 2.54.0