[codex] Structure relay activity-row persistence errors

infra/relay/src/agentActivity/AgentActivityRows.test.ts

@@ -0,0 +1,84 @@
+import type { RelayAgentActivityState } from "@t3tools/contracts/relay";
+import { describe, expect, it } from "@effect/vitest";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+
+import * as RelayDb from "../db.ts";
+import * as AgentActivityRows from "./AgentActivityRows.ts";
+
+const state: RelayAgentActivityState = {
+  environmentId: "env-1" as RelayAgentActivityState["environmentId"],
+  threadId: "thread-1" as RelayAgentActivityState["threadId"],
+  projectTitle: "Project",
+  threadTitle: "Thread",
+  modelTitle: "gpt-5.4",
+  phase: "running",
+  headline: "Running",
+  updatedAt: "2026-06-20T00:00:00.000Z",
+  deepLink: "/threads/env-1/thread-1",
+};
+
+describe("AgentActivityRows", () => {
+  it.effect("preserves activity context on persistence failures", () => {
+    const cause = new Error("database unavailable");
+    const failingDb = {
+      insert: () => ({
+        values: () => ({
+          onConflictDoUpdate: () => Effect.fail(cause),
+        }),
+      }),
+      delete: () => ({
+        where: () => Effect.fail(cause),
+      }),
+      select: () => ({
+        from: () => ({
+          innerJoin: () => ({
+            where: () => ({
+              orderBy: () => Effect.fail(cause),
+            }),
+          }),
+        }),
+      }),
+    } as unknown as RelayDb.RelayDb["Service"];
+
+    return Effect.gen(function* () {
+      const rows = yield* AgentActivityRows.AgentActivityRows;
+
+      const upsertError = yield* rows
+        .upsert({ environmentPublicKey: "public-key", state })
+        .pipe(Effect.flip);
+      expect(upsertError).toMatchObject({
+        environmentId: "env-1",
+        threadId: "thread-1",
+        cause,
+      });
+      expect(upsertError.message).toBe(
+        "Failed to persist agent activity state for environment env-1, thread thread-1.",
+      );
+
+      const deleteError = yield* rows
+        .remove({
+          environmentId: "env-1",
+          environmentPublicKey: "public-key",
+          threadId: "thread-1",
+        })
+        .pipe(Effect.flip);
+      expect(deleteError).toMatchObject({
+        environmentId: "env-1",
+        threadId: "thread-1",
+        cause,
+      });
+      expect(deleteError.message).toBe(
+        "Failed to delete agent activity state for environment env-1, thread thread-1.",
+      );
+
+      const listError = yield* rows.listForUser({ userId: "user-2" }).pipe(Effect.flip);
+      expect(listError).toMatchObject({ userId: "user-2", cause });
+      expect(listError.message).toBe("Failed to list agent activity state for user user-2.");
+    }).pipe(
+      Effect.provide(
+        AgentActivityRows.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, failingDb))),
+      ),
+    );
+  });
+});

infra/relay/src/agentActivity/AgentActivityRows.ts

@@ -14,28 +14,39 @@ import { relayAgentActivityRows, relayEnvironmentLinks } from "../persistence/sc
 
 export class AgentActivityRowUpsertPersistenceError extends Schema.TaggedErrorClass<AgentActivityRowUpsertPersistenceError>()(
   "AgentActivityRowUpsertPersistenceError",
-  { cause: Schema.Defect() },
+  {
+    environmentId: Schema.String,
+    threadId: Schema.String,
+    cause: Schema.Defect(),
+  },
 ) {
   override get message(): string {
-    return "Failed to persist agent activity state";
+    return `Failed to persist agent activity state for environment ${this.environmentId}, thread ${this.threadId}.`;
   }
 }
 
 export class AgentActivityRowDeletePersistenceError extends Schema.TaggedErrorClass<AgentActivityRowDeletePersistenceError>()(
   "AgentActivityRowDeletePersistenceError",
-  { cause: Schema.Defect() },
+  {
+    environmentId: Schema.String,
+    threadId: Schema.String,
+    cause: Schema.Defect(),
+  },
 ) {
   override get message(): string {
-    return "Failed to delete agent activity state";
+    return `Failed to delete agent activity state for environment ${this.environmentId}, thread ${this.threadId}.`;
   }
 }
 
 export class AgentActivityRowListPersistenceError extends Schema.TaggedErrorClass<AgentActivityRowListPersistenceError>()(
   "AgentActivityRowListPersistenceError",
-  { cause: Schema.Defect() },
+  {
+    userId: Schema.String,
+    cause: Schema.Defect(),
+  },
 ) {
   override get message(): string {
-    return "Failed to list agent activity state";
+    return `Failed to list agent activity state for user ${this.userId}.`;
   }
 }
 
