[codex] Structure client RPC unavailable errors

apps/mobile/src/state/thread-outbox-model.ts

@@ -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";
@@ -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 &&

apps/mobile/src/state/thread-outbox.test.ts

@@ -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";
 
@@ -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({

packages/client-runtime/src/connection/errors.test.ts

@@ -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:environment-password@environment.example.test/private/session?access_token=environment-secret#environment-fragment";
@@ -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:",
@@ -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:environment-password@environment.example.test/private/session?access_token=environment-secret#environment-fragment";
@@ -114,7 +53,6 @@ describe("connection error mapping", () => {
       requestUrlProtocol: "https:",
       requestUrlHostname: "environment.example.test",
     });
-    expect(error.cause).toBe(cause);
     expect(error).not.toHaveProperty("requestUrl");
   });
 });

packages/client-runtime/src/rpc/client.test.ts

@@ -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"),
@@ -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[] = [];

packages/client-runtime/src/rpc/client.ts

@@ -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";
@@ -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}.`;
+  }
+}
+
+export const isEnvironmentRpcUnavailableError = Schema.is(EnvironmentRpcUnavailableError);
 
 export interface EnvironmentRpcRequestObservation {
   readonly environmentId: string;
@@ -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(
@@ -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,
@@ -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>,
@@ -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>,
@@ -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,
                               },