1import {
2 type EnvironmentId,
3 type ServerConfig,
4 type ServerConfigStreamEvent,
5 type ServerLifecycleWelcomePayload,
6 type ServerLifecycleStreamReadyEvent,
7 type ServerSelfUpdateProgressEvent,
8 type ServerSelfUpdateResult,
9 WS_METHODS,
10} from "@t3tools/contracts";
11import * as Cause from "effect/Cause";
12import * as Duration from "effect/Duration";
13import * as Deferred from "effect/Deferred";
14import * as Effect from "effect/Effect";
15import * as Exit from "effect/Exit";
16import * as Fiber from "effect/Fiber";
17import * as Option from "effect/Option";
18import * as Queue from "effect/Queue";
19import * as Ref from "effect/Ref";
20import * as Result from "effect/Result";
21import * as Schema from "effect/Schema";
22import * as Stream from "effect/Stream";
23import * as SubscriptionRef from "effect/SubscriptionRef";
24import { AsyncResult, Atom } from "effect/reactivity";
25
26import {
27 createAtomCommandScheduler,
28 createEnvironmentRpcCommand,
29 createEnvironmentQueryAtomFamily,
30 createEnvironmentRpcQueryAtomFamily,
31 createEnvironmentRpcSubscriptionAtomFamily,
32 createEnvironmentSubscriptionAtomFamily,
33 createRuntimeCommand,
34 scheduleAtomCommandEffect,
35} from "./runtime.ts";
36import * as EnvironmentRegistry from "../connection/registry.ts";
37import * as EnvironmentSupervisor from "../connection/supervisor.ts";
38import { safeErrorLogAttributes } from "../errors/safeLog.ts";
39import * as Persistence from "../platform/persistence.ts";
40import { runCachePersistence } from "./cachePersistence.ts";
41import {
42 isRpcClientError,
43 request,
44 runStream,
45 subscribe,
46 subscribeDynamicWithSession,
47 type EnvironmentRpcInput,
48} from "../rpc/client.ts";
49import type { RpcSession } from "../rpc/session.ts";
50import { followStreamInEnvironment } from "./runtime.ts";
51import {
52 applyServerConfigProjection,
53 type ServerConfigProjection,
54 withoutEnvironmentThemes,
55} from "./serverConfigProjection.ts";
56
57// Exported server state includes this type in its inferred public return type.
58export type { ServerConfigProjection } from "./serverConfigProjection.ts";
59
60export type ServerUpdateStage = "downloading" | "installing" | "resuming";
61
62export type ServerUpdateState =
63 | { readonly status: "idle" }
64 | {
65 readonly status: "running";
66 readonly stage: ServerUpdateStage;
67 readonly fromVersion: string;
68 readonly targetVersion: string;
69 }
70 | {
71 readonly status: "failed";
72 readonly stage: ServerUpdateStage;
73 readonly fromVersion: string;
74 readonly targetVersion: string;
75 readonly message: string;
76 };
77
78export interface ServerUpdateTarget {
79 readonly environmentId: EnvironmentId;
80 readonly input: EnvironmentRpcInput<typeof WS_METHODS.serverUpdateServer>;
81}
82
83const IDLE_SERVER_UPDATE_STATE: ServerUpdateState = { status: "idle" };
84const EMPTY_SERVER_UPDATE_STATE_ATOM = Atom.make<ServerUpdateState>(IDLE_SERVER_UPDATE_STATE).pipe(
85 Atom.withLabel("environment-data:server:update-state:empty"),
86);
87/** Shared with the outdated-host update, which reports through the same state. */
88export const serverUpdateStateAtom = Atom.family((environmentId: EnvironmentId) =>
89 Atom.make<ServerUpdateState>(IDLE_SERVER_UPDATE_STATE).pipe(
90 Atom.withLabel(`environment-data:server:update-state:${environmentId}`),
91 ),
92);
93
94export class ServerUpdateResumeTimeoutError extends Schema.TaggedError<ServerUpdateResumeTimeoutError>()(
95 "ServerUpdateResumeTimeoutError",
96 {
97 environmentId: Schema.String,
98 targetVersion: Schema.String,
99 },
100) {
101 override get message(): string {
102 return `The server did not resume on t3@${this.targetVersion}.`;
103 }
104}
105
106export class ServerUpdateProgressIncompleteError extends Schema.TaggedError<ServerUpdateProgressIncompleteError>()(
107 "ServerUpdateProgressIncompleteError",
108 {
109 targetVersion: Schema.String,
110 },
111) {
112 override get message(): string {
113 return `The t3@${this.targetVersion} update ended before the server accepted the restart.`;
114 }
115}
116
117export class ServerUpdateTerminalError extends Schema.TaggedError<ServerUpdateTerminalError>()(
118 "ServerUpdateTerminalError",
119 {
120 targetVersion: Schema.String,
121 status: Schema.Literals(["committed", "rolled-back", "failed"]),
122 reason: Schema.optional(Schema.String),
123 },
124) {
125 override get message(): string {
126 return this.reason ?? `The t3@${this.targetVersion} update ${this.status}.`;
127 }
128}
129
130// Covers the service launcher's 30-minute trial deadline (PREPARED_TIMEOUT_MS)
131// and a final restart of the previous version when the trial rolls back.
132const SERVER_UPDATE_RESUME_TIMEOUT = Duration.minutes(32);
133
134export function matchesServerUpdateReadyEvent(
135 result: ServerSelfUpdateResult,
136 event: ServerLifecycleStreamReadyEvent,
137): boolean {
138 return result.updateId === undefined
139 ? event.payload.environment.serverVersion === result.targetVersion
140 : event.payload.updateOutcome?.id === result.updateId;
141}
142
143export function matchesServerUpdateResumeEvent(
144 result: ServerSelfUpdateResult,
145 event: ServerLifecycleStreamReadyEvent,
146): boolean {
147 return (
148 (result.method === "desktop-app" && result.desktopUpdateToken !== undefined) ||
149 matchesServerUpdateReadyEvent(result, event)
150 );
151}
152
153export function validateServerUpdateReadyEvent(
154 result: ServerSelfUpdateResult,
155 event: ServerLifecycleStreamReadyEvent,
156): Effect.Effect<void, ServerUpdateTerminalError> {
157 if (result.updateId === undefined) return Effect.void;
158 const outcome = event.payload.updateOutcome;
159 if (
160 outcome?.id === result.updateId &&
161 outcome.status === "committed" &&
162 outcome.targetVersion === result.targetVersion &&
163 event.payload.environment.serverVersion === result.targetVersion
164 ) {
165 return Effect.void;
166 }
167 return Effect.fail(
168 new ServerUpdateTerminalError({
169 targetVersion: result.targetVersion,
170 status: outcome?.status ?? "failed",
171 reason:
172 outcome?.reason ??
173 "The service launcher resumed without committing the requested server version.",
174 }),
175 );
176}
177
178/**
179 * Keeps reconnect attempts ~1s apart for the whole update restart.
180 *
181 * A restart takes the server down for ~15 seconds, but the supervisor's normal
182 * backoff assumes an unexpected failure and doubles its delay after each failed
183 * attempt, so a 15-second restart can be observed as a ~30-second "Resuming".
184 * Nudging on every backoff entry (not just the first)
185 * holds the retry cadence flat until the server answers again. The sleep before
186 * each nudge is the pacer: a connection that fails instantly re-enters backoff
187 * immediately and would otherwise spin a tight retry loop.
188 *
189 * A newly restarted server can also reject the first environment credential.
190 * Authentication blocks need the same paced retry during this known restart;
191 * permission and configuration failures remain blocked.
192 *
193 * Callers fork this as a child of the update command so it is interrupted as
194 * soon as the update settles, whether it succeeds, fails, or times out.
195 */
196export function nudgeReconnectDuringUpdateRestart(input: {
197 readonly stateChanges: Stream.Stream<
198 {
199 readonly phase: string;
200 readonly lastFailure?: { readonly reason: string } | null;
201 },
202 unknown
203 >;
204 readonly retryNow: Effect.Effect<void>;
205 readonly interval?: Duration.Duration;
206}): Effect.Effect<void> {
207 return input.stateChanges.pipe(
208 Stream.filter(
209 (state) =>
210 state.phase === "backoff" ||
211 (state.phase === "blocked" && state.lastFailure?.reason === "authentication"),
212 ),
213 Stream.runForEach(() =>
214 Effect.sleep(input.interval ?? Duration.seconds(1)).pipe(Effect.andThen(input.retryNow)),
215 ),
216 Effect.timeoutOption(SERVER_UPDATE_RESUME_TIMEOUT),
217 Effect.ignore,
218 );
219}
220
221export function waitForNextEnvironmentReconnect<E>(
222 stateChanges: Stream.Stream<{ readonly phase: string }, E>,
223): Effect.Effect<void, E> {
224 return stateChanges.pipe(
225 Stream.dropWhile((state) => state.phase === "connected"),
226 Stream.filter((state) => state.phase === "connected"),
227 Stream.runHead,
228 Effect.asVoid,
229 );
230}
231
232export const runDesktopCommitWithReconnectObserver = Effect.fn(
233 "runDesktopCommitWithReconnectObserver",
234)(function* <EState, ECommit>(
235 stateChanges: Stream.Stream<{ readonly phase: string }, EState>,
236 commit: Effect.Effect<void, ECommit>,
237) {
238 const armed = yield* Deferred.make<void>();
239 const reconnected = yield* Deferred.make<void>();
240 const observer = yield* stateChanges.pipe(
241 Stream.tap(() => Deferred.succeed(armed, undefined)),
242 waitForNextEnvironmentReconnect,
243 Effect.andThen(Deferred.succeed(reconnected, undefined)),
244 Effect.forkChild,
245 );
246 yield* Deferred.await(armed);
247 const commitExit = yield* commit.pipe(Effect.exit);
248 if (Exit.isSuccess(commitExit)) {
249 yield* Fiber.interrupt(observer);
250 return;
251 }
252 if (!isLegacyUpdateHandoffLoss(commitExit.cause)) {
253 yield* Fiber.interrupt(observer);
254 return yield* Effect.failCause(commitExit.cause);
255 }
256 yield* Deferred.await(reconnected).pipe(Effect.timeout(SERVER_UPDATE_RESUME_TIMEOUT));
257 return yield* Effect.failCause(commitExit.cause);
258});
259
260export const waitForDesktopUpdateTarget = Effect.fn("waitForDesktopUpdateTarget")(function* <
261 EReady,
262 ECommit,
263>(
264 targetVersion: string,
265 nextReady: Effect.Effect<ServerLifecycleStreamReadyEvent, EReady>,
266 retryCommit: Effect.Effect<void, ECommit>,
267 maxCommitAttempts = 3,
268): Effect.fn.Return<ServerLifecycleStreamReadyEvent, EReady | ECommit | ServerUpdateTerminalError> {
269 for (let attempt = 1; attempt <= maxCommitAttempts; attempt += 1) {
270 const ready = yield* nextReady;
271 if (ready.payload.environment.serverVersion === targetVersion) return ready;
272 if (attempt === maxCommitAttempts) break;
273 const retryExit = yield* retryCommit.pipe(Effect.exit);
274 if (Exit.isSuccess(retryExit)) break;
275 if (!isLegacyUpdateHandoffLoss(retryExit.cause)) {
276 return yield* Effect.failCause(retryExit.cause);
277 }
278 }
279 return yield* new ServerUpdateTerminalError({
280 targetVersion,
281 status: "failed",
282 reason: "The desktop app resumed without installing the prepared update.",
283 });
284});
285
286export function serverUpdateStateForProgressEvent(
287 fromVersion: string,
288 targetVersion: string,
289 event: ServerSelfUpdateProgressEvent,
290): Extract<ServerUpdateState, { status: "running" }> {
291 return {
292 status: "running",
293 stage: event.type === "complete" ? "resuming" : event.stage,
294 fromVersion,
295 targetVersion,
296 };
297}
298
299export function serverUpdateStateForServerVersion(
300 state: ServerUpdateState,
301 serverVersion: string | null,
302): ServerUpdateState {
303 return state.status === "idle" ||
304 state.status === "running" ||
305 serverVersion === null ||
306 state.fromVersion === serverVersion
307 ? state
308 : IDLE_SERVER_UPDATE_STATE;
309}
310
311export function serverUpdateFailureMessage(error: unknown): string {
312 return error instanceof Error ? error.message : "Server update failed.";
313}
314
315function isRpcSocketError(error: unknown): boolean {
316 if (!isRpcClientError(error)) {
317 return false;
318 }
319 switch (error.reason._tag) {
320 case "SocketReadError":
321 case "SocketWriteError":
322 case "SocketCloseError":
323 return true;
324 default:
325 return false;
326 }
327}
328
329export function isLegacyUpdateHandoffLoss(cause: Cause.Cause<unknown>): boolean {
330 if (Cause.hasInterruptsOnly(cause)) {
331 return true;
332 }
333 return (
334 cause.reasons.length > 0 &&
335 cause.reasons.every((reason) => Cause.isFailReason(reason) && isRpcSocketError(reason.error))
336 );
337}
338
339export function resolveServerUpdateProgressResult<E>(
340 targetVersion: string,
341 terminal: Option.Option<ServerSelfUpdateResult>,
342 streamExit: Exit.Exit<void, E>,
343): Effect.Effect<ServerSelfUpdateResult, E | ServerUpdateProgressIncompleteError> {
344 if (
345 Option.isSome(terminal) &&
346 (Exit.isSuccess(streamExit) || isLegacyUpdateHandoffLoss(streamExit.cause))
347 ) {
348 return Effect.succeed(terminal.value);
349 }
350 if (Exit.isFailure(streamExit)) {
351 return Effect.failCause(streamExit.cause);
352 }
353 return Effect.fail(new ServerUpdateProgressIncompleteError({ targetVersion }));
354}
355
356const cachedConfigSnapshotEvent = (config: ServerConfig): ServerConfigStreamEvent => ({
357 version: 1,
358 type: "snapshot",
359 config,
360});
361
362export interface ServerConfigSubscriptionOptions {
363 readonly environmentThemes?: boolean;
364 readonly usageLimitSources?: boolean;
365 readonly usageLimitsCommand?: boolean;
366}
367
368export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConfigState.make")(
369 function* (subscription: ServerConfigSubscriptionOptions) {
370 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
371 const cache = yield* Persistence.EnvironmentCacheStore;
372 const environmentId = supervisor.target.environmentId;
373 const cachedConfig = yield* cache.loadServerConfig(environmentId).pipe(
374 Effect.catch((error) =>
375 Effect.logWarning("Could not load cached server configuration.").pipe(
376 Effect.annotateLogs({
377 environmentId,
378 ...safeErrorLogAttributes(error),
379 }),
380 Effect.as(Option.none<ServerConfig>()),
381 ),
382 ),
383 );
384 const state = yield* SubscriptionRef.make<Option.Option<ServerConfigProjection>>(
385 // Stripped on load as well as on save: a cache written by an earlier
386 // build can still carry published themes.
387 Option.map(cachedConfig, (cached) => ({
388 config: withoutEnvironmentThemes(cached),
389 latestEvent: cachedConfigSnapshotEvent(withoutEnvironmentThemes(cached)),
390 source: "cache" as const,
391 })),
392 );
393 const persistence = yield* Queue.sliding<ServerConfig>(1);
394 const pendingPersistence = yield* Ref.make<Option.Option<ServerConfig>>(Option.none());
395
396 const persist = Effect.fn("EnvironmentServerConfigState.persist")(function* (
397 config: ServerConfig,
398 ) {
399 return yield* cache.saveServerConfig(environmentId, withoutEnvironmentThemes(config)).pipe(
400 Effect.as(true),
401 Effect.catch((error) =>
402 Effect.logWarning("Could not persist cached server configuration.").pipe(
403 Effect.annotateLogs({
404 environmentId,
405 ...safeErrorLogAttributes(error),
406 }),
407 Effect.as(false),
408 ),
409 ),
410 );
411 });
412
413 const persistPending = Effect.fn("EnvironmentServerConfigState.persistPending")(function* (
414 config: ServerConfig,
415 ) {
416 if (!(yield* persist(config))) {
417 return;
418 }
419 yield* Ref.update(pendingPersistence, (pending) =>
420 Option.isSome(pending) && pending.value === config ? Option.none() : pending,
421 );
422 });
423
424 yield* Effect.addFinalizer(() =>
425 Ref.get(pendingPersistence).pipe(
426 Effect.flatMap(
427 Option.match({
428 onNone: () => Effect.void,
429 onSome: (config) => persist(config).pipe(Effect.asVoid),
430 }),
431 ),
432 ),
433 );
434
435 yield* runCachePersistence(persistence, persistPending).pipe(Effect.forkScoped);
436
437 yield* subscribe(WS_METHODS.subscribeServerConfig, {
438 ...(subscription.environmentThemes === true ? { environmentThemes: true } : {}),
439 ...(subscription.usageLimitSources === true ? { usageLimitSources: true } : {}),
440 ...(subscription.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}),
441 }).pipe(
442 Stream.runForEach((event) =>
443 Effect.gen(function* () {
444 const next = applyServerConfigProjection(yield* SubscriptionRef.get(state), event);
445 if (Option.isNone(next)) {
446 return;
447 }
448 yield* Ref.set(pendingPersistence, Option.some(next.value.config));
449 yield* SubscriptionRef.set(state, next);
450 yield* Queue.offer(persistence, next.value.config);
451 }),
452 ),
453 Effect.forkScoped,
454 );
455
456 return state;
457 },
458);
459
460function serverConfigStateChanges(
461 environmentId: EnvironmentId,
462 subscription: ServerConfigSubscriptionOptions,
463) {
464 return followStreamInEnvironment(
465 environmentId,
466 Stream.unwrap(
467 makeEnvironmentServerConfigState(subscription).pipe(
468 Effect.map((state) =>
469 SubscriptionRef.changes(state).pipe(
470 Stream.filterMap((projection) =>
471 Option.match(projection, {
472 onNone: () => Result.failVoid,
473 onSome: (value) => Result.succeed(value),
474 }),
475 ),
476 ),
477 ),
478 ),
479 ),
480 );
481}
482
483export function applyServerWelcomeEvent(
484 current: EnvironmentServerWelcomeState,
485 session: RpcSession,
486 event: {
487 readonly type: "welcome" | "ready" | "legacyThreadMigration";
488 readonly payload: unknown;
489 },
490): EnvironmentServerWelcomeState {
491 return event.type === "welcome" && current.currentSession === session
492 ? {
493 ...current,
494 welcomeSession: session,
495 welcome: event.payload as ServerLifecycleWelcomePayload,
496 }
497 : current;
498}
499
500export interface EnvironmentServerWelcomeState {
501 readonly currentSession: RpcSession | null;
502 readonly welcomeSession: RpcSession | null;
503 readonly welcome: ServerLifecycleWelcomePayload | null;
504}
505
506export function resolveServerWelcomeState(
507 state: EnvironmentServerWelcomeState,
508): ServerLifecycleWelcomePayload | null {
509 return state.currentSession === state.welcomeSession ? state.welcome : null;
510}
511
512export const makeEnvironmentServerWelcomeState = Effect.fn("EnvironmentServerWelcomeState.make")(
513 function* () {
514 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
515 const initialSession = Option.getOrNull(yield* SubscriptionRef.get(supervisor.session));
516 const state = yield* SubscriptionRef.make<EnvironmentServerWelcomeState>({
517 currentSession: initialSession,
518 welcomeSession: null,
519 welcome: null,
520 });
521
522 const updateWithCurrentSession = Effect.fn(
523 "EnvironmentServerWelcomeState.updateWithCurrentSession",
524 )(function* (
525 update: (
526 current: EnvironmentServerWelcomeState,
527 currentSession: RpcSession | null,
528 ) => EnvironmentServerWelcomeState,
529 ) {
530 return yield* SubscriptionRef.modifyEffect(state, (current) =>
531 SubscriptionRef.get(supervisor.session).pipe(
532 Effect.map(
533 (latestSession) =>
534 [undefined, update(current, Option.getOrNull(latestSession))] as const,
535 ),
536 ),
537 );
538 });
539
540 yield* SubscriptionRef.changes(supervisor.session).pipe(
541 Stream.runForEach(() =>
542 updateWithCurrentSession((current, currentSession) => ({
543 ...current,
544 currentSession,
545 })),
546 ),
547 Effect.forkScoped,
548 );
549
550 yield* subscribeDynamicWithSession(
551 WS_METHODS.subscribeServerLifecycle,
552 Effect.fn("EnvironmentServerWelcomeState.makeSubscribeInput")(function* (session) {
553 yield* updateWithCurrentSession((current, currentSession) =>
554 currentSession === session
555 ? {
556 ...current,
557 currentSession,
558 welcomeSession: session,
559 welcome: null,
560 }
561 : { ...current, currentSession },
562 );
563 return {};
564 }),
565 ).pipe(
566 Stream.runForEach(([session, event]) =>
567 updateWithCurrentSession((current, currentSession) =>
568 applyServerWelcomeEvent(
569 {
570 ...current,
571 currentSession,
572 },
573 session,
574 event,
575 ),
576 ),
577 ),
578 Effect.forkScoped,
579 );
580
581 return state;
582 },
583);
584
585function serverWelcomeStateChanges(environmentId: EnvironmentId) {
586 return followStreamInEnvironment(
587 environmentId,
588 Stream.unwrap(
589 makeEnvironmentServerWelcomeState().pipe(
590 Effect.map((state) =>
591 SubscriptionRef.changes(state).pipe(Stream.map(resolveServerWelcomeState)),
592 ),
593 ),
594 ),
595 );
596}
597
598export function resolveServerConfigValue(
599 projection: ServerConfigProjection | null,
600 initialConfig: ServerConfig | null,
601): ServerConfig | null {
602 if (
603 projection?.source === "live" &&
604 (initialConfig === null ||
605 projection.config.environment.serverVersion === initialConfig.environment.serverVersion)
606 ) {
607 return projection.config;
608 }
609 return initialConfig ?? projection?.config ?? null;
610}
611
612export function createServerEnvironmentAtoms<R, E>(
613 runtime: Atom.AtomRuntime<
614 EnvironmentRegistry.EnvironmentRegistry | Persistence.EnvironmentCacheStore | R,
615 E
616 >,
617 options: {
618 readonly initialConfigValueAtom: (
619 environmentId: EnvironmentId,
620 ) => Atom.Atom<ServerConfig | null>;
621 /**
622 * Whether this surface renders themes the environment publishes. Mobile
623 * keeps its own appearance settings, so it neither asks for the stream nor
624 * receives the payload.
625 */
626 readonly environmentThemes?: boolean;
627 /** Whether this surface renders quota from configured usage-limit sources. */
628 readonly usageLimitSources?: boolean;
629 readonly usageLimitsCommand?: boolean;
630 },
631) {
632 const configScheduler = createAtomCommandScheduler();
633 // Updates stay serial end-to-end, but only their handoff phase occupies the config lane.
634 const updateScheduler = createAtomCommandScheduler();
635 const configConcurrency = {
636 mode: "serial" as const,
637 key: ({ environmentId }: { readonly environmentId: string }) => environmentId,
638 };
639 const configProjectionFamily = Atom.family((environmentId: EnvironmentId) =>
640 runtime
641 .atom(
642 serverConfigStateChanges(environmentId, {
643 ...(options.environmentThemes === true ? { environmentThemes: true } : {}),
644 ...(options.usageLimitSources === true ? { usageLimitSources: true } : {}),
645 ...(options.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}),
646 }),
647 )
648 .pipe(
649 Atom.setIdleTTL(5 * 60_000),
650 Atom.withLabel(`environment-data:server:config-projection:${environmentId}`),
651 ),
652 );
653 const configProjection = (target: {
654 readonly environmentId: EnvironmentId;
655 readonly input: EnvironmentRpcInput<typeof WS_METHODS.subscribeServerConfig>;
656 }) => configProjectionFamily(target.environmentId);
657 const emptyConfigAtom = Atom.make<ServerConfig | null>(null).pipe(
658 Atom.withLabel("environment-data:server:config:empty"),
659 );
660 const configValueAtom = Atom.family((environmentId: EnvironmentId | null) => {
661 if (environmentId === null) {
662 return emptyConfigAtom;
663 }
664 return Atom.make((get): ServerConfig | null => {
665 const projection = Option.getOrNull(
666 AsyncResult.value(get(configProjection({ environmentId, input: {} }))),
667 );
668 return resolveServerConfigValue(
669 projection,
670 get(options.initialConfigValueAtom(environmentId)),
671 );
672 }).pipe(Atom.withLabel(`environment-data:server:config:${environmentId}`));
673 });
674 const updateStateValueAtom = Atom.family((environmentId: EnvironmentId) =>
675 Atom.make((get) =>
676 serverUpdateStateForServerVersion(
677 get(serverUpdateStateAtom(environmentId)),
678 get(configValueAtom(environmentId))?.environment.serverVersion ?? null,
679 ),
680 ).pipe(Atom.withLabel(`environment-data:server:update-state-value:${environmentId}`)),
681 );
682 const updateStateAtom = (environmentId: EnvironmentId | null) =>
683 environmentId === null ? EMPTY_SERVER_UPDATE_STATE_ATOM : updateStateValueAtom(environmentId);
684 const updateServer = createRuntimeCommand<
685 EnvironmentRegistry.EnvironmentRegistry | Persistence.EnvironmentCacheStore | R,
686 E,
687 ServerUpdateTarget,
688 ServerSelfUpdateResult,
689 unknown
690 >(runtime, {
691 label: "environment-data:server:update-server",
692 scheduler: updateScheduler,
693 concurrency: configConcurrency,
694 execute: (target, atomRegistry) => {
695 const stateAtom = serverUpdateStateAtom(target.environmentId);
696 let targetVersion = target.input.targetVersion;
697 let fromVersion =
698 atomRegistry.get(configValueAtom(target.environmentId))?.environment.serverVersion ??
699 targetVersion;
700 let currentStage: ServerUpdateStage = "downloading";
701 let desktopCommitLostTransport = false;
702 atomRegistry.set(stateAtom, {
703 status: "running",
704 stage: currentStage,
705 fromVersion,
706 targetVersion,
707 });
708
709 return Effect.gen(function* () {
710 const environmentRegistry = yield* EnvironmentRegistry.EnvironmentRegistry;
711 const desktopCommitStarting = yield* Deferred.make<void>();
712 const desktopReconnectObserverArmed = yield* Deferred.make<void>();
713 const desktopReconnected = yield* Deferred.make<void>();
714 yield* Deferred.await(desktopCommitStarting).pipe(
715 Effect.andThen(
716 environmentRegistry.stateChanges(target.environmentId).pipe(
717 Stream.tap(() => Deferred.succeed(desktopReconnectObserverArmed, undefined)),
718 Stream.dropWhile((state) => state.phase === "connected"),
719 Stream.filter((state) => state.phase === "connected"),
720 Stream.runHead,
721 ),
722 ),
723 Effect.andThen(Deferred.succeed(desktopReconnected, undefined)),
724 Effect.forkChild,
725 );
726 const result = yield* scheduleAtomCommandEffect(
727 atomRegistry,
728 configScheduler,
729 configConcurrency,
730 target,
731 Effect.gen(function* () {
732 const currentConfig = atomRegistry.get(configValueAtom(target.environmentId));
733 fromVersion = currentConfig?.environment.serverVersion ?? targetVersion;
734 atomRegistry.set(stateAtom, {
735 status: "running",
736 stage: currentStage,
737 fromVersion,
738 targetVersion,
739 });
740
741 const supportsProgress =
742 currentConfig?.environment.capabilities.serverSelfUpdateProgress === true;
743 const updateResult: ServerSelfUpdateResult = supportsProgress
744 ? yield* Effect.gen(function* () {
745 const terminal = yield* Ref.make<Option.Option<ServerSelfUpdateResult>>(
746 Option.none(),
747 );
748 const streamExit = yield* environmentRegistry
749 .runStream(
750 target.environmentId,
751 runStream(WS_METHODS.serverUpdateServerWithProgress, target.input),
752 )
753 .pipe(
754 Stream.runForEach((event) =>
755 Effect.sync(() => {
756 currentStage = event.type === "complete" ? "resuming" : event.stage;
757 atomRegistry.set(
758 stateAtom,
759 serverUpdateStateForProgressEvent(fromVersion, targetVersion, event),
760 );
761 }).pipe(
762 Effect.andThen(
763 event.type === "complete"
764 ? Ref.set(terminal, Option.some(event.result))
765 : Effect.void,
766 ),
767 ),
768 ),
769 Effect.exit,
770 );
771 return yield* resolveServerUpdateProgressResult(
772 targetVersion,
773 yield* Ref.get(terminal),
774 streamExit,
775 );
776 })
777 : yield* Effect.gen(function* () {
778 const selfUpdateMethod = currentConfig?.environment.capabilities.serverSelfUpdate;
779 const exit = yield* environmentRegistry
780 .run(target.environmentId, request(WS_METHODS.serverUpdateServer, target.input))
781 .pipe(Effect.exit);
782 if (Exit.isSuccess(exit)) {
783 return exit.value;
784 }
785 if (
786 (selfUpdateMethod === "boot-service" || selfUpdateMethod === "respawn") &&
787 isLegacyUpdateHandoffLoss(exit.cause)
788 ) {
789 // Older servers can tear down the transport before their
790 // unary acknowledgement arrives. Treat only that transport
791 // loss as a handoff, then prove it by waiting for target ready.
792 return { targetVersion, method: selfUpdateMethod };
793 }
794 return yield* Effect.failCause(exit.cause);
795 });
796
797 if (
798 updateResult.method === "desktop-app" &&
799 updateResult.desktopUpdateToken !== undefined
800 ) {
801 yield* Deferred.succeed(desktopCommitStarting, undefined);
802 yield* Deferred.await(desktopReconnectObserverArmed);
803 const commitExit = yield* environmentRegistry
804 .run(
805 target.environmentId,
806 request(WS_METHODS.serverCommitDesktopUpdate, {
807 requestId: updateResult.desktopUpdateToken,
808 }),
809 )
810 .pipe(Effect.exit);
811 if (Exit.isFailure(commitExit) && !isLegacyUpdateHandoffLoss(commitExit.cause)) {
812 return yield* Effect.failCause(commitExit.cause);
813 }
814 desktopCommitLostTransport = Exit.isFailure(commitExit);
815 }
816
817 targetVersion = updateResult.targetVersion;
818
819 currentStage = "resuming";
820 atomRegistry.set(stateAtom, {
821 status: "running",
822 stage: currentStage,
823 fromVersion,
824 targetVersion,
825 });
826 return updateResult;
827 }),
828 );
829
830 // The update restart is intentional and the server stays unreachable
831 // for the whole restart, so hold the retry cadence flat instead of
832 // letting the supervisor climb its backoff ladder.
833 yield* nudgeReconnectDuringUpdateRestart({
834 stateChanges: environmentRegistry.stateChanges(target.environmentId),
835 retryNow: environmentRegistry.retryNow(target.environmentId),
836 }).pipe(Effect.forkChild);
837
838 if (result.method === "desktop-app" && desktopCommitLostTransport) {
839 yield* Deferred.await(desktopReconnected).pipe(
840 Effect.timeout(SERVER_UPDATE_RESUME_TIMEOUT),
841 );
842 }
843
844 const waitForReady = environmentRegistry
845 .followStream(target.environmentId, subscribe(WS_METHODS.subscribeServerLifecycle, {}))
846 .pipe(
847 Stream.filter(
848 (event): event is ServerLifecycleStreamReadyEvent =>
849 event.type === "ready" && matchesServerUpdateResumeEvent(result, event),
850 ),
851 Stream.runHead,
852 Effect.timeoutOption(SERVER_UPDATE_RESUME_TIMEOUT),
853 Effect.map(Option.flatten),
854 );
855 const nextReady = waitForReady.pipe(
856 Effect.flatMap(
857 Option.match({
858 onNone: () =>
859 new ServerUpdateResumeTimeoutError({
860 environmentId: target.environmentId,
861 targetVersion,
862 }),
863 onSome: Effect.succeed,
864 }),
865 ),
866 );
867 const desktopUpdateToken = result.desktopUpdateToken;
868 const resumed =
869 result.method === "desktop-app" && desktopUpdateToken !== undefined
870 ? yield* waitForDesktopUpdateTarget(
871 result.targetVersion,
872 nextReady,
873 runDesktopCommitWithReconnectObserver(
874 environmentRegistry.stateChanges(target.environmentId),
875 environmentRegistry
876 .run(
877 target.environmentId,
878 request(WS_METHODS.serverCommitDesktopUpdate, {
879 requestId: desktopUpdateToken,
880 }),
881 )
882 .pipe(Effect.asVoid),
883 ),
884 )
885 : yield* nextReady;
886 yield* validateServerUpdateReadyEvent(result, resumed);
887
888 atomRegistry.set(stateAtom, IDLE_SERVER_UPDATE_STATE);
889 return result;
890 }).pipe(
891 Effect.onExit((exit) =>
892 Effect.sync(() => {
893 if (Exit.isSuccess(exit)) {
894 return;
895 }
896 if (Cause.hasInterruptsOnly(exit.cause)) {
897 atomRegistry.set(stateAtom, IDLE_SERVER_UPDATE_STATE);
898 return;
899 }
900 atomRegistry.set(stateAtom, {
901 status: "failed",
902 stage: currentStage,
903 fromVersion,
904 targetVersion,
905 message: serverUpdateFailureMessage(Cause.squash(exit.cause)),
906 });
907 }),
908 ),
909 );
910 },
911 });
912 const settingsValueAtom = Atom.family((environmentId: EnvironmentId) =>
913 Atom.make((get) => get(configValueAtom(environmentId))?.settings ?? null).pipe(
914 Atom.withLabel(`environment-data:server:settings:${environmentId}`),
915 ),
916 );
917 const usagePricesAtom = Atom.family((environmentId: EnvironmentId) =>
918 Atom.make((get) => {
919 const overrides = get(settingsValueAtom(environmentId))?.usagePriceOverrides ?? {};
920 // Only changed prices should trigger another transcript scan. Settings
921 // snapshots can recreate the same mapping in a different property order.
922 return JSON.stringify(
923 Object.keys(overrides)
924 .sort()
925 .map((model) => {
926 const price = overrides[model]!;
927 return [
928 model,
929 price.inputCostPerMillionTokens,
930 price.outputCostPerMillionTokens,
931 price.cacheReadCostPerMillionTokens,
932 price.cacheWriteCostPerMillionTokens,
933 ];
934 }),
935 );
936 }).pipe(Atom.withLabel(`environment-data:server:usage-prices:${environmentId}`)),
937 );
938 const usageScanSettingsAtom = Atom.family((environmentId: EnvironmentId) =>
939 Atom.make((get) => {
940 const settings = get(settingsValueAtom(environmentId));
941 const aliases = settings?.usageModelAliases ?? {};
942 return JSON.stringify([
943 get(usagePricesAtom(environmentId)),
944 Object.keys(aliases)
945 .sort()
946 .map((model) => [model, aliases[model]]),
947 settings?.cursorKeychainUsageEnabled ?? false,
948 ]);
949 }).pipe(Atom.withLabel(`environment-data:server:usage-scan-settings:${environmentId}`)),
950 );
951 const providersValueAtom = Atom.family((environmentId: EnvironmentId) =>
952 Atom.make((get) => get(configValueAtom(environmentId))?.providers ?? null).pipe(
953 Atom.withLabel(`environment-data:server:providers:${environmentId}`),
954 ),
955 );
956 const welcomeStateFamily = Atom.family((environmentId: EnvironmentId) =>
957 runtime
958 .atom(serverWelcomeStateChanges(environmentId), { initialValue: null })
959 .pipe(
960 Atom.setIdleTTL(5 * 60_000),
961 Atom.withLabel(`environment-data:server:welcome-state:${environmentId}`),
962 ),
963 );
964 const welcomeFamily = Atom.family((environmentId: EnvironmentId) =>
965 Atom.make((get) => {
966 const result = get(welcomeStateFamily(environmentId));
967 if (result._tag !== "Success") return result;
968 return result.value === null
969 ? AsyncResult.initial<ServerLifecycleWelcomePayload, never>(result.waiting)
970 : AsyncResult.success(result.value, result);
971 }).pipe(Atom.withLabel(`environment-data:server:welcome:${environmentId}`)),
972 );
973 const welcome = (target: {
974 readonly environmentId: EnvironmentId;
975 readonly input: EnvironmentRpcInput<typeof WS_METHODS.subscribeServerLifecycle>;
976 }) => welcomeFamily(target.environmentId);
977 const updateSettings = createEnvironmentRpcCommand(runtime, {
978 label: "environment-data:server:update-settings",
979 tag: WS_METHODS.serverUpdateSettings,
980 scheduler: configScheduler,
981 concurrency: configConcurrency,
982 });
983
984 return {
985 configValueAtom,
986 runStorageCleanup: createEnvironmentRpcCommand(runtime, {
987 label: "environment-data:server:run-storage-cleanup",
988 tag: WS_METHODS.serverRunStorageCleanup,
989 concurrency: { mode: "singleFlight", key: ({ environmentId }) => environmentId },
990 }),
991 storageCleanupReport: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
992 label: "environment-data:server:storage-cleanup-report",
993 tag: WS_METHODS.serverGetStorageCleanupReport,
994 idleTtlMs: 0,
995 }),
996 updateStateAtom,
997 settingsValueAtom,
998 providersValueAtom,
999 providerAuthState: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
1000 label: "environment-data:provider:auth-state",
1001 tag: WS_METHODS.providerAuthSubscribe,
1002 idleTtlMs: 0,
1003 }),
1004 startProviderAuth: createEnvironmentRpcCommand(runtime, {
1005 label: "environment-data:provider:auth-start",
1006 tag: WS_METHODS.providerAuthStart,
1007 concurrency: {
1008 mode: "singleFlight",
1009 key: ({ environmentId, input }) => JSON.stringify([environmentId, input]),
1010 },
1011 }),
1012 respondProviderAuth: createEnvironmentRpcCommand(runtime, {
1013 label: "environment-data:provider:auth-respond",
1014 tag: WS_METHODS.providerAuthRespond,
1015 }),
1016 completeProviderAuth: createEnvironmentRpcCommand(runtime, {
1017 label: "environment-data:provider:auth-complete",
1018 tag: WS_METHODS.providerAuthComplete,
1019 }),
1020 chatGptReconnectProfile: createEnvironmentRpcCommand(runtime, {
1021 label: "environment-data:chatgpt:reconnect-profile",
1022 tag: WS_METHODS.chatGptReconnectProfile,
1023 }),
1024 chatGptImportProfile: createEnvironmentRpcCommand(runtime, {
1025 label: "environment-data:chatgpt:import-profile",
1026 tag: WS_METHODS.chatGptImportProfile,
1027 }),
1028 chatGptHandoffState: createEnvironmentSubscriptionAtomFamily(runtime, {
1029 label: "environment-data:chatgpt:handoff",
1030 sensitiveInput: true,
1031 // OAuth must not be replayed when the connection recovers.
1032 subscribe: (input: EnvironmentRpcInput<typeof WS_METHODS.chatGptHandoffSubscribe>) =>
1033 runStream(WS_METHODS.chatGptHandoffSubscribe, input),
1034 idleTtlMs: 0,
1035 }),
1036 codexAuthCallbackState: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
1037 label: "environment-data:codex:auth-callback",
1038 tag: WS_METHODS.codexAuthCallbackSubscribe,
1039 idleTtlMs: 0,
1040 }),
1041 cancelProviderAuth: createEnvironmentRpcCommand(runtime, {
1042 label: "environment-data:provider:auth-cancel",
1043 tag: WS_METHODS.providerAuthCancel,
1044 }),
1045 logoutProviderAuth: createEnvironmentRpcCommand(runtime, {
1046 label: "environment-data:provider:auth-logout",
1047 tag: WS_METHODS.providerAuthLogout,
1048 }),
1049 providerInstallState: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
1050 label: "environment-data:provider:install-state",
1051 tag: WS_METHODS.providerInstallSubscribe,
1052 idleTtlMs: 0,
1053 }),
1054 startProviderInstall: createEnvironmentRpcCommand(runtime, {
1055 label: "environment-data:provider:install-start",
1056 tag: WS_METHODS.providerInstallStart,
1057 concurrency: {
1058 mode: "singleFlight",
1059 key: ({ environmentId }) => environmentId,
1060 },
1061 }),
1062 cancelProviderInstall: createEnvironmentRpcCommand(runtime, {
1063 label: "environment-data:provider:install-cancel",
1064 tag: WS_METHODS.providerInstallCancel,
1065 }),
1066 removeProviderInstallation: createEnvironmentRpcCommand(runtime, {
1067 label: "environment-data:provider:install-remove",
1068 tag: WS_METHODS.providerInstallRemove,
1069 }),
1070 traceDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, {
1071 label: "environment-data:server:trace-diagnostics",
1072 tag: WS_METHODS.serverGetTraceDiagnostics,
1073 }),
1074 processDiagnostics: createEnvironmentRpcQueryAtomFamily(runtime, {
1075 label: "environment-data:server:process-diagnostics",
1076 tag: WS_METHODS.serverGetProcessDiagnostics,
1077 }),
1078 hostResources: createEnvironmentQueryAtomFamily(runtime, {
1079 label: "environment-data:server:host-resources",
1080 idleTtlMs: 0,
1081 staleTimeMs: 5_000,
1082 execute: (input: EnvironmentRpcInput<typeof WS_METHODS.serverGetHostResources>) =>
1083 request(WS_METHODS.serverGetHostResources, input).pipe(Effect.timeout("5 seconds")),
1084 }),
1085 processResourceHistory: createEnvironmentRpcQueryAtomFamily(runtime, {
1086 label: "environment-data:server:process-resource-history",
1087 tag: WS_METHODS.serverGetProcessResourceHistory,
1088 }),
1089 scheduledTaskWebhookDeliveries: createEnvironmentRpcQueryAtomFamily(runtime, {
1090 label: "environment-data:server:scheduled-task:webhook-deliveries",
1091 tag: WS_METHODS.scheduledTasksListWebhookDeliveries,
1092 }),
1093 scheduledTaskWebhookDelivery: createEnvironmentRpcQueryAtomFamily(runtime, {
1094 label: "environment-data:server:scheduled-task:webhook-delivery",
1095 tag: WS_METHODS.scheduledTasksGetWebhookDelivery,
1096 }),
1097 /** Live scheduled-task list: snapshot on subscribe, fresh list after every server-side change. */
1098 scheduledTasksLive: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
1099 label: "environment-data:server:scheduled-tasks:live",
1100 tag: WS_METHODS.scheduledTasksSubscribe,
1101 }),
1102 // A cold transcript scan is measured in seconds, so keep the result around
1103 // long enough that switching windows or re-rendering does not rescan.
1104 // Slow sources answer from cache first: that summary stays on screen, with
1105 // the query waiting, until one `awaitRefresh` request replaces it.
1106 usageSummary: createEnvironmentRpcQueryAtomFamily(runtime, {
1107 label: "environment-data:server:usage-summary",
1108 tag: WS_METHODS.serverGetUsageSummary,
1109 staleTimeMs: 60_000,
1110 refreshTrigger: ({ environmentId }) => usageScanSettingsAtom(environmentId),
1111 execute: (input, emit) =>
1112 request(WS_METHODS.serverGetUsageSummary, input).pipe(
1113 Effect.flatMap((summary) =>
1114 summary.sources.some((source) => source.refreshing === true)
1115 ? emit(summary).pipe(
1116 Effect.andThen(
1117 request(WS_METHODS.serverGetUsageSummary, { ...input, awaitRefresh: true }),
1118 ),
1119 // The cached summary is still the best data, and a failure
1120 // would read as the environment not reporting usage at all.
1121 Effect.catch((error) =>
1122 Effect.logWarning("Could not refresh slow usage sources.").pipe(
1123 Effect.annotateLogs({ ...safeErrorLogAttributes(error) }),
1124 Effect.as(summary),
1125 ),
1126 ),
1127 )
1128 : Effect.succeed(summary),
1129 ),
1130 ),
1131 }),
1132 resourceTelemetry: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
1133 label: "environment-data:server:resource-telemetry",
1134 tag: WS_METHODS.subscribeResourceTelemetry,
1135 idleTtlMs: 0,
1136 }),
1137 resourceTelemetryHistory: createEnvironmentRpcQueryAtomFamily(runtime, {
1138 label: "environment-data:server:resource-telemetry-history",
1139 tag: WS_METHODS.serverGetResourceTelemetryHistory,
1140 staleTimeMs: 5_000,
1141 }),
1142 searchAcpRegistry: createEnvironmentRpcQueryAtomFamily(runtime, {
1143 label: "environment-data:server:acp-registry:search",
1144 tag: WS_METHODS.serverSearchAcpRegistry,
1145 // Each submitted search refreshes the server-side registry. Dropping an
1146 // abandoned query immediately also interrupts stale in-flight requests.
1147 staleTimeMs: 0,
1148 idleTtlMs: 0,
1149 }),
1150 configProjection,
1151 welcome,
1152 legacyThreadMigration: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
1153 label: "environment-data:server:legacy-thread-migration",
1154 tag: WS_METHODS.subscribeServerLifecycle,
1155 transform: (stream) =>
1156 stream.pipe(
1157 Stream.filterMap((event) =>
1158 event.type === "legacyThreadMigration"
1159 ? Result.succeed(event.payload)
1160 : Result.failVoid,
1161 ),
1162 ),
1163 }),
1164 consumeResetCredit: createEnvironmentRpcCommand(runtime, {
1165 label: "environment-data:server:consume-reset-credit",
1166 tag: WS_METHODS.providerConsumeResetCredit,
1167 concurrency: {
1168 mode: "singleFlight",
1169 // Both ids are free-form strings; a delimiter could collide.
1170 key: ({ environmentId, input }) => JSON.stringify([environmentId, input]),
1171 },
1172 }),
1173 refreshProviders: createEnvironmentRpcCommand(runtime, {
1174 label: "environment-data:server:refresh-providers",
1175 tag: WS_METHODS.serverRefreshProviders,
1176 concurrency: {
1177 mode: "singleFlight",
1178 key: ({ environmentId, input }) =>
1179 JSON.stringify([
1180 environmentId,
1181 input.instanceId ?? null,
1182 input.cwd ?? null,
1183 input.fresh ?? false,
1184 input.refreshModels ?? false,
1185 ]),
1186 },
1187 }),
1188 updateProvider: createEnvironmentRpcCommand(runtime, {
1189 label: "environment-data:server:update-provider",
1190 tag: WS_METHODS.serverUpdateProvider,
1191 scheduler: configScheduler,
1192 concurrency: configConcurrency,
1193 }),
1194 updateServer,
1195 upsertKeybinding: createEnvironmentRpcCommand(runtime, {
1196 label: "environment-data:server:upsert-keybinding",
1197 tag: WS_METHODS.serverUpsertKeybinding,
1198 scheduler: configScheduler,
1199 concurrency: configConcurrency,
1200 }),
1201 removeKeybinding: createEnvironmentRpcCommand(runtime, {
1202 label: "environment-data:server:remove-keybinding",
1203 tag: WS_METHODS.serverRemoveKeybinding,
1204 scheduler: configScheduler,
1205 concurrency: configConcurrency,
1206 }),
1207 updateSettings,
1208 // Provider-instance mutations share the settings command and its
1209 // environment-serial scheduler. The named boundary keeps clients on the
1210 // atomic map-entry payload instead of rebuilding a stale whole map.
1211 mutateProviderInstance: updateSettings,
1212 prepareAcpRegistryAgent: createEnvironmentRpcCommand(runtime, {
1213 label: "environment-data:server:acp-registry:prepare",
1214 tag: WS_METHODS.serverPrepareAcpRegistryAgent,
1215 concurrency: {
1216 mode: "singleFlight",
1217 key: ({ environmentId, input }) => `${environmentId}:${input.agentId}`,
1218 },
1219 }),
1220 uninstallAcpRegistryManagedBinary: createEnvironmentRpcCommand(runtime, {
1221 label: "environment-data:server:acp-registry:uninstall-managed-binary",
1222 tag: WS_METHODS.serverUninstallAcpRegistryManagedBinary,
1223 concurrency: {
1224 mode: "singleFlight",
1225 key: ({ environmentId, input }) => `${environmentId}:${input.agentId}`,
1226 },
1227 }),
1228 acceptAcpRegistryUrlAuth: createEnvironmentRpcCommand(runtime, {
1229 label: "environment-data:server:acp-registry:accept-url-auth",
1230 tag: WS_METHODS.serverAcceptAcpRegistryUrlAuth,
1231 concurrency: {
1232 mode: "singleFlight",
1233 key: ({ environmentId, input }) =>
1234 `${environmentId}:${input.instanceId}:${input.elicitationId}`,
1235 },
1236 }),
1237 listAcpRegistrySessions: createEnvironmentRpcCommand(runtime, {
1238 label: "environment-data:server:acp-registry:list-sessions",
1239 tag: WS_METHODS.serverListAcpRegistrySessions,
1240 concurrency: {
1241 mode: "singleFlight",
1242 key: ({ environmentId, input }) =>
1243 `${environmentId}:${input.instanceId}:${input.projectId}:${input.cursor ?? "first"}`,
1244 },
1245 }),
1246 importAcpRegistrySession: createEnvironmentRpcCommand(runtime, {
1247 label: "environment-data:server:acp-registry:import-session",
1248 tag: WS_METHODS.serverImportAcpRegistrySession,
1249 concurrency: {
1250 mode: "singleFlight",
1251 key: ({ environmentId, input }) =>
1252 `${environmentId}:${input.instanceId}:${input.projectId}:${input.sessionId}`,
1253 },
1254 }),
1255 deleteAcpRegistrySession: createEnvironmentRpcCommand(runtime, {
1256 label: "environment-data:server:acp-registry:delete-session",
1257 tag: WS_METHODS.serverDeleteAcpRegistrySession,
1258 concurrency: {
1259 mode: "singleFlight",
1260 key: ({ environmentId, input }) =>
1261 `${environmentId}:${input.instanceId}:${input.projectId}:${input.sessionId}`,
1262 },
1263 }),
1264 listAcpRegistryProviders: createEnvironmentRpcCommand(runtime, {
1265 label: "environment-data:server:acp-registry:list-providers",
1266 tag: WS_METHODS.serverListAcpRegistryProviders,
1267 concurrency: {
1268 mode: "singleFlight",
1269 key: ({ environmentId, input }) =>
1270 `${environmentId}:${input.instanceId}:${input.projectId}`,
1271 },
1272 }),
1273 setAcpRegistryProvider: createEnvironmentRpcCommand(runtime, {
1274 label: "environment-data:server:acp-registry:set-provider",
1275 tag: WS_METHODS.serverSetAcpRegistryProvider,
1276 concurrency: {
1277 mode: "singleFlight",
1278 key: ({ environmentId, input }) =>
1279 `${environmentId}:${input.instanceId}:${input.projectId}:${input.providerId}`,
1280 },
1281 }),
1282 disableAcpRegistryProvider: createEnvironmentRpcCommand(runtime, {
1283 label: "environment-data:server:acp-registry:disable-provider",
1284 tag: WS_METHODS.serverDisableAcpRegistryProvider,
1285 concurrency: {
1286 mode: "singleFlight",
1287 key: ({ environmentId, input }) =>
1288 `${environmentId}:${input.instanceId}:${input.projectId}:${input.providerId}`,
1289 },
1290 }),
1291 logoutAcpRegistry: createEnvironmentRpcCommand(runtime, {
1292 label: "environment-data:server:acp-registry:logout",
1293 tag: WS_METHODS.serverLogoutAcpRegistry,
1294 concurrency: {
1295 mode: "singleFlight",
1296 key: ({ environmentId, input }) => `${environmentId}:${input.instanceId}`,
1297 },
1298 }),
1299 signalProcess: createEnvironmentRpcCommand(runtime, {
1300 label: "environment-data:server:signal-process",
1301 tag: WS_METHODS.serverSignalProcess,
1302 }),
1303 upsertScheduledTask: createEnvironmentRpcCommand(runtime, {
1304 label: "environment-data:server:scheduled-task:upsert",
1305 tag: WS_METHODS.scheduledTasksUpsert,
1306 scheduler: configScheduler,
1307 concurrency: configConcurrency,
1308 }),
1309 setScheduledTaskEnabled: createEnvironmentRpcCommand(runtime, {
1310 label: "environment-data:server:scheduled-task:set-enabled",
1311 tag: WS_METHODS.scheduledTasksSetEnabled,
1312 scheduler: configScheduler,
1313 concurrency: configConcurrency,
1314 }),
1315 deleteScheduledTask: createEnvironmentRpcCommand(runtime, {
1316 label: "environment-data:server:scheduled-task:delete",
1317 tag: WS_METHODS.scheduledTasksDelete,
1318 scheduler: configScheduler,
1319 concurrency: configConcurrency,
1320 }),
1321 // Deliberately not on the config lane: run-now blocks until the run is
1322 // dispatched, and a slow run must not stall settings/keybinding/provider
1323 // mutations (or other scheduled-task edits) queued behind it.
1324 runScheduledTaskNow: createEnvironmentRpcCommand(runtime, {
1325 label: "environment-data:server:scheduled-task:run-now",
1326 tag: WS_METHODS.scheduledTasksRunNow,
1327 }),
1328 rotateScheduledTaskWebhookToken: createEnvironmentRpcCommand(runtime, {
1329 label: "environment-data:server:scheduled-task:rotate-webhook-token",
1330 tag: WS_METHODS.scheduledTasksRotateWebhookToken,
1331 scheduler: configScheduler,
1332 concurrency: configConcurrency,
1333 }),
1334 // Off the config lane: answering a card must not queue behind settings
1335 // edits. One answer per card at a time.
1336 answerSecretRequest: createEnvironmentRpcCommand(runtime, {
1337 label: "environment-data:server:secrets:answer-request",
1338 tag: WS_METHODS.secretsAnswerRequest,
1339 concurrency: {
1340 mode: "singleFlight",
1341 key: ({ environmentId, input }) =>
1342 JSON.stringify([environmentId, input.threadId, input.turnItemId]),
1343 },
1344 }),
1345 refreshUsageRates: createEnvironmentRpcCommand(runtime, {
1346 label: "environment-data:server:refresh-usage-rates",
1347 tag: WS_METHODS.serverRefreshUsageRates,
1348 concurrency: {
1349 mode: "singleFlight",
1350 key: ({ environmentId }) => environmentId,
1351 },
1352 }),
1353 checkOtlpEndpoint: createEnvironmentRpcCommand(runtime, {
1354 label: "environment-data:server:check-otlp-endpoint",
1355 tag: WS_METHODS.serverCheckOtlpEndpoint,
1356 }),
1357 retryResourceTelemetry: createEnvironmentRpcCommand(runtime, {
1358 label: "environment-data:server:retry-resource-telemetry",
1359 tag: WS_METHODS.serverRetryResourceTelemetry,
1360 concurrency: {
1361 mode: "singleFlight",
1362 key: ({ environmentId }) => environmentId,
1363 },
1364 }),
1365 };
1366}