@@ -75,41 +86,56 @@ export const make = Effect.gen(function* () {
   const db = yield* RelayDb.RelayDb;
 
   return AgentActivityRows.of({
-    upsert: Effect.fn("relay.agent_activity_rows.upsert")(
-      function* (input) {
-        yield* Effect.annotateCurrentSpan({
-          "relay.environment_id": input.state.environmentId,
-          "relay.thread_id": input.state.threadId,
-        });
-        const now = yield* DateTime.now;
-        const stateJson = yield* encodeRelayAgentActivityStateJson(input.state).pipe(
-          Effect.flatMap(decodeJsonString),
-          Effect.map(Function.cast<unknown, RelayAgentActivityState>),
-        );
-        yield* db
-          .insert(relayAgentActivityRows)
-          .values({
-            environmentId: input.state.environmentId,
-            environmentPublicKey: input.environmentPublicKey,
-            threadId: input.state.threadId,
+    upsert: Effect.fn("relay.agent_activity_rows.upsert")(function* (input) {
+      yield* Effect.annotateCurrentSpan({
+        "relay.environment_id": input.state.environmentId,
+        "relay.thread_id": input.state.threadId,
+      });
+      const now = yield* DateTime.now;
+      const stateJson = yield* encodeRelayAgentActivityStateJson(input.state).pipe(
+        Effect.flatMap(decodeJsonString),
+        Effect.map(Function.cast<unknown, RelayAgentActivityState>),
+        Effect.mapError(
+          (cause) =>
+            new AgentActivityRowUpsertPersistenceError({
+              environmentId: input.state.environmentId,
+              threadId: input.state.threadId,
+              cause,
+            }),
+        ),
+      );
+      yield* db
+        .insert(relayAgentActivityRows)
+        .values({
+          environmentId: input.state.environmentId,
+          environmentPublicKey: input.environmentPublicKey,
+          threadId: input.state.threadId,
+          stateJson,
+          updatedAt: input.state.updatedAt,
+          createdAt: DateTime.formatIso(now),
+        })
+        .onConflictDoUpdate({
+          target: [
+            relayAgentActivityRows.environmentId,
+            relayAgentActivityRows.environmentPublicKey,
+            relayAgentActivityRows.threadId,
+          ],
+          set: {
             stateJson,
             updatedAt: input.state.updatedAt,
-            createdAt: DateTime.formatIso(now),
-          })
-          .onConflictDoUpdate({
-            target: [
-              relayAgentActivityRows.environmentId,
-              relayAgentActivityRows.environmentPublicKey,
-              relayAgentActivityRows.threadId,
-            ],
-            set: {
-              stateJson,
-              updatedAt: input.state.updatedAt,
-            },
-          });
-      },
-      Effect.mapError((cause) => new AgentActivityRowUpsertPersistenceError({ cause })),
-    ),
+          },
+        })
+        .pipe(
+          Effect.mapError(
+            (cause) =>
+              new AgentActivityRowUpsertPersistenceError({
+                environmentId: input.state.environmentId,
+                threadId: input.state.threadId,
+                cause,
+              }),
+          ),
+        );
+    }),
 
     remove: Effect.fn("relay.agent_activity_rows.remove")(function* (input) {
       yield* Effect.annotateCurrentSpan({
@@ -125,7 +151,16 @@ export const make = Effect.gen(function* () {
             eq(relayAgentActivityRows.threadId, input.threadId),
           ),
         )
-        .pipe(Effect.mapError((cause) => new AgentActivityRowDeletePersistenceError({ cause })));
+        .pipe(
+          Effect.mapError(
+            (cause) =>
+              new AgentActivityRowDeletePersistenceError({
+                environmentId: input.environmentId,
+                threadId: input.threadId,
+                cause,
+              }),
+          ),
+        );
     }),
 
     listForUser: Effect.fn("relay.agent_activity_rows.list_for_user")(function* (input) {
@@ -159,7 +194,13 @@ export const make = Effect.gen(function* () {
           Effect.map((rows) =>
             rows.flatMap((row) => Option.toArray(decodeRelayAgentActivityStateJson(row))),
           ),
-          Effect.mapError((cause) => new AgentActivityRowListPersistenceError({ cause })),
+          Effect.mapError(
+            (cause) =>
+              new AgentActivityRowListPersistenceError({
+                userId: input.userId,
+                cause,
+              }),
+          ),
         );
     }),
   });

infra/relay/src/agentActivity/MobileRegistrations.test.ts

@@ -249,6 +249,7 @@ describe("MobileRegistrations", () => {
                     replayForLiveActivityRegistration: () =>
                       Effect.fail(
                         new AgentActivityRows.AgentActivityRowListPersistenceError({
+                          userId: "dev:julius",
                           cause: "replay failed",
                         }),
                       ),