Skip to content

[codex] Structure client RPC unavailable errors #3260

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
juliusmarminge merged 4 commits into from
Jun 20, 2026
Merged

[codex] Structure client RPC unavailable errors #3260

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
f255391
structure client RPC unavailable errors
juliusmarminge Jun 20, 2026
ee6201f
Preserve typed RPC retry classification
juliusmarminge Jun 20, 2026
3a308b8
fix(client-runtime): bound RPC transport diagnostics
juliusmarminge Jun 20, 2026
0c5fe3c
test(client-runtime): focus error mapping coverage
juliusmarminge Jun 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions apps/mobile/src/state/thread-outbox-model.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { isTransportConnectionErrorMessage } from "@t3tools/client-runtime/errors";
import { isEnvironmentRpcUnavailableError } from "@t3tools/client-runtime/rpc";
import type { EnvironmentShellStatus } from "@t3tools/client-runtime/state/shell";
import { CommandId, EnvironmentId, IsoDateTime, MessageId, ThreadId } from "@t3tools/contracts";
import * as Schema from "effect/Schema";
Expand Down Expand Up @@ -109,6 +110,9 @@ function errorMessage(error: unknown): string | null {
}

export function shouldRetryThreadOutboxDelivery(error: unknown): boolean {
if (isEnvironmentRpcUnavailableError(error)) {
return true;
}
if (
typeof error === "object" &&
error !== null &&
Expand Down
10 changes: 10 additions & 0 deletions apps/mobile/src/state/thread-outbox.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { describe, expect, it } from "@effect/vitest";
import { EnvironmentRpcUnavailableError } from "@t3tools/client-runtime/rpc";
import { CommandId, EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
import { AtomRegistry } from "effect/unstable/reactivity";

Expand Down Expand Up @@ -215,6 +216,15 @@ describe("thread outbox", () => {
});

it("retries transport failures but drops deterministic command failures", () => {
expect(
shouldRetryThreadOutboxDelivery(
new EnvironmentRpcUnavailableError({
environmentId: EnvironmentId.make("environment-1"),
environmentLabel: "Test environment",
method: "thread.turn.start",
}),
),
).toBe(true);
expect(shouldRetryThreadOutboxDelivery(new Error("Socket is not connected"))).toBe(true);
expect(
shouldRetryThreadOutboxDelivery({
Expand Down
68 changes: 3 additions & 65 deletions packages/client-runtime/src/connection/errors.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,72 +1,13 @@
import { EnvironmentAuthInvalidError } from "@t3tools/contracts";
import { RelayAuthInvalidError } from "@t3tools/contracts/relay";
import { describe, expect, it } from "@effect/vitest";

import { mapManagedRelayError, mapRemoteEnvironmentError } from "./errors.ts";
import * as ManagedRelay from "../relay/managedRelay.ts";
import { mapRemoteEnvironmentError } from "./errors.ts";
import {
RemoteEnvironmentAuthFetchError,
RemoteEnvironmentAuthUndeclaredStatusError,
} from "../rpc/http.ts";

describe("connection error mapping", () => {
it("retains the managed relay request as the cause when classifying a protected error", () => {
const relayError = new RelayAuthInvalidError({
code: "auth_invalid",
reason: "invalid_bearer",
traceId: "relay-trace-id",
});
const source = new ManagedRelay.ManagedRelayRequestFailedError({
action: "connect relay environment",
cause: relayError,
relayError,
traceId: relayError.traceId,
});

const error = mapManagedRelayError(source);

expect(error).toMatchObject({
_tag: "ConnectionBlockedError",
reason: "authentication",
traceId: "relay-trace-id",
});
expect(error.cause).toBe(source);
});

it("retains a managed relay timeout and its structured activity", () => {
const source = new ManagedRelay.ManagedRelayRequestTimeoutError({
activity: "Relay environment connection",
timeoutMs: 10_000,
});

const error = mapManagedRelayError(source);

expect(error).toMatchObject({
_tag: "ConnectionTransientError",
reason: "timeout",
});
expect(error.cause).toBe(source);
expect(source.activity).toBe("Relay environment connection");
});

it("retains structured remote authorization failures", () => {
const source = new EnvironmentAuthInvalidError({
code: "auth_invalid",
reason: "invalid_credential",
traceId: "environment-trace-id",
});

const error = mapRemoteEnvironmentError(source);

expect(error).toMatchObject({
_tag: "ConnectionBlockedError",
reason: "authentication",
traceId: "environment-trace-id",
});
expect(error.cause).toBe(source);
});

it("retains local transport failures without deriving their message from the cause", () => {
it("keeps transport diagnostics structured and redacted", () => {
const transportCause = new Error("sensitive transport implementation detail");
const requestUrl =
"https://environment-user:[email protected]/private/session?access_token=environment-secret#environment-fragment";
Expand All @@ -82,8 +23,6 @@ describe("connection error mapping", () => {
_tag: "ConnectionTransientError",
reason: "network",
});
expect(error.cause).toBe(source);
expect(source.cause).toBe(transportCause);
expect(source).toMatchObject({
requestUrlInputLength: requestUrl.length,
requestUrlProtocol: "https:",
Expand All @@ -102,7 +41,7 @@ describe("connection error mapping", () => {
}
});

it("retains the HTTP client cause for undeclared statuses", () => {
it("keeps undeclared status diagnostics structured and redacted", () => {
const cause = new Error("upstream response metadata");
const requestUrl =
"https://environment-user:[email protected]/private/session?access_token=environment-secret#environment-fragment";
Expand All @@ -114,7 +53,6 @@ describe("connection error mapping", () => {
requestUrlProtocol: "https:",
requestUrlHostname: "environment.example.test",
});
expect(error.cause).toBe(cause);
expect(error).not.toHaveProperty("requestUrl");
});
});
29 changes: 28 additions & 1 deletion packages/client-runtime/src/rpc/client.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,13 @@ import {
import * as EnvironmentSupervisor from "../connection/supervisor.ts";
import * as RpcSession from "../rpc/session.ts";
import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
import { EnvironmentRpcRequestObserver, request, runStream, subscribe } from "./client.ts";
import {
EnvironmentRpcRequestObserver,
EnvironmentRpcUnavailableError,
request,
runStream,
subscribe,
} from "./client.ts";

const TARGET = new PrimaryConnectionTarget({
environmentId: EnvironmentId.make("environment-1"),
Expand Down Expand Up @@ -77,6 +83,27 @@ const makeHarness = Effect.fn("TestEnvironmentRpc.makeHarness")(function* () {
});

describe("environment RPC", () => {
it.effect("reports the disconnected environment and requested method", () =>
Effect.gen(function* () {
const { supervisor } = yield* makeHarness();

const error = yield* request(WS_METHODS.cloudGetRelayClientStatus, {}).pipe(
Effect.flip,
Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor),
);

expect(error).toBeInstanceOf(EnvironmentRpcUnavailableError);
expect(error).toMatchObject({
environmentId: TARGET.environmentId,
environmentLabel: TARGET.label,
method: WS_METHODS.cloudGetRelayClientStatus,
});
expect(error.message).toBe(
`Test environment is not connected for RPC method ${WS_METHODS.cloudGetRelayClientStatus}.`,
);
}),
);

it.effect("observes unary requests until they complete", () =>
Effect.gen(function* () {
const observations: string[] = [];
Expand Down
28 changes: 19 additions & 9 deletions packages/client-runtime/src/rpc/client.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,4 @@
import { ORCHESTRATION_WS_METHODS, WS_METHODS } from "@t3tools/contracts";
import { EnvironmentId, ORCHESTRATION_WS_METHODS, WS_METHODS } from "@t3tools/contracts";
import * as Cause from "effect/Cause";
import * as Context from "effect/Context";
import type * as Duration from "effect/Duration";
Expand All @@ -15,10 +15,18 @@ import type { WsRpcProtocolClient } from "../rpc/protocol.ts";
export class EnvironmentRpcUnavailableError extends Schema.TaggedErrorClass<EnvironmentRpcUnavailableError>()(
"EnvironmentRpcUnavailableError",
{
environmentId: Schema.String,
message: Schema.String,
environmentId: EnvironmentId,
environmentLabel: Schema.String,
method: Schema.optionalKey(Schema.String),
},
) {}
) {
override get message(): string {
const method = this.method === undefined ? "" : ` for RPC method ${this.method}`;
return `${this.environmentLabel} is not connected${method}.`;
}
Comment thread
cursor[bot] marked this conversation as resolved.
Resolved
}

export const isEnvironmentRpcUnavailableError = Schema.is(EnvironmentRpcUnavailableError);

export interface EnvironmentRpcRequestObservation {
readonly environmentId: string;
Expand Down Expand Up @@ -85,7 +93,7 @@ export type EnvironmentRpcStreamFailure<TTag extends EnvironmentStreamRpcTag> =
? E
: never;

const currentSession = Effect.fn("EnvironmentRpc.currentSession")(function* () {
const currentSession = Effect.fn("EnvironmentRpc.currentSession")(function* (method?: string) {
const supervisor = yield* EnvironmentSupervisor;
return yield* SubscriptionRef.get(supervisor.session).pipe(
Effect.flatMap(
Expand All @@ -94,7 +102,8 @@ const currentSession = Effect.fn("EnvironmentRpc.currentSession")(function* () {
Effect.fail(
new EnvironmentRpcUnavailableError({
environmentId: supervisor.target.environmentId,
message: `${supervisor.target.label} is not connected.`,
environmentLabel: supervisor.target.label,
...(method === undefined ? {} : { method }),
}),
),
onSome: Effect.succeed,
Expand All @@ -111,7 +120,7 @@ export const request = Effect.fn("EnvironmentRpc.request")(function* <
"environment.id": supervisor.target.environmentId,
"rpc.method": tag,
});
const session = yield* currentSession();
const session = yield* currentSession(tag);
const observer = yield* EnvironmentRpcRequestObserver;
const method = session.client[tag] as (
input: EnvironmentRpcInput<TTag>,
Expand All @@ -132,7 +141,7 @@ export function runStream<TTag extends EnvironmentStreamCommandRpcTag>(
EnvironmentSupervisor
> {
return Stream.unwrap(
currentSession().pipe(
currentSession(tag).pipe(
Effect.map((session) => {
const method = session.client[tag] as (
input: EnvironmentRpcInput<TTag>,
Expand Down Expand Up @@ -195,7 +204,8 @@ export function subscribe<TTag extends EnvironmentSubscriptionRpcTag>(
Effect.logWarning(
"Durable RPC subscription lost its transport; waiting for the next session.",
{
cause: Cause.pretty(cause),
errorTag: "RpcClientError",
reasonCount: cause.reasons.length,
method: tag,
environmentId: supervisor.target.environmentId,
},
Expand Down
Loading