1import * as UiSettings from "./uiSettings.ts";
2import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
3import { OrchestrationDispatchCommandError } from "@t3tools/contracts";
4import * as Crypto from "effect/Crypto";
5import * as Orchestrator from "./orchestration-v2/Orchestrator.ts";
6
7import * as DateTime from "effect/DateTime";
8import * as Duration from "effect/Duration";
9import * as Base64 from "effect/encoding/Base64";
10import * as Effect from "effect/Effect";
11import * as FileSystem from "effect/FileSystem";
12import * as Layer from "effect/Layer";
13import * as Option from "effect/Option";
14import * as Path from "effect/Path";
15import * as Queue from "effect/Queue";
16import * as Ref from "effect/Ref";
17import * as Schema from "effect/Schema";
18import * as Result from "effect/Result";
19import * as Stream from "effect/Stream";
20import { rpcInitialItems } from "./rpcInitialItems.ts";
21import { subscribeChatGptHandoff } from "./provider/CodexChatGptHandoff.ts";
22import { subscribeCodexAuthCallback } from "./provider/CodexAuthCallback.ts";
23import {
24 DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL,
25 AcpRegistryOperationError,
26 CommandId,
27 authScopeResponse,
28 AuthAccessStreamError,
29 type AuthAccessStreamEvent,
30 AuthOrchestrationOperateScope,
31 type AuthEnvironmentScope,
32 type ScheduledTaskListResult,
33 AuthSessionId,
34 ClientSurface,
35 type DiscoveredLocalServerList,
36 type EditorId,
37 type FileManagerRevealKind,
38 type OrchestrationClientOrigin,
39 type GitActionProgressEvent,
40 type GitManagerServiceError,
41 type MessageId,
42 type AcpRegistryImportSessionInput,
43 type AcpRegistryDeleteSessionInput,
44 type AcpRegistryDisableProviderInput,
45 type AcpRegistryListProvidersInput,
46 type AcpRegistryListSessionsInput,
47 type AcpRegistrySetProviderInput,
48 OrchestrationGetFullThreadDiffError,
49 OrchestrationSearchThreadsError,
50 OrchestrationV2SearchThreadError,
51 OrchestrationGetTurnDiffError,
52 ORCHESTRATION_V2_WS_METHODS,
53 ORCHESTRATION_PROTOCOL_QUERY_PARAM,
54 ORCHESTRATION_PROTOCOL_VERSION,
55 OrchestrationV2DispatchCommandError,
56 OrchestrationV2GetShellSnapshotError,
57 OrchestrationV2GetThreadProjectionError,
58 OrchestrationV2ThreadLaunchError,
59 type OrchestrationProjectShell,
60 type OrchestrationV2ShellSnapshot,
61 type ProjectEntriesFailure,
62 type ProjectFileFailure,
63 type ProjectFileOperation,
64 type ProjectMutation,
65 ProjectListEntriesError,
66 ProjectReadFileError,
67 ProjectSearchContentsError,
68 ProjectSearchEntriesError,
69 ProjectWriteFileError,
70 ProjectMutationError,
71 ProviderUploadFeedbackError,
72 ProviderSetupError,
73 RelayClientInstallFailedError,
74 type RelayClientInstallProgressEvent,
75 type ServerSelfUpdateError,
76 type ServerSelfUpdateProgressEvent,
77 type ServerConfig as ClientServerConfig,
78 type ServerConfigStreamEvent,
79 type ServerLifecycleStreamEvent,
80 type FilesystemBrowseFailure,
81 FilesystemBrowseError,
82 AssetWorkspaceContextNotFoundError,
83 AssetWorkspaceContextResolutionError,
84 ChatAttachmentId,
85 PersistChatAttachmentsError,
86 RpcClientId,
87 EnvironmentAuthorizationError,
88 type ProjectId,
89 type ProviderDriverKind,
90 type ProviderInstanceId,
91 ThreadId,
92 type TerminalMetadataStreamEvent,
93 type PullRequestRef,
94 WS_METHODS,
95 WsRpcGroup,
96} from "@t3tools/contracts";
97import { resolveServerBackgroundActivitySettings } from "@t3tools/shared/backgroundActivitySettings";
98import {
99 HttpRouter,
100 HttpServerRequest,
101 HttpServerRespondable,
102 HttpServerResponse,
103} from "effect/http";
104import { RpcSerialization, RpcServer } from "effect/rpc";
105
106import * as CheckpointDiffQuery from "./checkpointing/CheckpointDiffQuery.ts";
107import * as ServerConfig from "./config.ts";
108import * as EnvironmentTheme from "./environmentTheme.ts";
109import * as Keybindings from "./keybindings.ts";
110import * as ExternalLauncher from "./process/externalLauncher.ts";
111import * as ThreadManagementService from "./orchestration-v2/ThreadManagementService.ts";
112import * as McpAppRequests from "./mcpApps/McpAppRequests.ts";
113import * as ProviderSessionManager from "./orchestration-v2/ProviderSessionManager.ts";
114import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts";
115import * as ThreadMessageIntake from "./orchestration-v2/ThreadMessageIntake.ts";
116import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
117import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts";
118import * as SecretRequests from "./secrets/SecretRequests.ts";
119import {
120 archivedShellStreamItemFromThreadShell,
121 buildActiveShellSnapshot,
122 coalesceShellApplicationEvents,
123 coalesceStoredThreadEvents,
124 composeShellStreamWithEnrichment,
125 dedupeShellEnrichment,
126 loadShellSnapshotParts,
127 shellStreamItemFromEnrichmentRefresh,
128 shellStreamItemFromThreadShell,
129 shellStreamItemsFromInitialSnapshot,
130 shellStreamItemsFromResumeSnapshot,
131 skipUnchangedThreadShells,
132 toShellApplicationEvent,
133 type ShellApplicationEvent,
134} from "./orchestration-v2/ShellStream.ts";
135import { ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION } from "./orchestration-v2/ProjectionStore.ts";
136import { bufferLiveStream } from "./orchestration-v2/LiveStreamBudget.ts";
137import { coalesceThreadLiveStream } from "./orchestration-v2/ThreadLiveEventCoalescer.ts";
138import {
139 buildBoundedThreadStreamSnapshot,
140 decideThreadResume,
141 isThreadReplayRawPayloadSafe,
142 threadReplayEncodedBytes,
143 THREAD_RESUME_MAX_REPLAY_EVENTS,
144} from "./orchestration-v2/ThreadStream.ts";
145import {
146 buildBoundedThreadProjection,
147 THREAD_HISTORY_PAGE_POLICY,
148 THREAD_HISTORY_SNAPSHOT_ROW_LIMIT,
149} from "./orchestration-v2/threadHistoryPaging.ts";
150import {
151 projectDomainEventForWire,
152 projectThreadProjectionForWire,
153} from "./orchestration-v2/WireProjection.ts";
154import * as ProjectStore from "./orchestration-v2/ProjectStore.ts";
155import * as ThreadSearch from "./orchestration-v2/ThreadSearch.ts";
156import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts";
157import { userFacingDispatchErrorMessage } from "./orchestration-v2/UserFacingErrors.ts";
158import * as ProviderRegistry from "./provider/ProviderRegistry.ts";
159import * as ProviderInstanceRegistry from "./provider/ProviderInstanceRegistry.ts";
160import * as AcpRegistrySupport from "@t3tools/provider-acp-registry/server/AcpRegistrySupport";
161import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator";
162import * as ModelManifest from "./provider/ModelManifest.ts";
163import * as ProviderLatestVersions from "@t3tools/provider-core/server/ProviderLatestVersions";
164import * as ProviderMaintenanceRunner from "./provider/providerMaintenanceRunner.ts";
165import * as ProviderAuthService from "./provider/ProviderAuthService.ts";
166import { makeProviderInstallation } from "./provider/providerInstallation.ts";
167import * as ServerSelfUpdate from "./cloud/selfUpdate.ts";
168import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts";
169import * as ServerRuntimeStartup from "./serverRuntimeStartup.ts";
170import * as ServerSettings from "./serverSettings.ts";
171import * as StorageCleanup from "./storageCleanup.ts";
172import * as TerminalManager from "./terminal/Manager.ts";
173import { makeTerminalRpcHandlers } from "./terminal/Rpc.ts";
174import { withTerminalOutputWindow } from "./terminal/OutputProtocol.ts";
175import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts";
176import * as ServerBrowser from "./preview/ServerBrowser.ts";
177import * as DeviceService from "./device/DeviceService.ts";
178import { remoteSshDeviceHosts } from "./device/localSshDeviceHost.ts";
179import * as PreviewManager from "./preview/Manager.ts";
180import { issueAssetUrl } from "./assets/AssetAccess.ts";
181import {
182 attachmentRelativePath,
183 chatAttachmentStableKey,
184 createDeterministicAttachmentId,
185} from "./attachmentStore.ts";
186import { parseBase64DataUrl } from "./imageMime.ts";
187import { deletePendingAttachment, issueAttachmentUploadUrl } from "./assets/AttachmentUpload.ts";
188import * as PortScanner from "./preview/PortScanner.ts";
189import * as WorkspaceEntries from "./workspace/WorkspaceEntries.ts";
190import * as MergebotQueue from "./mergebot/MergebotQueue.ts";
191import * as JjStackService from "./vcs/JjStackService.ts";
192import * as DevServer from "./workspace/DevServer.ts";
193import * as ManagedWorkspaces from "./workspace/ManagedWorkspaces.ts";
194import * as WorkspaceFileSystem from "./workspace/WorkspaceFileSystem.ts";
195import { readWorkflowScript } from "./orchestration-v2/workflowScriptQuery.ts";
196import * as WorkspacePaths from "./workspace/WorkspacePaths.ts";
197import * as VcsStatusBroadcaster from "./vcs/VcsStatusBroadcaster.ts";
198import * as VcsProvisioningService from "./vcs/VcsProvisioningService.ts";
199import * as GitWorkflowService from "./git/GitWorkflowService.ts";
200import { refreshPushedPullRequests } from "./git/refreshPushedPullRequests.ts";
201import { linkCreatedPullRequest } from "./git/linkCreatedPullRequest.ts";
202import * as ReviewService from "./review/ReviewService.ts";
203import * as ProjectEnrichmentService from "./project/ProjectEnrichmentService.ts";
204import * as ProjectService from "./project/ProjectService.ts";
205import * as ManagedProjectFolders from "./project/ManagedProjectFolders.ts";
206import { projectMutationOperation } from "./project/ProjectMutation.ts";
207import * as ProjectCloneTracker from "./project/ProjectCloneTracker.ts";
208import * as RepositoryIdentityResolver from "./project/RepositoryIdentityResolver.ts";
209import * as WorktreeSetupTracker from "./project/WorktreeSetupTracker.ts";
210import * as ServerEnvironment from "./environment/ServerEnvironment.ts";
211import * as DirectEndpoints from "./environment/DirectEndpoints.ts";
212import * as RemoteOpenTargets from "./environment/RemoteOpenTargets.ts";
213import * as DefectReporter from "./observability/DefectReporter.ts";
214import * as BackgroundPolicy from "./background/BackgroundPolicy.ts";
215import * as EnvironmentAuth from "./auth/EnvironmentAuth.ts";
216import { requiredScopeForDeviceList, rpcAuthorizationError } from "./auth/RpcAuthorization.ts";
217import * as RpcAuthorization from "./auth/RpcAuthorization.ts";
218import { RpcInstrumentation, rpcInstrumentationLayer } from "./observability/RpcInstrumentation.ts";
219import * as ProcessDiagnostics from "./diagnostics/ProcessDiagnostics.ts";
220import * as ProcessResourceMonitor from "./diagnostics/ProcessResourceMonitor.ts";
221import * as ResourceTelemetry from "./resourceTelemetry/ResourceTelemetry.ts";
222import * as HostResources from "./resourceTelemetry/HostResources.ts";
223import * as UsageService from "./usage/UsageService.ts";
224import * as TraceDiagnostics from "./diagnostics/TraceDiagnostics.ts";
225import * as PullRequestService from "./pullRequest/PullRequestService.ts";
226import * as PullRequestReviewGuideService from "./pullRequest/PullRequestReviewGuideService.ts";
227import { listLinkedPullRequestThreads } from "./pullRequest/linkedThreads.ts";
228import { pullRequestSyncKey } from "./pullRequest/pullRequestSyncKey.ts";
229import * as SqlClient from "effect/sql/SqlClient";
230import * as PullRequestSyncReactor from "./orchestration-v2/PullRequestSyncReactor.ts";
231import * as SourceControlDiscovery from "./sourceControl/SourceControlDiscovery.ts";
232import * as OtlpEndpointCheck from "./observability/OtlpEndpointCheck.ts";
233import * as SourceControlRepositoryService from "./sourceControl/SourceControlRepositoryService.ts";
234import * as SourceControlBuiltInDrivers from "./sourceControl/builtInDrivers.ts";
235import * as SourceControlProviderRegistry from "./sourceControl/SourceControlProviderRegistry.ts";
236import * as GitVcsDriver from "./vcs/GitVcsDriver.ts";
237import * as VcsDriverRegistry from "./vcs/VcsDriverRegistry.ts";
238import * as VcsProjectConfig from "./vcs/VcsProjectConfig.ts";
239import * as PairingGrantStore from "./auth/PairingGrantStore.ts";
240import * as SessionStore from "./auth/SessionStore.ts";
241import { failEnvironmentAuthInvalid, failEnvironmentInternal } from "./auth/http.ts";
242import * as RelayClient from "@t3tools/shared/relayClient";
243import {
244 sameUsageLimitCommandCoverage,
245 withUsageLimitsCommands,
246} from "@t3tools/shared/usageLimits";
247import * as AgentSessionScanner from "./project/AgentSessionScanner.ts";
248import * as AgentSessionImporter from "./project/AgentSessionImporter.ts";
249import * as UsageLimitSources from "./usage/UsageLimitSources.ts";
250
251const CONFIG_DISCOVERY_TIMEOUT = Duration.seconds(5);
252const isProviderUploadFeedbackError = Schema.is(ProviderUploadFeedbackError);
253
254const resolveDiscoveryForConfig = <A, E, R>(
255 discovery: Effect.Effect<A, E, R>,
256 onTimeout: () => A,
257) =>
258 discovery.pipe(
259 Effect.timeoutOption(CONFIG_DISCOVERY_TIMEOUT),
260 Effect.map(Option.getOrElse(onTimeout)),
261 );
262
263export const resolveAvailableEditorsForConfig = <A, E, R>(
264 discovery: Effect.Effect<ReadonlyArray<A>, E, R>,
265) => resolveDiscoveryForConfig(discovery, () => []);
266
267const resolveFileManagerRevealKindForConfig = <E, R>(
268 discovery: Effect.Effect<FileManagerRevealKind | undefined, E, R>,
269) => resolveDiscoveryForConfig(discovery, () => undefined);
270
271type EditorDiscovery = Pick<
272 ExternalLauncher.ExternalLauncher["Service"],
273 "resolveAvailableEditors" | "resolveFileManagerRevealKind"
274>;
275
276// The config fields that follow from which editors are installed.
277const resolveEditorConfig = <E, R>(
278 availableEditors: ReadonlyArray<EditorId>,
279 revealKind: Effect.Effect<FileManagerRevealKind | undefined, E, R>,
280) =>
281 Effect.gen(function* () {
282 const fileManagerRevealKind = availableEditors.includes("file-manager")
283 ? yield* revealKind
284 : undefined;
285 return {
286 availableEditors,
287 ...(fileManagerRevealKind === undefined
288 ? {}
289 : {
290 shellRevealInFileManager: true,
291 shellRevealInFileManagerKind: fileManagerRevealKind,
292 }),
293 };
294 });
295
296/**
297 * Live config updates that follow a snapshot of `config`. A busy host can
298 * outlast the snapshot's discovery timeouts, which send no editors, or no
299 * reveal kind for the file manager. The scan keeps running, so once it lands
300 * this resends the config: clients replace theirs on any snapshot. The resent
301 * config is folded from the live updates already sent, so it cannot roll back
302 * a change that landed while the scan ran.
303 */
304export const withLateEditorConfig = <E, R>(
305 config: ClientServerConfig,
306 liveUpdates: Stream.Stream<ServerConfigStreamEvent, E, R>,
307 launcher: EditorDiscovery,
308) => {
309 const lateEditorConfig = Stream.fromEffect(launcher.resolveAvailableEditors()).pipe(
310 Stream.filter(
311 (editors) =>
312 editors.join() !== config.availableEditors.join() ||
313 (editors.includes("file-manager") && config.shellRevealInFileManagerKind === undefined),
314 ),
315 // Unbounded, unlike the snapshot: the reveal-kind probe is not shared, so
316 // a timeout here would cancel a probe that outlasts it every time.
317 Stream.mapEffect((editors) =>
318 resolveEditorConfig(editors, launcher.resolveFileManagerRevealKind()),
319 ),
320 Stream.filter(
321 (editorConfig) =>
322 editorConfig.availableEditors.join() !== config.availableEditors.join() ||
323 editorConfig.shellRevealInFileManagerKind !== config.shellRevealInFileManagerKind,
324 ),
325 Stream.map((editorConfig) => ({ type: "editorsResolved" as const, editorConfig })),
326 );
327
328 return Stream.merge(liveUpdates, lateEditorConfig).pipe(
329 Stream.mapAccum(
330 (): ClientServerConfig => config,
331 (current, event): readonly [ClientServerConfig, ReadonlyArray<ServerConfigStreamEvent>] => {
332 switch (event.type) {
333 case "editorsResolved": {
334 const {
335 availableEditors: _editors,
336 shellRevealInFileManager: _reveal,
337 shellRevealInFileManagerKind: _revealKind,
338 ...rest
339 } = current;
340 const next = { ...rest, ...event.editorConfig };
341 return [next, [{ version: 1, type: "snapshot", config: next }]];
342 }
343 case "keybindingsUpdated":
344 return [{ ...current, ...event.payload }, [event]];
345 case "providerStatuses":
346 return [{ ...current, providers: event.payload.providers }, [event]];
347 case "settingsUpdated":
348 return [{ ...current, settings: event.payload.settings }, [event]];
349 // Themes and usage-limit sources never ride in a snapshot; clients
350 // carry their projected values across one.
351 default:
352 return [current, [event]];
353 }
354 },
355 ),
356 );
357};
358
359function unexpectedCompatibilityError(error: never): never {
360 throw new Error(`Unhandled compatibility error: ${String(error)}`);
361}
362
363const persistChatAttachments = Effect.fn("ws.assets.persistChatAttachments")(function* (input: {
364 readonly threadId: ThreadId;
365 readonly messageId: MessageId;
366 readonly attachments: ReadonlyArray<{
367 readonly type: "image";
368 readonly name: string;
369 readonly mimeType: string;
370 readonly sizeBytes: number;
371 readonly dataUrl: string;
372 }>;
373}) {
374 const config = yield* ServerConfig.ServerConfig;
375 const fileSystem = yield* FileSystem.FileSystem;
376 const path = yield* Path.Path;
377 return yield* Effect.forEach(
378 input.attachments.map((attachment, index) => ({ attachment, index })),
379 Effect.fn("ws.assets.persistChatAttachment")(function* ({ attachment, index }) {
380 const parsed = parseBase64DataUrl(attachment.dataUrl);
381 if (parsed === null || parsed.mimeType !== attachment.mimeType.toLowerCase()) {
382 return yield* new PersistChatAttachmentsError({
383 message: `Attachment ${attachment.name} has an invalid image payload.`,
384 });
385 }
386 const bytes = yield* Effect.fromResult(Base64.decode(parsed.base64)).pipe(
387 Effect.mapError(
388 (cause) =>
389 new PersistChatAttachmentsError({
390 message: `Attachment ${attachment.name} is not valid base64.`,
391 cause,
392 }),
393 ),
394 );
395 if (bytes.byteLength !== attachment.sizeBytes) {
396 return yield* new PersistChatAttachmentsError({
397 message: `Attachment ${attachment.name} size does not match its payload.`,
398 });
399 }
400 const rawId = createDeterministicAttachmentId(
401 input.threadId,
402 chatAttachmentStableKey(input.messageId, index, bytes),
403 );
404 if (rawId === null) {
405 return yield* new PersistChatAttachmentsError({
406 message: "Could not allocate an attachment identifier.",
407 });
408 }
409 const persisted = {
410 type: "image" as const,
411 id: ChatAttachmentId.make(rawId),
412 name: attachment.name,
413 mimeType: attachment.mimeType,
414 sizeBytes: attachment.sizeBytes,
415 };
416 yield* fileSystem
417 .writeFile(path.join(config.attachmentsDir, attachmentRelativePath(persisted)!), bytes)
418 .pipe(
419 Effect.mapError(
420 (cause) =>
421 new PersistChatAttachmentsError({
422 message: `Could not persist attachment ${attachment.name}.`,
423 cause,
424 }),
425 ),
426 );
427 return persisted;
428 }),
429 { concurrency: 2 },
430 );
431});
432
433function projectEntriesFailureContext(error: WorkspaceEntries.WorkspaceEntriesError): {
434 readonly failure: ProjectEntriesFailure;
435 readonly normalizedCwd?: string;
436 readonly timeout?: string;
437 readonly detail?: string;
438} {
439 switch (error._tag) {
440 case "WorkspaceRootNotExistsError":
441 return {
442 failure: "workspace_root_not_found",
443 normalizedCwd: error.normalizedWorkspaceRoot,
444 };
445 case "WorkspaceRootCreateFailedError":
446 return {
447 failure: "workspace_root_create_failed",
448 normalizedCwd: error.normalizedWorkspaceRoot,
449 };
450 case "WorkspaceRootStatFailedError":
451 return {
452 failure: "workspace_root_stat_failed",
453 normalizedCwd: error.normalizedWorkspaceRoot,
454 detail: error.phase,
455 };
456 case "WorkspaceRootNotDirectoryError":
457 return {
458 failure: "workspace_root_not_directory",
459 normalizedCwd: error.normalizedWorkspaceRoot,
460 };
461 case "WorkspaceEntriesReadDirectoryError":
462 return {
463 failure: "directory_list_failed",
464 ...(error.cwd !== undefined ? { normalizedCwd: error.cwd } : {}),
465 detail: error.message,
466 };
467 case "WorkspaceSearchIndexCreateFailed":
468 return {
469 failure: "search_index_create_failed",
470 normalizedCwd: error.cwd,
471 detail: error.reason,
472 };
473 case "WorkspaceSearchIndexScanTimedOut":
474 return {
475 failure: "search_index_scan_timed_out",
476 normalizedCwd: error.cwd,
477 timeout: error.timeout,
478 };
479 case "WorkspaceSearchIndexSearchFailed":
480 return {
481 failure: "search_index_search_failed",
482 normalizedCwd: error.cwd,
483 detail: error.reason,
484 };
485 default:
486 return unexpectedCompatibilityError(error);
487 }
488}
489
490function filesystemBrowseFailureContext(error: WorkspaceEntries.WorkspaceEntriesBrowseError): {
491 readonly failure: FilesystemBrowseFailure;
492 readonly parentPath?: string;
493 readonly platform?: string;
494} {
495 switch (error._tag) {
496 case "WorkspaceEntriesWindowsPathUnsupportedError":
497 return { failure: "windows_path_unsupported", platform: error.platform };
498 case "WorkspaceEntriesCurrentProjectRequiredError":
499 return { failure: "current_project_required" };
500 case "WorkspaceEntriesReadDirectoryError":
501 return { failure: "read_directory_failed", parentPath: error.parentPath };
502 default:
503 return unexpectedCompatibilityError(error);
504 }
505}
506
507function projectFileFailureContext(
508 error:
509 | WorkspaceFileSystem.WorkspaceFileSystemError
510 | WorkspacePaths.WorkspacePathOutsideRootError,
511): {
512 readonly failure: ProjectFileFailure;
513 readonly resolvedPath?: string;
514 readonly resolvedWorkspaceRoot?: string;
515 readonly operation?: ProjectFileOperation;
516 readonly operationPath?: string;
517} {
518 switch (error._tag) {
519 case "WorkspacePathOutsideRootError":
520 return { failure: "workspace_path_outside_root" };
521 case "WorkspaceFileSystemOperationError":
522 return {
523 failure: "operation_failed",
524 resolvedPath: error.resolvedPath,
525 operation: error.operation,
526 operationPath: error.operationPath,
527 };
528 case "WorkspaceFilePathEscapeError":
529 return {
530 failure: "resolved_path_outside_root",
531 resolvedPath: error.resolvedPath,
532 resolvedWorkspaceRoot: error.resolvedWorkspaceRoot,
533 };
534 case "WorkspacePathNotFileError":
535 return { failure: "path_not_file", resolvedPath: error.resolvedPath };
536 case "WorkspaceBinaryFileError":
537 return { failure: "binary_file", resolvedPath: error.resolvedPath };
538 default:
539 return unexpectedCompatibilityError(error);
540 }
541}
542
543const PROVIDER_STATUS_DEBOUNCE_MS = 200;
544
545// Middleware added later wraps middleware added earlier, so instrumentation wraps authorization.
546const ServerWsRpcGroup = WsRpcGroup.middleware(RpcInstrumentation);
547// When a resuming client's cursor is more than this many events behind the
548// current head, skip the per-event catch-up replay and send a fresh shell
549// snapshot instead. Replaying each intervening event costs a shell refetch;
550// past this gap a single O(active-threads) snapshot is cheaper and bounded.
551// Matches the event store's default page size (DEFAULT_READ_FROM_SEQUENCE_LIMIT).
552const SHELL_RESUME_MAX_GAP = 1_000;
553// Row count alone does not bound replay memory: a few events with large tool
554// payloads can decode to gigabytes. Before replaying, sum the serialized
555// payload bytes of the range in SQL and reset with a snapshot past this budget.
556const ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES = 8 * 1024 * 1024;
557
558function toAuthAccessStreamEvent(
559 change: PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange,
560 revision: number,
561 currentSessionId: AuthSessionId,
562): AuthAccessStreamEvent {
563 switch (change.type) {
564 case "pairingLinkUpserted":
565 return {
566 version: 1,
567 revision,
568 type: "pairingLinkUpserted",
569 payload: { ...change.pairingLink, ...authScopeResponse(change.pairingLink.scopes) },
570 };
571 case "pairingLinkRemoved":
572 return {
573 version: 1,
574 revision,
575 type: "pairingLinkRemoved",
576 payload: { id: change.id },
577 };
578 case "clientUpserted":
579 return {
580 version: 1,
581 revision,
582 type: "clientUpserted",
583 payload: {
584 ...change.clientSession,
585 ...authScopeResponse(change.clientSession.scopes),
586 current: change.clientSession.sessionId === currentSessionId,
587 },
588 };
589 case "clientRemoved":
590 return {
591 version: 1,
592 revision,
593 type: "clientRemoved",
594 payload: { sessionId: change.sessionId },
595 };
596 }
597}
598
599const isClientSurface = Schema.is(ClientSurface);
600const MAX_CLIENT_APP_VERSION_LENGTH = 64;
601
602export function hasCompatibleOrchestrationProtocol(url: URL): boolean {
603 return (
604 url.searchParams.get(ORCHESTRATION_PROTOCOL_QUERY_PARAM) ===
605 String(ORCHESTRATION_PROTOCOL_VERSION)
606 );
607}
608
609export function shouldUseBoundedThreadSnapshot(input: {
610 readonly acceptBoundedSnapshot?: boolean;
611}): boolean {
612 return input.acceptBoundedSnapshot === true;
613}
614
615// Optional client identity announced on the /ws upgrade URL next to wsTicket.
616// Lenient by design: absent or malformed values degrade to {} so a connection
617// never fails over attribution metadata.
618function readClientConnectionOrigin(
619 request: HttpServerRequest.HttpServerRequest,
620): OrchestrationClientOrigin {
621 const url = HttpServerRequest.toURL(request);
622 if (Option.isNone(url)) {
623 return {};
624 }
625 const surface = url.value.searchParams.get("clientSurface");
626 const appVersion = url.value.searchParams.get("clientAppVersion")?.trim() ?? "";
627 return {
628 ...(isClientSurface(surface) ? { surface } : {}),
629 ...(appVersion !== "" && appVersion.length <= MAX_CLIENT_APP_VERSION_LENGTH
630 ? { appVersion }
631 : {}),
632 };
633}
634
635const canReplayPersistedRange = Effect.fnUntraced(function* (
636 afterSequence: number,
637 headSequence: number,
638 maxGap: number,
639) {
640 const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
641
642 const replayGap = headSequence - afterSequence;
643 if (replayGap < 0 || replayGap > maxGap) {
644 return false;
645 }
646 const stats = yield* applicationEvents.getReplayStats({
647 afterSequence,
648 throughSequence: headSequence,
649 });
650 if (stats.rawPayloadBytes > ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES) {
651 yield* Effect.logDebug("orchestration replay replaced by snapshot", {
652 afterSequence,
653 headSequence,
654 replayGap,
655 eventCount: stats.eventCount,
656 payloadBytes: stats.rawPayloadBytes,
657 payloadBudgetBytes: ORCHESTRATION_REPLAY_PAYLOAD_BUDGET_BYTES,
658 });
659 return false;
660 }
661 return true;
662});
663
664const enrichProjectShells = Effect.fn("ws.orchestrationV2.enrichProjectShells")(
665 (projects: ReadonlyArray<OrchestrationProjectShell>) =>
666 Effect.flatMap(ProjectEnrichmentService.ProjectEnrichmentService, (projectEnrichment) =>
667 Effect.forEach(
668 projects,
669 (project) =>
670 // Non-blocking: emit with cached identity (or null) and schedule
671 // background resolution. subscribeChanges is attached before
672 // loadSnapshot, so later identity completions push refreshed
673 // shells for multi-env grouping without blocking the initial
674 // snapshot or completion marker on slow git probes.
675 projectEnrichment.getAvailable(project.workspaceRoot).pipe(
676 Effect.map((enrichment) => ({
677 project: {
678 ...project,
679 repositoryIdentity: enrichment.repositoryIdentity,
680 },
681 repositoryIdentityResolved: enrichment.repositoryIdentityResolved,
682 })),
683 ),
684 { concurrency: 16 },
685 ).pipe(
686 Effect.map((enriched) => ({
687 projects: enriched.map((entry) => entry.project),
688 resolvedRepositoryIdentityRoots: enriched
689 .filter((entry) => entry.repositoryIdentityResolved)
690 .map((entry) => entry.project.workspaceRoot),
691 })),
692 ),
693 ),
694);
695
696export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subscribeThread")(
697 function* (input: {
698 readonly threadId: ThreadId;
699 readonly afterSequence?: number;
700 readonly requestCompletionMarker?: boolean;
701 readonly acceptBoundedSnapshot?: boolean;
702 readonly acceptCompactTurnItems?: boolean;
703 }) {
704 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
705 const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
706
707 yield* Effect.annotateCurrentSpan({
708 "orchestration_v2.thread_id": input.threadId,
709 });
710 yield* threadManagement.ensureLegacyTranscript(input.threadId).pipe(
711 Effect.mapError(
712 (cause) =>
713 new OrchestrationV2GetThreadProjectionError({
714 threadId: input.threadId,
715 message: `Failed to hydrate migrated thread ${input.threadId}`,
716 cause,
717 }),
718 ),
719 );
720
721 const eventStreamFrom = (afterSequence: number) =>
722 threadManagement
723 .streamStoredEventsFrom({
724 threadId: input.threadId,
725 afterSequence,
726 })
727 .pipe(
728 Stream.map((stored) => ({
729 kind: "event" as const,
730 sequence: stored.sequence,
731 event: projectDomainEventForWire(stored.event),
732 })),
733 coalesceThreadLiveStream,
734 Stream.mapError(
735 (cause) =>
736 new OrchestrationV2GetThreadProjectionError({
737 threadId: input.threadId,
738 message: `Failed while streaming orchestration V2 thread ${input.threadId}`,
739 cause,
740 }),
741 ),
742 );
743
744 const loadReplayThrough = (afterSequence: number, throughSequence: number) =>
745 applicationEvents
746 .readAgentEvents({
747 threadId: input.threadId,
748 afterSequence,
749 throughSequence,
750 limit: THREAD_RESUME_MAX_REPLAY_EVENTS + 1,
751 // A row a newer build wrote (e.g. after a downgrade) must not fail
752 // every resubscribe. The live tail resumes from `highWater`, so a
753 // dropped row never hides later events.
754 skipUnknownEventTypes: true,
755 })
756 .pipe(
757 Stream.map((stored) => ({
758 kind: "event" as const,
759 sequence: stored.sequence,
760 event: projectDomainEventForWire(stored.event),
761 })),
762 Stream.runCollect,
763 Effect.map((items) => Array.from(items)),
764 Effect.mapError(
765 (cause) =>
766 new OrchestrationV2GetThreadProjectionError({
767 threadId: input.threadId,
768 message: `Failed while replaying orchestration V2 thread ${input.threadId}`,
769 cause,
770 }),
771 ),
772 );
773
774 const completionMarker =
775 input.requestCompletionMarker === true
776 ? Stream.make({ kind: "synchronized" as const })
777 : Stream.empty;
778
779 const snapshotThenLive = Effect.fn("ws.orchestrationV2.threadSnapshotThenLive")(function* () {
780 const useBoundedSnapshot = shouldUseBoundedThreadSnapshot(input);
781 const snapshot = yield* (
782 useBoundedSnapshot
783 ? threadManagement.getThreadSnapshotWindow(input.threadId, {
784 rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT,
785 userTurnLimit: THREAD_HISTORY_PAGE_POLICY.maxUserTurns,
786 })
787 : threadManagement.getThreadSnapshot(input.threadId)
788 ).pipe(
789 Effect.mapError(
790 (cause) =>
791 new OrchestrationV2GetThreadProjectionError({
792 threadId: input.threadId,
793 message: `Failed to load orchestration V2 thread ${input.threadId}`,
794 cause,
795 }),
796 ),
797 );
798 const { snapshotSequence } = snapshot;
799 const snapshotItem = useBoundedSnapshot
800 ? buildBoundedThreadStreamSnapshot({
801 ...snapshot,
802 compactTurnItems: input.acceptCompactTurnItems === true,
803 })
804 : {
805 kind: "snapshot" as const,
806 snapshotSequence,
807 projection: projectThreadProjectionForWire(snapshot.projection),
808 };
809 return Stream.concat(
810 Stream.concat(rpcInitialItems([snapshotItem]), completionMarker),
811 eventStreamFrom(snapshotSequence),
812 );
813 });
814
815 // When the client already holds the projection (cached, or loaded over
816 // HTTP) it passes that snapshot's sequence, and we resume by replaying
817 // persisted events after it instead of re-sending the (potentially
818 // multi-KB) snapshot frame over the socket. The event sink subscribes
819 // to live events before reading the persisted tail, so no event
820 // published during the replay window is lost; overlapping events are
821 // deduped by sequence on the client.
822 if (input.afterSequence !== undefined) {
823 const highWater = yield* applicationEvents.latestAgentSequence(input.threadId).pipe(
824 Effect.mapError(
825 (cause) =>
826 new OrchestrationV2GetThreadProjectionError({
827 threadId: input.threadId,
828 message: `Failed to prepare orchestration V2 thread ${input.threadId} replay`,
829 cause,
830 }),
831 ),
832 );
833 if (input.afterSequence > highWater) {
834 return yield* snapshotThenLive();
835 }
836 const stats = yield* applicationEvents
837 .getAgentReplayStats({
838 threadId: input.threadId,
839 afterSequence: input.afterSequence,
840 throughSequence: highWater,
841 maxEvents: THREAD_RESUME_MAX_REPLAY_EVENTS,
842 })
843 .pipe(
844 Effect.mapError(
845 (cause) =>
846 new OrchestrationV2GetThreadProjectionError({
847 threadId: input.threadId,
848 message: `Failed to measure orchestration V2 thread ${input.threadId} replay`,
849 cause,
850 }),
851 ),
852 );
853 // Bound stored JSON before decoding, then check projected event
854 // size separately. Neither byte count is a bound on process memory.
855 if (
856 stats.eventCount > THREAD_RESUME_MAX_REPLAY_EVENTS ||
857 !isThreadReplayRawPayloadSafe(stats.rawPayloadBytes)
858 ) {
859 return yield* snapshotThenLive();
860 }
861 if (stats.hasCreateEvent) {
862 const shell = yield* threadManagement.getThreadShell(input.threadId).pipe(
863 Effect.mapError(
864 (cause) =>
865 new OrchestrationV2GetThreadProjectionError({
866 threadId: input.threadId,
867 message: `Failed to locate recreated orchestration V2 thread ${input.threadId}`,
868 cause,
869 }),
870 ),
871 );
872 // A retained creation can belong to a thread already deleted.
873 // Only replace its bounded replay when a snapshot can exist.
874 if (shell !== null) return yield* snapshotThenLive();
875 }
876 const replay = yield* loadReplayThrough(input.afterSequence, highWater);
877 const plan = decideThreadResume({
878 afterSequence: input.afterSequence,
879 highWater,
880 replayEventCount: replay.length,
881 replayEncodedBytes: threadReplayEncodedBytes(replay),
882 });
883 if (plan.mode === "snapshot") {
884 return yield* snapshotThenLive();
885 }
886 return Stream.concat(
887 Stream.concat(rpcInitialItems(replay), completionMarker),
888 eventStreamFrom(highWater),
889 );
890 }
891
892 return yield* snapshotThenLive();
893 },
894);
895
896export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subscribeShell")(
897 function* (input: {
898 readonly afterSequence?: number;
899 readonly requestCompletionMarker?: boolean;
900 }) {
901 const sql = yield* SqlClient.SqlClient;
902 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
903 const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
904 const projects = yield* ProjectStore.ProjectStoreV2;
905 const projectService = yield* ProjectService.ProjectService;
906 const projectEnrichment = yield* ProjectEnrichmentService.ProjectEnrichmentService;
907
908 const enrichmentChanges = yield* projectEnrichment.subscribeChanges;
909 const loadProjectMetadataSnapshot = Effect.fn("ws.orchestrationV2.loadProjectMetadataSnapshot")(
910 function* (snapshotSequence: number) {
911 const enriched = yield* enrichProjectShells(yield* projects.listShells());
912 return {
913 snapshot: {
914 schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,
915 snapshotSequence,
916 projects: enriched.projects,
917 threads: [],
918 archivedThreads: [],
919 } as OrchestrationV2ShellSnapshot,
920 resolvedRepositoryIdentityRoots: enriched.resolvedRepositoryIdentityRoots,
921 };
922 },
923 );
924 const loadSnapshot = Effect.fn("ws.orchestrationV2.loadShellSnapshot")(function* () {
925 const base = buildActiveShellSnapshot(
926 yield* loadShellSnapshotParts({
927 sql,
928 readThreads: threadManagement.readShellSnapshot({ location: "active" }),
929 listProjects: projects.listShells(),
930 latestSequence: applicationEvents.latestApplicationSequence,
931 }),
932 );
933 const enriched = yield* enrichProjectShells(base.projects);
934 return {
935 snapshot: { ...base, projects: enriched.projects } as OrchestrationV2ShellSnapshot,
936 resolvedRepositoryIdentityRoots: enriched.resolvedRepositoryIdentityRoots,
937 };
938 });
939 const projectItem = Effect.fn("ws.orchestrationV2.projectShellItem")(function* (
940 stored: Extract<ShellApplicationEvent, { readonly aggregateKind: "project" }>,
941 ) {
942 if (stored.type === "project.deleted") {
943 return {
944 kind: "project.removed" as const,
945 sequence: stored.sequence,
946 projectId: stored.aggregateId,
947 };
948 }
949 const project = yield* projectService.getShell(stored.aggregateId);
950 return Option.match(project, {
951 onNone: () => ({
952 kind: "project.removed" as const,
953 sequence: stored.sequence,
954 projectId: stored.aggregateId,
955 }),
956 onSome: (value) => ({
957 kind: "project.updated" as const,
958 sequence: stored.sequence,
959 project: value,
960 }),
961 });
962 });
963
964 // Coalescing makes each per-thread shell read represent every event
965 // for that thread in the current window; reading only the affected
966 // threads keeps the cost of a busy stream independent of how many
967 // threads exist overall.
968 const projectShellItems = Effect.fn("ws.orchestrationV2.projectShellItems")(function* (
969 events: ReadonlyArray<ShellApplicationEvent>,
970 ) {
971 return yield* Effect.forEach(
972 coalesceShellApplicationEvents(events),
973 (stored) =>
974 Effect.gen(function* () {
975 if ("aggregateKind" in stored) {
976 return yield* projectItem(stored);
977 }
978 const shell = yield* threadManagement.getThreadShell(stored.event.threadId);
979 return shellStreamItemFromThreadShell({ stored, shell });
980 }),
981 { concurrency: 8 },
982 );
983 });
984
985 const toShellStream = <E, R>(stream: Stream.Stream<ShellApplicationEvent, E, R>) =>
986 stream.pipe(
987 Stream.groupedWithin(512, Duration.millis(50)),
988 Stream.mapEffect((events) => projectShellItems(Array.from(events))),
989 Stream.flatMap(Stream.fromIterable),
990 skipUnchangedThreadShells,
991 );
992
993 const liveFrom = (afterSequence: number) =>
994 bufferLiveStream(
995 toShellStream(
996 applicationEvents.streamProjectedApplicationEvents({
997 afterSequence,
998 project: toShellApplicationEvent,
999 }),
1000 ),
1001 );
1002
1003 const enrichmentRefreshes = Stream.fromSubscription(enrichmentChanges).pipe(
1004 Stream.filter((change) => change.repositoryIdentityResolved),
1005 Stream.groupedWithin(64, Duration.millis(25)),
1006 // Build the refresh from the identities the changes carry. Re-enriching
1007 // every project here re-requested each expired root, whose resolution
1008 // published again, so one expiry kept every subscriber reloading every
1009 // project's metadata once a minute.
1010 Stream.mapEffect((changes) =>
1011 Effect.gen(function* () {
1012 const identities = new Map(
1013 Array.from(changes, (change) => [
1014 change.workspaceRoot,
1015 change.enrichment.repositoryIdentity,
1016 ]),
1017 );
1018 const snapshotSequence = yield* applicationEvents.latestApplicationSequence;
1019 const changedProjects = (yield* projects.listShells()).flatMap((project) =>
1020 identities.has(project.workspaceRoot)
1021 ? [{ ...project, repositoryIdentity: identities.get(project.workspaceRoot) ?? null }]
1022 : [],
1023 );
1024 return shellStreamItemFromEnrichmentRefresh({
1025 snapshot: {
1026 schemaVersion: ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION,
1027 snapshotSequence,
1028 projects: changedProjects,
1029 threads: [],
1030 archivedThreads: [],
1031 } as OrchestrationV2ShellSnapshot,
1032 changes: Array.from(changes),
1033 });
1034 }),
1035 ),
1036 );
1037
1038 // Always attach the enrichment subscription before the first load so
1039 // completions that race HTTP snapshot fetch still push a refresh.
1040 // When the client already holds a shell snapshot (cached, or loaded
1041 // over HTTP) it passes that snapshot's sequence. We still emit one
1042 // compact metadata refresh up front: getAvailable may have been cold on the
1043 // HTTP path (null identity), and enrichment PubSub events published
1044 // before this subscribe attached are dropped. Rehydrating here fills
1045 // repositoryIdentity for cross-environment project grouping even on
1046 // afterSequence resumes. Application events after the sequence still
1047 // stream as deltas; overlapping events are deduped by sequence on the
1048 // client.
1049 //
1050 // After the unmarked authoritative frame, emit a same-sequence
1051 // metadata-only frame for roots that already resolved successfully
1052 // (including cached null). Cold/failed roots stay unmarked and use
1053 // the PubSub enrichment path when they complete later.
1054 const completionMarker =
1055 input.requestCompletionMarker === true
1056 ? Stream.make({ kind: "synchronized" as const })
1057 : Stream.empty;
1058 const initialSnapshotItems = (loaded: {
1059 readonly snapshot: OrchestrationV2ShellSnapshot;
1060 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
1061 }) =>
1062 rpcInitialItems(
1063 shellStreamItemsFromInitialSnapshot({
1064 snapshot: loaded.snapshot,
1065 resolvedRepositoryIdentityRoots: loaded.resolvedRepositoryIdentityRoots,
1066 }),
1067 );
1068 const initialEnrichmentItems = (loaded: {
1069 readonly snapshot: OrchestrationV2ShellSnapshot;
1070 readonly resolvedRepositoryIdentityRoots: ReadonlyArray<string>;
1071 }) =>
1072 rpcInitialItems(
1073 shellStreamItemsFromResumeSnapshot({
1074 snapshot: loaded.snapshot,
1075 resolvedRepositoryIdentityRoots: loaded.resolvedRepositoryIdentityRoots,
1076 }),
1077 );
1078 // Initial unmarked (+ optional same-load marked) always drains first.
1079 // Enrichment merges only with the post-prefix tail so a ready marked
1080 // refresh cannot interleave before the authoritative initial frame.
1081 const completionThenLive = (afterSequence: number) =>
1082 Stream.concat(completionMarker, liveFrom(afterSequence));
1083
1084 const stream = yield* Effect.gen(function* () {
1085 if (input.afterSequence === undefined) {
1086 const loaded = yield* loadSnapshot();
1087 return composeShellStreamWithEnrichment({
1088 initial: initialSnapshotItems(loaded),
1089 tail: completionThenLive(loaded.snapshot.snapshotSequence),
1090 enrichment: enrichmentRefreshes,
1091 });
1092 }
1093
1094 const highWater = yield* applicationEvents.latestApplicationSequence;
1095 if (!(yield* canReplayPersistedRange(input.afterSequence, highWater, SHELL_RESUME_MAX_GAP))) {
1096 const loaded = yield* loadSnapshot();
1097 return composeShellStreamWithEnrichment({
1098 initial: initialSnapshotItems(loaded),
1099 tail: completionThenLive(loaded.snapshot.snapshotSequence),
1100 enrichment: enrichmentRefreshes,
1101 });
1102 }
1103
1104 const loaded = yield* loadProjectMetadataSnapshot(highWater);
1105 const replay = toShellStream(
1106 applicationEvents.readApplicationEvents({
1107 afterSequence: input.afterSequence,
1108 throughSequence: highWater,
1109 // A row a newer build wrote (e.g. after a downgrade) must not fail
1110 // every shell resubscribe. The live tail below resumes from
1111 // highWater, so a dropped row never hides later events.
1112 skipUnknownEventTypes: true,
1113 }),
1114 );
1115 return composeShellStreamWithEnrichment({
1116 initial: initialEnrichmentItems(loaded),
1117 tail: Stream.concat(Stream.concat(replay, completionMarker), liveFrom(highWater)),
1118 enrichment: enrichmentRefreshes,
1119 });
1120 }).pipe(
1121 Effect.mapError(
1122 (cause) =>
1123 new OrchestrationV2GetShellSnapshotError({
1124 message: "Failed to prepare the application shell stream",
1125 cause,
1126 }),
1127 ),
1128 );
1129
1130 return stream.pipe(
1131 dedupeShellEnrichment,
1132 Stream.mapError(
1133 (cause) =>
1134 new OrchestrationV2GetShellSnapshotError({
1135 message: "Failed while streaming the application shell",
1136 cause,
1137 }),
1138 ),
1139 );
1140 },
1141);
1142
1143const layerWsRpc = (
1144 currentSession: EnvironmentAuth.AuthenticatedSession,
1145 clientOrigin: OrchestrationClientOrigin,
1146 previewAutomationBroker: PreviewAutomationBroker.PreviewAutomationBroker["Service"],
1147 serverBrowser: ServerBrowser.ServerBrowser["Service"],
1148) =>
1149 ServerWsRpcGroup.toLayer(
1150 Effect.gen(function* () {
1151 const currentSessionId = currentSession.sessionId;
1152 const sql = yield* SqlClient.SqlClient;
1153 const threadManagement = yield* ThreadManagementService.ThreadManagementService;
1154 const intakeContext = yield* Effect.context<
1155 | ThreadManagementService.ThreadManagementService
1156 | ThreadLaunchService.ThreadLaunchService
1157 | FileSystem.FileSystem
1158 | ServerConfig.ServerConfig
1159 | ServerSettings.ServerSettingsService
1160 >();
1161 const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore;
1162 const projectStore = yield* ProjectStore.ProjectStoreV2;
1163 const projectService = yield* ProjectService.ProjectService;
1164 const managedFolders = yield* ManagedProjectFolders.ManagedProjectFolders;
1165 const threadSearch = yield* ThreadSearch.ThreadSearch;
1166
1167 const providerSessionsV2 = yield* ProviderSessionManager.ProviderSessionManagerV2;
1168 const mcpAppRequests = yield* McpAppRequests.McpAppRequests;
1169 const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService;
1170 const providerSessionManager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1171 const scheduledTasks = yield* ScheduledTasks.ScheduledTaskService;
1172 const secretRequests = yield* SecretRequests.SecretRequests;
1173 const mergebotQueue = yield* MergebotQueue.MergebotQueue;
1174 const scriptProgress = yield* ScriptProgress.ScriptProgress;
1175 const pullRequests = yield* PullRequestService.PullRequestService;
1176 const reviewGuides = yield* PullRequestReviewGuideService.PullRequestReviewGuideService;
1177 const pullRequestSync = yield* PullRequestSyncReactor.PullRequestSyncReactor;
1178 const deviceService = yield* DeviceService.DeviceService;
1179 const deviceHostContext =
1180 yield* Effect.context<Effect.Services<ReturnType<typeof remoteSshDeviceHosts>>>();
1181 const orchestrationEngine = yield* Orchestrator.OrchestratorV2;
1182 const crypto = yield* Crypto.Crypto;
1183 const serverCommandId = (tag: string) =>
1184 crypto.randomUUIDv4().pipe(
1185 Effect.orDie,
1186 Effect.map((id) => CommandId.make(`server:${tag}:${id}`)),
1187 );
1188 const resolvePullRequestSyncKey = (reference: PullRequestRef) =>
1189 reference.host !== undefined && reference.repository.includes("/")
1190 ? Effect.succeed(pullRequestSyncKey(reference))
1191 : projectService.getShell(reference.projectId).pipe(
1192 Effect.map((project) =>
1193 pullRequestSyncKey(reference, Option.getOrUndefined(project)?.repositoryIdentity),
1194 ),
1195 Effect.orElseSucceed(() => null),
1196 );
1197 const usage = yield* UsageService.UsageService;
1198 const usageLimitSources = yield* UsageLimitSources.UsageLimitSources;
1199 const worktreeSetupTracker = yield* WorktreeSetupTracker.WorktreeSetupTracker;
1200 const projectCloneTracker = yield* ProjectCloneTracker.ProjectCloneTracker;
1201 const repositoryIdentityResolver =
1202 yield* RepositoryIdentityResolver.RepositoryIdentityResolver;
1203 const agentSessionScanner = yield* AgentSessionScanner.AgentSessionScanner;
1204 const agentSessionImporter = yield* AgentSessionImporter.AgentSessionImporter;
1205 const checkpointDiffQuery = yield* CheckpointDiffQuery.CheckpointDiffQuery;
1206 const uiSettings = yield* UiSettings.UiSettings;
1207 const keybindings = yield* Keybindings.Keybindings;
1208 const environmentTheme = yield* EnvironmentTheme.EnvironmentThemeService;
1209 const externalLauncher = yield* ExternalLauncher.ExternalLauncher;
1210 const remoteOpenTargets = yield* RemoteOpenTargets.RemoteOpenTargets;
1211 const directEndpoints = yield* DirectEndpoints.DirectEndpoints;
1212 const gitWorkflow = yield* GitWorkflowService.GitWorkflowService;
1213 const review = yield* ReviewService.ReviewService;
1214 const vcsProvisioning = yield* VcsProvisioningService.VcsProvisioningService;
1215 const vcsStatusBroadcaster = yield* VcsStatusBroadcaster.VcsStatusBroadcaster;
1216 const terminalManager = yield* TerminalManager.TerminalManager;
1217 const previewManager = yield* PreviewManager.PreviewManager;
1218 const portDiscovery = yield* PortScanner.PortDiscovery;
1219 const providerRegistry = yield* ProviderRegistry.ProviderRegistry;
1220 const modelManifest = yield* ModelManifest.ModelManifest;
1221 const providerLatestVersions = yield* ProviderLatestVersions.ProviderLatestVersions;
1222 const providerInstances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry;
1223 const acpRegistryCatalog = yield* AcpRegistrySupport.AcpRegistryCatalog;
1224 const acpRegistryRuntimeCoordinator =
1225 yield* AcpRegistryRuntimeCoordinator.AcpRegistryRuntimeCoordinator;
1226 const providerMaintenanceRunner = yield* ProviderMaintenanceRunner.ProviderMaintenanceRunner;
1227 const providerAuth = yield* ProviderAuthService.ProviderAuthService;
1228 const providerInstallation = yield* makeProviderInstallation();
1229 const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate;
1230 const config = yield* ServerConfig.ServerConfig;
1231 const lifecycleEvents = yield* ServerLifecycleEvents.ServerLifecycleEvents;
1232 const storageCleanup = yield* StorageCleanup.StorageCleanup;
1233 const serverSettings = yield* ServerSettings.ServerSettingsService;
1234 const startup = yield* ServerRuntimeStartup.ServerRuntimeStartup;
1235 const workspaceEntries = yield* WorkspaceEntries.WorkspaceEntries;
1236 const devServer = yield* DevServer.DevServer;
1237 const jjStack = yield* JjStackService.JjStackService;
1238 const managedWorkspaces = yield* ManagedWorkspaces.ManagedWorkspaces;
1239 const workspaceFileSystem = yield* WorkspaceFileSystem.WorkspaceFileSystem;
1240 const serverEnvironment = yield* ServerEnvironment.ServerEnvironment;
1241 const backgroundPolicy = yield* BackgroundPolicy.BackgroundPolicy;
1242 const rpcClientIds = yield* Ref.make(new Set<RpcClientId>());
1243 yield* Effect.addFinalizer(() =>
1244 Ref.get(rpcClientIds).pipe(
1245 Effect.flatMap((clientIds) =>
1246 Effect.forEach(
1247 clientIds,
1248 (clientId) => backgroundPolicy.removeRpcClient(currentSessionId, clientId),
1249 {
1250 discard: true,
1251 },
1252 ),
1253 ),
1254 Effect.ignore,
1255 ),
1256 );
1257 const serverAuth = yield* EnvironmentAuth.EnvironmentAuth;
1258 const sourceControlDiscovery = yield* SourceControlDiscovery.SourceControlDiscovery;
1259 const otlpEndpointCheck = yield* OtlpEndpointCheck.OtlpEndpointCheck;
1260 const automaticGitFetchInterval = serverSettings.getSettings.pipe(
1261 Effect.map(
1262 (settings) => resolveServerBackgroundActivitySettings(settings).automaticGitFetchInterval,
1263 ),
1264 Effect.catch((cause) =>
1265 Effect.logWarning("Failed to read automatic Git fetch interval setting", {
1266 detail: cause.message,
1267 }).pipe(Effect.as(DEFAULT_AUTOMATIC_GIT_FETCH_INTERVAL)),
1268 ),
1269 );
1270 const sourceControlRepositories =
1271 yield* SourceControlRepositoryService.SourceControlRepositoryService;
1272 const withPullRequestViewer = pullRequests.withRoutingCredential;
1273 const bootstrapCredentials = yield* PairingGrantStore.PairingGrantStore;
1274 const sessions = yield* SessionStore.SessionStore;
1275 const processDiagnostics = yield* ProcessDiagnostics.ProcessDiagnostics;
1276 const hostResources = yield* HostResources.HostResources;
1277 const processResourceMonitor = yield* ProcessResourceMonitor.ProcessResourceMonitor;
1278 const resourceTelemetry = yield* ResourceTelemetry.ResourceTelemetry;
1279 const relayClient = yield* RelayClient.RelayClient;
1280 // A webhook URL starts agent runs, so only sessions that may operate
1281 // see it; read-only sessions still see the task itself.
1282 const withVisibleWebhookUrls = (result: ScheduledTaskListResult): ScheduledTaskListResult =>
1283 currentSession.scopes.includes(AuthOrchestrationOperateScope)
1284 ? result
1285 : {
1286 tasks: result.tasks.map(({ webhook: _webhook, ...task }) => task),
1287 };
1288 // RpcScopeAuthorization checks each RPC's declared scope before its handler
1289 // runs. This covers the one RPC whose scope depends on its input.
1290 const authorizeEffect = <A, E, R>(
1291 requiredScope: AuthEnvironmentScope,
1292 effect: Effect.Effect<A, E, R>,
1293 ): Effect.Effect<A, E | EnvironmentAuthorizationError, R> =>
1294 currentSession.scopes.includes(requiredScope)
1295 ? effect
1296 : Effect.fail(rpcAuthorizationError(requiredScope));
1297
1298 const acpRegistryProject = Effect.fn("ws.acpRegistry.project")(function* (
1299 projectId: ProjectId,
1300 ) {
1301 const project = yield* projectService.getById(projectId).pipe(
1302 Effect.mapError(
1303 (cause) =>
1304 new AcpRegistryOperationError({
1305 reason: "project_not_found",
1306 message: `Project ${projectId} is unavailable.`,
1307 cause,
1308 }),
1309 ),
1310 );
1311 return yield* Option.match(project, {
1312 onNone: () =>
1313 Effect.fail(
1314 new AcpRegistryOperationError({
1315 reason: "project_not_found",
1316 message: `Project ${projectId} was not found.`,
1317 }),
1318 ),
1319 onSome: Effect.succeed,
1320 });
1321 });
1322
1323 const acpSessionManager = Effect.fn("ws.acpRegistry.sessionManager")(function* (
1324 instanceId: ProviderInstanceId,
1325 ) {
1326 const instance = yield* providerInstances.getInstance(instanceId);
1327 if (instance === undefined) {
1328 return yield* new AcpRegistryOperationError({
1329 reason: "instance_not_found",
1330 message: `Provider instance ${instanceId} was not found.`,
1331 });
1332 }
1333 if (instance.acpSessionManagement === undefined) {
1334 return yield* new AcpRegistryOperationError({
1335 reason: "session_list_unsupported",
1336 message: `Provider instance ${instanceId} does not expose ACP session management.`,
1337 });
1338 }
1339 return { instance, manager: instance.acpSessionManagement };
1340 });
1341
1342 const importedAcpThreadId = (input: {
1343 readonly driver: ProviderDriverKind;
1344 readonly instanceId: ProviderInstanceId;
1345 readonly sessionId: string;
1346 }) =>
1347 IdAllocator.deriveThreadFromProviderThread({
1348 driver: input.driver,
1349 providerInstanceId: input.instanceId,
1350 nativeThreadId: input.sessionId,
1351 });
1352
1353 const listAcpRegistrySessions = Effect.fn("ws.acpRegistry.listSessions")(function* (
1354 input: AcpRegistryListSessionsInput,
1355 ) {
1356 const project = yield* acpRegistryProject(input.projectId);
1357 const { instance, manager } = yield* acpSessionManager(input.instanceId);
1358 const listed = yield* manager.listSessions({
1359 cwd: project.workspaceRoot,
1360 ...(input.cursor === undefined ? {} : { cursor: input.cursor }),
1361 });
1362 const sessions = yield* Effect.forEach(
1363 listed.sessions,
1364 (session) => {
1365 const threadId = importedAcpThreadId({
1366 driver: instance.driverKind,
1367 instanceId: input.instanceId,
1368 sessionId: session.sessionId,
1369 });
1370 return threadManagement.getThreadShell(threadId).pipe(
1371 Effect.map((thread) => ({
1372 ...session,
1373 importedThreadId: thread === null ? null : threadId,
1374 })),
1375 Effect.mapError(
1376 (cause) =>
1377 new AcpRegistryOperationError({
1378 reason: "session_import_failed",
1379 message: "Could not inspect existing imported ACP sessions.",
1380 cause,
1381 }),
1382 ),
1383 );
1384 },
1385 { concurrency: 16 },
1386 );
1387 return { ...listed, sessions };
1388 });
1389
1390 const importAcpRegistrySession = Effect.fn("ws.acpRegistry.importSession")(function* (
1391 input: AcpRegistryImportSessionInput,
1392 ) {
1393 return yield* acpRegistryRuntimeCoordinator.withSessionMutation(
1394 Effect.gen(function* () {
1395 yield* acpRegistryProject(input.projectId);
1396 const { instance } = yield* acpSessionManager(input.instanceId);
1397 const providerSnapshot = yield* instance.snapshot.getSnapshot;
1398 if (
1399 providerSnapshot.nativeSessions?.canLoad !== true &&
1400 providerSnapshot.nativeSessions?.canResume !== true
1401 ) {
1402 return yield* new AcpRegistryOperationError({
1403 reason: "session_resume_unsupported",
1404 message: "The ACP agent cannot load or resume native sessions.",
1405 });
1406 }
1407 const threadId = importedAcpThreadId({
1408 driver: instance.driverKind,
1409 instanceId: input.instanceId,
1410 sessionId: input.sessionId,
1411 });
1412 const existing = yield* threadManagement.getThreadShell(threadId).pipe(
1413 Effect.mapError(
1414 (cause) =>
1415 new AcpRegistryOperationError({
1416 reason: "session_import_failed",
1417 message: "Could not inspect the imported ACP session mapping.",
1418 cause,
1419 }),
1420 ),
1421 );
1422 if (existing !== null) return { threadId, imported: false } as const;
1423
1424 const provider = (yield* providerRegistry.getProviders).find(
1425 (candidate) => candidate.instanceId === input.instanceId,
1426 );
1427 const model =
1428 provider?.models.find((candidate) => candidate.isDefault)?.slug ??
1429 provider?.models[0]?.slug ??
1430 "default";
1431 const commandId = CommandId.make(yield* crypto.randomUUIDv4().pipe(Effect.orDie));
1432 const launched = yield* Effect.result(
1433 startup.enqueueCommand(
1434 threadLaunch.launch({
1435 commandId,
1436 threadId,
1437 projectId: input.projectId,
1438 title: input.title ?? "Imported ACP session",
1439 modelSelection: { instanceId: input.instanceId, model },
1440 runtimeMode: "approval-required",
1441 interactionMode: "default",
1442 workspaceStrategy: { type: "root" },
1443 importedNativeThread: {
1444 ref: {
1445 driver: instance.driverKind,
1446 nativeId: input.sessionId,
1447 strength: "strong",
1448 },
1449 metadata: {
1450 itemIdentityVersion: 2,
1451 ...(input.title === undefined ? {} : { title: input.title }),
1452 ...(input.updatedAt === undefined ? {} : { updatedAt: input.updatedAt }),
1453 },
1454 },
1455 createdBy: "user",
1456 creationSource: "web",
1457 }),
1458 ),
1459 );
1460 if (Result.isFailure(launched)) {
1461 const racedImport = yield* threadManagement.getThreadShell(threadId).pipe(
1462 Effect.mapError(
1463 (cause) =>
1464 new AcpRegistryOperationError({
1465 reason: "session_import_failed",
1466 message: "Could not inspect the imported ACP session after launch failed.",
1467 cause,
1468 }),
1469 ),
1470 );
1471 if (racedImport !== null) return { threadId, imported: false } as const;
1472 return yield* new AcpRegistryOperationError({
1473 reason: "session_import_failed",
1474 message: "Could not create a T3 thread for the ACP session.",
1475 cause: launched.failure,
1476 });
1477 }
1478 return { threadId, imported: true } as const;
1479 }),
1480 );
1481 });
1482
1483 const deleteAcpRegistrySession = Effect.fn("ws.acpRegistry.deleteSession")(function* (
1484 input: AcpRegistryDeleteSessionInput,
1485 ) {
1486 return yield* acpRegistryRuntimeCoordinator.withSessionMutation(
1487 Effect.gen(function* () {
1488 const project = yield* acpRegistryProject(input.projectId);
1489 const { instance, manager } = yield* acpSessionManager(input.instanceId);
1490 const snapshot = yield* instance.snapshot.getSnapshot;
1491 if (snapshot.nativeSessions?.canDelete !== true) {
1492 return yield* new AcpRegistryOperationError({
1493 reason: "session_delete_unsupported",
1494 message: "The ACP agent does not advertise session deletion.",
1495 });
1496 }
1497 const threadId = importedAcpThreadId({
1498 driver: instance.driverKind,
1499 instanceId: input.instanceId,
1500 sessionId: input.sessionId,
1501 });
1502 const importedThread = yield* threadManagement.getThreadShell(threadId).pipe(
1503 Effect.mapError(
1504 (cause) =>
1505 new AcpRegistryOperationError({
1506 reason: "session_delete_failed",
1507 message: "Could not inspect the imported ACP session mapping.",
1508 cause,
1509 }),
1510 ),
1511 );
1512 if (importedThread !== null) {
1513 return yield* new AcpRegistryOperationError({
1514 reason: "session_delete_failed",
1515 message: "Delete the imported T3 thread before deleting its native ACP session.",
1516 });
1517 }
1518 yield* manager.deleteSession({
1519 cwd: project.workspaceRoot,
1520 sessionId: input.sessionId,
1521 });
1522 return { deleted: true } as const;
1523 }),
1524 );
1525 });
1526
1527 const listAcpRegistryProviders = Effect.fn("ws.acpRegistry.listProviders")(function* (
1528 input: AcpRegistryListProvidersInput,
1529 ) {
1530 const project = yield* acpRegistryProject(input.projectId);
1531 const { instance, manager } = yield* acpSessionManager(input.instanceId);
1532 const snapshot = yield* instance.snapshot.getSnapshot;
1533 if (snapshot.configurableProviders !== true) {
1534 return yield* new AcpRegistryOperationError({
1535 reason: "providers_unsupported",
1536 message: "The ACP agent does not advertise provider configuration.",
1537 });
1538 }
1539 return yield* manager.listProviders(project.workspaceRoot);
1540 });
1541
1542 const setAcpRegistryProvider = Effect.fn("ws.acpRegistry.setProvider")(function* (
1543 input: AcpRegistrySetProviderInput,
1544 ) {
1545 const project = yield* acpRegistryProject(input.projectId);
1546 const { manager } = yield* acpSessionManager(input.instanceId);
1547 if (input.headers !== undefined && Object.keys(input.headers).length > 32) {
1548 return yield* new AcpRegistryOperationError({
1549 reason: "provider_configuration_failed",
1550 message: "ACP provider configuration accepts at most 32 headers.",
1551 });
1552 }
1553 const listed = yield* manager.listProviders(project.workspaceRoot);
1554 const provider = listed.providers.find(
1555 (candidate) => candidate.providerId === input.providerId,
1556 );
1557 if (provider === undefined || !provider.supported.includes(input.apiType)) {
1558 return yield* new AcpRegistryOperationError({
1559 reason: "provider_configuration_failed",
1560 message: `Provider ${input.providerId} does not support ${input.apiType}.`,
1561 });
1562 }
1563 yield* providerSessionManager.closeInstance(input.instanceId).pipe(
1564 Effect.mapError(
1565 (cause) =>
1566 new AcpRegistryOperationError({
1567 reason: "provider_configuration_failed",
1568 message: "Could not stop live sessions before updating the ACP provider.",
1569 cause,
1570 }),
1571 ),
1572 );
1573 yield* manager.setProvider({
1574 cwd: project.workspaceRoot,
1575 providerId: input.providerId,
1576 apiType: input.apiType,
1577 baseUrl: input.baseUrl,
1578 ...(input.headers === undefined ? {} : { headers: input.headers }),
1579 });
1580 yield* providerRegistry.refreshInstance(input.instanceId);
1581 return { configured: true } as const;
1582 });
1583
1584 const disableAcpRegistryProvider = Effect.fn("ws.acpRegistry.disableProvider")(function* (
1585 input: AcpRegistryDisableProviderInput,
1586 ) {
1587 const project = yield* acpRegistryProject(input.projectId);
1588 const { manager } = yield* acpSessionManager(input.instanceId);
1589 const listed = yield* manager.listProviders(project.workspaceRoot);
1590 const provider = listed.providers.find(
1591 (candidate) => candidate.providerId === input.providerId,
1592 );
1593 if (provider === undefined || provider.required) {
1594 return yield* new AcpRegistryOperationError({
1595 reason: "provider_configuration_failed",
1596 message:
1597 provider === undefined
1598 ? `Provider ${input.providerId} was not advertised by the ACP agent.`
1599 : `Provider ${input.providerId} is required and cannot be disabled.`,
1600 });
1601 }
1602 yield* providerSessionManager.closeInstance(input.instanceId).pipe(
1603 Effect.mapError(
1604 (cause) =>
1605 new AcpRegistryOperationError({
1606 reason: "provider_configuration_failed",
1607 message: "Could not stop live sessions before disabling the ACP provider.",
1608 cause,
1609 }),
1610 ),
1611 );
1612 yield* manager.disableProvider({
1613 cwd: project.workspaceRoot,
1614 providerId: input.providerId,
1615 });
1616 yield* providerRegistry.refreshInstance(input.instanceId);
1617 return { disabled: true } as const;
1618 });
1619 const loadAuthAccessSnapshot = () =>
1620 Effect.all({
1621 pairingLinks: serverAuth.listPairingLinks(),
1622 clientSessions: serverAuth.listClientSessions(currentSessionId),
1623 }).pipe(
1624 Effect.mapError(
1625 (error) =>
1626 new AuthAccessStreamError({
1627 message: error.message,
1628 }),
1629 ),
1630 );
1631
1632 const loadServerConfig = (options: { readonly usageLimitsCommand: boolean }) =>
1633 Effect.gen(function* () {
1634 const keybindingsConfig = yield* keybindings.loadConfigState;
1635 const currentProviders = yield* providerRegistry.getProviders;
1636 const providers = options.usageLimitsCommand
1637 ? withUsageLimitsCommands(currentProviders, yield* usageLimitSources.current)
1638 : currentProviders;
1639 const settings = ServerSettings.redactServerSettingsForClient(
1640 yield* serverSettings.getSettings,
1641 );
1642 const environment = yield* serverEnvironment.getDescriptor;
1643 const auth = yield* serverAuth.getDescriptor();
1644 const scratchWorkspaceRoot = yield* managedFolders.scratchRoot;
1645 const editorConfig = yield* resolveEditorConfig(
1646 yield* resolveAvailableEditorsForConfig(externalLauncher.resolveAvailableEditors()),
1647 resolveFileManagerRevealKindForConfig(externalLauncher.resolveFileManagerRevealKind()),
1648 );
1649
1650 return {
1651 environment,
1652 auth,
1653 cwd: config.cwd,
1654 keybindingsConfigPath: config.keybindingsConfigPath,
1655 keybindings: keybindingsConfig.keybindings,
1656 issues: keybindingsConfig.issues,
1657 providers,
1658 ...editorConfig,
1659 // Same discovery-with-timeout treatment as editors: a slow probe
1660 // must not stall server.getConfig, so it degrades to no targets.
1661 remoteOpenTargets: yield* resolveAvailableEditorsForConfig(
1662 remoteOpenTargets.resolveTargets(),
1663 ),
1664 directEndpoints: yield* resolveAvailableEditorsForConfig(directEndpoints.resolve()),
1665 observability: {
1666 logsDirectoryPath: config.logsDir,
1667 localTracingEnabled: true,
1668 ...(config.otlpTracesUrl !== undefined
1669 ? { otlpTracesUrl: config.otlpTracesUrl }
1670 : {}),
1671 otlpTracesEnabled: config.otlpTracesUrl !== undefined,
1672 ...(config.otlpMetricsUrl !== undefined
1673 ? { otlpMetricsUrl: config.otlpMetricsUrl }
1674 : {}),
1675 otlpMetricsEnabled: config.otlpMetricsUrl !== undefined,
1676 ...(config.otlpLogsUrl !== undefined ? { otlpLogsUrl: config.otlpLogsUrl } : {}),
1677 otlpLogsEnabled: config.otlpLogsUrl !== undefined,
1678 },
1679 settings,
1680 devServerStatus: true,
1681 managedWorkspaces: true,
1682 ...(mergebotQueue.configuration ? { mergebot: mergebotQueue.configuration } : {}),
1683 shellResumeCompletionMarker: true,
1684 threadResumeCompletionMarker: true,
1685 threadSnapshotPagination: true,
1686 threadFind: true,
1687 threadFindProgressive: true,
1688 ...Option.match(scratchWorkspaceRoot, {
1689 onNone: () => ({}),
1690 onSome: (root) => ({ scratchWorkspaceRoot: root }),
1691 }),
1692 newProjectsRoot: managedFolders.namedProjectsRoot,
1693 };
1694 });
1695
1696 const refreshGitStatus = (cwd: string) =>
1697 vcsStatusBroadcaster
1698 .refreshStatus(cwd)
1699 .pipe(Effect.ignoreCause({ log: true }), Effect.forkDetach, Effect.asVoid);
1700
1701 const getOrchestrationV2ArchivedShellSnapshot = Effect.gen(function* () {
1702 const { threads, projects, snapshotSequence } = yield* loadShellSnapshotParts({
1703 sql,
1704 readThreads: threadManagement.readShellSnapshot({ location: "archive" }),
1705 listProjects: projectStore.listShells(),
1706 latestSequence: applicationEvents.latestApplicationSequence,
1707 });
1708 return {
1709 schemaVersion: threads.schemaVersion,
1710 snapshotSequence,
1711 projects,
1712 threads: threads.archivedThreads,
1713 } as const;
1714 }).pipe(
1715 Effect.flatMap((snapshot) =>
1716 enrichProjectShells(snapshot.projects).pipe(
1717 Effect.map(({ projects }) => ({ ...snapshot, projects })),
1718 ),
1719 ),
1720 Effect.mapError(
1721 (cause) =>
1722 new OrchestrationV2GetShellSnapshotError({
1723 message: "Failed to load archived thread snapshot",
1724 cause,
1725 }),
1726 ),
1727 );
1728
1729 const subscribeOrchestrationV2ArchivedShell = Effect.fn(
1730 "ws.orchestrationV2.subscribeArchivedShell",
1731 )(function* () {
1732 const snapshot = yield* getOrchestrationV2ArchivedShellSnapshot;
1733 const live = threadManagement
1734 .streamStoredEventsFrom({ afterSequence: snapshot.snapshotSequence })
1735 .pipe(
1736 Stream.groupedWithin(512, Duration.millis(50)),
1737 Stream.mapEffect((events) =>
1738 Effect.forEach(
1739 coalesceStoredThreadEvents(Array.from(events)),
1740 (stored) =>
1741 threadManagement
1742 .getThreadShell(stored.event.threadId)
1743 .pipe(
1744 Effect.map((shell) =>
1745 archivedShellStreamItemFromThreadShell({ stored, shell }),
1746 ),
1747 ),
1748 { concurrency: 8 },
1749 ),
1750 ),
1751 Stream.flatMap(Stream.fromIterable),
1752 Stream.filterMap((item) => (item === null ? Result.failVoid : Result.succeed(item))),
1753 (stream) => bufferLiveStream(stream),
1754 Stream.mapError(
1755 (cause) =>
1756 new OrchestrationV2GetShellSnapshotError({
1757 message: "Failed while streaming archived threads",
1758 cause,
1759 }),
1760 ),
1761 );
1762 return Stream.concat(rpcInitialItems([{ kind: "snapshot" as const, snapshot }]), live);
1763 });
1764
1765 const mutateProject = Effect.fn("ws.projects.mutate")(function* (mutation: ProjectMutation) {
1766 const result = yield* projectMutationOperation(projectService, mutation);
1767 if (mutation.type === "project.delete")
1768 yield* projectCloneTracker.discard(mutation.projectId);
1769 return result;
1770 });
1771
1772 const handlers = ServerWsRpcGroup.of({
1773 [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command) =>
1774 Effect.annotateCurrentSpan({
1775 "orchestration_v2.command_id": command.commandId,
1776 "orchestration_v2.command_type": command.type,
1777 "orchestration_v2.thread_id":
1778 command.type === "thread.fork" || command.type === "thread.merge_back"
1779 ? command.targetThreadId
1780 : command.type === "delegated_task.request" ||
1781 command.type === "delegated_task.wake-policy" ||
1782 command.type === "delegated_task.completion-delivery.acknowledge" ||
1783 command.type === "delegated_task.completion-delivery.dispose" ||
1784 command.type === "thread.created.record"
1785 ? command.parentThreadId
1786 : command.threadId,
1787 ...(command.type === "thread.fork" || command.type === "thread.merge_back"
1788 ? { "orchestration_v2.source_thread_id": command.sourceThreadId }
1789 : {}),
1790 }).pipe(
1791 Effect.andThen(
1792 startup
1793 .enqueueCommand(
1794 // A retry also restarts the preparation work the launch owns.
1795 (command.type === "prepared-run.retry"
1796 ? threadLaunch.retryPreparation(command)
1797 : ThreadMessageIntake.dispatchCommand(
1798 ThreadManagementService.withCreationProvenance(command, {
1799 createdBy: "user",
1800 creationSource:
1801 "creationSource" in command ? command.creationSource : "web",
1802 }),
1803 )
1804 ).pipe(Effect.provide(intakeContext)),
1805 )
1806 .pipe(
1807 Effect.map((result) => ({ sequence: result.sequence })),
1808 Effect.mapError((cause) => {
1809 const detail = userFacingDispatchErrorMessage(cause);
1810 return new OrchestrationV2DispatchCommandError({
1811 commandId: command.commandId,
1812 commandType: command.type,
1813 ...(cause._tag === "OrchestratorDispatchError" && cause.reason !== undefined
1814 ? { reason: cause.reason }
1815 : {}),
1816 message: detail ?? "Failed to dispatch orchestration V2 command",
1817 ...(detail === undefined ? {} : { detail }),
1818 cause,
1819 });
1820 }),
1821 ),
1822 ),
1823 ),
1824 [ORCHESTRATION_V2_WS_METHODS.stopWorkflow]: (input) => threadManagement.stopWorkflow(input),
1825 [ORCHESTRATION_V2_WS_METHODS.getWorkflowScript]: (input) =>
1826 readWorkflowScript({ scriptPath: input.scriptPath }),
1827 [ORCHESTRATION_V2_WS_METHODS.getTurnItem]: (input) =>
1828 threadManagement.getTurnItem(input).pipe(
1829 Effect.mapError(
1830 (cause) =>
1831 new OrchestrationV2GetThreadProjectionError({
1832 threadId: input.threadId,
1833 message: "Failed to load turn item",
1834 cause,
1835 }),
1836 ),
1837 ),
1838 [ORCHESTRATION_V2_WS_METHODS.getTurnDiff]: (input) =>
1839 checkpointDiffQuery.getTurnDiff(input).pipe(
1840 Effect.mapError(
1841 (cause) =>
1842 new OrchestrationGetTurnDiffError({
1843 message: "Failed to load turn diff",
1844 cause,
1845 }),
1846 ),
1847 ),
1848 [ORCHESTRATION_V2_WS_METHODS.getFullThreadDiff]: (input) =>
1849 checkpointDiffQuery.getFullThreadDiff(input).pipe(
1850 Effect.mapError(
1851 (cause) =>
1852 new OrchestrationGetFullThreadDiffError({
1853 message: "Failed to load full thread diff",
1854 cause,
1855 }),
1856 ),
1857 ),
1858 [ORCHESTRATION_V2_WS_METHODS.searchThread]: (input) =>
1859 threadManagement
1860 .searchThread(input)
1861 .pipe(Effect.mapError((cause) => new OrchestrationV2SearchThreadError({ cause }))),
1862 [ORCHESTRATION_V2_WS_METHODS.searchThreadStream]: (input) =>
1863 threadManagement
1864 .searchThreadStream(input)
1865 .pipe(Stream.mapError((cause) => new OrchestrationV2SearchThreadError({ cause }))),
1866 [ORCHESTRATION_V2_WS_METHODS.searchThreads]: (input) =>
1867 threadSearch.search(input).pipe(
1868 Effect.mapError(
1869 (cause) =>
1870 new OrchestrationSearchThreadsError({
1871 message: "Failed to search threads",
1872 cause,
1873 }),
1874 ),
1875 ),
1876 [ORCHESTRATION_V2_WS_METHODS.getArchivedShellSnapshot]: (_input) =>
1877 getOrchestrationV2ArchivedShellSnapshot,
1878 [ORCHESTRATION_V2_WS_METHODS.getThreadProjection]: (input) =>
1879 Effect.annotateCurrentSpan({ "orchestration_v2.thread_id": input.threadId }).pipe(
1880 Effect.andThen(
1881 // Pre-pagination clients still call this compatibility endpoint.
1882 // Keep stale clients from materializing an unbounded transcript.
1883 threadManagement
1884 .getThreadSnapshotWindow(input.threadId, {
1885 rowLimit: THREAD_HISTORY_SNAPSHOT_ROW_LIMIT,
1886 })
1887 .pipe(
1888 Effect.map((snapshot) =>
1889 projectThreadProjectionForWire(
1890 buildBoundedThreadProjection({
1891 projection: snapshot.projection,
1892 snapshotSequence: snapshot.snapshotSequence,
1893 }).projection,
1894 ),
1895 ),
1896 Effect.mapError(
1897 (cause) =>
1898 new OrchestrationV2GetThreadProjectionError({
1899 threadId: input.threadId,
1900 message: `Failed to load orchestration V2 thread ${input.threadId}`,
1901 cause,
1902 }),
1903 ),
1904 ),
1905 ),
1906 ),
1907 [ORCHESTRATION_V2_WS_METHODS.launchThread]: (input) =>
1908 Effect.annotateCurrentSpan({
1909 "orchestration_v2.command_id": input.commandId,
1910 "orchestration_v2.project_id": input.projectId,
1911 }).pipe(
1912 Effect.andThen(
1913 startup
1914 .enqueueCommand(
1915 ThreadMessageIntake.launchThread({
1916 commandId: input.commandId,
1917 ...(input.threadId === undefined ? {} : { threadId: input.threadId }),
1918 ...(input.reuseExistingThread === undefined
1919 ? {}
1920 : { reuseExistingThread: input.reuseExistingThread }),
1921 projectId: input.projectId,
1922 title: input.title,
1923 ...(input.generateTitle === undefined
1924 ? {}
1925 : { generateTitle: input.generateTitle }),
1926 modelSelection: input.modelSelection,
1927 runtimeMode: input.runtimeMode,
1928 interactionMode: input.interactionMode,
1929 workspaceStrategy: input.workspaceStrategy,
1930 ...(input.initialMessage === undefined
1931 ? {}
1932 : {
1933 initialMessage: {
1934 ...(input.initialMessage.messageId === undefined
1935 ? {}
1936 : { messageId: input.initialMessage.messageId }),
1937 text: input.initialMessage.text,
1938 attachments: input.initialMessage.attachments,
1939 ...(input.initialMessage.context === undefined
1940 ? {}
1941 : { context: input.initialMessage.context }),
1942 },
1943 }),
1944 createdBy: "user",
1945 creationSource: input.creationSource ?? "web",
1946 }).pipe(Effect.provide(intakeContext)),
1947 )
1948 .pipe(
1949 Effect.map((result) => ({
1950 ...result,
1951 projection: projectThreadProjectionForWire(result.projection),
1952 })),
1953 Effect.catchTags({
1954 AttachmentClaimError: (cause) =>
1955 new OrchestrationV2ThreadLaunchError({
1956 commandId: input.commandId,
1957 projectId: input.projectId,
1958 message: cause.message,
1959 cause,
1960 }),
1961 ThreadLaunchError: (cause) =>
1962 new OrchestrationV2ThreadLaunchError({
1963 commandId: input.commandId,
1964 projectId: input.projectId,
1965 message: "Failed to launch thread",
1966 cause,
1967 }),
1968 ServerRuntimeStartupError: (cause) =>
1969 new OrchestrationV2ThreadLaunchError({
1970 commandId: input.commandId,
1971 projectId: input.projectId,
1972 message: "Failed to launch thread",
1973 cause,
1974 }),
1975 ThreadManagementProviderInstanceDisabledError: (cause) =>
1976 new OrchestrationV2ThreadLaunchError({
1977 commandId: input.commandId,
1978 projectId: input.projectId,
1979 message: cause.message,
1980 cause,
1981 }),
1982 }),
1983 ),
1984 ),
1985 ),
1986 [ORCHESTRATION_V2_WS_METHODS.subscribeArchivedShell]: (_input) =>
1987 Stream.unwrap(subscribeOrchestrationV2ArchivedShell()),
1988 [ORCHESTRATION_V2_WS_METHODS.subscribeShell]: (input) =>
1989 Stream.unwrap(subscribeOrchestrationV2Shell(input)),
1990 [ORCHESTRATION_V2_WS_METHODS.subscribeThread]: (input) =>
1991 Stream.unwrap(
1992 Effect.annotateCurrentSpan({ "orchestration_v2.thread_id": input.threadId }).pipe(
1993 Effect.andThen(subscribeOrchestrationV2Thread(input)),
1994 ),
1995 ),
1996 [WS_METHODS.scheduledTasksList]: (_input) =>
1997 scheduledTasks.list().pipe(Effect.map(withVisibleWebhookUrls)),
1998 [WS_METHODS.scheduledTasksSubscribe]: (_input) =>
1999 scheduledTasks.subscribeList().pipe(Stream.map(withVisibleWebhookUrls)),
2000 [WS_METHODS.scheduledTasksUpsert]: (input) => scheduledTasks.upsert(input),
2001 [WS_METHODS.scheduledTasksSetEnabled]: (input) =>
2002 Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe(
2003 Effect.andThen(scheduledTasks.setEnabled(input)),
2004 ),
2005 [WS_METHODS.scheduledTasksDelete]: (input) =>
2006 Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe(
2007 Effect.andThen(scheduledTasks.delete(input)),
2008 ),
2009 [WS_METHODS.scheduledTasksRunNow]: (input) =>
2010 Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe(
2011 Effect.andThen(scheduledTasks.runNow(input)),
2012 ),
2013 [WS_METHODS.scheduledTasksRotateWebhookToken]: (input) =>
2014 Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe(
2015 Effect.andThen(scheduledTasks.rotateWebhookToken(input)),
2016 ),
2017 [WS_METHODS.secretsAnswerRequest]: (input) =>
2018 Effect.annotateCurrentSpan({ "orchestration_v2.thread_id": input.threadId }).pipe(
2019 Effect.andThen(secretRequests.answer(input)),
2020 ),
2021 [WS_METHODS.scheduledTasksListWebhookDeliveries]: (input) =>
2022 Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe(
2023 Effect.andThen(scheduledTasks.listWebhookDeliveries(input)),
2024 ),
2025 [WS_METHODS.scheduledTasksGetWebhookDelivery]: (input) =>
2026 Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe(
2027 Effect.andThen(scheduledTasks.getWebhookDelivery(input)),
2028 ),
2029 [WS_METHODS.serverProbe]: (_input) => Effect.succeed({}),
2030 [WS_METHODS.serverGetConfig]: (_input) => loadServerConfig({ usageLimitsCommand: false }),
2031 [WS_METHODS.serverSearchAcpRegistry]: (input) =>
2032 acpRegistryCatalog
2033 .search(input)
2034 .pipe(Effect.mapError(AcpRegistrySupport.toAcpRegistryOperationError)),
2035 [WS_METHODS.serverPrepareAcpRegistryAgent]: (input) =>
2036 Effect.annotateCurrentSpan({ "acp_registry.agent_id": input.agentId }).pipe(
2037 Effect.andThen(
2038 acpRegistryCatalog
2039 .prepare(input)
2040 .pipe(Effect.mapError(AcpRegistrySupport.toAcpRegistryOperationError)),
2041 ),
2042 ),
2043 [WS_METHODS.serverUninstallAcpRegistryManagedBinary]: (input) =>
2044 Effect.annotateCurrentSpan({ "acp_registry.agent_id": input.agentId }).pipe(
2045 Effect.andThen(
2046 acpRegistryCatalog
2047 .uninstallManagedBinary(input)
2048 .pipe(Effect.mapError(AcpRegistrySupport.toAcpRegistryOperationError)),
2049 ),
2050 ),
2051 [WS_METHODS.serverAcceptAcpRegistryUrlAuth]: (input) =>
2052 Effect.annotateCurrentSpan({ "provider.instance_id": input.instanceId }).pipe(
2053 Effect.andThen(
2054 acpRegistryRuntimeCoordinator
2055 .acceptUrlAuthentication(input)
2056 .pipe(Effect.map((accepted) => ({ accepted }))),
2057 ),
2058 ),
2059 [WS_METHODS.serverListAcpRegistrySessions]: (input) =>
2060 Effect.annotateCurrentSpan({
2061 "provider.instance_id": input.instanceId,
2062 "project.id": input.projectId,
2063 }).pipe(Effect.andThen(listAcpRegistrySessions(input))),
2064 [WS_METHODS.serverImportAcpRegistrySession]: (input) =>
2065 Effect.annotateCurrentSpan({
2066 "provider.instance_id": input.instanceId,
2067 "project.id": input.projectId,
2068 }).pipe(Effect.andThen(importAcpRegistrySession(input))),
2069 [WS_METHODS.serverDeleteAcpRegistrySession]: (input) =>
2070 Effect.annotateCurrentSpan({
2071 "provider.instance_id": input.instanceId,
2072 "project.id": input.projectId,
2073 }).pipe(Effect.andThen(deleteAcpRegistrySession(input))),
2074 [WS_METHODS.serverListAcpRegistryProviders]: (input) =>
2075 Effect.annotateCurrentSpan({
2076 "provider.instance_id": input.instanceId,
2077 "project.id": input.projectId,
2078 }).pipe(Effect.andThen(listAcpRegistryProviders(input))),
2079 [WS_METHODS.serverSetAcpRegistryProvider]: (input) =>
2080 Effect.annotateCurrentSpan({
2081 "provider.instance_id": input.instanceId,
2082 "project.id": input.projectId,
2083 }).pipe(Effect.andThen(setAcpRegistryProvider(input))),
2084 [WS_METHODS.serverDisableAcpRegistryProvider]: (input) =>
2085 Effect.annotateCurrentSpan({
2086 "provider.instance_id": input.instanceId,
2087 "project.id": input.projectId,
2088 }).pipe(Effect.andThen(disableAcpRegistryProvider(input))),
2089 [WS_METHODS.serverLogoutAcpRegistry]: (input) =>
2090 Effect.annotateCurrentSpan({ "provider.instance_id": input.instanceId }).pipe(
2091 Effect.andThen(
2092 Effect.gen(function* () {
2093 const { instance, manager } = yield* acpSessionManager(input.instanceId);
2094 const snapshot = yield* instance.snapshot.getSnapshot;
2095 if (snapshot.auth.canLogout !== true) {
2096 return yield* new AcpRegistryOperationError({
2097 reason: "logout_unsupported",
2098 message: "The ACP agent does not advertise logout.",
2099 });
2100 }
2101 if (instance.auth) {
2102 yield* providerAuth.logout(input).pipe(
2103 Effect.mapError(
2104 (cause) =>
2105 new AcpRegistryOperationError({
2106 reason: "logout_failed",
2107 message: "Could not sign out of the ACP agent.",
2108 cause,
2109 }),
2110 ),
2111 );
2112 } else {
2113 yield* providerSessionManager.closeInstance(input.instanceId).pipe(
2114 Effect.mapError(
2115 (cause) =>
2116 new AcpRegistryOperationError({
2117 reason: "logout_failed",
2118 message: "Could not stop live sessions before ACP logout.",
2119 cause,
2120 }),
2121 ),
2122 );
2123 yield* manager.logout(config.cwd);
2124 }
2125 yield* providerRegistry.refreshInstance(input.instanceId);
2126 return { loggedOut: true } as const;
2127 }),
2128 ),
2129 ),
2130 [WS_METHODS.serverRefreshProviders]: (input) =>
2131 Effect.gen(function* () {
2132 // Only explicit catalog refreshes bypass T3's caches. Workspace
2133 // discovery and background status checks retain their timers.
2134 if (input.refreshModels) {
2135 yield* modelManifest.forceRefresh;
2136 const instances = yield* providerInstances.listInstances;
2137 yield* Effect.forEach(
2138 instances.filter(
2139 (instance) =>
2140 input.instanceId === undefined || input.instanceId === instance.instanceId,
2141 ),
2142 (instance) =>
2143 Effect.gen(function* () {
2144 yield* instance.invalidateCaches ?? Effect.void;
2145 const maintenance = yield* instance.snapshot.resolveMaintenance({
2146 fresh: true,
2147 });
2148 if (maintenance.packageName)
2149 yield* providerLatestVersions.invalidate(maintenance.packageName);
2150 }),
2151 { concurrency: "unbounded", discard: true },
2152 );
2153 }
2154 // An untargeted refresh is "re-read everything's status", which
2155 // includes quota from configured usage-limit sources. Awaited,
2156 // not forked: the RPC scope closes on return and would
2157 // interrupt a fork before the hub answered.
2158 if (input.instanceId === undefined) {
2159 yield* usageLimitSources.refresh;
2160 }
2161 let providers = yield* input.cwd !== undefined && input.instanceId !== undefined
2162 ? providerRegistry.refreshWorkspaceSnapshot({
2163 instanceId: input.instanceId,
2164 cwd: input.cwd,
2165 fresh: input.fresh === true,
2166 })
2167 : input.instanceId !== undefined
2168 ? providerRegistry.refreshInstance(input.instanceId)
2169 : providerRegistry.refresh();
2170 if (input.refreshModels) {
2171 const instances = yield* providerInstances.listInstances;
2172 for (const instance of instances) {
2173 if (
2174 !instance.refreshModels ||
2175 (input.instanceId !== undefined && input.instanceId !== instance.instanceId) ||
2176 !providers.some(
2177 (provider) =>
2178 provider.instanceId === instance.instanceId &&
2179 provider.enabled &&
2180 provider.installed,
2181 )
2182 )
2183 continue;
2184 yield* instance.refreshModels().pipe(
2185 Effect.mapError(
2186 (error) =>
2187 new ProviderSetupError({
2188 instanceId: instance.instanceId,
2189 operation: "refresh-models",
2190 detail: error.detail,
2191 }),
2192 ),
2193 );
2194 providers = yield* providerRegistry.refreshInstance(instance.instanceId);
2195 }
2196 }
2197 return { providers };
2198 }),
2199 [WS_METHODS.mcpAppsCallTool]: (input) => mcpAppRequests.callTool(input),
2200 [WS_METHODS.mcpAppsToolInfo]: (input) => mcpAppRequests.toolInfo(input),
2201 [WS_METHODS.mcpAppsUpdateModelContext]: (input) => mcpAppRequests.updateModelContext(input),
2202 [WS_METHODS.mcpAppsReadResource]: (input) => mcpAppRequests.readResource(input),
2203 [WS_METHODS.providerUploadFeedback]: (input) =>
2204 Effect.gen(function* () {
2205 const projection = yield* threadManagement.getThreadRecords(input.threadId, [
2206 "providerThreads",
2207 ]);
2208 const providerThread =
2209 projection.providerThreads.find(
2210 (candidate) => candidate.id === projection.thread.activeProviderThreadId,
2211 ) ?? projection.providerThreads.at(-1);
2212 const providerSessionId = providerThread?.providerSessionId ?? null;
2213 if (providerThread === undefined || providerSessionId === null) {
2214 return yield* Effect.fail(
2215 new ProviderUploadFeedbackError({
2216 threadId: input.threadId,
2217 cause: "No provider session has run in this thread yet.",
2218 }),
2219 );
2220 }
2221 const runtime = Option.getOrNull(yield* providerSessionsV2.get(providerSessionId));
2222 if (runtime === null) {
2223 return yield* Effect.fail(
2224 new ProviderUploadFeedbackError({
2225 threadId: input.threadId,
2226 cause: "The provider session is no longer running. Send a message first.",
2227 }),
2228 );
2229 }
2230 if (runtime.uploadFeedback === undefined) {
2231 return yield* Effect.fail(
2232 new ProviderUploadFeedbackError({
2233 threadId: input.threadId,
2234 cause: `Provider '${runtime.driver}' does not support feedback uploads.`,
2235 }),
2236 );
2237 }
2238 return yield* runtime.uploadFeedback({
2239 providerThread,
2240 ...(input.reason === undefined ? {} : { reason: input.reason }),
2241 });
2242 }).pipe(
2243 Effect.mapError((cause) =>
2244 isProviderUploadFeedbackError(cause)
2245 ? cause
2246 : new ProviderUploadFeedbackError({
2247 threadId: input.threadId,
2248 cause,
2249 }),
2250 ),
2251 ),
2252 [WS_METHODS.serverUpdateProvider]: (input) =>
2253 providerMaintenanceRunner.updateProvider(input),
2254 [WS_METHODS.providerConsumeResetCredit]: (input) =>
2255 Effect.gen(function* () {
2256 if ("sourceId" in input) return yield* usageLimitSources.consumeResetCredit(input);
2257 const instance = yield* providerInstances.getInstance(input.instanceId);
2258 // A disabled instance must not spend anything on its account.
2259 if (instance === undefined || !instance.enabled) {
2260 return yield* new ProviderSetupError({
2261 instanceId: input.instanceId,
2262 operation: "consume-reset-credit",
2263 detail: instance ? "This provider is disabled." : "Provider instance not found.",
2264 });
2265 }
2266 if (instance.consumeResetCredit === undefined) {
2267 return yield* new ProviderSetupError({
2268 instanceId: input.instanceId,
2269 operation: "consume-reset-credit",
2270 detail: "This provider does not bank reset credits.",
2271 });
2272 }
2273 const outcome = yield* instance.consumeResetCredit().pipe(
2274 Effect.mapError(
2275 (error) =>
2276 new ProviderSetupError({
2277 instanceId: input.instanceId,
2278 operation: "consume-reset-credit",
2279 detail: error.detail,
2280 cause: error,
2281 }),
2282 ),
2283 );
2284 return { outcome };
2285 }),
2286 [WS_METHODS.providerAuthStart]: (input) => providerAuth.start(input, currentSessionId),
2287 [WS_METHODS.providerAuthRespond]: (input) =>
2288 Effect.annotateCurrentSpan({ instanceId: input.instanceId }).pipe(
2289 Effect.andThen(providerAuth.respond(input, currentSessionId)),
2290 ),
2291 [WS_METHODS.providerAuthComplete]: (input) =>
2292 providerAuth.complete(input, currentSessionId),
2293 [WS_METHODS.chatGptReconnectProfile]: (input) => providerAuth.reconnectProfile(input),
2294 [WS_METHODS.chatGptImportProfile]: (input) => providerAuth.importProfile(input),
2295 [WS_METHODS.chatGptHandoffSubscribe]: (input) =>
2296 subscribeChatGptHandoff(input, currentSessionId),
2297 [WS_METHODS.codexAuthCallbackSubscribe]: (input) => subscribeCodexAuthCallback(input),
2298 [WS_METHODS.providerAuthCancel]: (input) => providerAuth.cancel(input, currentSessionId),
2299 [WS_METHODS.providerAuthLogout]: (input) => providerAuth.logout(input),
2300 [WS_METHODS.providerAuthSubscribe]: (input) =>
2301 providerAuth.subscribe(input, currentSessionId),
2302 [WS_METHODS.providerInstallStart]: (input) => providerInstallation.start(input),
2303 [WS_METHODS.providerInstallCancel]: (input) => providerInstallation.cancel(input),
2304 [WS_METHODS.providerInstallSubscribe]: (input) => providerInstallation.subscribe(input),
2305 [WS_METHODS.providerInstallRemove]: (input) => providerInstallation.remove(input),
2306 [WS_METHODS.serverUpdateServer]: (input) => serverSelfUpdate.update(input),
2307 [WS_METHODS.serverUpdateServerWithProgress]: (input) =>
2308 Stream.callback<ServerSelfUpdateProgressEvent, ServerSelfUpdateError>((queue) =>
2309 serverSelfUpdate
2310 .update(input, (stage) =>
2311 Queue.offer(queue, {
2312 type: "progress",
2313 stage,
2314 }).pipe(Effect.asVoid),
2315 )
2316 .pipe(
2317 Effect.flatMap((result) =>
2318 Queue.offer(queue, {
2319 type: "complete",
2320 result,
2321 }),
2322 ),
2323 Effect.catchTags({
2324 ServerSelfUpdateError: (error) => Queue.fail(queue, error),
2325 }),
2326 Effect.andThen(Queue.end(queue)),
2327 Effect.forkScoped,
2328 ),
2329 ),
2330 [WS_METHODS.serverCommitDesktopUpdate]: (input) =>
2331 serverSelfUpdate.commitDesktopUpdate(input.requestId),
2332 [WS_METHODS.serverUpsertKeybinding]: (rule) =>
2333 Effect.gen(function* () {
2334 const keybindingsConfig = yield* keybindings.upsertKeybindingRule(rule);
2335 return { keybindings: keybindingsConfig, issues: [] };
2336 }),
2337 [WS_METHODS.serverRemoveKeybinding]: (rule) =>
2338 Effect.gen(function* () {
2339 const keybindingsConfig = yield* keybindings.removeKeybindingRule(rule);
2340 return { keybindings: keybindingsConfig, issues: [] };
2341 }),
2342 [WS_METHODS.serverRunStorageCleanup]: () => storageCleanup.runNow,
2343 [WS_METHODS.serverGetStorageCleanupReport]: () => storageCleanup.reports,
2344 [WS_METHODS.uiSettingsGet]: () => uiSettings.current,
2345 [WS_METHODS.uiSettingsPatch]: (input) => uiSettings.patch(input),
2346 [WS_METHODS.uiSettingsSubscribe]: () => uiSettings.changes,
2347 [WS_METHODS.serverGetSettings]: (_input) =>
2348 serverSettings.getSettings.pipe(Effect.map(ServerSettings.redactServerSettingsForClient)),
2349 [WS_METHODS.serverUpdateSettings]: ({ patch, providerInstanceMutation }) =>
2350 Effect.gen(function* () {
2351 const deviceHosts = patch.deviceHosts
2352 ? yield* remoteSshDeviceHosts(patch.deviceHosts).pipe(
2353 Effect.provide(deviceHostContext),
2354 )
2355 : undefined;
2356 const nextPatch = { ...patch, ...(deviceHosts ? { deviceHosts } : {}) };
2357 const settings = yield* providerInstanceMutation === undefined
2358 ? serverSettings.updateSettings(nextPatch)
2359 : serverSettings.updateProviderInstance(providerInstanceMutation, nextPatch);
2360 return ServerSettings.redactServerSettingsForClient(settings);
2361 }),
2362 [WS_METHODS.serverDiscoverSourceControl]: (_input) => sourceControlDiscovery.discover,
2363 [WS_METHODS.serverGetTraceDiagnostics]: (_input) =>
2364 TraceDiagnostics.readTraceDiagnostics({
2365 traceFilePath: config.serverTracePath,
2366 maxFiles: config.traceMaxFiles,
2367 }),
2368 [WS_METHODS.serverGetProcessDiagnostics]: (_input) => processDiagnostics.read,
2369 [WS_METHODS.serverGetHostResources]: (_input) => hostResources.read,
2370 [WS_METHODS.serverGetProcessResourceHistory]: (input) =>
2371 processResourceMonitor.readHistory(input),
2372 [WS_METHODS.serverGetResourceTelemetryHistory]: (input) =>
2373 resourceTelemetry.readHistory(input),
2374 [WS_METHODS.serverGetUsageSummary]: (input) => usage.readSummary(input),
2375 [WS_METHODS.serverRefreshUsageRates]: (_input) => usage.refreshRates,
2376 [WS_METHODS.serverRetryResourceTelemetry]: (_input) => resourceTelemetry.retry,
2377 [WS_METHODS.serverCheckOtlpEndpoint]: (input) => otlpEndpointCheck.check(input),
2378 [WS_METHODS.serverSignalProcess]: (input) => processDiagnostics.signal(input),
2379 [WS_METHODS.serverReportClientActivity]: (input, metadata) =>
2380 Ref.update(rpcClientIds, (clientIds) => {
2381 const next = new Set(clientIds);
2382 next.add(RpcClientId.make(metadata.client.id));
2383 return next;
2384 }).pipe(
2385 Effect.andThen(
2386 backgroundPolicy.reportClientActivity(
2387 currentSessionId,
2388 RpcClientId.make(metadata.client.id),
2389 input,
2390 ),
2391 ),
2392 ),
2393 [WS_METHODS.serverReportHostPowerState]: (input) =>
2394 backgroundPolicy.reportHostPowerState(input),
2395 [WS_METHODS.serverGetBackgroundPolicy]: (_input) => backgroundPolicy.snapshot,
2396 [WS_METHODS.cloudGetRelayClientStatus]: (_input) => relayClient.resolve,
2397 [WS_METHODS.cloudInstallRelayClient]: (_input) =>
2398 Stream.callback<RelayClientInstallProgressEvent, RelayClientInstallFailedError>((queue) =>
2399 relayClient
2400 .installWithProgress((event) => Queue.offer(queue, event).pipe(Effect.asVoid))
2401 .pipe(
2402 Effect.flatMap((status) =>
2403 Queue.offer(queue, {
2404 type: "complete",
2405 status,
2406 }),
2407 ),
2408 Effect.catchTags({
2409 RelayClientInstallError: (error) =>
2410 Queue.fail(
2411 queue,
2412 new RelayClientInstallFailedError({
2413 reason: error.reason,
2414 message: error.message,
2415 }),
2416 ),
2417 }),
2418 Effect.andThen(Queue.end(queue)),
2419 Effect.forkScoped,
2420 ),
2421 ),
2422 [WS_METHODS.mergebotQueue]: (input) => mergebotQueue.read(input),
2423 [WS_METHODS.mergebotSubscribe]: (input) => mergebotQueue.watchQueue(input),
2424 [WS_METHODS.mergebotSubscribePullRequestStatus]: (input) =>
2425 mergebotQueue.watchPullRequestStatus(input),
2426 [WS_METHODS.mergebotSubscribeDeployments]: (input) => mergebotQueue.watchDeployments(input),
2427 [WS_METHODS.mergebotOpenPullRequest]: (input) => mergebotQueue.openPullRequest(input),
2428 [WS_METHODS.mergebotSetAutoMerge]: (input) => mergebotQueue.setAutoMerge(input),
2429 [WS_METHODS.mergebotUpdatePullRequest]: (input) => mergebotQueue.updatePullRequest(input),
2430 [WS_METHODS.mergebotDeployProduction]: (input) => mergebotQueue.deployProduction(input),
2431 [WS_METHODS.pullRequestsList]: (input) => pullRequests.list(input),
2432 [WS_METHODS.pullRequestsListStats]: (input) => pullRequests.listStats(input),
2433 [WS_METHODS.pullRequestsRoutingIdentity]: (input) => pullRequests.routingIdentity(input),
2434 [WS_METHODS.pullRequestsRouting]: (input) => pullRequests.routing(input),
2435 [WS_METHODS.pullRequestsSummary]: (input) =>
2436 withPullRequestViewer(input, pullRequests.summary(input)),
2437 [WS_METHODS.pullRequestsStack]: (input) =>
2438 withPullRequestViewer(input, pullRequests.stack(input)),
2439 [WS_METHODS.pullRequestsLinkedThreads]: (input) =>
2440 resolvePullRequestSyncKey(input).pipe(
2441 Effect.flatMap((key) =>
2442 key === null
2443 ? Effect.succeed({ threads: [] })
2444 : listLinkedPullRequestThreads(key).pipe(
2445 Effect.provideService(SqlClient.SqlClient, sql),
2446 ),
2447 ),
2448 ),
2449 [WS_METHODS.pullRequestsDetail]: (input) =>
2450 withPullRequestViewer(input, pullRequests.detail(input)),
2451 [WS_METHODS.pullRequestsPreview]: (input) =>
2452 withPullRequestViewer(input, pullRequests.preview(input)),
2453 [WS_METHODS.pullRequestsChecks]: (input) =>
2454 withPullRequestViewer(input, pullRequests.checks(input)),
2455 [WS_METHODS.pullRequestsActivity]: (input) =>
2456 withPullRequestViewer(input, pullRequests.activity(input)),
2457 [WS_METHODS.pullRequestsReviewGuide]: (input) =>
2458 withPullRequestViewer(input, reviewGuides.get(input)),
2459 [WS_METHODS.pullRequestsGenerateReviewGuide]: (input) =>
2460 withPullRequestViewer(input, reviewGuides.generate(input)),
2461 [WS_METHODS.pullRequestsThreadComments]: (input) =>
2462 withPullRequestViewer(input, pullRequests.threadComments(input)),
2463 [WS_METHODS.pullRequestsDiffFileContents]: (input) =>
2464 withPullRequestViewer(input, pullRequests.diffFileContents(input)),
2465 [WS_METHODS.pullRequestsFilesViewed]: (input) =>
2466 withPullRequestViewer(input, pullRequests.filesViewed(input)),
2467 [WS_METHODS.pullRequestsSetFilesViewed]: (input) =>
2468 withPullRequestViewer(input, pullRequests.setFilesViewed(input)),
2469 [WS_METHODS.pullRequestsRunAction]: (input) =>
2470 withPullRequestViewer(input, pullRequests.runAction(input)).pipe(
2471 Effect.tap(() =>
2472 resolvePullRequestSyncKey(input).pipe(
2473 Effect.flatMap((key) =>
2474 key === null ? Effect.void : pullRequestSync.requestSync(key),
2475 ),
2476 ),
2477 ),
2478 ),
2479 [WS_METHODS.pullRequestsUpdate]: (input) =>
2480 withPullRequestViewer(input, pullRequests.update(input)),
2481 [WS_METHODS.pullRequestsComment]: (input) =>
2482 withPullRequestViewer(input, pullRequests.comment(input)),
2483 [WS_METHODS.pullRequestsUpdateComment]: (input) =>
2484 withPullRequestViewer(input, pullRequests.updateComment(input)),
2485 [WS_METHODS.pullRequestsSubmitReview]: (input) =>
2486 withPullRequestViewer(input, pullRequests.submitReview(input)),
2487 [WS_METHODS.pullRequestsReplyToThread]: (input) =>
2488 withPullRequestViewer(input, pullRequests.replyToThread(input)),
2489 [WS_METHODS.pullRequestsSetThreadResolution]: (input) =>
2490 withPullRequestViewer(input, pullRequests.setThreadResolution(input)),
2491 [WS_METHODS.pullRequestsSetReaction]: (input) =>
2492 withPullRequestViewer(input, pullRequests.setReaction(input)),
2493 [WS_METHODS.pullRequestsReportState]: (input) => pullRequests.reportState(input),
2494 [WS_METHODS.pullRequestsInvalidate]: (input) =>
2495 pullRequests.invalidate(input, { notifyReaders: true }).pipe(
2496 // A reader asking for fresh host state also wants the thread badges it feeds to
2497 // catch up, including a merged link the sweep would otherwise never revisit.
2498 Effect.andThen(
2499 input.reference === undefined || input.filesViewedOnly === true
2500 ? Effect.void
2501 : resolvePullRequestSyncKey(input.reference).pipe(
2502 Effect.flatMap((key) =>
2503 key === null ? Effect.void : pullRequestSync.requestSync(key),
2504 ),
2505 ),
2506 ),
2507 ),
2508 [WS_METHODS.pullRequestsSubscribeRefreshes]: () => pullRequests.subscribeRefreshes,
2509 [WS_METHODS.pullRequestsReviewerCandidates]: (input) =>
2510 withPullRequestViewer(input, pullRequests.reviewerCandidates(input)),
2511 [WS_METHODS.pullRequestsRequestReviewers]: (input) =>
2512 withPullRequestViewer(input, pullRequests.requestReviewers(input)),
2513 [WS_METHODS.pullRequestsLabelCandidates]: (input) =>
2514 withPullRequestViewer(input, pullRequests.labelCandidates(input)),
2515 [WS_METHODS.pullRequestsSetLabels]: (input) =>
2516 withPullRequestViewer(input, pullRequests.setLabels(input)),
2517 [WS_METHODS.sourceControlLookupRepository]: (input) =>
2518 sourceControlRepositories.lookupRepository(input),
2519 [WS_METHODS.sourceControlCloneRepository]: (input) =>
2520 sourceControlRepositories.cloneRepository(input),
2521 [WS_METHODS.projectCloneStart]: (input) =>
2522 projectCloneTracker.start(input, {
2523 createProject: (project) =>
2524 projectService
2525 .create({
2526 commandId: CommandId.make(`project-clone-create:${project.projectId}`),
2527 projectId: project.projectId,
2528 title: project.title,
2529 workspaceRoot: project.workspaceRoot,
2530 createWorkspaceRootIfMissing: true,
2531 })
2532 .pipe(
2533 Effect.asVoid,
2534 Effect.mapError(
2535 (cause) =>
2536 new OrchestrationDispatchCommandError({
2537 message: "Failed to create clone project.",
2538 cause,
2539 }),
2540 ),
2541 ),
2542 onCloned: (project) =>
2543 repositoryIdentityResolver.resolve(project.workspaceRoot, { refresh: true }).pipe(
2544 Effect.andThen(
2545 projectService.update({
2546 commandId: CommandId.make(`project-clone-done:${project.projectId}`),
2547 projectId: project.projectId,
2548 }),
2549 ),
2550 Effect.andThen(refreshGitStatus(project.workspaceRoot)),
2551 Effect.ignoreCause({ log: true }),
2552 ),
2553 }),
2554 [WS_METHODS.projectsEnsureScratch]: () =>
2555 managedFolders.ensureScratchProject.pipe(
2556 Effect.mapError(
2557 (cause) => new OrchestrationDispatchCommandError({ message: cause.message, cause }),
2558 ),
2559 ),
2560 [WS_METHODS.projectsCreateNew]: (input) =>
2561 managedFolders
2562 .createNamedProject(input)
2563 .pipe(
2564 Effect.mapError(
2565 (cause) => new OrchestrationDispatchCommandError({ message: cause.message, cause }),
2566 ),
2567 ),
2568 [WS_METHODS.projectCloneCancel]: (input) =>
2569 projectCloneTracker.cancel(input.projectId).pipe(Effect.map((applied) => ({ applied }))),
2570 [WS_METHODS.projectCloneRetry]: (input) =>
2571 projectCloneTracker.retry(input.projectId).pipe(Effect.map((applied) => ({ applied }))),
2572 [WS_METHODS.subscribeProjectClones]: () => projectCloneTracker.stream,
2573 [WS_METHODS.sourceControlPublishRepository]: (input) =>
2574 sourceControlRepositories.publishRepository(input).pipe(
2575 // A new remote can change the cached identity. Only the `cwd` entry
2576 // refreshes, so after a publish from a linked worktree the project
2577 // root entry waits for its TTL.
2578 Effect.tap(() => repositoryIdentityResolver.resolve(input.cwd, { refresh: true })),
2579 Effect.tap(() => refreshGitStatus(input.cwd)),
2580 ),
2581 [WS_METHODS.projectsSearchEntries]: (input) =>
2582 workspaceEntries.search(input).pipe(
2583 Effect.mapError(
2584 (cause) =>
2585 new ProjectSearchEntriesError({
2586 cwd: input.cwd,
2587 queryLength: input.query.length,
2588 limit: input.limit,
2589 ...projectEntriesFailureContext(cause),
2590 cause,
2591 }),
2592 ),
2593 ),
2594 [WS_METHODS.projectsSearchContents]: (input) =>
2595 workspaceEntries.searchContents(input).pipe(
2596 Effect.mapError(
2597 (cause) =>
2598 new ProjectSearchContentsError({
2599 cwd: input.cwd,
2600 queryLength: input.query.length,
2601 limit: input.limit,
2602 ...projectEntriesFailureContext(cause),
2603 cause,
2604 }),
2605 ),
2606 ),
2607 [WS_METHODS.projectsListEntries]: (input) =>
2608 workspaceEntries.list(input).pipe(
2609 Effect.mapError(
2610 (cause) =>
2611 new ProjectListEntriesError({
2612 ...input,
2613 ...projectEntriesFailureContext(cause),
2614 cause,
2615 }),
2616 ),
2617 ),
2618 [WS_METHODS.managedWorkspacesStatus]: () => managedWorkspaces.status(),
2619 [WS_METHODS.managedWorkspacesCreate]: () => managedWorkspaces.create(),
2620 [WS_METHODS.managedWorkspacesArchiveCompleted]: (input) =>
2621 managedWorkspaces.archiveCompleted(input),
2622 [WS_METHODS.devServerStatus]: (input) => devServer.status(input),
2623 [WS_METHODS.jjSubscribeStack]: (input) => jjStack.watchStack(input),
2624 [WS_METHODS.jjSnapshot]: (input) => jjStack.snapshot(input),
2625 [WS_METHODS.jjDiff]: (input) => jjStack.diff(input),
2626 [WS_METHODS.jjDescribe]: (input) => jjStack.describe(input),
2627 [WS_METHODS.jjSquash]: (input) => jjStack.squash(input),
2628 [WS_METHODS.jjPush]: (input) => jjStack.push(input),
2629 [WS_METHODS.devServerStart]: (input) => devServer.start(input),
2630 [WS_METHODS.projectsReadFile]: (input) =>
2631 workspaceFileSystem.readFile(input).pipe(
2632 Effect.mapError(
2633 (cause) =>
2634 new ProjectReadFileError({
2635 ...input,
2636 ...projectFileFailureContext(cause),
2637 cause,
2638 }),
2639 ),
2640 ),
2641 [WS_METHODS.projectsWatch]: (input) =>
2642 workspaceFileSystem.watch(input).pipe(
2643 Stream.mapError(
2644 (cause) =>
2645 new ProjectReadFileError({
2646 ...input,
2647 ...projectFileFailureContext(cause),
2648 cause,
2649 }),
2650 ),
2651 ),
2652 [WS_METHODS.projectsWriteFile]: (input) =>
2653 workspaceFileSystem.writeFile(input).pipe(
2654 Effect.mapError(
2655 (cause) =>
2656 new ProjectWriteFileError({
2657 cwd: input.cwd,
2658 relativePath: input.relativePath,
2659 ...projectFileFailureContext(cause),
2660 cause,
2661 }),
2662 ),
2663 ),
2664 [WS_METHODS.projectsMutate]: (mutation) =>
2665 startup.enqueueCommand(mutateProject(mutation)).pipe(
2666 Effect.mapError(
2667 (cause) =>
2668 new ProjectMutationError({
2669 commandId: mutation.commandId,
2670 message:
2671 cause._tag === "ProjectNotEmptyError"
2672 ? cause.message
2673 : "Failed to mutate project.",
2674 cause,
2675 }),
2676 ),
2677 ),
2678 [WS_METHODS.shellOpenInEditor]: (input) => externalLauncher.launchEditor(input),
2679 [WS_METHODS.filesystemGetMetadata]: (input) => workspaceFileSystem.getMetadata(input),
2680 [WS_METHODS.filesystemBrowse]: (input) =>
2681 workspaceEntries.browse(input).pipe(
2682 Effect.mapError(
2683 (cause) =>
2684 new FilesystemBrowseError({
2685 ...input,
2686 ...filesystemBrowseFailureContext(cause),
2687 cause,
2688 }),
2689 ),
2690 ),
2691 [WS_METHODS.attachmentsCreateUploadUrl]: (input) => issueAttachmentUploadUrl(input),
2692 [WS_METHODS.attachmentsDelete]: (input) => deletePendingAttachment(input.attachmentId),
2693 [WS_METHODS.agentSessionsScan]: () => agentSessionScanner.scan,
2694 [WS_METHODS.agentSessionsImport]: (input) =>
2695 agentSessionImporter.importRecentAgentThreads(input),
2696 [WS_METHODS.assetsCreateUrl]: (input) =>
2697 Effect.gen(function* () {
2698 const path = yield* Path.Path;
2699 // An absolute media path can be linked from a thread on another environment.
2700 if (
2701 input.resource._tag === "attachment" ||
2702 input.resource._tag === "native-app-icon" ||
2703 input.resource._tag === "tool-output-image" ||
2704 // GitHub media names the repository it authenticates through itself.
2705 input.resource._tag === "github-media" ||
2706 (input.resource._tag === "media-file" && path.isAbsolute(input.resource.path))
2707 ) {
2708 return yield* issueAssetUrl({ resource: input.resource });
2709 }
2710 if (input.resource._tag === "draft-workspace-file") {
2711 // A project draft names its workspace directly; there is no
2712 // thread to resolve one from.
2713 return yield* issueAssetUrl({
2714 resource: input.resource,
2715 workspaceRoot: input.resource.cwd,
2716 });
2717 }
2718 if (input.resource._tag === "project-favicon") {
2719 const project = yield* projectStore
2720 .findActiveByWorkspaceRoot(input.resource.cwd)
2721 .pipe(
2722 Effect.mapError(
2723 (cause) =>
2724 new AssetWorkspaceContextResolutionError({
2725 resource: input.resource,
2726 cause,
2727 }),
2728 ),
2729 );
2730 if (Option.isNone(project)) {
2731 return yield* new AssetWorkspaceContextNotFoundError({
2732 resource: input.resource,
2733 });
2734 }
2735 // A cloned project exists before its files do. Clients ask again
2736 // when the clone lands (see createProjectFaviconUrlAtomFamily).
2737 const clone = yield* projectCloneTracker.get(project.value.projectId);
2738 return yield* issueAssetUrl({
2739 resource: input.resource,
2740 ...(project.value.faviconPath
2741 ? { projectFaviconPath: project.value.faviconPath }
2742 : {}),
2743 projectCheckoutPending:
2744 clone !== null &&
2745 clone.phase !== "done" &&
2746 clone.destinationPath === project.value.workspaceRoot,
2747 });
2748 }
2749 const thread = yield* threadManagement
2750 .getThreadRecords(input.resource.threadId, [])
2751 .pipe(
2752 Effect.mapError(
2753 (cause) =>
2754 new AssetWorkspaceContextResolutionError({
2755 resource: input.resource,
2756 cause,
2757 }),
2758 ),
2759 );
2760 const project = yield* projectService.getById(thread.thread.projectId).pipe(
2761 Effect.mapError(
2762 (cause) =>
2763 new AssetWorkspaceContextResolutionError({
2764 resource: input.resource,
2765 cause,
2766 }),
2767 ),
2768 );
2769 if (Option.isNone(project)) {
2770 return yield* new AssetWorkspaceContextNotFoundError({
2771 resource: input.resource,
2772 });
2773 }
2774 return yield* issueAssetUrl({
2775 resource: input.resource,
2776 workspaceRoot: thread.thread.worktreePath ?? project.value.workspaceRoot,
2777 });
2778 }),
2779 [WS_METHODS.assetsPersistChatAttachments]: (input) =>
2780 persistChatAttachments(input).pipe(Effect.map((attachments) => ({ attachments }))),
2781 [WS_METHODS.subscribeVcsStatus]: (input) =>
2782 vcsStatusBroadcaster.streamStatus(input, {
2783 automaticRemoteRefreshInterval: automaticGitFetchInterval,
2784 }),
2785 [WS_METHODS.subscribeWorktreeSetup]: (input) => worktreeSetupTracker.stream(input.threadId),
2786 [WS_METHODS.worktreeSetupCancel]: (input) =>
2787 worktreeSetupTracker
2788 .cancel(input.threadId)
2789 .pipe(Effect.map((cancelled) => ({ cancelled }))),
2790 [WS_METHODS.vcsRefreshStatus]: (input) => vcsStatusBroadcaster.refreshStatus(input.cwd),
2791 [WS_METHODS.vcsPull]: (input) =>
2792 gitWorkflow.pullCurrentBranch(input.cwd).pipe(
2793 Effect.matchCauseEffect({
2794 onFailure: (cause) => Effect.failCause(cause),
2795 onSuccess: (result) =>
2796 refreshGitStatus(input.cwd).pipe(Effect.ignore({ log: true }), Effect.as(result)),
2797 }),
2798 ),
2799 [WS_METHODS.gitRunStackedAction]: (input) =>
2800 Stream.callback<GitActionProgressEvent, GitManagerServiceError>((queue) =>
2801 gitWorkflow
2802 .runStackedAction(input, {
2803 actionId: input.actionId,
2804 progressReporter: {
2805 publish: (event) => Queue.offer(queue, event).pipe(Effect.asVoid),
2806 },
2807 })
2808 .pipe(
2809 Effect.matchCauseEffect({
2810 onFailure: (cause) => Queue.failCause(queue, cause),
2811 onSuccess: (result) =>
2812 (input.threadId === undefined
2813 ? Effect.void
2814 : linkCreatedPullRequest({
2815 threadId: input.threadId,
2816 result,
2817 commandId: serverCommandId("pr-created-link"),
2818 }).pipe(
2819 Effect.provideService(Orchestrator.OrchestratorV2, orchestrationEngine),
2820 Effect.provideService(ProjectService.ProjectService, projectService),
2821 )
2822 ).pipe(
2823 Effect.andThen(
2824 refreshPushedPullRequests(input, result).pipe(
2825 Effect.provideService(Orchestrator.OrchestratorV2, orchestrationEngine),
2826 Effect.provideService(ProjectStore.ProjectStoreV2, projectStore),
2827 Effect.provideService(
2828 PullRequestService.PullRequestService,
2829 pullRequests,
2830 ),
2831 ),
2832 ),
2833 Effect.andThen(refreshGitStatus(input.cwd)),
2834 Effect.andThen(Queue.end(queue).pipe(Effect.asVoid)),
2835 ),
2836 }),
2837 ),
2838 ),
2839 [WS_METHODS.gitResolvePullRequest]: (input) => gitWorkflow.resolvePullRequest(input),
2840 [WS_METHODS.gitPreparePullRequestThread]: (input) =>
2841 gitWorkflow
2842 .preparePullRequestThread(input)
2843 .pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
2844 [WS_METHODS.vcsListRefs]: (input) => gitWorkflow.listRefs(input),
2845 [WS_METHODS.vcsCreateWorktree]: (input) =>
2846 gitWorkflow.createWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
2847 [WS_METHODS.vcsRemoveWorktree]: (input) =>
2848 gitWorkflow.removeWorktree(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
2849 [WS_METHODS.vcsCreateRef]: (input) =>
2850 gitWorkflow.createRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
2851 [WS_METHODS.vcsSwitchRef]: (input) =>
2852 gitWorkflow.switchRef(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
2853 [WS_METHODS.vcsInit]: (input) =>
2854 vcsProvisioning.initRepository(input).pipe(Effect.tap(() => refreshGitStatus(input.cwd))),
2855 [WS_METHODS.reviewGetDiffPreview]: (input) => review.getDiffPreview(input),
2856 [WS_METHODS.reviewGetDiffFileContents]: (input) => review.getDiffFileContents(input),
2857 ...makeTerminalRpcHandlers(terminalManager),
2858 [WS_METHODS.subscribeScriptProgress]: (input) => scriptProgress.subscribe(input.threadId),
2859 [WS_METHODS.subscribeTerminalMetadata]: (_input) =>
2860 Stream.callback<TerminalMetadataStreamEvent>((queue) =>
2861 Effect.acquireRelease(
2862 terminalManager.subscribeMetadata((event) => Queue.offer(queue, event)),
2863 (unsubscribe) => Effect.sync(unsubscribe),
2864 ),
2865 ),
2866 [WS_METHODS.previewOpen]: (input) => previewManager.open(input),
2867 [WS_METHODS.previewNavigate]: (input) => previewManager.navigate(input),
2868 [WS_METHODS.previewResize]: (input) => previewManager.resize(input),
2869 [WS_METHODS.previewAdjust]: (input) => previewManager.adjust(input),
2870 [WS_METHODS.previewRefresh]: (input) => previewManager.refresh(input),
2871 [WS_METHODS.previewClose]: (input) => previewManager.close(input),
2872 [WS_METHODS.previewList]: (input) => previewManager.list(input),
2873 [WS_METHODS.previewClearProfile]: (input) => serverBrowser.clearProfile(input.profileId),
2874 [WS_METHODS.previewReportProfiles]: (input) => serverBrowser.reportProfiles(input),
2875 [WS_METHODS.previewReportStatus]: (input) => previewManager.reportStatus(input),
2876 [WS_METHODS.subscribePreviewEvents]: (_input) => previewManager.events,
2877 [WS_METHODS.deviceConfigure]: (input) => deviceService.configure(input),
2878 [WS_METHODS.deviceTestHost]: (input) => deviceService.testHost(input),
2879 [WS_METHODS.deviceList]: (input) =>
2880 input.inspectOnly && !input.updateTool
2881 ? deviceService.inspect
2882 : authorizeEffect(
2883 requiredScopeForDeviceList(input),
2884 input.updateTool
2885 ? deviceService.updateTool(input.updateTool)
2886 : input.retryHostId
2887 ? deviceService.retryHost(input.retryHostId)
2888 : deviceService.list,
2889 ),
2890 [WS_METHODS.deviceOpen]: (input) => deviceService.open(input),
2891 [WS_METHODS.deviceClose]: (input) => deviceService.close(input),
2892 [WS_METHODS.deviceShutdown]: (input) => deviceService.shutdown(input),
2893 [WS_METHODS.deviceDetail]: (input) => deviceService.detail(input),
2894 [WS_METHODS.deviceAction]: (input) => deviceService.action(input),
2895 [WS_METHODS.subscribeDeviceState]: (_input) => DeviceService.stateStream(deviceService),
2896 [WS_METHODS.subscribeDiscoveredLocalServers]: (input) =>
2897 Stream.callback<DiscoveredLocalServerList>((queue) =>
2898 Effect.gen(function* () {
2899 const configuredUrls = input.configuredUrls ?? [];
2900 yield* portDiscovery.retain;
2901 const initial = yield* portDiscovery.scan(configuredUrls);
2902 const initialScannedAt = DateTime.formatIso(yield* DateTime.now);
2903 yield* Queue.offer(queue, {
2904 servers: initial,
2905 scannedAt: initialScannedAt,
2906 configuredUrlProbing: true,
2907 });
2908 yield* portDiscovery.subscribe(
2909 { configuredUrls, initialSnapshot: initial },
2910 (servers) =>
2911 Effect.gen(function* () {
2912 const scannedAt = DateTime.formatIso(yield* DateTime.now);
2913 yield* Queue.offer(queue, {
2914 servers,
2915 scannedAt,
2916 configuredUrlProbing: true,
2917 });
2918 }),
2919 );
2920 }),
2921 ),
2922 [WS_METHODS.subscribeServerConfig]: (input) =>
2923 Stream.unwrap(
2924 Effect.gen(function* () {
2925 const usageLimitsCommand = input.usageLimitsCommand === true;
2926 const config = yield* loadServerConfig({ usageLimitsCommand });
2927 const keybindingsUpdates = keybindings.streamChanges.pipe(
2928 Stream.map((event) => ({
2929 version: 1 as const,
2930 type: "keybindingsUpdated" as const,
2931 payload: {
2932 keybindings: event.keybindings,
2933 issues: event.issues,
2934 },
2935 })),
2936 );
2937 const providerStatuses = Stream.zipLatestWith(
2938 // The registry stream carries changes only. Seed it with the current
2939 // providers so a source refresh that lands before any provider change
2940 // still pairs up and reaches the client.
2941 Stream.concat(
2942 Stream.fromEffect(providerRegistry.getProviders),
2943 providerRegistry.streamChanges,
2944 ),
2945 usageLimitSources.streamChanges.pipe(
2946 // Quota updates already have their own stream. Republish the model
2947 // catalog only when the set of providers offered the command changes.
2948 Stream.changesWith(
2949 usageLimitsCommand ? sameUsageLimitCommandCoverage : () => true,
2950 ),
2951 ),
2952 (providers, sources) =>
2953 usageLimitsCommand ? withUsageLimitsCommands(providers, sources) : providers,
2954 ).pipe(
2955 // Both sides replay their current value, so the first pairing normally
2956 // repeats the snapshot the client already holds. Compare against that
2957 // snapshot rather than dropping blindly: a refresh that landed between
2958 // the snapshot and the subscription still goes out.
2959 (updates) => Stream.concat(rpcInitialItems([config.providers]), updates),
2960 Stream.changesWith(
2961 (previous, next) => JSON.stringify(previous) === JSON.stringify(next),
2962 ),
2963 Stream.drop(1),
2964 Stream.map((providers) => ({
2965 version: 1 as const,
2966 type: "providerStatuses" as const,
2967 payload: { providers },
2968 })),
2969 Stream.debounce(Duration.millis(PROVIDER_STATUS_DEBOUNCE_MS)),
2970 );
2971 // The only source of published themes: the stream emits the
2972 // current set before any change, so the snapshot carrying it too
2973 // would just send every client the same array twice per connect.
2974 // Gated on the subscriber's capability flag because an
2975 // already-shipped client decodes this stream against the old
2976 // event union and its whole config subscription dies on an
2977 // unknown member.
2978 const environmentThemeUpdates =
2979 input.environmentThemes === true
2980 ? environmentTheme.streamChanges.pipe(
2981 Stream.map((themes) => ({
2982 version: 1 as const,
2983 type: "environmentThemesUpdated" as const,
2984 payload: { themes },
2985 })),
2986 )
2987 : Stream.empty;
2988 const usageLimitSourceUpdates =
2989 input.usageLimitSources === true
2990 ? usageLimitSources.streamChanges.pipe(
2991 Stream.map((sources) => ({
2992 version: 1 as const,
2993 type: "usageLimitSourcesUpdated" as const,
2994 payload: { sources },
2995 })),
2996 )
2997 : Stream.empty;
2998 const settingsUpdates = serverSettings.streamChanges.pipe(
2999 Stream.map((settings) => ServerSettings.redactServerSettingsForClient(settings)),
3000 Stream.map((settings) => ({
3001 version: 1 as const,
3002 type: "settingsUpdated" as const,
3003 payload: { settings },
3004 })),
3005 );
3006
3007 const liveUpdates = Stream.merge(
3008 keybindingsUpdates,
3009 Stream.merge(
3010 providerStatuses,
3011 Stream.merge(
3012 settingsUpdates,
3013 Stream.merge(environmentThemeUpdates, usageLimitSourceUpdates),
3014 ),
3015 ),
3016 );
3017
3018 return Stream.concat(
3019 rpcInitialItems([{ version: 1 as const, type: "snapshot" as const, config }]),
3020 withLateEditorConfig(config, liveUpdates, externalLauncher),
3021 );
3022 }),
3023 ),
3024 [WS_METHODS.subscribeServerLifecycle]: (_input) =>
3025 Stream.unwrap(
3026 Effect.gen(function* () {
3027 const liveBuffer = yield* Queue.unbounded<ServerLifecycleStreamEvent>();
3028 yield* Effect.forkScoped(
3029 lifecycleEvents.stream.pipe(
3030 Stream.runForEach((event) => Queue.offer(liveBuffer, event)),
3031 ),
3032 { startImmediately: true },
3033 );
3034 const snapshot = yield* lifecycleEvents.snapshot;
3035 const snapshotEvents = Array.from(snapshot.events).toSorted(
3036 (left, right) => left.sequence - right.sequence,
3037 );
3038 const liveEvents = Stream.fromQueue(liveBuffer).pipe(
3039 Stream.filter((event) => event.sequence > snapshot.sequence),
3040 );
3041 return Stream.concat(rpcInitialItems(snapshotEvents), liveEvents);
3042 }),
3043 ),
3044 [WS_METHODS.subscribeAuthAccess]: (_input) =>
3045 Stream.unwrap(
3046 Effect.gen(function* () {
3047 const initialSnapshot = yield* loadAuthAccessSnapshot();
3048 const revisionRef = yield* Ref.make(1);
3049 const accessChanges: Stream.Stream<
3050 PairingGrantStore.BootstrapCredentialChange | SessionStore.SessionCredentialChange
3051 > = Stream.merge(bootstrapCredentials.streamChanges, sessions.streamChanges);
3052
3053 const liveEvents: Stream.Stream<AuthAccessStreamEvent> = accessChanges.pipe(
3054 Stream.mapEffect((change) =>
3055 Ref.updateAndGet(revisionRef, (revision) => revision + 1).pipe(
3056 Effect.map((revision) =>
3057 toAuthAccessStreamEvent(change, revision, currentSessionId),
3058 ),
3059 ),
3060 ),
3061 );
3062
3063 return Stream.concat(
3064 rpcInitialItems([
3065 {
3066 version: 1 as const,
3067 revision: 1,
3068 type: "snapshot" as const,
3069 payload: initialSnapshot,
3070 },
3071 ]),
3072 liveEvents,
3073 );
3074 }),
3075 ),
3076 [WS_METHODS.subscribeBackgroundPolicy]: (_input) =>
3077 Stream.unwrap(
3078 Effect.map(backgroundPolicy.subscribe, ({ latest, changes }) =>
3079 Stream.concat(Stream.make(latest), changes),
3080 ),
3081 ),
3082 [WS_METHODS.subscribeResourceTelemetry]: (_input) =>
3083 Stream.unwrap(
3084 Effect.map(resourceTelemetry.subscribe, ({ latest, changes }) =>
3085 Stream.concat(Stream.make(latest), changes),
3086 ),
3087 ),
3088 });
3089 return handlers;
3090 }),
3091 );
3092
3093// A defect in a handler's effect fails only its own request. RpcServer's default
3094// sends a socket-level Defect frame instead, and the client ends every pending
3095// request on the socket with it. DefectReporter logs these defects.
3096export const WS_RPC_SERVER_OPTIONS = {
3097 disableTracing: true,
3098 disableFatalDefects: true,
3099} as const;
3100
3101export const layer = Layer.unwrap(
3102 Effect.gen(function* () {
3103 const previewAutomationBroker = yield* PreviewAutomationBroker.PreviewAutomationBroker;
3104 const serverBrowser = yield* ServerBrowser.ServerBrowser;
3105 const serverSelfUpdate = yield* ServerSelfUpdate.ServerSelfUpdate;
3106 const pullRequests = yield* PullRequestService.PullRequestService;
3107 const reviewGuides = yield* PullRequestReviewGuideService.PullRequestReviewGuideService;
3108 const sql = yield* SqlClient.SqlClient;
3109 return HttpRouter.add(
3110 "GET",
3111 "/ws",
3112 Effect.gen(function* () {
3113 const request = yield* HttpServerRequest.HttpServerRequest;
3114 const requestUrl = HttpServerRequest.toURL(request);
3115 if (Option.isNone(requestUrl) || !hasCompatibleOrchestrationProtocol(requestUrl.value)) {
3116 return HttpServerResponse.jsonUnsafe(
3117 {
3118 code: "orchestration_protocol_incompatible",
3119 message: `Update this client to one that supports orchestration protocol ${ORCHESTRATION_PROTOCOL_VERSION}.`,
3120 orchestrationProtocolVersion: ORCHESTRATION_PROTOCOL_VERSION,
3121 },
3122 { status: 426 },
3123 );
3124 }
3125 const serverAuth = yield* EnvironmentAuth.EnvironmentAuth;
3126 const sessions = yield* SessionStore.SessionStore;
3127 const session = yield* serverAuth.authenticateWebSocketUpgrade(request).pipe(
3128 Effect.catchIf(EnvironmentAuth.isServerAuthCredentialError, (error) =>
3129 failEnvironmentAuthInvalid(
3130 EnvironmentAuth.serverAuthCredentialReason(error),
3131 EnvironmentAuth.serverAuthDpopFailureReason(error),
3132 ),
3133 ),
3134 Effect.catchIf(EnvironmentAuth.isServerAuthInternalError, (error) =>
3135 failEnvironmentInternal("internal_error", error),
3136 ),
3137 );
3138 const clientOrigin = readClientConnectionOrigin(request);
3139 yield* sessions.recordClientConnection(session.sessionId, clientOrigin);
3140 const rpcWebSocketHttpEffect = yield* Effect.gen(function* () {
3141 const { protocol, httpEffect } = yield* RpcServer.makeProtocolWithHttpEffectWebsocket;
3142 yield* RpcServer.make(ServerWsRpcGroup, WS_RPC_SERVER_OPTIONS).pipe(
3143 Effect.provideService(RpcServer.Protocol, withTerminalOutputWindow(protocol)),
3144 Effect.provide(
3145 Layer.merge(RpcAuthorization.layer(session.scopes), rpcInstrumentationLayer),
3146 ),
3147 Effect.forkScoped,
3148 );
3149 // @effect-diagnostics-next-line returnEffectInGen:off
3150 return httpEffect;
3151 }).pipe(
3152 Effect.provide(
3153 layerWsRpc(session, clientOrigin, previewAutomationBroker, serverBrowser).pipe(
3154 Layer.provideMerge(RpcSerialization.layerJson),
3155 // Request fibers run in the handlers' context, so this reporter sees
3156 // their defects, not the rest of the server's.
3157 Layer.provide(DefectReporter.layer),
3158 Layer.provide(Layer.succeed(SqlClient.SqlClient, sql)),
3159 Layer.provide(AgentSessionScanner.layer),
3160 Layer.provide(ProviderMaintenanceRunner.layer),
3161 Layer.provide(Layer.succeed(ServerSelfUpdate.ServerSelfUpdate, serverSelfUpdate)),
3162 // One server-lifetime service means clients share the same PR caches, and a WS
3163 // mutation invalidates the HTTP diff cache that every client reads from.
3164 Layer.provide(Layer.succeed(PullRequestService.PullRequestService, pullRequests)),
3165 Layer.provide(OtlpEndpointCheck.layer),
3166 // Shared like the pull request service, so concurrent generations single-flight.
3167 Layer.provide(
3168 Layer.succeed(
3169 PullRequestReviewGuideService.PullRequestReviewGuideService,
3170 reviewGuides,
3171 ),
3172 ),
3173 Layer.provide(
3174 SourceControlDiscovery.layer.pipe(
3175 Layer.provide(
3176 SourceControlProviderRegistry.layer.pipe(
3177 Layer.provide(SourceControlBuiltInDrivers.layer),
3178 Layer.provideMerge(GitVcsDriver.layer),
3179 Layer.provide(
3180 VcsDriverRegistry.layer.pipe(Layer.provide(VcsProjectConfig.layer)),
3181 ),
3182 ),
3183 ),
3184 ),
3185 ),
3186 ),
3187 ),
3188 );
3189 return yield* Effect.acquireUseRelease(
3190 sessions.markConnected(session.sessionId),
3191 () =>
3192 Effect.raceFirst(
3193 rpcWebSocketHttpEffect,
3194 sessions.awaitInvalidation(session.sessionId).pipe(
3195 Effect.as(HttpServerResponse.empty()),
3196 Effect.catchTags({
3197 SessionCredentialVerificationError: (error) =>
3198 failEnvironmentInternal("internal_error", error),
3199 }),
3200 ),
3201 ),
3202 () => sessions.markDisconnected(session.sessionId),
3203 );
3204 }).pipe(
3205 Effect.catchTags({
3206 EnvironmentAuthInvalidError: HttpServerRespondable.toResponse,
3207 EnvironmentInternalError: HttpServerRespondable.toResponse,
3208 }),
3209 ),
3210 );
3211 }),
3212);