From 80b8395bfafba723cccdcd5eb5888d3dca4e672d Mon Sep 17 00:00:00 2001 From: JamesOBrien2 Date: Wed, 28 Oct 2026 02:06:34 +0000 Subject: [PATCH 1/4] fix(relay): stop repeated dings for a finished job A completed job could keep vibrating the phone: the relay decided "newly finished" by diffing the Live Activity card's five visible rows, so a completion that dropped out and came back looked new again, and two overlapping jobs could both ring. Each completion is now claimed in the relay's durable delivery-attempt table before APNs is called, keyed by user, device, thread, phase and run. Duplicate and concurrent deliveries still update the card, silently. A new run in the same thread has a new identity and still alerts. Created with DeepSeek V4.1 Flash Free in OpenCode. --- .../src/agentActivity/ApnsDeliveries.test.ts | 233 ++++++++++++++++++ .../relay/src/agentActivity/ApnsDeliveries.ts | 73 ++++-- .../agentActivity/DeliveryAttempts.test.ts | 48 ++++ .../src/agentActivity/DeliveryAttempts.ts | 71 +++++- .../agentActivity/MobileRegistrations.test.ts | 1 + .../agentActivity/agentActivityAggregate.ts | 1 + .../src/agentActivity/agentActivityAlerts.ts | 25 +- .../agentActivity/agentActivityPayloads.ts | 1 + .../src/agentActivity/apnsDeliveryJobs.ts | 1 + packages/contracts/src/relay.ts | 2 + packages/shared/src/agentAwareness.test.ts | 18 +- packages/shared/src/agentAwareness.ts | 5 +- pnpm-lock.yaml | 25 ++ 13 files changed, 475 insertions(+), 29 deletions(-) diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index b2445b0ab4c08..064c125ee3027 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -26,6 +26,7 @@ import * as AgentActivityRows from "./AgentActivityRows.ts"; import * as ApnsDeliveries from "./ApnsDeliveries.ts"; import * as ApnsClient from "./ApnsClient.ts"; import * as ApnsProviderTokens from "./ApnsProviderTokens.ts"; +import { makeAggregateState } from "./agentActivityAggregate.ts"; const config = RelayConfiguration.RelayConfiguration.of({ relayIssuer: "https://relay.example.test", @@ -141,6 +142,7 @@ const target: LiveActivities.TargetRow = { function layerFor(input: { readonly attempts: Array; readonly sourceJobClaims?: ReadonlyMap; + readonly claimedActivityAlerts?: Set; readonly queuedJobs?: Array; readonly markedDeliveries?: Array< Parameters[0] @@ -185,6 +187,21 @@ function layerFor(input: { }), }), Layer.succeed(DeliveryAttempts.DeliveryAttempts, { + claimActivityAlert: (alert) => + Effect.sync(() => { + const key = JSON.stringify([ + alert.userId, + alert.deviceId, + alert.environmentId, + alert.threadId, + alert.phase, + alert.runId ?? alert.updatedAt, + ]); + const claims = input.claimedActivityAlerts; + if (claims?.has(key)) return false; + claims?.add(key); + return true; + }), pruneExpired: Effect.succeed(0), claimSourceJob: (attempt) => Effect.sync(() => { @@ -1110,6 +1127,222 @@ describe("live activity alert decisions", () => { }); describe("fast completion delivery", () => { + it.effect("rings each completion once across display eviction and delivery restarts", () => { + const states = Array.from({ length: 6 }, (_, index) => ({ + ...state, + threadId: ThreadId.make(`thread-${index}`), + phase: "completed" as const, + runId: `run-${index}`, + })); + const device = { ...target, last_aggregate_json: JSON.stringify(aggregate) }; + const claimedActivityAlerts = new Set(); + const requests: Array> = []; + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + if (request.body._tag === "Uint8Array") { + requests.push(JSON.parse(new TextDecoder().decode(request.body.body))); + } + return HttpClientResponse.fromWeb(request, new Response("", { status: 200 })); + }); + + return Effect.gen(function* () { + for (const index of [0, 1, 2, 3, 4, 5, 4, 5, 4, 5]) { + const jobs: SignedApnsDeliveryJob[] = []; + const marked: Array< + Parameters[0] + > = []; + yield* Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + yield* deliveries.sendForTarget({ + target: device, + aggregate: makeAggregateState({ + activeStates: states, + terminalState: states[index]!, + nowMs: 0, + }), + nowMs: 0, + }); + for (const job of jobs) yield* deliveries.processSignedJob(job); + }).pipe( + Effect.provide( + layerFor({ + attempts: [], + queuedJobs: jobs, + markedDeliveries: marked, + claimedActivityAlerts, + currentTargets: [device], + activityStates: states, + config: signingConfig, + execute, + }), + ), + ); + const delivered = marked.at(-1); + if (delivered) device.last_aggregate_json = JSON.stringify(delivered.aggregate); + } + expect( + requests.filter((request) => (request.aps as Record).alert), + ).toHaveLength(6); + }); + }); + + it.effect("rings once when separate completion jobs are consumed concurrently", () => { + const done = { ...state, phase: "completed" as const }; + const completedAggregate = makeAggregateState({ + activeStates: [done], + terminalState: done, + nowMs: 0, + })!; + const device = { ...target, last_aggregate_json: JSON.stringify(aggregate) }; + const payload = { + kind: "live_activity_update" as const, + userId: device.user_id, + deviceId: device.device_id, + token: "activity-token", + aggregate: completedAggregate, + alert: { title: "Thread", body: "Done: Project" }, + createdAt: state.updatedAt, + expiresAt: "1970-01-01T00:10:00.000Z", + }; + const requests: Array> = []; + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + if (request.body._tag === "Uint8Array") + requests.push(JSON.parse(new TextDecoder().decode(request.body.body))); + return HttpClientResponse.fromWeb(request, new Response("", { status: 200 })); + }); + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + yield* Effect.forEach( + ["first", "second"], + (jobId) => + deliveries.processSignedJob( + signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload: makeApnsDeliveryJobPayload({ ...payload, jobId }), + }), + ), + { concurrency: 2 }, + ); + expect(requests).toHaveLength(2); + expect( + requests.filter((request) => (request.aps as Record).alert), + ).toHaveLength(1); + }).pipe( + Effect.provide( + layerFor({ + attempts: [], + claimedActivityAlerts: new Set(), + currentTargets: [device], + activityStates: [done], + config: signingConfig, + execute, + }), + ), + ); + }); + + it.effect("alerts a new run in the same thread even when the previous card was completed", () => { + const done = { ...state, phase: "completed" as const, runId: "run-new" }; + const previous = makeAggregateState({ + activeStates: [], + terminalState: { ...done, runId: "run-old" }, + nowMs: 0, + })!; + const device = { + ...target, + last_aggregate_json: JSON.stringify(previous), + last_live_activity_delivery_at: state.updatedAt, + }; + const jobs: SignedApnsDeliveryJob[] = []; + const requests: Array> = []; + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + if (request.body._tag === "Uint8Array") + requests.push(JSON.parse(new TextDecoder().decode(request.body.body))); + return HttpClientResponse.fromWeb(request, new Response("", { status: 200 })); + }); + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + yield* deliveries.sendForTarget({ + target: device, + aggregate: makeAggregateState({ activeStates: [], terminalState: done, nowMs: 0 }), + nowMs: 0, + }); + expect(jobs).toHaveLength(1); + yield* deliveries.processSignedJob(jobs[0]!); + expect( + requests.filter((request) => (request.aps as Record).alert), + ).toHaveLength(1); + }).pipe( + Effect.provide( + layerFor({ + attempts: [], + queuedJobs: jobs, + claimedActivityAlerts: new Set(), + currentTargets: [device], + activityStates: [done], + config: signingConfig, + execute, + }), + ), + ); + }); + + it.effect("shares completion deduplication between Live Activities and ordinary pushes", () => { + const done = { ...state, phase: "completed" as const, runId: "run-1" }; + const completed = makeAggregateState({ activeStates: [], terminalState: done, nowMs: 0 })!; + const device = { + ...target, + push_token: "push-token", + last_aggregate_json: JSON.stringify(aggregate), + }; + const jobs: SignedApnsDeliveryJob[] = []; + const requests: Array> = []; + const execute = (request: HttpClientRequest.HttpClientRequest) => + Effect.sync(() => { + if (request.body._tag === "Uint8Array") + requests.push(JSON.parse(new TextDecoder().decode(request.body.body))); + return HttpClientResponse.fromWeb(request, new Response("", { status: 200 })); + }); + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + yield* deliveries.sendForTarget({ target: device, aggregate: completed, nowMs: 0 }); + for (const job of jobs) yield* deliveries.processSignedJob(job); + yield* deliveries.sendPushNotification({ + target: device, + token: "push-token", + sourceJobId: "push-job-1", + notification: { + title: "Thread", + body: "Done: Project", + environmentId: done.environmentId, + threadId: done.threadId, + phase: done.phase, + runId: done.runId, + updatedAt: done.updatedAt, + deepLink: "/", + }, + }); + expect(requests).toHaveLength(1); + expect(requests[0]?.aps).toMatchObject({ + alert: expect.objectContaining({ title: "Thread", body: "Done: Project" }), + }); + }).pipe( + Effect.provide( + layerFor({ + attempts: [], + queuedJobs: jobs, + claimedActivityAlerts: new Set(), + currentTargets: [device], + activityStates: [done], + config: signingConfig, + execute, + }), + ), + ); + }); + it.effect("keeps a completion alert when work finishes before running delivery", () => { const queuedJobs: SignedApnsDeliveryJob[] = []; const old = { diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 6e0e4bdc6e236..0b43220d367d2 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -1,5 +1,6 @@ import type { RelayAgentActivityAggregateState, + RelayAgentActivityAggregateRow, RelayAgentAwarenessPreferences, RelayDeliveryKind, RelayDeliveryResult, @@ -48,6 +49,8 @@ import { alertForTerminalAggregate, rowPhasesChanged, shouldAlertForActivity, + alertForActivityRows, + terminalTransitionRows, } from "./agentActivityAlerts.ts"; const MIN_LIVE_ACTIVITY_UPDATE_INTERVAL_MS = 15_000; const FRESHLY_ARMED_GRACE_MS = 2 * 60 * 1_000; @@ -649,6 +652,7 @@ const make = Effect.gen(function* () { const aggregate = input.aggregate === null ? null : sanitizeAgentActivityAggregateState(input.aggregate); let alert = input.alert ?? null; + let terminalAlertRows: ReadonlyArray = []; const recoverTransportError = (cause: Apns.ApnsError) => recoverApnsDeliveryTransportError( { @@ -689,21 +693,24 @@ const make = Effect.gen(function* () { if (alert) { const preferences = parsePreferences(currentTarget.preferences_json); const previousAggregate = parseAggregate(currentTarget.last_aggregate_json); - alert = - !preferences?.notificationsEnabled || !aggregate - ? null - : (alertForAttentionTransition({ - previousAggregate, - nextAggregate: aggregate, - preferences, - }) ?? - alertForNewlyTerminal({ - previousAggregate, - nextAggregate: aggregate, - preferences, - nowMs: now.epochMilliseconds, - includeUnobserved: true, - })); + alert = null; + if (preferences?.notificationsEnabled && aggregate) { + alert = alertForAttentionTransition({ + previousAggregate, + nextAggregate: aggregate, + preferences, + }); + if (!alert) { + terminalAlertRows = terminalTransitionRows({ + previousAggregate, + nextAggregate: aggregate, + preferences, + nowMs: now.epochMilliseconds, + includeUnobserved: true, + }); + alert = alertForActivityRows(terminalAlertRows); + } + } } if ( aggregate !== null && @@ -732,6 +739,22 @@ const make = Effect.gen(function* () { }); return staleJobResult({ deviceId: input.target.device_id, kind: input.kind }); } + if (terminalAlertRows.length > 0) { + const unclaimed = yield* Effect.filter(terminalAlertRows, (row) => + row.phase === "completed" || row.phase === "failed" + ? attempts.claimActivityAlert({ + userId: input.target.user_id, + deviceId: input.target.device_id, + environmentId: row.environmentId, + threadId: row.threadId, + phase: row.phase, + ...(row.runId ? { runId: row.runId } : {}), + updatedAt: row.updatedAt, + }) + : Effect.succeed(false), + ); + alert = alertForActivityRows(unclaimed); + } const { epochSeconds, iso, request } = makeLiveActivityDeliveryRequest( apns, { ...input, aggregate, alert } as SendLiveActivityDeliveryInput, @@ -886,6 +909,26 @@ const make = Effect.gen(function* () { }); return staleJobResult({ deviceId: input.target.device_id, kind: "push_notification" }); } + if ( + (notification.phase === "completed" || notification.phase === "failed") && + notification.updatedAt !== undefined && + !(yield* attempts.claimActivityAlert({ + userId: input.target.user_id, + deviceId: input.target.device_id, + environmentId: notification.environmentId, + threadId: notification.threadId, + phase: notification.phase, + ...(notification.runId ? { runId: notification.runId } : {}), + updatedAt: notification.updatedAt, + })) + ) { + if (input.sourceJobId) + yield* attempts.completeSourceJob({ + sourceJobId: input.sourceJobId, + apnsReason: "Duplicate activity alert skipped.", + }); + return duplicateJobResult({ deviceId: input.target.device_id, kind: "push_notification" }); + } const result = yield* apns .sendPushNotificationRequest({ credentials: credentialsForTarget(config.apns, deliveryTarget), diff --git a/infra/relay/src/agentActivity/DeliveryAttempts.test.ts b/infra/relay/src/agentActivity/DeliveryAttempts.test.ts index a7b95e30600d2..4f0454e00e649 100644 --- a/infra/relay/src/agentActivity/DeliveryAttempts.test.ts +++ b/infra/relay/src/agentActivity/DeliveryAttempts.test.ts @@ -9,6 +9,54 @@ import * as RelayDb from "../db.ts"; import * as DeliveryAttempts from "./DeliveryAttempts.ts"; describe("DeliveryAttempts", () => { + it.effect("claims a completion once across workers while allowing other runs and devices", () => { + const receipts = new Set(); + const fakeDb = { + insert: () => ({ + values: (values: { sourceJobId: string }) => ({ + onConflictDoNothing: () => ({ + returning: () => + Effect.sync(() => { + if (receipts.has(values.sourceJobId)) return []; + receipts.add(values.sourceJobId); + return [{ id: values.sourceJobId }]; + }), + }), + }), + }), + } as unknown as RelayDb.RelayDb["Service"]; + const layer = DeliveryAttempts.layer.pipe( + Layer.provide(NodeCryptoLayer.layer), + Layer.provide(Layer.succeed(RelayDb.RelayDb, fakeDb)), + ); + const alert = { + userId: "user-1", + deviceId: "device-1", + environmentId: "env-1", + threadId: "thread-1", + phase: "completed" as const, + runId: "run-1", + updatedAt: "1970-01-01T00:00:00.000Z", + }; + const claim = ( + input: Parameters[0], + ) => + Effect.flatMap(DeliveryAttempts.DeliveryAttempts, (attempts) => + attempts.claimActivityAlert(input), + ).pipe(Effect.provide(layer)); + + return Effect.gen(function* () { + const concurrent = yield* Effect.all([claim(alert), claim(alert)], { concurrency: 2 }); + expect(concurrent.toSorted()).toEqual([false, true]); + expect(yield* claim({ ...alert, updatedAt: "1970-01-01T00:01:00.000Z" })).toBe(false); + expect(yield* claim({ ...alert, runId: "run-2" })).toBe(true); + expect(yield* claim({ ...alert, deviceId: "device-2" })).toBe(true); + const { runId: _runId, ...legacyAlert } = alert; + expect(yield* claim(legacyAlert)).toBe(true); + expect(yield* claim(legacyAlert)).toBe(false); + }); + }); + it.effect("reports completed source jobs when the durable claim already exists", () => { const fakeDb = { insert: () => ({ diff --git a/infra/relay/src/agentActivity/DeliveryAttempts.ts b/infra/relay/src/agentActivity/DeliveryAttempts.ts index 161fa48cbce84..5fe73def5961b 100644 --- a/infra/relay/src/agentActivity/DeliveryAttempts.ts +++ b/infra/relay/src/agentActivity/DeliveryAttempts.ts @@ -7,6 +7,7 @@ import { and, eq, inArray, isNull, lt } from "drizzle-orm"; import { QueryBuilder } from "drizzle-orm/pg-core"; import * as Crypto from "effect/Crypto"; import * as Schema from "effect/Schema"; +import { stableStringify } from "@t3tools/shared/relaySigning"; import * as RelayDb from "../db.ts"; import { relayDeliveryAttempts } from "../persistence/schema.ts"; @@ -14,7 +15,12 @@ import { relayDeliveryAttempts } from "../persistence/schema.ts"; export class DeliveryAttemptRecordPersistenceError extends Schema.TaggedError()( "DeliveryAttemptRecordPersistenceError", { - operation: Schema.Literals(["claim-source-job", "complete-source-job", "prune-expired"]), + operation: Schema.Literals([ + "claim-source-job", + "complete-source-job", + "prune-expired", + "claim-activity-alert", + ]), sourceJobId: Schema.NullOr(Schema.String), userId: Schema.NullOr(Schema.String), environmentId: Schema.NullOr(Schema.String), @@ -56,6 +62,15 @@ export type DeliverySourceJobClaimResult = "claimed" | "completed" | "in_flight" export class DeliveryAttempts extends Context.Service< DeliveryAttempts, { + readonly claimActivityAlert: (input: { + readonly userId: string; + readonly deviceId: string; + readonly environmentId: string; + readonly threadId: string; + readonly phase: "completed" | "failed"; + readonly runId?: string; + readonly updatedAt: string; + }) => Effect.Effect; readonly claimSourceJob: ( input: DeliveryAttemptInput & { readonly sourceJobId: string }, ) => Effect.Effect; @@ -112,6 +127,60 @@ const make = Effect.gen(function* () { }; return DeliveryAttempts.of({ + claimActivityAlert: Effect.fn("relay.delivery_attempts.claim_activity_alert")( + function* (input) { + return yield* Effect.gen(function* () { + const identity = stableStringify([ + "activity-alert", + input.userId, + input.deviceId, + input.environmentId, + input.threadId, + input.phase, + input.runId ? ["run", input.runId] : ["updatedAt", input.updatedAt], + ]); + const digest = yield* crypto.digest("SHA-256", new TextEncoder().encode(identity)); + const sourceJobId = Array.from(digest, (byte) => byte.toString(16).padStart(2, "0")).join( + "", + ); + const id = yield* crypto.randomUUIDv4(); + const createdAt = DateTime.formatIso(yield* DateTime.now); + // Reserve the sound before sending: an uncertain APNs response must not let another job ring again. + const inserted = yield* db + .insert(relayDeliveryAttempts) + .values( + insertValues( + { + ...input, + sourceJobId, + kind: "activity_alert", + token: null, + apnsReason: "Activity alert reserved before delivery.", + }, + id, + createdAt, + ), + ) + .onConflictDoNothing({ target: relayDeliveryAttempts.sourceJobId }) + .returning({ id: relayDeliveryAttempts.id }); + return inserted.length > 0; + }).pipe( + Effect.mapError( + (cause) => + new DeliveryAttemptRecordPersistenceError({ + operation: "claim-activity-alert", + sourceJobId: null, + userId: input.userId, + environmentId: input.environmentId, + threadId: input.threadId, + deviceId: input.deviceId, + kind: "activity_alert", + cause, + }), + ), + ); + }, + ), claimSourceJob: Effect.fn("relay.delivery_attempts.claim_source_job")(function* (input) { yield* Effect.annotateCurrentSpan({ "relay.delivery.kind": input.kind, diff --git a/infra/relay/src/agentActivity/MobileRegistrations.test.ts b/infra/relay/src/agentActivity/MobileRegistrations.test.ts index 6f5997c931bdb..e6e676d82ff56 100644 --- a/infra/relay/src/agentActivity/MobileRegistrations.test.ts +++ b/infra/relay/src/agentActivity/MobileRegistrations.test.ts @@ -127,6 +127,7 @@ function makeDeliveryAttempts( overrides: Partial = {}, ): DeliveryAttempts.DeliveryAttempts["Service"] { return { + claimActivityAlert: () => Effect.succeed(true), claimSourceJob: () => Effect.succeed("claimed"), completeSourceJob: () => Effect.void, pruneExpired: Effect.succeed(0), diff --git a/infra/relay/src/agentActivity/agentActivityAggregate.ts b/infra/relay/src/agentActivity/agentActivityAggregate.ts index 99fdcf71ef90d..5b0df38d83473 100644 --- a/infra/relay/src/agentActivity/agentActivityAggregate.ts +++ b/infra/relay/src/agentActivity/agentActivityAggregate.ts @@ -35,6 +35,7 @@ function aggregateRowForState(state: RelayAgentActivityState) { return { environmentId: state.environmentId, threadId: state.threadId, + ...(state.runId ? { runId: state.runId } : {}), projectTitle: state.projectTitle, threadTitle: state.threadTitle, modelTitle: state.modelTitle, diff --git a/infra/relay/src/agentActivity/agentActivityAlerts.ts b/infra/relay/src/agentActivity/agentActivityAlerts.ts index c6008950e95e8..1015f784d0452 100644 --- a/infra/relay/src/agentActivity/agentActivityAlerts.ts +++ b/infra/relay/src/agentActivity/agentActivityAlerts.ts @@ -35,10 +35,14 @@ export function rowPhasesChanged( nextAggregate: RelayAgentActivityAggregateState, ): boolean { if (previousAggregate.activities.length !== nextAggregate.activities.length) return true; - const previousPhases = new Map( - previousAggregate.activities.map((row) => [rowKey(row), row.phase]), - ); - return nextAggregate.activities.some((row) => previousPhases.get(rowKey(row)) !== row.phase); + const previousRows = new Map(previousAggregate.activities.map((row) => [rowKey(row), row])); + return nextAggregate.activities.some((row) => { + const previous = previousRows.get(rowKey(row)); + return ( + previous?.phase !== row.phase || + (previous?.runId !== undefined && row.runId !== undefined && previous.runId !== row.runId) + ); + }); } function isAttentionPhase(phase: string): boolean { @@ -84,16 +88,15 @@ function newlyTerminalRows( includeUnobserved = false, ): ReadonlyArray { if (previousAggregate === null) return []; - const previousPhases = new Map( - previousAggregate.activities.map((row) => [rowKey(row), row.phase]), - ); + const previousRows = new Map(previousAggregate.activities.map((row) => [rowKey(row), row])); return nextAggregate.activities.filter((row) => { if (row.phase !== "completed" && row.phase !== "failed") return false; - const previousPhase = previousPhases.get(rowKey(row)); + const previous = previousRows.get(rowKey(row)); + if (previous?.runId && row.runId && previous.runId !== row.runId) return true; return ( - (includeUnobserved || previousPhase !== undefined) && - previousPhase !== "completed" && - previousPhase !== "failed" + (includeUnobserved || previous !== undefined) && + previous?.phase !== "completed" && + previous?.phase !== "failed" ); }); } diff --git a/infra/relay/src/agentActivity/agentActivityPayloads.ts b/infra/relay/src/agentActivity/agentActivityPayloads.ts index b2086426a9ec0..d6040f67b6360 100644 --- a/infra/relay/src/agentActivity/agentActivityPayloads.ts +++ b/infra/relay/src/agentActivity/agentActivityPayloads.ts @@ -109,6 +109,7 @@ export function notificationForActivity( body: `${activity.status}: ${activity.projectTitle}`, environmentId: activity.environmentId, threadId: activity.threadId, + ...(activity.runId ? { runId: activity.runId } : {}), deepLink: activity.deepLink, phase: activity.phase, updatedAt: activity.updatedAt, diff --git a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts index a1fe2aa0bce58..eddece494b684 100644 --- a/infra/relay/src/agentActivity/apnsDeliveryJobs.ts +++ b/infra/relay/src/agentActivity/apnsDeliveryJobs.ts @@ -29,6 +29,7 @@ const ApnsDeliveryJobContext = { }; export const ApnsNotificationPayload = Schema.Struct({ + runId: Schema.optional(Schema.String), title: Schema.String, body: Schema.String, environmentId: Schema.String, diff --git a/packages/contracts/src/relay.ts b/packages/contracts/src/relay.ts index 983f74de418fa..f672c92b11a2c 100644 --- a/packages/contracts/src/relay.ts +++ b/packages/contracts/src/relay.ts @@ -119,6 +119,7 @@ export type RelayDeviceUnregistrationParams = typeof RelayDeviceUnregistrationPa export const RelayAgentActivityState = Schema.Struct({ environmentId: EnvironmentId, threadId: ThreadId, + runId: Schema.optional(TrimmedNonEmptyString), projectTitle: TrimmedNonEmptyString, threadTitle: TrimmedNonEmptyString, phase: RelayAgentAwarenessPhase, @@ -133,6 +134,7 @@ export type RelayAgentActivityState = typeof RelayAgentActivityState.Type; export const RelayAgentActivityAggregateRow = Schema.Struct({ environmentId: EnvironmentId, threadId: ThreadId, + runId: Schema.optional(TrimmedNonEmptyString), projectTitle: TrimmedNonEmptyString, threadTitle: TrimmedNonEmptyString, modelTitle: TrimmedNonEmptyString, diff --git a/packages/shared/src/agentAwareness.test.ts b/packages/shared/src/agentAwareness.test.ts index caf261bfbb1e9..3eea2892f442c 100644 --- a/packages/shared/src/agentAwareness.test.ts +++ b/packages/shared/src/agentAwareness.test.ts @@ -6,7 +6,7 @@ import type { Project, ThreadId, } from "@t3tools/contracts"; -import { ProviderInstanceId, RuntimeRequestId } from "@t3tools/contracts"; +import { ProviderInstanceId, RuntimeRequestId, RunId } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import { projectThreadAwarenessV2 } from "./agentAwareness.ts"; @@ -63,6 +63,22 @@ describe("projectThreadAwarenessV2", () => { ).toBeNull(); }); + it("preserves the run identity when completed thread metadata changes", () => { + const thread = { ...v2Thread({ status: "completed" }), latestRunId: RunId.make("run-1") }; + for (const updatedThread of [ + thread, + { ...thread, title: "Renamed", updatedAt: DateTime.makeUnsafe("2026-05-22T12:01:00.000Z") }, + ]) { + expect( + projectThreadAwarenessV2({ + environmentId: "env-1" as EnvironmentId, + project, + thread: updatedThread, + }), + ).toMatchObject({ phase: "completed", runId: "run-1" }); + } + }); + it.each(["running", "completed", "failed"] as const)( "does not publish %s subagent activity", (status) => { diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index df912b4bb0f1b..10bd79c586342 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -20,6 +20,7 @@ export type AgentAwarenessPhase = export interface AgentAwarenessState { readonly environmentId: EnvironmentId; readonly threadId: ThreadId; + readonly runId?: string; readonly projectTitle: string; readonly threadTitle: string; readonly phase: AgentAwarenessPhase; @@ -53,7 +54,8 @@ export interface ProjectThreadAwarenessV2Input { | "status" | "title" | "updatedAt" - >; + > & + Partial>; } export function projectThreadAwarenessV2( @@ -83,6 +85,7 @@ export function projectThreadAwarenessV2( return { environmentId, threadId: thread.id, + ...(thread.latestRunId ? { runId: thread.latestRunId } : {}), projectTitle: project.title, threadTitle: thread.title, phase, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 6bdece6bd1839..33f48e6a7c6cb 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -6,6 +6,9 @@ importers: .: configDependencies: {} packageManagerDependencies: + '@pnpm/exe': + specifier: 12.14.0 + version: 12.14.0 pnpm: specifier: 12.14.0 version: 12.14.0 @@ -89,6 +92,11 @@ packages: cpu: [x64] os: [win32] + '@pnpm/exe@12.14.0': + resolution: {integrity: sha512-iZCbiKegYUapNVkZY4O8vjXt9EF/j34LbDvew6qUM5g/HrxMY38cFJaoE8+5ZWQfbt2YCsLfufI5TTh+GxUbWw==} + engines: {node: '>=18.*'} + hasBin: true + pnpm@12.14.0: resolution: {integrity: sha512-wYum2HmT09Fky16JSPMrLAbuhUSH4GzY2GeBNC+z5btGNDZwNHEehoArqpw3vqkGzhkprQjXCITJfYAd8TOJpg==} engines: {node: '>=18.*'} @@ -138,6 +146,23 @@ snapshots: '@pnpm/exe.win32-x64@12.14.0': optional: true + '@pnpm/exe@12.14.0': + optionalDependencies: + '@pnpm/exe.android-arm64': 12.14.0 + '@pnpm/exe.android-x64': 12.14.0 + '@pnpm/exe.darwin-arm64': 12.14.0 + '@pnpm/exe.darwin-x64': 12.14.0 + '@pnpm/exe.freebsd-x64': 12.14.0 + '@pnpm/exe.linux-arm64': 12.14.0 + '@pnpm/exe.linux-arm64-musl': 12.14.0 + '@pnpm/exe.linux-ppc64': 12.14.0 + '@pnpm/exe.linux-riscv64': 12.14.0 + '@pnpm/exe.linux-s390x': 12.14.0 + '@pnpm/exe.linux-x64': 12.14.0 + '@pnpm/exe.linux-x64-musl': 12.14.0 + '@pnpm/exe.win32-arm64': 12.14.0 + '@pnpm/exe.win32-x64': 12.14.0 + pnpm@12.14.0: optionalDependencies: '@pnpm/exe.android-arm64': 12.14.0 From 3401cff429ed96fc472de9e78511a361780439bd Mon Sep 17 00:00:00 2001 From: JamesOBrien2 Date: Wed, 28 Oct 2026 15:30:14 +0000 Subject: [PATCH 2/4] fix(shared): require run identity in activity projection --- packages/shared/src/agentAwareness.test.ts | 1 + packages/shared/src/agentAwareness.ts | 4 ++-- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/packages/shared/src/agentAwareness.test.ts b/packages/shared/src/agentAwareness.test.ts index 3eea2892f442c..1250ac649934c 100644 --- a/packages/shared/src/agentAwareness.test.ts +++ b/packages/shared/src/agentAwareness.test.ts @@ -42,6 +42,7 @@ describe("projectThreadAwarenessV2", () => { title: "Integrate orchestration", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, status: "running" as const, + latestRunId: null, pendingRuntimeRequest: null, settledOverride: null, updatedAt, diff --git a/packages/shared/src/agentAwareness.ts b/packages/shared/src/agentAwareness.ts index 10bd79c586342..505d6a125a25b 100644 --- a/packages/shared/src/agentAwareness.ts +++ b/packages/shared/src/agentAwareness.ts @@ -46,6 +46,7 @@ export interface ProjectThreadAwarenessV2Input { | "activityRunStatus" | "id" | "lineage" + | "latestRunId" | "latestRunStopRequested" | "modelSelection" | "pendingBackgroundTasks" @@ -54,8 +55,7 @@ export interface ProjectThreadAwarenessV2Input { | "status" | "title" | "updatedAt" - > & - Partial>; + >; } export function projectThreadAwarenessV2( From 63bc6ec4785ec97c86f052ee07affdf6b4b898a9 Mon Sep 17 00:00:00 2001 From: JamesOBrien2 Date: Wed, 28 Oct 2026 21:38:56 +0000 Subject: [PATCH 3/4] fix(relay): detect transitions from legacy run identities --- .../src/agentActivity/agentActivityAlerts.ts | 7 ++----- .../agentActivity/agentActivityPolicy.test.ts | 20 +++++++++++++++++++ 2 files changed, 22 insertions(+), 5 deletions(-) diff --git a/infra/relay/src/agentActivity/agentActivityAlerts.ts b/infra/relay/src/agentActivity/agentActivityAlerts.ts index 1015f784d0452..be6df95b534a1 100644 --- a/infra/relay/src/agentActivity/agentActivityAlerts.ts +++ b/infra/relay/src/agentActivity/agentActivityAlerts.ts @@ -38,10 +38,7 @@ export function rowPhasesChanged( const previousRows = new Map(previousAggregate.activities.map((row) => [rowKey(row), row])); return nextAggregate.activities.some((row) => { const previous = previousRows.get(rowKey(row)); - return ( - previous?.phase !== row.phase || - (previous?.runId !== undefined && row.runId !== undefined && previous.runId !== row.runId) - ); + return previous?.phase !== row.phase || (previous?.runId ?? null) !== (row.runId ?? null); }); } @@ -92,7 +89,7 @@ function newlyTerminalRows( return nextAggregate.activities.filter((row) => { if (row.phase !== "completed" && row.phase !== "failed") return false; const previous = previousRows.get(rowKey(row)); - if (previous?.runId && row.runId && previous.runId !== row.runId) return true; + if (previous !== undefined && (previous.runId ?? null) !== (row.runId ?? null)) return true; return ( (includeUnobserved || previous !== undefined) && previous?.phase !== "completed" && diff --git a/infra/relay/src/agentActivity/agentActivityPolicy.test.ts b/infra/relay/src/agentActivity/agentActivityPolicy.test.ts index e73cfd85145a5..97ce2fafdea92 100644 --- a/infra/relay/src/agentActivity/agentActivityPolicy.test.ts +++ b/infra/relay/src/agentActivity/agentActivityPolicy.test.ts @@ -4,6 +4,7 @@ import type { RelayAgentActivityState } from "@t3tools/contracts/relay"; import { makeAggregateState } from "./agentActivityAggregate.ts"; import { attentionTransitionRows, + rowPhasesChanged, terminalTransitionRows, shouldAlertForActivity, } from "./agentActivityAlerts.ts"; @@ -73,6 +74,25 @@ describe("shared agent activity policy", () => { ); }); + it.each([ + [undefined, "run-new", true], + ["run-old", undefined, true], + ["run-old", "run-new", true], + [undefined, undefined, false], + ["run-same", "run-same", false], + ] as const)("detects terminal run identity %s -> %s", (before, after, changed) => { + const previousAggregate = aggregate([ + { ...state, phase: "completed", ...(before === undefined ? {} : { runId: before }) }, + ]); + const nextAggregate = aggregate([ + { ...state, phase: "completed", ...(after === undefined ? {} : { runId: after }) }, + ]); + expect(rowPhasesChanged(previousAggregate, nextAggregate)).toBe(changed); + expect( + terminalTransitionRows({ previousAggregate, nextAggregate, preferences, nowMs: 0 }), + ).toHaveLength(changed ? 1 : 0); + }); + it("distinguishes equal thread IDs across environments", () => { const waiting = { ...state, phase: "waiting_for_input" as const }; const other = { ...waiting, environmentId: EnvironmentId.make("other") }; From 96fab796daa642e3827a1bd3260784992c7bb9e5 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Sat, 14 Nov 2026 09:27:57 +0000 Subject: [PATCH 4/4] fix(relay): adapt completion claims to main's SecureRandom and signed jobs Co-Authored-By: Claude Opus 5.5 --- .../src/agentActivity/AgentActivityPublisher.test.ts | 1 + infra/relay/src/agentActivity/ApnsDeliveries.test.ts | 10 ++++------ infra/relay/src/agentActivity/DeliveryAttempts.ts | 11 +++++------ 3 files changed, 10 insertions(+), 12 deletions(-) diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts index 679c410470397..8cb35a9f2bd04 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.test.ts @@ -402,6 +402,7 @@ describe("AgentActivityPublisher", () => { }), }), Layer.succeed(DeliveryAttempts.DeliveryAttempts, { + claimActivityAlert: () => Effect.succeed(true), pruneExpired: Effect.succeed(0), claimSourceJob: () => Effect.succeed("claimed" as const), completeSourceJob: () => Effect.void, diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index ae2120fcf1b5b..6bd668bbd828d 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -1295,12 +1295,10 @@ describe("fast completion delivery", () => { yield* Effect.forEach( ["first", "second"], (jobId) => - deliveries.processSignedJob( - signApnsDeliveryJob({ - secret: config.apnsDeliveryJobSigningSecret, - payload: makeApnsDeliveryJobPayload({ ...payload, jobId }), - }), - ), + signApnsDeliveryJob({ + secret: config.apnsDeliveryJobSigningSecret, + payload: makeApnsDeliveryJobPayload({ ...payload, jobId }), + }).pipe(Effect.flatMap(deliveries.processSignedJob)), { concurrency: 2 }, ); expect(requests).toHaveLength(2); diff --git a/infra/relay/src/agentActivity/DeliveryAttempts.ts b/infra/relay/src/agentActivity/DeliveryAttempts.ts index 21335491182dc..f7d4ed7152526 100644 --- a/infra/relay/src/agentActivity/DeliveryAttempts.ts +++ b/infra/relay/src/agentActivity/DeliveryAttempts.ts @@ -5,9 +5,11 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import { and, eq, inArray, isNull, lt } from "drizzle-orm"; import { QueryBuilder } from "drizzle-orm/pg-core"; +import { sha256 } from "@noble/hashes/sha2"; +import { bytesToHex } from "@noble/hashes/utils"; +import { stableStringify } from "@t3tools/shared/relaySigning"; import * as SecureRandom from "effect/crypto/SecureRandom"; import * as Schema from "effect/Schema"; -import { stableStringify } from "@t3tools/shared/relaySigning"; import * as RelayDb from "../db.ts"; import { relayDeliveryAttempts } from "../persistence/schema.ts"; @@ -139,11 +141,8 @@ const make = Effect.gen(function* () { input.phase, input.runId ? ["run", input.runId] : ["updatedAt", input.updatedAt], ]); - const digest = yield* crypto.digest("SHA-256", new TextEncoder().encode(identity)); - const sourceJobId = Array.from(digest, (byte) => byte.toString(16).padStart(2, "0")).join( - "", - ); - const id = yield* crypto.randomUUIDv4(); + const sourceJobId = bytesToHex(sha256(new TextEncoder().encode(identity))); + const id = yield* random.uuidV4(); const createdAt = DateTime.formatIso(yield* DateTime.now); // Reserve the sound before sending: an uncertain APNs response must not let another job ring again. const inserted = yield* db