diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts index 1e7cdc3f00e6..6370034036bb 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.test.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts @@ -12,21 +12,29 @@ import type { ThreadId, TurnId, } from "@t3tools/contracts"; +import { RelayAgentActivityPublishRequest } from "@t3tools/contracts/relay"; import type { RelayAgentActivityPublishProofPayload, RelayAgentActivityState, } from "@t3tools/contracts/relay"; import { CommandId, ProviderInstanceId } from "@t3tools/contracts"; import { RelayClientTracer } from "@t3tools/shared/relayTracing"; -import { RELAY_ACTIVITY_PUBLISH_TYP, verifyRelayJwt } from "@t3tools/shared/relayJwt"; +import { + decodeRelayJwt, + RELAY_ACTIVITY_PUBLISH_TYP, + verifyRelayJwt, +} from "@t3tools/shared/relayJwt"; import { describe, expect, it } from "@effect/vitest"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Queue from "effect/Queue"; +import * as Schema from "effect/Schema"; +import * as TestClock from "effect/testing/TestClock"; import * as Stream from "effect/Stream"; import * as Tracer from "effect/Tracer"; +import * as FetchHttpClient from "effect/unstable/http/FetchHttpClient"; import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; @@ -59,6 +67,8 @@ const state: RelayAgentActivityState = { deepLink: "/threads/env/thread", }; +const decodePublishRequest = Schema.decodeUnknownSync(RelayAgentActivityPublishRequest); + const encodeSecret = (value: string): Uint8Array => new TextEncoder().encode(value); function makeMemorySecretStore() { @@ -232,7 +242,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { type: "thread.settled", payload: { threadId: "thread-1" as ThreadId }, } as unknown as OrchestrationEvent), - ).toBe(true); + ).toBe(false); }); it("deduplicates awareness state updates whose only change is their event timestamp", () => { @@ -349,46 +359,48 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { } satisfies Omit; expect( - AgentAwarenessRelay.resolveAgentAwarenessRelayActiveThreadIds({ - environmentId, - projects: [ - { - id: projectId, - title: "T3 Code", - }, - ], - threads: [ - { - ...baseThread, - id: activeThreadId, - latestTurn: { - turnId: "turn-1" as TurnId, - state: "running", - requestedAt: now, - startedAt: now, - completedAt: null, - assistantMessageId: null, + Array.from( + AgentAwarenessRelay.resolveAgentAwarenessRelayActiveStates({ + environmentId, + projects: [ + { + id: projectId, + title: "T3 Code", }, - }, - { - ...baseThread, - id: idleThreadId, - }, - { - ...baseThread, - id: "thread-missing-project" as ThreadId, - projectId: "missing-project" as ProjectId, - latestTurn: { - turnId: "turn-2" as TurnId, - state: "running", - requestedAt: now, - startedAt: now, - completedAt: null, - assistantMessageId: null, + ], + threads: [ + { + ...baseThread, + id: activeThreadId, + latestTurn: { + turnId: "turn-1" as TurnId, + state: "running", + requestedAt: now, + startedAt: now, + completedAt: null, + assistantMessageId: null, + }, }, - }, - ], - }), + { + ...baseThread, + id: idleThreadId, + }, + { + ...baseThread, + id: "thread-missing-project" as ThreadId, + projectId: "missing-project" as ProjectId, + latestTurn: { + turnId: "turn-2" as TurnId, + state: "running", + requestedAt: now, + startedAt: now, + completedAt: null, + assistantMessageId: null, + }, + }, + ], + }).keys(), + ), ).toEqual([activeThreadId]); }); @@ -621,15 +633,23 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { ), ); - it.effect("publishes agent activity to the relay transport URL, not the relay issuer", () => + const sources = [ + "event", + "running", + "waiting_for_input", + "completed", + "failed", + "changed", + "transient-null", + ] as const; + it.effect.each(sources)("publishes %s with silent startup replay", (source) => Effect.scoped( Effect.gen(function* () { - const originalFetch = globalThis.fetch; const events = yield* Queue.unbounded(); - let resolveFetchSeen: (url: URL) => void = () => {}; - const fetchSeen = new Promise((resolve) => { - resolveFetchSeen = resolve; - }); + const requests = yield* Queue.unbounded<{ + url: URL; + payload: RelayAgentActivityPublishRequest; + }>(); const userSpans: Array = []; const productSpans: Array = []; const collectingTracer = (spans: Array) => @@ -661,7 +681,8 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { updatedAt: now, } satisfies OrchestrationProjectShell; - const thread = { + let shellReads = 0; + let thread: OrchestrationThreadShell = { id: threadId, projectId, title: "Run remote agent", @@ -673,10 +694,10 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { pullRequests: [], latestTurn: { turnId: "turn-1" as TurnId, - state: "running", + state: source === "completed" ? "completed" : source === "failed" ? "error" : "running", requestedAt: now, startedAt: now, - completedAt: null, + completedAt: source === "completed" || source === "failed" ? now : null, assistantMessageId: null, }, createdAt: now, @@ -686,18 +707,19 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { settledAt: null, session: { threadId, - status: "running", + status: source === "completed" ? "ready" : source === "failed" ? "error" : "running", providerName: "Codex", runtimeMode: "full-access", - activeTurnId: "turn-1" as TurnId, + activeTurnId: + source === "completed" || source === "failed" ? null : ("turn-1" as TurnId), lastError: null, updatedAt: now, }, latestUserMessageAt: now, hasPendingApprovals: false, - hasPendingUserInput: false, + hasPendingUserInput: source === "waiting_for_input", hasActionableProposedPlan: false, - } satisfies OrchestrationThreadShell; + }; const descriptor = { environmentId, @@ -712,20 +734,12 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { }, } satisfies ExecutionEnvironmentDescriptor; - globalThis.fetch = ((input: Parameters[0]) => { - const url = new URL( - typeof input === "string" || input instanceof URL - ? input - : (input as unknown as { readonly url: string }).url, - ); - resolveFetchSeen(url); - return Promise.resolve(Response.json({ ok: true, deliveries: [] })); - }) as unknown as typeof fetch; - yield* Effect.addFinalizer(() => - Effect.sync(() => { - globalThis.fetch = originalFetch; - }), - ); + const fetch: typeof globalThis.fetch = async (input, init) => { + const request = new Request(input, init); + const payload = decodePublishRequest(await request.json()); + Queue.offerUnsafe(requests, { url: new URL(request.url), payload }); + return Response.json({ ok: true, deliveries: [] }); + }; const layer = Layer.mergeAll( Layer.succeed(ServerSecretStore.ServerSecretStore, secrets.store), @@ -750,7 +764,13 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { threads: [thread], updatedAt: now, } satisfies OrchestrationShellSnapshot), - getThreadShellById: () => Effect.succeed(Option.some(thread)), + getThreadShellById: () => + Effect.sync(() => { + if (source === "transient-null" && shellReads++ === 0) return Option.none(); + return Option.some( + source === "changed" ? { ...thread, hasPendingUserInput: true } : thread, + ); + }), getProjectShellById: () => Effect.succeed(Option.some(project)), } as unknown as ProjectionSnapshotQueryShape), ); @@ -762,26 +782,47 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { yield* secrets.setString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET, "relay-credential"); yield* secrets.setString(PUBLISH_AGENT_ACTIVITY_SECRET, "true"); yield* relay.start(); - yield* Queue.offer(events, { - type: "thread.activity-appended", - sequence: 1, - eventId: "evt-1", - commandId: CommandId.make("cmd-1"), - aggregateKind: "thread", - aggregateId: threadId, - actor: { kind: "server" }, - metadata: {}, - payload: { - threadId, - activity: { - kind: "approval.requested", + if (source === "event") { + yield* Queue.offer(events, { + type: "thread.activity-appended", + sequence: 1, + eventId: "evt-1", + commandId: CommandId.make("cmd-1"), + aggregateKind: "thread", + aggregateId: threadId, + actor: { kind: "server" }, + metadata: {}, + payload: { + threadId, + activity: { + kind: "approval.requested", + }, }, - }, - occurredAt: now, - } as unknown as OrchestrationEvent); - - const url = yield* Effect.promise(() => fetchSeen).pipe(Effect.timeout("2 seconds")); + occurredAt: now, + } as unknown as OrchestrationEvent); + } else { + yield* TestClock.adjust("1 second"); + yield* TestClock.adjust("5 seconds"); + } + const { url, payload } = yield* Queue.take(requests); expect(url.origin).toBe("https://transport.example.test"); + if (source !== "event") { + expect(payload.state).not.toBeNull(); + if (source === "changed") { + expect(payload.state?.phase).toBe("waiting_for_input"); + expect(payload).not.toMatchObject({ replay: true }); + } else { + expect(payload).toMatchObject({ replay: true }); + expect(decodeRelayJwt(payload.proof)).toMatchObject({ replay: true }); + } + thread = { ...thread, hasPendingUserInput: false, hasPendingApprovals: true }; + yield* relay.publishThread(threadId); + const live = yield* Queue.take(requests); + expect(live.payload.state?.phase).toBe("waiting_for_approval"); + expect(live.payload).not.toMatchObject({ replay: true }); + } else { + expect(payload).not.toMatchObject({ replay: true }); + } expect(productSpans).toContain("makePublishProof"); expect(userSpans).not.toContain("makePublishProof"); }).pipe( @@ -791,6 +832,7 @@ describe.sequential("signRelayAgentActivityPublishProof", () => { Layer.provideMerge(NodeServices.layer), ), ), + Effect.provideService(FetchHttpClient.Fetch, fetch), Effect.provideService(RelayClientTracer, Option.some(collectingTracer(productSpans))), Effect.withTracer(collectingTracer(userSpans)), ); diff --git a/apps/server/src/relay/AgentAwarenessRelay.ts b/apps/server/src/relay/AgentAwarenessRelay.ts index 052a67959ad1..a214eb2ba7de 100644 --- a/apps/server/src/relay/AgentAwarenessRelay.ts +++ b/apps/server/src/relay/AgentAwarenessRelay.ts @@ -79,6 +79,7 @@ export function shouldPublishAgentAwarenessEvent(event: OrchestrationEvent): boo // before the real running state arrives. Provider lifecycle events publish // the authoritative starting/running state instead. return false; + case "thread.settled": case "thread.proposed-plan-upserted": case "thread.runtime-mode-set": case "thread.interaction-mode-set": @@ -193,6 +194,7 @@ const makePublishProof = Effect.fn("makePublishProof")(function* (input: { readonly threadId: ThreadId; readonly state: RelayAgentActivityState | null; readonly jti: string; + readonly replay: boolean; }) { const now = yield* DateTime.now; const expiresAt = DateTime.add(now, { minutes: 5 }); @@ -206,6 +208,7 @@ const makePublishProof = Effect.fn("makePublishProof")(function* (input: { environmentId: input.environmentId as RelayAgentActivityPublishProofPayload["environmentId"], threadId: input.threadId, state: input.state, + ...(input.replay ? { replay: true } : {}), } satisfies RelayAgentActivityPublishProofPayload; return yield* signRelayAgentActivityPublishProof({ privateKey: input.privateKey, payload }); }); @@ -267,27 +270,26 @@ export function resolveAgentAwarenessRelayPublishSnapshot(input: { }; } -export function resolveAgentAwarenessRelayActiveThreadIds(input: { +export function resolveAgentAwarenessRelayActiveStates(input: { readonly environmentId: EnvironmentId; readonly projects: ReadonlyArray>; readonly threads: ReadonlyArray; -}): ReadonlyArray { +}): ReadonlyMap { const projectById = new Map(input.projects.map((project) => [project.id, project])); - return input.threads - .filter((thread) => { - const project = projectById.get(thread.projectId); - if (!project) { - return false; - } - return ( - projectThreadAwareness({ - environmentId: input.environmentId, - project, - thread, - }) !== null - ); - }) - .map((thread) => thread.id); + const states = new Map(); + for (const thread of input.threads) { + const project = projectById.get(thread.projectId); + if (!project) continue; + const state = sanitizeRelayAgentActivityState( + projectThreadAwareness({ + environmentId: input.environmentId, + project, + thread, + }), + ); + if (state !== null) states.set(thread.id, state); + } + return states; } /** @public Service construction is part of the canonical Effect module API. */ @@ -340,9 +342,15 @@ export const make = Effect.gen(function* () { // tombstone can never race an in-flight live update; a recovered state // clears the deadline. Assigned after the worker exists. const publishConfirmDeadlines = new Map(); - let schedulePublishConfirm: (threadId: ThreadId) => Effect.Effect = () => Effect.void; - - const publishThreadUnsafe = Effect.fn("publishThreadUnsafe")(function* (threadId: ThreadId) { + let schedulePublishConfirm: ( + threadId: ThreadId, + replayIdentity?: string, + ) => Effect.Effect = () => Effect.void; + + const publishThreadUnsafe = Effect.fn("publishThreadUnsafe")(function* ( + threadId: ThreadId, + replayIdentity?: string, + ) { const publishAgentActivity = yield* readPublishAgentActivityEnabled.pipe( Effect.orElseSucceed(() => false), ); @@ -375,6 +383,7 @@ export const make = Effect.gen(function* () { threadId, state: input.state, jti: yield* crypto.randomUUIDv4, + replay, }); yield* Effect.logInfo("publishing agent activity for thread", { @@ -394,6 +403,7 @@ export const make = Effect.gen(function* () { payload: { state: input.state, proof, + ...(replay ? { replay: true } : {}), }, }); @@ -416,6 +426,7 @@ export const make = Effect.gen(function* () { project, }); const publishIdentity = agentAwarenessPublishIdentity(snapshot.state); + const replay = publishIdentity === replayIdentity; const publishedStateByThread = yield* Ref.get(publishedStateByThreadRef); if (publishedStateByThread.get(threadId) === publishIdentity) { // The projection is back at (or never left) the last published state, so @@ -457,7 +468,7 @@ export const make = Effect.gen(function* () { statePhase: snapshot.state?.phase ?? null, shell: describeThreadShellForAwareness(thread), }); - yield* schedulePublishConfirm(threadId); + yield* schedulePublishConfirm(threadId, replayIdentity); return; } if (nowMs < deadline) { @@ -500,8 +511,8 @@ export const make = Effect.gen(function* () { }); }); - const publishThread: AgentAwarenessRelay["Service"]["publishThread"] = (threadId) => - publishThreadUnsafe(threadId).pipe( + const publishThread = (threadId: ThreadId, replayIdentity?: string) => + publishThreadUnsafe(threadId, replayIdentity).pipe( Effect.catchCause((cause) => { return Effect.logWarning("agent activity publish failed", { threadId, @@ -512,6 +523,11 @@ export const make = Effect.gen(function* () { withRelayClientTracing, ); + const worker = yield* makeDrainableWorker( + (input: { readonly threadId: ThreadId; readonly replayIdentity?: string }) => + publishThread(input.threadId, input.replayIdentity), + ); + const publishActiveThreadsUnsafe = Effect.gen(function* () { const publishAgentActivity = yield* readPublishAgentActivityEnabled.pipe( Effect.orElseSucceed(() => false), @@ -527,19 +543,27 @@ export const make = Effect.gen(function* () { } const environmentId = yield* serverEnvironment.getEnvironmentId; const snapshot = yield* snapshotQuery.getShellSnapshot(); - const activeThreadIds = resolveAgentAwarenessRelayActiveThreadIds({ + const activeStates = resolveAgentAwarenessRelayActiveStates({ environmentId, projects: snapshot.projects, threads: snapshot.threads, }); - if (activeThreadIds.length === 0) { + if (activeStates.size === 0) { yield* Effect.logDebug("agent activity snapshot has no publishable threads"); return true; } yield* Effect.logInfo("publishing active agent activity snapshot", { - count: activeThreadIds.length, + count: activeStates.size, }); - yield* Effect.forEach(activeThreadIds, publishThread, { concurrency: 4, discard: true }); + // Serialize catch-up with live events. Only the captured state is silent; + // a thread that changes before this drains still gets its live alert. + yield* Effect.forEach(activeStates, ([threadId, state]) => + worker.enqueue({ + threadId, + replayIdentity: agentAwarenessPublishIdentity(state), + }), + ); + yield* worker.drain; return true; }); @@ -561,12 +585,12 @@ export const make = Effect.gen(function* () { } }); - const worker = yield* makeDrainableWorker(publishThread); - - schedulePublishConfirm = (threadId) => + schedulePublishConfirm = (threadId, replayIdentity) => Effect.forkDetach( Effect.sleep("5 seconds").pipe( - Effect.andThen(worker.enqueue(threadId)), + Effect.andThen( + worker.enqueue({ threadId, ...(replayIdentity === undefined ? {} : { replayIdentity }) }), + ), Effect.catchCause((cause) => Effect.logWarning("deferred agent activity confirmation failed", { threadId, @@ -626,7 +650,7 @@ export const make = Effect.gen(function* () { return Effect.logDebug("agent activity publishing queued thread publish", { eventType: event.type, threadId, - }).pipe(Effect.andThen(worker.enqueue(threadId))); + }).pipe(Effect.andThen(worker.enqueue({ threadId }))); }), ); }, diff --git a/infra/relay/src/agentActivity/AgentActivityPublisher.ts b/infra/relay/src/agentActivity/AgentActivityPublisher.ts index 61a420dd7858..5e3453be8d06 100644 --- a/infra/relay/src/agentActivity/AgentActivityPublisher.ts +++ b/infra/relay/src/agentActivity/AgentActivityPublisher.ts @@ -39,6 +39,7 @@ export class AgentActivityPublisher extends Context.Service< readonly environmentPublicKey: string; readonly threadId: string; readonly state: RelayAgentActivityState | null; + readonly replay?: boolean; }) => Effect.Effect; readonly replayForLiveActivityRegistration: (input: { readonly userId: string; @@ -58,6 +59,7 @@ export const make = Effect.gen(function* () { readonly deliveryUser: EnvironmentLinks.AgentAwarenessDeliveryUserRecord; readonly state: RelayAgentActivityState | null; readonly nowMs: number; + readonly replay?: boolean; }) { const activeStates = input.deliveryUser.liveActivitiesEnabled ? yield* rows.listForUser({ userId: input.deliveryUser.userId }) @@ -84,7 +86,13 @@ export const make = Effect.gen(function* () { targets, Effect.fnUntraced(function* (target) { if (target.platform === "android") { - return [yield* fcmDeliveries.enqueue({ target, state: input.state })]; + return [ + yield* fcmDeliveries.enqueue({ + target, + state: input.state, + ...(input.replay ? { replay: true } : {}), + }), + ]; } return yield* Effect.all( [ @@ -92,8 +100,9 @@ export const make = Effect.gen(function* () { target, aggregate: liveActivityAggregate, nowMs: input.nowMs, + ...(input.replay ? { replay: true } : {}), }), - notificationOnlyAggregate === null + input.replay || notificationOnlyAggregate === null ? Effect.succeed(null) : apnsDeliveries.sendPushNotificationForTarget({ target, @@ -178,6 +187,7 @@ export const make = Effect.gen(function* () { deliveryUser, state: input.state, nowMs: now.epochMilliseconds, + ...(input.replay ? { replay: true } : {}), }), { concurrency: 4 }, ); diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts index e98c2b639598..01c97f69da34 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.test.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.test.ts @@ -34,6 +34,9 @@ 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 * as AgentActivityPublisher from "./AgentActivityPublisher.ts"; +import * as EnvironmentLinks from "../environments/EnvironmentLinks.ts"; +import * as FcmDeliveries from "./FcmDeliveries.ts"; const config = RelayConfiguration.RelayConfiguration.of({ relayIssuer: "https://relay.example.test", @@ -179,7 +182,7 @@ function makeLayer(input: { Layer.provide(ApnsClient.layer), Layer.provide(ApnsProviderTokens.layer), Layer.provide(ApnsDeliveryQueue.layer.pipe(Layer.provide(NodeCryptoLayer.layer))), - Layer.provide( + Layer.provideMerge( Layer.mergeAll( Layer.succeed(AgentActivityRows.AgentActivityRows, { upsert: () => Effect.void, @@ -257,6 +260,79 @@ function makeLayer(input: { } describe("ApnsDeliveries", () => { + it.effect.each(["notification-only", "unarmed", "armed"] as const)( + "keeps environment startup replay silent for %s devices and delivers live input", + (mode) => { + const waiting = { ...state, phase: "waiting_for_input" as const }; + const queuedJobs: SignedApnsDeliveryJob[] = []; + const device = { + ...target, + push_token: "push-token", + activity_push_token: mode === "armed" ? "activity-token" : null, + preferences_json: mode === "notification-only" ? disabledPreferences : enabledPreferences, + last_aggregate_json: JSON.stringify(aggregate), + }; + return Effect.gen(function* () { + const publisher = yield* AgentActivityPublisher.AgentActivityPublisher; + const input = { + environmentId: state.environmentId, + environmentPublicKey: "key", + threadId: state.threadId, + state: waiting, + }; + yield* publisher.publish({ ...input, replay: true }); + expect(queuedJobs).toHaveLength(mode === "armed" ? 1 : 0); + expect(queuedJobs.every((job) => !job.payload.alert && !job.payload.notification)).toBe( + true, + ); + queuedJobs.length = 0; + yield* publisher.publish(input); + expect(queuedJobs).toHaveLength(1); + expect(queuedJobs[0]?.payload.alert ?? queuedJobs[0]?.payload.notification).toMatchObject({ + title: "Thread", + body: "Input: Project", + }); + }).pipe( + Effect.provide( + AgentActivityPublisher.layer.pipe( + Layer.provide( + makeLayer({ + attempts: [], + queuedJobs, + currentTargets: [device], + activityStates: [waiting], + }), + ), + Layer.provide( + Layer.succeed(FcmDeliveries.FcmDeliveries, { + enqueue: () => Effect.succeed(null), + process: () => Effect.void, + }), + ), + Layer.provide( + Layer.succeed(EnvironmentLinks.EnvironmentLinks, { + upsert: () => Effect.void, + listUsersForEnvironment: () => Effect.succeed([device.user_id]), + listDeliveryUsersForEnvironment: () => + Effect.succeed([ + { + userId: device.user_id, + notificationsEnabled: true, + liveActivitiesEnabled: mode !== "notification-only", + }, + ]), + listPublicKeysForEnvironment: () => Effect.succeed([]), + listForUser: () => Effect.succeed([]), + getForUser: () => Effect.succeed(null), + revokeForUser: () => Effect.succeed(false), + }), + ), + ), + ), + ); + }, + ); + it.effect("skips Apple delivery when an Android-only relay disables APNs", () => { const attempts: Array = []; const queuedJobs: Array = []; @@ -1767,6 +1843,7 @@ describe("live activity alert decisions", () => { activities: [...aggregate.activities, attentionRow], }, preferences, + nowMs: 0, }); expect(alert).toEqual({ title: "Blocked thread", body: "Approval: Project" }); }); @@ -1782,6 +1859,7 @@ describe("live activity alert decisions", () => { previousAggregate: withAttention, nextAggregate: withAttention, preferences, + nowMs: 0, }), ).toBeNull(); }); @@ -1792,6 +1870,7 @@ describe("live activity alert decisions", () => { previousAggregate: null, nextAggregate: { ...aggregate, activities: [attentionRow] }, preferences, + nowMs: 0, }), ).toBeNull(); }); @@ -1806,6 +1885,7 @@ describe("live activity alert decisions", () => { activities: [...aggregate.activities, attentionRow], }, preferences: { ...preferences, notifyOnApproval: false }, + nowMs: 0, }), ).toBeNull(); }); @@ -1826,6 +1906,7 @@ describe("live activity alert decisions", () => { activities: [...aggregate.activities, attentionRow, secondAttentionRow], }, preferences, + nowMs: 0, }); expect(alert).toEqual({ title: "2 agents need attention", @@ -1909,6 +1990,54 @@ describe("live activity alert decisions", () => { }); describe("queued iOS alert policy", () => { + for (const phase of ["waiting_for_input", "waiting_for_approval"] as const) { + for (const scenario of ["fresh", "stale at enqueue", "stale at delivery"] as const) { + it.effect(`${phase} keeps card updates and checks alerts that are ${scenario}`, () => { + const queuedJobs: SignedApnsDeliveryJob[] = []; + const requests: string[] = []; + const waiting = { ...state, phase }; + const nextAggregate = { + ...aggregate, + activities: [{ ...aggregate.activities[0]!, phase }], + }; + const device = { ...target, last_aggregate_json: JSON.stringify(aggregate) }; + return Effect.gen(function* () { + const deliveries = yield* ApnsDeliveries.ApnsDeliveries; + const staleAtEnqueue = scenario === "stale at enqueue"; + yield* TestClock.adjust(staleAtEnqueue ? 120_001 : 120_000); + yield* deliveries.sendForTarget({ + target: device, + aggregate: nextAggregate, + nowMs: staleAtEnqueue ? 120_001 : 120_000, + }); + expect(queuedJobs).toHaveLength(1); + expect(Boolean(queuedJobs[0]?.payload.alert)).toBe(!staleAtEnqueue); + if (scenario === "stale at delivery") yield* TestClock.adjust(1); + yield* deliveries.processSignedJob(queuedJobs[0]); + expect(requests).toHaveLength(1); + expect(requests[0]).toContain('"content-state"'); + expect(requests[0]?.includes('"alert":')).toBe(scenario === "fresh"); + }).pipe( + Effect.provide( + makeLayer({ + attempts: [], + queuedJobs, + config: signingConfig, + currentTargets: [device], + activityStates: [waiting], + execute: (request) => + Effect.sync(() => { + if (request.body._tag === "Uint8Array") + requests.push(new TextDecoder().decode(request.body.body)); + return HttpClientResponse.fromWeb(request, new Response("", { status: 200 })); + }), + }), + ), + ); + }); + } + } + for (const scenario of ["enabled", "muted", "late"] as const) { it.effect(`checks the current policy for a ${scenario} completion`, () => { let sent = 0; diff --git a/infra/relay/src/agentActivity/ApnsDeliveries.ts b/infra/relay/src/agentActivity/ApnsDeliveries.ts index 651f031f0efa..d2026d7d7b7d 100644 --- a/infra/relay/src/agentActivity/ApnsDeliveries.ts +++ b/infra/relay/src/agentActivity/ApnsDeliveries.ts @@ -293,6 +293,7 @@ function chooseLiveActivityDelivery(input: { previousAggregate, nextAggregate, preferences, + nowMs: input.nowMs, }) ?? alertForNewlyTerminal({ previousAggregate, @@ -744,6 +745,7 @@ export const make = Effect.gen(function* () { previousAggregate, nextAggregate: aggregate, preferences, + nowMs: now.epochMilliseconds, }) ?? alertForNewlyTerminal({ previousAggregate, diff --git a/infra/relay/src/agentActivity/FcmDeliveries.test.ts b/infra/relay/src/agentActivity/FcmDeliveries.test.ts index 73b90f99b0a1..350e7fd722ae 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.test.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.test.ts @@ -324,6 +324,26 @@ describe("Android delivery routing", () => { }); } + it.effect.each([false, true])( + "startup replay stays silent with notification-only mode %s", + (notificationOnly) => { + const h = harness(); + h.current.state = { ...state, phase: "waiting_for_input" }; + h.current.target.last_aggregate_json = encodeJson(aggregateFor([state])); + if (notificationOnly) h.current.notificationOnlyEnvironments = [state.environmentId]; + return Effect.gen(function* () { + const delivery = yield* FcmDeliveries; + yield* delivery.process({ ...h.job, state: h.current.state, replay: true }); + expect(h.sent.every((message) => !message.alert)).toBe(true); + h.current.state = { ...state, phase: "running" }; + yield* delivery.process({ ...h.job, state: h.current.state }); + h.current.state = { ...state, phase: "waiting_for_input" }; + yield* delivery.process({ ...h.job, state: h.current.state }); + expect(h.sent.filter((message) => message.alert)).toHaveLength(1); + }).pipe(Effect.provide(h.layer)); + }, + ); + it.effect("registration replay establishes a baseline without alerting", () => { const h = harness(); h.current.state = { ...state, phase: "waiting_for_approval" }; diff --git a/infra/relay/src/agentActivity/FcmDeliveries.ts b/infra/relay/src/agentActivity/FcmDeliveries.ts index 8f998d9d9574..96ea51f71756 100644 --- a/infra/relay/src/agentActivity/FcmDeliveries.ts +++ b/infra/relay/src/agentActivity/FcmDeliveries.ts @@ -215,7 +215,7 @@ export const make = Effect.gen(function* () { // update the card, but must leave that transition for its own alert job. // Registration replay deliberately establishes a silent baseline. let acknowledgeAggregate = job.state !== null || job.replay === true || aggregate === null; - if (job.state && preferences.value.notificationsEnabled) { + if (!job.replay && job.state && preferences.value.notificationsEnabled) { const state = yield* rows.getForUserThread({ userId: job.userId, environmentId: job.state.environmentId, diff --git a/infra/relay/src/agentActivity/agentActivityAlerts.ts b/infra/relay/src/agentActivity/agentActivityAlerts.ts index c8e7814eed1d..82f0ea351849 100644 --- a/infra/relay/src/agentActivity/agentActivityAlerts.ts +++ b/infra/relay/src/agentActivity/agentActivityAlerts.ts @@ -11,19 +11,18 @@ export interface AgentActivityAlert { readonly body: string; } -export const TERMINAL_NOTIFICATION_FRESHNESS_MS = 2 * 60 * 1_000; +export const NOTIFICATION_FRESHNESS_MS = 2 * 60 * 1_000; -export function isFreshTerminalNotification(updatedAt: string, nowMs: number): boolean { +export function isFreshNotification(updatedAt: string, nowMs: number): boolean { const timestamp = Option.getOrNull(DateTime.make(updatedAt)); - return ( - timestamp !== null && nowMs - timestamp.epochMilliseconds <= TERMINAL_NOTIFICATION_FRESHNESS_MS - ); + return timestamp !== null && nowMs - timestamp.epochMilliseconds <= NOTIFICATION_FRESHNESS_MS; } type TransitionInput = { readonly previousAggregate: RelayAgentActivityAggregateState | null; readonly nextAggregate: RelayAgentActivityAggregateState; readonly preferences: RelayAgentAwarenessPreferences | null; + readonly nowMs: number; }; function rowKey(row: RelayAgentActivityAggregateRow): string { @@ -63,7 +62,8 @@ export function attentionTransitionRows(input: TransitionInput) { (row) => isAttentionPhase(row.phase) && !previouslyAttention.has(rowKey(row)) && - alertAllowedForPhase(input.preferences, row.phase), + alertAllowedForPhase(input.preferences, row.phase) && + isFreshNotification(row.updatedAt, input.nowMs), ); } @@ -90,7 +90,7 @@ export function newlyTerminalRows( } export function terminalTransitionRows( - input: TransitionInput & { readonly nowMs: number; readonly includeUnobserved?: boolean }, + input: TransitionInput & { readonly includeUnobserved?: boolean }, ) { return newlyTerminalRows( input.previousAggregate, @@ -99,7 +99,7 @@ export function terminalTransitionRows( ).filter((row) => { return ( alertAllowedForPhase(input.preferences, row.phase) && - isFreshTerminalNotification(row.updatedAt, input.nowMs) + isFreshNotification(row.updatedAt, input.nowMs) ); }); } @@ -123,7 +123,7 @@ export function alertForAttentionTransition(input: TransitionInput): AgentActivi } export function alertForNewlyTerminal( - input: TransitionInput & { readonly nowMs: number; readonly includeUnobserved?: boolean }, + input: TransitionInput & { readonly includeUnobserved?: boolean }, ): AgentActivityAlert | null { return alertForActivityRows(terminalTransitionRows(input)); } @@ -146,7 +146,6 @@ export function shouldAlertForActivity(input: { return ( input.preferences?.notificationsEnabled === true && alertAllowedForPhase(input.preferences, input.phase) && - ((input.phase !== "completed" && input.phase !== "failed") || - isFreshTerminalNotification(input.updatedAt, input.nowMs)) + isFreshNotification(input.updatedAt, input.nowMs) ); } diff --git a/infra/relay/src/agentActivity/agentActivityPolicy.test.ts b/infra/relay/src/agentActivity/agentActivityPolicy.test.ts index e73cfd85145a..35163532e17c 100644 --- a/infra/relay/src/agentActivity/agentActivityPolicy.test.ts +++ b/infra/relay/src/agentActivity/agentActivityPolicy.test.ts @@ -48,6 +48,7 @@ describe("shared agent activity policy", () => { previousAggregate: aggregate(running), nextAggregate: next, preferences, + nowMs: 0, }), ).toMatchObject([{ threadId: state.threadId }]); }, @@ -81,10 +82,20 @@ describe("shared agent activity policy", () => { previousAggregate: aggregate([waiting]), nextAggregate: aggregate([waiting, other]), preferences, + nowMs: 0, }), ).toMatchObject([{ environmentId: other.environmentId }]); }); + it.each(["waiting_for_input", "waiting_for_approval"] as const)( + "does not alert on stale %s state after reconnect", + (phase) => { + const input = { ...state, phase, preferences }; + expect(shouldAlertForActivity({ ...input, nowMs: 120_000 })).toBe(true); + expect(shouldAlertForActivity({ ...input, nowMs: 120_001 })).toBe(false); + }, + ); + it("checks current permission, event preferences, and freshness together", () => { const input = { ...state, phase: "completed" as const, preferences, nowMs: 0 }; expect(shouldAlertForActivity(input)).toBe(true); diff --git a/infra/relay/src/environments/EnvironmentPublishSignatures.test.ts b/infra/relay/src/environments/EnvironmentPublishSignatures.test.ts index f61c5a27d5bc..2a93c7e32c0e 100644 --- a/infra/relay/src/environments/EnvironmentPublishSignatures.test.ts +++ b/infra/relay/src/environments/EnvironmentPublishSignatures.test.ts @@ -102,6 +102,25 @@ function layer(replay?: Partial) { } describe("EnvironmentPublishSignatures", () => { + it.effect("rejects an unsigned change to the startup replay flag", () => + Effect.gen(function* () { + const signatures = yield* EnvironmentPublishSignatures.EnvironmentPublishSignatures; + const request = yield* freshRequest; + const result = yield* signatures + .verify({ + environmentId: state.environmentId, + environmentPublicKey: keyPair.publicKey, + threadId: state.threadId, + request: { ...request, replay: true }, + }) + .pipe(Effect.result); + expect(result).toMatchObject({ + _tag: "Failure", + failure: { _tag: "EnvironmentPublishSignatureInvalid", stage: "validate_claims" }, + }); + }).pipe(Effect.provide(layer())), + ); + it.effect("verifies activity JWTs and scopes replay storage to the environment key", () => { let replayThumbprint: string | null = null; return Effect.gen(function* () { diff --git a/infra/relay/src/environments/EnvironmentPublishSignatures.ts b/infra/relay/src/environments/EnvironmentPublishSignatures.ts index 962d55b050bc..88e63877f1c7 100644 --- a/infra/relay/src/environments/EnvironmentPublishSignatures.ts +++ b/infra/relay/src/environments/EnvironmentPublishSignatures.ts @@ -159,6 +159,7 @@ const make = Effect.gen(function* () { proof.environmentId !== input.environmentId || proof.threadId !== input.threadId || proof.sub !== input.environmentId || + (proof.replay ?? false) !== (input.request.replay ?? false) || stableStringify(proof.state) !== stableStringify(input.request.state) || (input.request.state !== null && (input.request.state.environmentId !== input.environmentId || diff --git a/infra/relay/src/http/Api.ts b/infra/relay/src/http/Api.ts index fcaeca640420..4bf286a1415e 100644 --- a/infra/relay/src/http/Api.ts +++ b/infra/relay/src/http/Api.ts @@ -893,6 +893,7 @@ export const serverApi = HttpApiBuilder.group( environmentPublicKey: principal.environmentPublicKey, threadId: params.threadId, state: payload.state, + ...(payload.replay ? { replay: true } : {}), }); }, mapErrorTags({ diff --git a/packages/contracts/src/relay.ts b/packages/contracts/src/relay.ts index 8b6562497681..e56f0bc56346 100644 --- a/packages/contracts/src/relay.ts +++ b/packages/contracts/src/relay.ts @@ -218,12 +218,14 @@ export const RelayAgentActivityPublishProofPayload = Schema.Struct({ environmentId: EnvironmentId, threadId: ThreadId, state: Schema.NullOr(RelayAgentActivityState), + replay: Schema.optional(Schema.Boolean), }); export type RelayAgentActivityPublishProofPayload = typeof RelayAgentActivityPublishProofPayload.Type; export type RelayAgentActivityPublishProof = string; export const RelayAgentActivityPublishRequest = Schema.Struct({ + replay: Schema.optional(Schema.Boolean), state: Schema.NullOr(RelayAgentActivityState).annotate({ description: "Current agent-awareness state, or null to remove the published state.", }),