From 57e0304996b95afe65095a6e0b38637eeac8eec3 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 20 Jun 2026 04:40:05 -0700 Subject: [PATCH] Structure relay activity persistence errors Co-authored-by: codex --- .../agentActivity/AgentActivityRows.test.ts | 84 ++++++++++++ .../src/agentActivity/AgentActivityRows.ts | 123 ++++++++++++------ .../agentActivity/MobileRegistrations.test.ts | 1 + 3 files changed, 167 insertions(+), 41 deletions(-) create mode 100644 infra/relay/src/agentActivity/AgentActivityRows.test.ts diff --git a/infra/relay/src/agentActivity/AgentActivityRows.test.ts b/infra/relay/src/agentActivity/AgentActivityRows.test.ts new file mode 100644 index 00000000000..be976d16bbb --- /dev/null +++ b/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))), + ), + ); + }); +}); diff --git a/infra/relay/src/agentActivity/AgentActivityRows.ts b/infra/relay/src/agentActivity/AgentActivityRows.ts index 7f19378633f..7e1a8c50f1b 100644 --- a/infra/relay/src/agentActivity/AgentActivityRows.ts +++ b/infra/relay/src/agentActivity/AgentActivityRows.ts @@ -14,28 +14,39 @@ import { relayAgentActivityRows, relayEnvironmentLinks } from "../persistence/sc export class AgentActivityRowUpsertPersistenceError extends Schema.TaggedErrorClass()( "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", - { 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", - { 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), - ); - 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), + 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, + }), + ), ); }), }); diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index eed330dd589..17a9c7bd417 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/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", }), ),