authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-09-26 01:35:35-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-09-26 01:35:35-07:00
log8872666957802b37c9fdd212e01f0024bcc102c6
tree63d56f0e77326073359da072b9dd3d7215aa3a5e
parent9151ea407accd1b7af575dcdd514e2b2e3fe6cbb
signature Signed by PGP key B5690EEEBB952194

perf(server): cut idle wakeups from the Connect relay and session reaper (#13774)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

12 files changed, 286 insertions(+), 31 deletions(-)

apps/server/integration/OrchestrationEngineHarness.integration.ts+1
...@@ -420,6 +420,7 @@ export const makeOrchestrationIntegrationHarness = (...@@ -420,6 +420,7 @@ export const makeOrchestrationIntegrationHarness = (
420 Layer.provideMerge(420 Layer.provideMerge(
421 Layer.succeed(AgentAwarenessRelay.AgentAwarenessRelay, {421 Layer.succeed(AgentAwarenessRelay.AgentAwarenessRelay, {
422 publishThread: () => Effect.void,422 publishThread: () => Effect.void,
423 requestCatchUp: () => Effect.void,
423 start: () => Effect.void,424 start: () => Effect.void,
424 }),425 }),
425 ),426 ),
apps/server/src/cloud/http.test.ts+9
...@@ -33,6 +33,7 @@ import {...@@ -33,6 +33,7 @@ import {
33 type ServiceUpdateRecord,33 type ServiceUpdateRecord,
34} from "./serviceProtocol.ts";34} from "./serviceProtocol.ts";
35import * as ServerEnvironment from "../environment/ServerEnvironment.ts";35import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
36import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";
36import { CLOUD_CLI_DESIRED_LINK_SECRET } from "./CliState.ts";37import { CLOUD_CLI_DESIRED_LINK_SECRET } from "./CliState.ts";
37import * as CliTokenManager from "./CliTokenManager.ts";38import * as CliTokenManager from "./CliTokenManager.ts";
38import {39import {
...@@ -81,6 +82,12 @@ const storeFailure = (tag: "AlreadyExists" | "PermissionDenied") =>...@@ -81,6 +82,12 @@ const storeFailure = (tag: "AlreadyExists" | "PermissionDenied") =>
81 });82 });
8283
83const unusedSecretStoreOperation = () => Effect.die("unused secret-store operation");84const unusedSecretStoreOperation = () => Effect.die("unused secret-store operation");
85// Linking wakes the awareness relay; these tests do not run it.
86const idleAwarenessRelay = AgentAwarenessRelay.AgentAwarenessRelay.of({
87 publishThread: () => Effect.void,
88 requestCatchUp: () => Effect.void,
89 start: () => Effect.void,
90});
84const decodeManagedTunnelRecoveryRegistration = Schema.decodeUnknownEffect(91const decodeManagedTunnelRecoveryRegistration = Schema.decodeUnknownEffect(
85 Schema.fromJsonString(RelayManagedEndpointRecoveryRegistrationRequest),92 Schema.fromJsonString(RelayManagedEndpointRecoveryRegistrationRequest),
86);93);
...@@ -262,6 +269,7 @@ describe("reconcileDesiredCloudLink", () => {...@@ -262,6 +269,7 @@ describe("reconcileDesiredCloudLink", () => {
262 HttpClient.HttpClient,269 HttpClient.HttpClient,
263 HttpClient.make(() => unusedSecretStoreOperation()),270 HttpClient.make(() => unusedSecretStoreOperation()),
264 ),271 ),
272 Effect.provideService(AgentAwarenessRelay.AgentAwarenessRelay, idleAwarenessRelay),
265 Effect.provide(NodeServices.layer),273 Effect.provide(NodeServices.layer),
266 ),274 ),
267 );275 );
...@@ -375,6 +383,7 @@ describe("releaseManagedTunnelOnShutdown", () => {...@@ -375,6 +383,7 @@ describe("releaseManagedTunnelOnShutdown", () => {
375 <A, E, R>(effect: Effect.Effect<A, E, R>) =>383 <A, E, R>(effect: Effect.Effect<A, E, R>) =>
376 effect.pipe(384 effect.pipe(
377 Effect.provideService(ServerSecretStore.ServerSecretStore, harness.store),385 Effect.provideService(ServerSecretStore.ServerSecretStore, harness.store),
386 Effect.provideService(AgentAwarenessRelay.AgentAwarenessRelay, idleAwarenessRelay),
378 Effect.provideService(387 Effect.provideService(
379 ServerEnvironment.ServerEnvironment,388 ServerEnvironment.ServerEnvironment,
380 ServerEnvironment.ServerEnvironment.of({389 ServerEnvironment.ServerEnvironment.of({
apps/server/src/cloud/http.ts+5
...@@ -69,6 +69,7 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts";...@@ -69,6 +69,7 @@ import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
69import { requireEnvironmentScope } from "../auth/http.ts";69import { requireEnvironmentScope } from "../auth/http.ts";
70import * as ServerConfig from "../config.ts";70import * as ServerConfig from "../config.ts";
71import * as ServerEnvironment from "../environment/ServerEnvironment.ts";71import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
72import * as AgentAwarenessRelay from "../relay/AgentAwarenessRelay.ts";
72import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts";73import * as ManagedEndpointRuntime from "./ManagedEndpointRuntime.ts";
73import {74import {
74 SERVICE_STATE_FILE,75 SERVICE_STATE_FILE,
...@@ -411,6 +412,7 @@ interface CloudHttpDependencies {...@@ -411,6 +412,7 @@ interface CloudHttpDependencies {
411 readonly environmentAuth: EnvironmentAuth.EnvironmentAuth["Service"];412 readonly environmentAuth: EnvironmentAuth.EnvironmentAuth["Service"];
412 readonly cliTokenManager: CliTokenManager.CloudCliTokenManager["Service"];413 readonly cliTokenManager: CliTokenManager.CloudCliTokenManager["Service"];
413 readonly httpClient: HttpClient.HttpClient;414 readonly httpClient: HttpClient.HttpClient;
415 readonly awarenessRelay: AgentAwarenessRelay.AgentAwarenessRelay["Service"];
414}416}
415417
416const cloudHttpDependencies = Effect.gen(function* () {418const cloudHttpDependencies = Effect.gen(function* () {
...@@ -421,6 +423,7 @@ const cloudHttpDependencies = Effect.gen(function* () {...@@ -421,6 +423,7 @@ const cloudHttpDependencies = Effect.gen(function* () {
421 environmentAuth: yield* EnvironmentAuth.EnvironmentAuth,423 environmentAuth: yield* EnvironmentAuth.EnvironmentAuth,
422 cliTokenManager: yield* CliTokenManager.CloudCliTokenManager,424 cliTokenManager: yield* CliTokenManager.CloudCliTokenManager,
423 httpClient: yield* HttpClient.HttpClient,425 httpClient: yield* HttpClient.HttpClient,
426 awarenessRelay: yield* AgentAwarenessRelay.AgentAwarenessRelay,
424 } satisfies CloudHttpDependencies;427 } satisfies CloudHttpDependencies;
425});428});
426429
...@@ -668,6 +671,7 @@ const applyCloudRelayConfig = Effect.fn("environment.cloud.applyRelayConfig")(fu...@@ -668,6 +671,7 @@ const applyCloudRelayConfig = Effect.fn("environment.cloud.applyRelayConfig")(fu
668 CLOUD_MINT_PUBLIC_KEY,671 CLOUD_MINT_PUBLIC_KEY,
669 stringToBytes(payload.cloudMintPublicKey),672 stringToBytes(payload.cloudMintPublicKey),
670 );673 );
674 yield* dependencies.awarenessRelay.requestCatchUp();
671 if (payload.endpointRuntime) {675 if (payload.endpointRuntime) {
672 const endpointRuntimeJson = yield* encodeEndpointRuntimeConfigJson(payload.endpointRuntime);676 const endpointRuntimeJson = yield* encodeEndpointRuntimeConfigJson(payload.endpointRuntime);
673 yield* dependencies.secrets.set(677 yield* dependencies.secrets.set(
...@@ -1349,6 +1353,7 @@ const cloudPreferencesHandler = Effect.fn("environment.cloud.preferences")(...@@ -1349,6 +1353,7 @@ const cloudPreferencesHandler = Effect.fn("environment.cloud.preferences")(
1349 PUBLISH_AGENT_ACTIVITY_SECRET,1353 PUBLISH_AGENT_ACTIVITY_SECRET,
1350 stringToBytes(String(payload.publishAgentActivity)),1354 stringToBytes(String(payload.publishAgentActivity)),
1351 );1355 );
1356 yield* dependencies.awarenessRelay.requestCatchUp();
1352 return yield* readCloudLinkState(dependencies);1357 return yield* readCloudLinkState(dependencies);
1353 },1358 },
1354 Effect.catchIf(1359 Effect.catchIf(
apps/server/src/orchestration/Layers/OrchestrationReactor.test.ts+1
...@@ -108,6 +108,7 @@ describe("OrchestrationReactor", () => {...@@ -108,6 +108,7 @@ describe("OrchestrationReactor", () => {
108 Layer.provideMerge(108 Layer.provideMerge(
109 Layer.succeed(AgentAwarenessRelay.AgentAwarenessRelay, {109 Layer.succeed(AgentAwarenessRelay.AgentAwarenessRelay, {
110 publishThread: () => Effect.void,110 publishThread: () => Effect.void,
111 requestCatchUp: () => Effect.void,
111 start: () => {112 start: () => {
112 started.push("agent-awareness-relay");113 started.push("agent-awareness-relay");
113 return Effect.void;114 return Effect.void;
apps/server/src/persistence/ProviderSessionRuntime.ts+11-7
...@@ -102,11 +102,14 @@ export class ProviderSessionRuntimeRepository extends Context.Service<...@@ -102,11 +102,14 @@ export class ProviderSessionRuntimeRepository extends Context.Service<
102 >;102 >;
103103
104 /**104 /**
105 * List all provider runtime rows.105 * List provider runtime rows.
106 *106 *
107 * Returned in ascending last-seen order.107 * Returned in ascending last-seen order. `excludeStopped` filters stopped
108 * rows in SQL. Long-lived installs keep thousands for their resume cursors.
108 */109 */
109 readonly list: () => Effect.Effect<110 readonly list: (options?: {
111 readonly excludeStopped?: boolean;
112 }) => Effect.Effect<
110 ReadonlyArray<ProviderSessionRuntime>,113 ReadonlyArray<ProviderSessionRuntime>,
111 ProviderSessionRuntimeRepositoryError114 ProviderSessionRuntimeRepositoryError
112 >;115 >;
...@@ -336,9 +339,9 @@ export const make = Effect.gen(function* () {...@@ -336,9 +339,9 @@ export const make = Effect.gen(function* () {
336 });339 });
337340
338 const listRuntimeRows = SqlSchema.findAll({341 const listRuntimeRows = SqlSchema.findAll({
339 Request: Schema.Void,342 Request: Schema.Struct({ excludeStopped: Schema.Boolean }),
340 Result: ProviderSessionRuntimeRawDbRowSchema,343 Result: ProviderSessionRuntimeRawDbRowSchema,
341 execute: () =>344 execute: ({ excludeStopped }) =>
342 sql`345 sql`
343 SELECT346 SELECT
344 thread_id AS "threadId",347 thread_id AS "threadId",
...@@ -351,6 +354,7 @@ export const make = Effect.gen(function* () {...@@ -351,6 +354,7 @@ export const make = Effect.gen(function* () {
351 resume_cursor_json AS "resumeCursor",354 resume_cursor_json AS "resumeCursor",
352 runtime_payload_json AS "runtimePayload"355 runtime_payload_json AS "runtimePayload"
353 FROM provider_session_runtime356 FROM provider_session_runtime
357 ${excludeStopped ? sql`WHERE status != 'stopped'` : sql``}
354 ORDER BY last_seen_at ASC, thread_id ASC358 ORDER BY last_seen_at ASC, thread_id ASC
355 `,359 `,
356 });360 });
...@@ -414,8 +418,8 @@ export const make = Effect.gen(function* () {...@@ -414,8 +418,8 @@ export const make = Effect.gen(function* () {
414 ),418 ),
415 );419 );
416420
417 const list: ProviderSessionRuntimeRepository["Service"]["list"] = () =>421 const list: ProviderSessionRuntimeRepository["Service"]["list"] = (options) =>
418 listRuntimeRows(undefined).pipe(422 listRuntimeRows({ excludeStopped: options?.excludeStopped === true }).pipe(
419 Effect.mapError(423 Effect.mapError(
420 toPersistenceSqlOrDecodeError(424 toPersistenceSqlOrDecodeError(
421 "ProviderSessionRuntimeRepository.list:query",425 "ProviderSessionRuntimeRepository.list:query",
apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts+35
...@@ -372,6 +372,41 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL...@@ -372,6 +372,41 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL
372 }),372 }),
373 );373 );
374374
375 it.effect("lists only bindings that are not stopped when asked", () =>
376 Effect.gen(function* () {
377 const directory = yield* ProviderSessionDirectory;
378 const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository;
379 const statuses = ["running", "starting", "error", "stopped"] as const;
380 const threadIds = new Set<string>();
381
382 for (const status of statuses) {
383 const threadId = ThreadId.make(`thread-exclude-stopped-${status}`);
384 threadIds.add(threadId);
385 yield* runtimeRepository.upsert({
386 threadId,
387 providerName: "codex",
388 providerInstanceId: ProviderInstanceId.make("codex"),
389 adapterKey: "codex",
390 runtimeMode: "full-access",
391 status,
392 lastSeenAt: "2026-04-14T12:00:00.000Z",
393 resumeCursor: null,
394 runtimePayload: null,
395 });
396 }
397
398 const liveStatuses = (yield* directory.listBindings({ excludeStopped: true }))
399 .filter((binding) => threadIds.has(binding.threadId))
400 .map((binding) => binding.status);
401 const allStatuses = (yield* directory.listBindings())
402 .filter((binding) => threadIds.has(binding.threadId))
403 .map((binding) => binding.status);
404
405 assert.deepEqual(liveStatuses.toSorted(), ["error", "running", "starting"]);
406 assert.deepEqual(allStatuses.toSorted(), ["error", "running", "starting", "stopped"]);
407 }),
408 );
409
375 it.effect(410 it.effect(
376 "resets adapterKey to the new provider when provider changes without an explicit adapter key",411 "resets adapterKey to the new provider when provider changes without an explicit adapter key",
377 () =>412 () =>
apps/server/src/provider/Layers/ProviderSessionDirectory.ts+2-2
...@@ -182,8 +182,8 @@ const makeProviderSessionDirectory = Effect.gen(function* () {...@@ -182,8 +182,8 @@ const makeProviderSessionDirectory = Effect.gen(function* () {
182 Effect.map((rows) => rows.map((row) => row.threadId)),182 Effect.map((rows) => rows.map((row) => row.threadId)),
183 );183 );
184184
185 const listBindings: ProviderSessionDirectoryShape["listBindings"] = () =>185 const listBindings: ProviderSessionDirectoryShape["listBindings"] = (options) =>
186 repository.list().pipe(186 repository.list(options).pipe(
187 Effect.mapError(toPersistenceError("ProviderSessionDirectory.listBindings:list")),187 Effect.mapError(toPersistenceError("ProviderSessionDirectory.listBindings:list")),
188 Effect.flatMap((rows) =>188 Effect.flatMap((rows) =>
189 Effect.forEach(189 Effect.forEach(
apps/server/src/provider/Layers/ProviderSessionReaper.ts+4-6
...@@ -35,15 +35,13 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =...@@ -35,15 +35,13 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =
35 const sweepIntervalMs = Math.max(1, options?.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS);35 const sweepIntervalMs = Math.max(1, options?.sweepIntervalMs ?? DEFAULT_SWEEP_INTERVAL_MS);
3636
37 const sweep = Effect.gen(function* () {37 const sweep = Effect.gen(function* () {
38 const bindings = yield* directory.listBindings();38 // Stopped rows stay for their resume cursors and far outnumber live
39 // ones, so the query skips them.
40 const bindings = yield* directory.listBindings({ excludeStopped: true });
39 const now = yield* Clock.currentTimeMillis;41 const now = yield* Clock.currentTimeMillis;
40 let reapedCount = 0;42 let reapedCount = 0;
4143
42 for (const binding of bindings) {44 for (const binding of bindings) {
43 if (binding.status === "stopped") {
44 continue;
45 }
46
47 const lastSeenMs = Date.parse(binding.lastSeenAt);45 const lastSeenMs = Date.parse(binding.lastSeenAt);
48 if (Number.isNaN(lastSeenMs)) {46 if (Number.isNaN(lastSeenMs)) {
49 yield* Effect.logWarning("provider.session.reaper.invalid-last-seen", {47 yield* Effect.logWarning("provider.session.reaper.invalid-last-seen", {
...@@ -122,7 +120,7 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =...@@ -122,7 +120,7 @@ const makeProviderSessionReaper = (options?: ProviderSessionReaperLiveOptions) =
122 if (reapedCount > 0) {120 if (reapedCount > 0) {
123 yield* Effect.logInfo("provider.session.reaper.sweep-complete", {121 yield* Effect.logInfo("provider.session.reaper.sweep-complete", {
124 reapedCount,122 reapedCount,
125 totalBindings: bindings.length,123 liveBindings: bindings.length,
126 });124 });
127 }125 }
128 });126 });
apps/server/src/provider/Services/ProviderSessionDirectory.ts+4-1
...@@ -70,7 +70,10 @@ export interface ProviderSessionDirectoryShape {...@@ -70,7 +70,10 @@ export interface ProviderSessionDirectoryShape {
70 ProviderSessionDirectoryPersistenceError70 ProviderSessionDirectoryPersistenceError
71 >;71 >;
7272
73 readonly listBindings: () => Effect.Effect<73 /** `excludeStopped` skips stopped rows in the query, not after decoding. */
74 readonly listBindings: (options?: {
75 readonly excludeStopped?: boolean;
76 }) => Effect.Effect<
74 ReadonlyArray<ProviderRuntimeBindingWithMetadata>,77 ReadonlyArray<ProviderRuntimeBindingWithMetadata>,
75 ProviderSessionDirectoryPersistenceError78 ProviderSessionDirectoryPersistenceError
76 >;79 >;
apps/server/src/relay/AgentAwarenessRelay.test.ts+112
...@@ -28,6 +28,7 @@ import * as Option from "effect/Option";...@@ -28,6 +28,7 @@ import * as Option from "effect/Option";
28import * as Queue from "effect/Queue";28import * as Queue from "effect/Queue";
29import * as Stream from "effect/Stream";29import * as Stream from "effect/Stream";
30import * as Tracer from "effect/Tracer";30import * as Tracer from "effect/Tracer";
31import * as TestClock from "effect/testing/TestClock";
3132
32import * as ServerSecretStore from "../auth/ServerSecretStore.ts";33import * as ServerSecretStore from "../auth/ServerSecretStore.ts";
33import * as ServerEnvironment from "../environment/ServerEnvironment.ts";34import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
...@@ -975,3 +976,114 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {...@@ -975,3 +976,114 @@ describe.sequential("signRelayAgentActivityPublishProof", () => {
975 ),976 ),
976 );977 );
977});978});
979
980describe.sequential("startup catch-up", () => {
981 // An unlinked relay with publishing off. `link` writes the link secrets and
982 // `enablePublishing` the opt-in. Counts link checks (relay URL reads) and
983 // catch-up publishes (shell snapshot reads).
984 function makeUnlinkedRelay() {
985 const secrets = makeMemorySecretStore();
986 const counts = { linkChecks: 0, catchUpPublishes: 0 };
987 const countingStore = {
988 ...secrets.store,
989 get: (name: string) =>
990 Effect.suspend(() => {
991 if (name === RELAY_URL_SECRET) counts.linkChecks += 1;
992 return secrets.store.get(name);
993 }),
994 } satisfies ServerSecretStore.ServerSecretStore["Service"];
995
996 const layer = AgentAwarenessRelay.layer.pipe(
997 Layer.provide(
998 Layer.mergeAll(
999 Layer.succeed(ServerSecretStore.ServerSecretStore, countingStore),
1000 Layer.succeed(ServerEnvironment.ServerEnvironment, {
1001 getEnvironmentId: Effect.succeed("env-1" as EnvironmentId),
1002 getDescriptor: Effect.die("unused descriptor"),
1003 }),
1004 Layer.succeed(OrchestrationEngineService, {
1005 streamDomainEvents: Stream.never,
1006 } as unknown as OrchestrationEngineShape),
1007 Layer.succeed(ProjectionSnapshotQuery, {
1008 getShellSnapshot: () =>
1009 Effect.sync(() => {
1010 counts.catchUpPublishes += 1;
1011 return {
1012 snapshotSequence: 1,
1013 projects: [],
1014 threads: [],
1015 updatedAt: "2026-05-25T00:00:00.000Z",
1016 } satisfies OrchestrationShellSnapshot;
1017 }),
1018 } as unknown as ProjectionSnapshotQueryShape),
1019 ),
1020 ),
1021 Layer.provideMerge(NodeServices.layer),
1022 );
1023 const link = Effect.all(
1024 [
1025 secrets.setString(RELAY_URL_SECRET, "https://relay.example.test"),
1026 secrets.setString(RELAY_ENVIRONMENT_CREDENTIAL_SECRET, "relay-credential"),
1027 ],
1028 { discard: true },
1029 );
1030 const enablePublishing = secrets.setString(PUBLISH_AGENT_ACTIVITY_SECRET, "true");
1031 return { counts, layer, link, enablePublishing };
1032 }
1033
1034 it.effect("checks an unlinked environment once a minute and still catches up once linked", () => {
1035 const { counts, layer, link, enablePublishing } = makeUnlinkedRelay();
1036 return Effect.gen(function* () {
1037 const relay = yield* AgentAwarenessRelay.AgentAwarenessRelay;
1038 yield* enablePublishing;
1039 yield* relay.start();
1040
1041 // Get past the backoff ramp, then count checks in a steady window.
1042 yield* TestClock.adjust("10 minutes");
1043 const checksBeforeWindow = counts.linkChecks;
1044 yield* TestClock.adjust("10 minutes");
1045 expect(counts.linkChecks - checksBeforeWindow).toBe(10);
1046 expect(counts.catchUpPublishes).toBe(0);
1047
1048 yield* link;
1049 yield* TestClock.adjust("1 minute");
1050 expect(counts.catchUpPublishes).toBe(1);
1051 }).pipe(Effect.provide(layer), Effect.scoped);
1052 });
1053
1054 it.effect("publishes at once when this process links while the check is backed off", () => {
1055 const { counts, layer, link, enablePublishing } = makeUnlinkedRelay();
1056 return Effect.gen(function* () {
1057 const relay = yield* AgentAwarenessRelay.AgentAwarenessRelay;
1058 yield* enablePublishing;
1059 yield* relay.start();
1060
1061 // Backed off to 60 s: the next check is still seconds away.
1062 yield* TestClock.adjust("10 minutes");
1063 yield* link;
1064 yield* TestClock.adjust("1 second");
1065 expect(counts.catchUpPublishes).toBe(0);
1066
1067 yield* relay.requestCatchUp();
1068 yield* TestClock.adjust("1 second");
1069 expect(counts.catchUpPublishes).toBe(1);
1070 }).pipe(Effect.provide(layer), Effect.scoped);
1071 });
1072
1073 it.effect("catches up within 5 s when another process enables publishing on a link", () => {
1074 const { counts, layer, link, enablePublishing } = makeUnlinkedRelay();
1075 return Effect.gen(function* () {
1076 const relay = yield* AgentAwarenessRelay.AgentAwarenessRelay;
1077 yield* link;
1078 yield* relay.start();
1079
1080 yield* TestClock.adjust("10 minutes");
1081 expect(counts.catchUpPublishes).toBe(0);
1082
1083 // `t3 connect publish` writes the opt-in without waking this process.
1084 yield* enablePublishing;
1085 yield* TestClock.adjust("5 seconds");
1086 expect(counts.catchUpPublishes).toBe(1);
1087 }).pipe(Effect.provide(layer), Effect.scoped);
1088 });
1089});
apps/server/src/relay/AgentAwarenessRelay.ts+37-7
...@@ -25,6 +25,7 @@ import * as DateTime from "effect/DateTime";...@@ -25,6 +25,7 @@ import * as DateTime from "effect/DateTime";
25import * as Effect from "effect/Effect";25import * as Effect from "effect/Effect";
26import * as Layer from "effect/Layer";26import * as Layer from "effect/Layer";
27import * as Option from "effect/Option";27import * as Option from "effect/Option";
28import * as Queue from "effect/Queue";
28import * as Ref from "effect/Ref";29import * as Ref from "effect/Ref";
29import type * as Scope from "effect/Scope";30import type * as Scope from "effect/Scope";
30import * as Stream from "effect/Stream";31import * as Stream from "effect/Stream";
...@@ -51,6 +52,8 @@ export class AgentAwarenessRelay extends Context.Service<...@@ -51,6 +52,8 @@ export class AgentAwarenessRelay extends Context.Service<
51 AgentAwarenessRelay,52 AgentAwarenessRelay,
52 {53 {
53 readonly publishThread: (threadId: ThreadId) => Effect.Effect<void>;54 readonly publishThread: (threadId: ThreadId) => Effect.Effect<void>;
55 /** Retries a pending catch-up publish now. Call after this process links or enables publishing. */
56 readonly requestCatchUp: () => Effect.Effect<void>;
54 readonly start: () => Effect.Effect<void, never, Scope.Scope>;57 readonly start: () => Effect.Effect<void, never, Scope.Scope>;
55 }58 }
56>()("t3/relay/AgentAwarenessRelay") {}59>()("t3/relay/AgentAwarenessRelay") {}
...@@ -309,6 +312,8 @@ export const make = Effect.gen(function* () {...@@ -309,6 +312,8 @@ export const make = Effect.gen(function* () {
309 const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets);312 const cloudLinkKeyPair = yield* getOrCreateEnvironmentKeyPairFromSecretStore(secrets);
310 const startedAt = (yield* DateTime.now).epochMilliseconds;313 const startedAt = (yield* DateTime.now).epochMilliseconds;
311 const activeSnapshotPublishedRef = yield* Ref.make(false);314 const activeSnapshotPublishedRef = yield* Ref.make(false);
315 // Holds at most one pending wake, so a burst of requests costs one retry.
316 const catchUpRequests = yield* Queue.dropping<void>(1);
312 const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>());317 const publishedStateByThreadRef = yield* Ref.make(new Map<ThreadId, string>());
313318
314 const readSecretString = (name: string) =>319 const readSecretString = (name: string) =>
...@@ -534,18 +539,29 @@ export const make = Effect.gen(function* () {...@@ -534,18 +539,29 @@ export const make = Effect.gen(function* () {
534 withRelayClientTracing,539 withRelayClientTracing,
535 );540 );
536541
542 // Publishes the active threads once. Returns why it did not, so the retry
543 // knows whether it is waiting on a link or on the publish setting.
537 const publishActiveThreadsUnsafe = Effect.gen(function* () {544 const publishActiveThreadsUnsafe = Effect.gen(function* () {
545 // One secret read settles the common never-linked case; the full link
546 // config is read only once publishing is on.
547 const relayUrl = yield* readSecretString(RELAY_URL_SECRET).pipe(
548 Effect.orElseSucceed(() => null),
549 );
550 if (!relayUrl) {
551 yield* Effect.logDebug("agent activity snapshot skipped; relay link credentials unavailable");
552 return "unlinked" as const;
553 }
538 const publishAgentActivity = yield* readPublishAgentActivityEnabled.pipe(554 const publishAgentActivity = yield* readPublishAgentActivityEnabled.pipe(
539 Effect.orElseSucceed(() => false),555 Effect.orElseSucceed(() => false),
540 );556 );
541 if (!publishAgentActivity) {557 if (!publishAgentActivity) {
542 yield* Effect.logDebug("agent activity snapshot skipped; publication disabled");558 yield* Effect.logDebug("agent activity snapshot skipped; publication disabled");
543 return false;559 return "disabled" as const;
544 }560 }
545 const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null));561 const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null));
546 if (!relayConfig) {562 if (!relayConfig) {
547 yield* Effect.logDebug("agent activity snapshot skipped; relay link credentials unavailable");563 yield* Effect.logDebug("agent activity snapshot skipped; relay link credentials unavailable");
548 return false;564 return "unlinked" as const;
549 }565 }
550 const environmentId = yield* serverEnvironment.getEnvironmentId;566 const environmentId = yield* serverEnvironment.getEnvironmentId;
551 const snapshot = yield* snapshotQuery.getShellSnapshot();567 const snapshot = yield* snapshotQuery.getShellSnapshot();
...@@ -557,20 +573,29 @@ export const make = Effect.gen(function* () {...@@ -557,20 +573,29 @@ export const make = Effect.gen(function* () {
557 });573 });
558 if (activeThreadIds.length === 0) {574 if (activeThreadIds.length === 0) {
559 yield* Effect.logDebug("agent activity snapshot has no publishable threads");575 yield* Effect.logDebug("agent activity snapshot has no publishable threads");
560 return true;576 return "published" as const;
561 }577 }
562 yield* Effect.logInfo("publishing active agent activity snapshot", {578 yield* Effect.logInfo("publishing active agent activity snapshot", {
563 count: activeThreadIds.length,579 count: activeThreadIds.length,
564 });580 });
565 yield* Effect.forEach(activeThreadIds, publishThread, { concurrency: 4, discard: true });581 yield* Effect.forEach(activeThreadIds, publishThread, { concurrency: 4, discard: true });
566 return true;582 return "published" as const;
567 });583 });
568584
585 // Publishes the catch-up snapshot of active threads once the environment is
586 // linked and publishing is enabled. Many environments never link, so while
587 // unlinked the retry backs off from 5 s to 60 s. Only this process writes
588 // the link, and it calls `requestCatchUp`, which ends the wait early. A
589 // linked environment keeps the 5 s retry, because `t3 connect publish` can
590 // turn publishing on from another process.
569 const publishActiveThreadsOnceWhenConfigured = (logEnabledWhenReady: boolean) =>591 const publishActiveThreadsOnceWhenConfigured = (logEnabledWhenReady: boolean) =>
570 Effect.gen(function* () {592 Effect.gen(function* () {
593 let unlinkedRetryDelayMs = 5_000;
571 while (!(yield* Ref.get(activeSnapshotPublishedRef))) {594 while (!(yield* Ref.get(activeSnapshotPublishedRef))) {
572 const published = yield* publishActiveThreadsUnsafe.pipe(Effect.orElseSucceed(() => false));595 const result = yield* publishActiveThreadsUnsafe.pipe(
573 if (published) {596 Effect.orElseSucceed(() => "failed" as const),
597 );
598 if (result === "published") {
574 yield* Ref.set(activeSnapshotPublishedRef, true);599 yield* Ref.set(activeSnapshotPublishedRef, true);
575 if (logEnabledWhenReady) {600 if (logEnabledWhenReady) {
576 const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null));601 const relayConfig = yield* readRelayConfig.pipe(Effect.orElseSucceed(() => null));
...@@ -580,7 +605,11 @@ export const make = Effect.gen(function* () {...@@ -580,7 +605,11 @@ export const make = Effect.gen(function* () {
580 }605 }
581 return;606 return;
582 }607 }
583 yield* Effect.sleep("5 seconds");608 const retryDelayMs = result === "unlinked" ? unlinkedRetryDelayMs : 5_000;
609 yield* Effect.race(Effect.sleep(retryDelayMs), Queue.take(catchUpRequests));
610 if (result === "unlinked") {
611 unlinkedRetryDelayMs = Math.min(unlinkedRetryDelayMs * 2, 60_000);
612 }
584 }613 }
585 });614 });
586615
...@@ -657,6 +686,7 @@ export const make = Effect.gen(function* () {...@@ -657,6 +686,7 @@ export const make = Effect.gen(function* () {
657686
658 return AgentAwarenessRelay.of({687 return AgentAwarenessRelay.of({
659 publishThread,688 publishThread,
689 requestCatchUp: () => Queue.offer(catchUpRequests, undefined).pipe(Effect.asVoid),
660 start,690 start,
661 });691 });
662});692});
apps/server/src/server.test.ts+65-8
...@@ -181,6 +181,7 @@ import * as ServerSecretStore from "./auth/ServerSecretStore.ts";...@@ -181,6 +181,7 @@ import * as ServerSecretStore from "./auth/ServerSecretStore.ts";
181import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";181import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";
182import * as PairingGrantStore from "./auth/PairingGrantStore.ts";182import * as PairingGrantStore from "./auth/PairingGrantStore.ts";
183import * as CloudManagedEndpointRuntime from "./cloud/ManagedEndpointRuntime.ts";183import * as CloudManagedEndpointRuntime from "./cloud/ManagedEndpointRuntime.ts";
184import * as AgentAwarenessRelay from "./relay/AgentAwarenessRelay.ts";
184import * as CloudCliTokenManager from "./cloud/CliTokenManager.ts";185import * as CloudCliTokenManager from "./cloud/CliTokenManager.ts";
185import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts";186import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts";
186import * as HostResources from "./resourceTelemetry/HostResources.ts";187import * as HostResources from "./resourceTelemetry/HostResources.ts";
...@@ -562,6 +563,7 @@ const buildAppUnderTest = (options?: {...@@ -562,6 +563,7 @@ const buildAppUnderTest = (options?: {
562 CloudManagedEndpointRuntime.CloudManagedEndpointRuntime["Service"]563 CloudManagedEndpointRuntime.CloudManagedEndpointRuntime["Service"]
563 >;564 >;
564 relayClient?: Partial<RelayClient.RelayClient["Service"]>;565 relayClient?: Partial<RelayClient.RelayClient["Service"]>;
566 agentAwarenessRelay?: Partial<AgentAwarenessRelay.AgentAwarenessRelay["Service"]>;
565 cloudCliTokenManager?: Partial<CloudCliTokenManager.CloudCliTokenManager["Service"]>;567 cloudCliTokenManager?: Partial<CloudCliTokenManager.CloudCliTokenManager["Service"]>;
566 httpClient?: HttpClient.HttpClient;568 httpClient?: HttpClient.HttpClient;
567 nativeTelemetryClient?: Partial<NativeTelemetryClient.NativeTelemetryClient["Service"]>;569 nativeTelemetryClient?: Partial<NativeTelemetryClient.NativeTelemetryClient["Service"]>;
...@@ -1173,14 +1175,20 @@ const buildAppUnderTest = (options?: {...@@ -1173,14 +1175,20 @@ const buildAppUnderTest = (options?: {
1173 }),1175 }),
1174 ),1176 ),
1175 Layer.provide(1177 Layer.provide(
1176 Layer.succeed(1178 Layer.mergeAll(
1177 CloudManagedEndpointRuntime.CloudManagedEndpointRuntime,1179 Layer.succeed(
1178 CloudManagedEndpointRuntime.CloudManagedEndpointRuntime.of({1180 CloudManagedEndpointRuntime.CloudManagedEndpointRuntime,
1179 applyConfig: () => Effect.succeed({ status: "disabled" }),1181 CloudManagedEndpointRuntime.CloudManagedEndpointRuntime.of({
1180 recoveryRequests: Stream.empty,1182 applyConfig: () => Effect.succeed({ status: "disabled" }),
1181 requestRecovery: () => Effect.void,1183 recoveryRequests: Stream.empty,
1182 withLinkStateLock: (effect) => effect,1184 requestRecovery: () => Effect.void,
1183 ...options?.layers?.cloudManagedEndpointRuntime,1185 withLinkStateLock: (effect) => effect,
1186 ...options?.layers?.cloudManagedEndpointRuntime,
1187 }),
1188 ),
1189 Layer.mock(AgentAwarenessRelay.AgentAwarenessRelay)({
1190 requestCatchUp: () => Effect.void,
1191 ...options?.layers?.agentAwarenessRelay,
1184 }),1192 }),
1185 ),1193 ),
1186 ),1194 ),
...@@ -3114,6 +3122,55 @@ it.layer(NodeServices.layer)("server router seam", (it) => {...@@ -3114,6 +3122,55 @@ it.layer(NodeServices.layer)("server router seam", (it) => {
3114 }).pipe(Effect.provide(NodeHttpServer.layerTest)),3122 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
3115 );3123 );
31163124
3125 it.effect("wakes the agent awareness relay when this server links or changes publishing", () =>
3126 Effect.gen(function* () {
3127 let catchUpRequests = 0;
3128 yield* buildAppUnderTest({
3129 layers: {
3130 agentAwarenessRelay: {
3131 requestCatchUp: () =>
3132 Effect.sync(() => {
3133 catchUpRequests += 1;
3134 }),
3135 },
3136 },
3137 });
3138
3139 const cloudKeyPair = NodeCrypto.generateKeyPairSync("ed25519", {
3140 privateKeyEncoding: { format: "pem", type: "pkcs8" },
3141 publicKeyEncoding: { format: "pem", type: "spki" },
3142 });
3143 const ownerCookie = yield* getAuthenticatedSessionCookieHeader();
3144 const relayConfigResponse = yield* fetchEffect(
3145 yield* getHttpServerUrl("/api/connect/relay-config"),
3146 {
3147 method: "POST",
3148 headers: { cookie: ownerCookie, "content-type": "application/json" },
3149 body: jsonRequestBody({
3150 relayUrl: "https://relay.example.test",
3151 cloudUserId: "user_123",
3152 environmentCredential: "t3env_test_credential",
3153 cloudMintPublicKey: cloudKeyPair.publicKey,
3154 endpointRuntime: null,
3155 }),
3156 },
3157 );
3158 assert.equal(relayConfigResponse.status, 200);
3159 assert.equal(catchUpRequests, 1);
3160
3161 const preferencesResponse = yield* fetchEffect(
3162 yield* getHttpServerUrl("/api/connect/preferences"),
3163 {
3164 method: "POST",
3165 headers: { cookie: ownerCookie, "content-type": "application/json" },
3166 body: jsonRequestBody({ publishAgentActivity: true }),
3167 },
3168 );
3169 assert.equal(preferencesResponse.status, 200);
3170 assert.equal(catchUpRequests, 2);
3171 }).pipe(Effect.provide(NodeHttpServer.layerTest)),
3172 );
3173
3117 it.effect("rejects relay config with an invalid cloud mint public key", () =>3174 it.effect("rejects relay config with an invalid cloud mint public key", () =>
3118 Effect.gen(function* () {3175 Effect.gen(function* () {
3119 yield* buildAppUnderTest();3176 yield* buildAppUnderTest();