authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-20 10:42:38-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-20 10:42:38-07:00
logb4fe8faa1d59aa602a04737aa4054ee8acb62193
treec07d170b9f134dcab87608a0058999ecf0549c46
parent515303edf2f83ab6ba15ebdae01ea2d5765756fc
signature Signed by PGP key B5690EEEBB952194

[codex] Structure relay activity-row persistence errors (#3305)

Co-authored-by: codex <codex@users.noreply.github.com>

3 files changed, 167 insertions(+), 41 deletions(-)

infra/relay/src/agentActivity/AgentActivityRows.test.ts created+84
......@@ -0,0 +1,84 @@
1import type { RelayAgentActivityState } from "@t3tools/contracts/relay";
2import { describe, expect, it } from "@effect/vitest";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5
6import * as RelayDb from "../db.ts";
7import * as AgentActivityRows from "./AgentActivityRows.ts";
8
9const state: RelayAgentActivityState = {
10 environmentId: "env-1" as RelayAgentActivityState["environmentId"],
11 threadId: "thread-1" as RelayAgentActivityState["threadId"],
12 projectTitle: "Project",
13 threadTitle: "Thread",
14 modelTitle: "gpt-5.4",
15 phase: "running",
16 headline: "Running",
17 updatedAt: "2026-06-20T00:00:00.000Z",
18 deepLink: "/threads/env-1/thread-1",
19};
20
21describe("AgentActivityRows", () => {
22 it.effect("preserves activity context on persistence failures", () => {
23 const cause = new Error("database unavailable");
24 const failingDb = {
25 insert: () => ({
26 values: () => ({
27 onConflictDoUpdate: () => Effect.fail(cause),
28 }),
29 }),
30 delete: () => ({
31 where: () => Effect.fail(cause),
32 }),
33 select: () => ({
34 from: () => ({
35 innerJoin: () => ({
36 where: () => ({
37 orderBy: () => Effect.fail(cause),
38 }),
39 }),
40 }),
41 }),
42 } as unknown as RelayDb.RelayDb["Service"];
43
44 return Effect.gen(function* () {
45 const rows = yield* AgentActivityRows.AgentActivityRows;
46
47 const upsertError = yield* rows
48 .upsert({ environmentPublicKey: "public-key", state })
49 .pipe(Effect.flip);
50 expect(upsertError).toMatchObject({
51 environmentId: "env-1",
52 threadId: "thread-1",
53 cause,
54 });
55 expect(upsertError.message).toBe(
56 "Failed to persist agent activity state for environment env-1, thread thread-1.",
57 );
58
59 const deleteError = yield* rows
60 .remove({
61 environmentId: "env-1",
62 environmentPublicKey: "public-key",
63 threadId: "thread-1",
64 })
65 .pipe(Effect.flip);
66 expect(deleteError).toMatchObject({
67 environmentId: "env-1",
68 threadId: "thread-1",
69 cause,
70 });
71 expect(deleteError.message).toBe(
72 "Failed to delete agent activity state for environment env-1, thread thread-1.",
73 );
74
75 const listError = yield* rows.listForUser({ userId: "user-2" }).pipe(Effect.flip);
76 expect(listError).toMatchObject({ userId: "user-2", cause });
77 expect(listError.message).toBe("Failed to list agent activity state for user user-2.");
78 }).pipe(
79 Effect.provide(
80 AgentActivityRows.layer.pipe(Layer.provide(Layer.succeed(RelayDb.RelayDb, failingDb))),
81 ),
82 );
83 });
84});
infra/relay/src/agentActivity/AgentActivityRows.ts+82-41
......@@ -14,28 +14,39 @@ import { relayAgentActivityRows, relayEnvironmentLinks } from "../persistence/sc
1414
1515export class AgentActivityRowUpsertPersistenceError extends Schema.TaggedErrorClass<AgentActivityRowUpsertPersistenceError>()(
1616 "AgentActivityRowUpsertPersistenceError",
17 { cause: Schema.Defect() },
17 {
18 environmentId: Schema.String,
19 threadId: Schema.String,
20 cause: Schema.Defect(),
21 },
1822) {
1923 override get message(): string {
20 return "Failed to persist agent activity state";
24 return `Failed to persist agent activity state for environment ${this.environmentId}, thread ${this.threadId}.`;
2125 }
2226}
2327
2428export class AgentActivityRowDeletePersistenceError extends Schema.TaggedErrorClass<AgentActivityRowDeletePersistenceError>()(
2529 "AgentActivityRowDeletePersistenceError",
26 { cause: Schema.Defect() },
30 {
31 environmentId: Schema.String,
32 threadId: Schema.String,
33 cause: Schema.Defect(),
34 },
2735) {
2836 override get message(): string {
29 return "Failed to delete agent activity state";
37 return `Failed to delete agent activity state for environment ${this.environmentId}, thread ${this.threadId}.`;
3038 }
3139}
3240
3341export class AgentActivityRowListPersistenceError extends Schema.TaggedErrorClass<AgentActivityRowListPersistenceError>()(
3442 "AgentActivityRowListPersistenceError",
35 { cause: Schema.Defect() },
43 {
44 userId: Schema.String,
45 cause: Schema.Defect(),
46 },
3647) {
3748 override get message(): string {
38 return "Failed to list agent activity state";
49 return `Failed to list agent activity state for user ${this.userId}.`;
3950 }
4051}
4152
......@@ -75,41 +86,56 @@ export const make = Effect.gen(function* () {
7586 const db = yield* RelayDb.RelayDb;
7687
7788 return AgentActivityRows.of({
78 upsert: Effect.fn("relay.agent_activity_rows.upsert")(
79 function* (input) {
80 yield* Effect.annotateCurrentSpan({
81 "relay.environment_id": input.state.environmentId,
82 "relay.thread_id": input.state.threadId,
83 });
84 const now = yield* DateTime.now;
85 const stateJson = yield* encodeRelayAgentActivityStateJson(input.state).pipe(
86 Effect.flatMap(decodeJsonString),
87 Effect.map(Function.cast<unknown, RelayAgentActivityState>),
88 );
89 yield* db
90 .insert(relayAgentActivityRows)
91 .values({
92 environmentId: input.state.environmentId,
93 environmentPublicKey: input.environmentPublicKey,
94 threadId: input.state.threadId,
89 upsert: Effect.fn("relay.agent_activity_rows.upsert")(function* (input) {
90 yield* Effect.annotateCurrentSpan({
91 "relay.environment_id": input.state.environmentId,
92 "relay.thread_id": input.state.threadId,
93 });
94 const now = yield* DateTime.now;
95 const stateJson = yield* encodeRelayAgentActivityStateJson(input.state).pipe(
96 Effect.flatMap(decodeJsonString),
97 Effect.map(Function.cast<unknown, RelayAgentActivityState>),
98 Effect.mapError(
99 (cause) =>
100 new AgentActivityRowUpsertPersistenceError({
101 environmentId: input.state.environmentId,
102 threadId: input.state.threadId,
103 cause,
104 }),
105 ),
106 );
107 yield* db
108 .insert(relayAgentActivityRows)
109 .values({
110 environmentId: input.state.environmentId,
111 environmentPublicKey: input.environmentPublicKey,
112 threadId: input.state.threadId,
113 stateJson,
114 updatedAt: input.state.updatedAt,
115 createdAt: DateTime.formatIso(now),
116 })
117 .onConflictDoUpdate({
118 target: [
119 relayAgentActivityRows.environmentId,
120 relayAgentActivityRows.environmentPublicKey,
121 relayAgentActivityRows.threadId,
122 ],
123 set: {
95124 stateJson,
96125 updatedAt: input.state.updatedAt,
97 createdAt: DateTime.formatIso(now),
98 })
99 .onConflictDoUpdate({
100 target: [
101 relayAgentActivityRows.environmentId,
102 relayAgentActivityRows.environmentPublicKey,
103 relayAgentActivityRows.threadId,
104 ],
105 set: {
106 stateJson,
107 updatedAt: input.state.updatedAt,
108 },
109 });
110 },
111 Effect.mapError((cause) => new AgentActivityRowUpsertPersistenceError({ cause })),
112 ),
126 },
127 })
128 .pipe(
129 Effect.mapError(
130 (cause) =>
131 new AgentActivityRowUpsertPersistenceError({
132 environmentId: input.state.environmentId,
133 threadId: input.state.threadId,
134 cause,
135 }),
136 ),
137 );
138 }),
113139
114140 remove: Effect.fn("relay.agent_activity_rows.remove")(function* (input) {
115141 yield* Effect.annotateCurrentSpan({
......@@ -125,7 +151,16 @@ export const make = Effect.gen(function* () {
125151 eq(relayAgentActivityRows.threadId, input.threadId),
126152 ),
127153 )
128 .pipe(Effect.mapError((cause) => new AgentActivityRowDeletePersistenceError({ cause })));
154 .pipe(
155 Effect.mapError(
156 (cause) =>
157 new AgentActivityRowDeletePersistenceError({
158 environmentId: input.environmentId,
159 threadId: input.threadId,
160 cause,
161 }),
162 ),
163 );
129164 }),
130165
131166 listForUser: Effect.fn("relay.agent_activity_rows.list_for_user")(function* (input) {
......@@ -159,7 +194,13 @@ export const make = Effect.gen(function* () {
159194 Effect.map((rows) =>
160195 rows.flatMap((row) => Option.toArray(decodeRelayAgentActivityStateJson(row))),
161196 ),
162 Effect.mapError((cause) => new AgentActivityRowListPersistenceError({ cause })),
197 Effect.mapError(
198 (cause) =>
199 new AgentActivityRowListPersistenceError({
200 userId: input.userId,
201 cause,
202 }),
203 ),
163204 );
164205 }),
165206 });
infra/relay/src/agentActivity/MobileRegistrations.test.ts+1
......@@ -249,6 +249,7 @@ describe("MobileRegistrations", () => {
249249 replayForLiveActivityRegistration: () =>
250250 Effect.fail(
251251 new AgentActivityRows.AgentActivityRowListPersistenceError({
252 userId: "dev:julius",
252253 cause: "replay failed",
253254 }),
254255 ),