1import { ArchiveError, ORCHESTRATION_V2_WS_METHODS, type ThreadId } from "@t3tools/contracts";
2import * as Effect from "effect/Effect";
3import * as Option from "effect/Option";
4import * as SubscriptionRef from "effect/SubscriptionRef";
5import type { HttpClient } from "effect/http";
6import { Atom } from "effect/reactivity";
7
8import {
9 createEnvironmentRpcCommand,
10 createEnvironmentQueryAtomFamily,
11 createEnvironmentRpcQueryAtomFamily,
12 createEnvironmentRpcSubscriptionAtomFamily,
13 createEnvironmentCommand,
14} from "./runtime.ts";
15import type { EnvironmentRegistry } from "../connection/registry.ts";
16import * as EnvironmentSupervisor from "../connection/supervisor.ts";
17import * as RemoteEnvironmentAuthorization from "../authorization/service.ts";
18import * as ManagedRelayDpopSigner from "../relay/managedRelay.ts";
19import * as HubConnection from "../connection/hub.ts";
20import {
21 request,
22 EnvironmentRpcUnavailableError,
23 type EnvironmentRpcInput,
24} from "../rpc/client.ts";
25import { fetchEnvironmentThreadTranscript } from "./threadTranscriptHttp.ts";
26
27export function createOrchestrationEnvironmentAtoms<R, E>(
28 runtime: Atom.AtomRuntime<EnvironmentRegistry | HttpClient.HttpClient | R, E>,
29) {
30 return {
31 threadTranscript: createEnvironmentCommand(runtime, {
32 label: "environment-data:orchestration:thread-transcript",
33 execute: (input: { readonly threadId: ThreadId }) =>
34 Effect.gen(function* () {
35 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
36 const prepared = yield* SubscriptionRef.get(supervisor.prepared);
37 if (Option.isNone(prepared)) {
38 return yield* new EnvironmentRpcUnavailableError({
39 environmentId: supervisor.target.environmentId,
40 message: "The source environment is not connected.",
41 });
42 }
43 return yield* fetchEnvironmentThreadTranscript({
44 prepared: prepared.value,
45 threadId: input.threadId,
46 signer: yield* Effect.serviceOption(ManagedRelayDpopSigner.ManagedRelayDpopSigner),
47 remoteAuthorization: yield* Effect.serviceOption(
48 RemoteEnvironmentAuthorization.RemoteEnvironmentAuthorization,
49 ),
50 });
51 }),
52 }),
53 v2: {
54 dispatchCommand: createEnvironmentRpcCommand(runtime, {
55 label: "environment-data:orchestration-v2:dispatch-command",
56 tag: ORCHESTRATION_V2_WS_METHODS.dispatchCommand,
57 }),
58 threadProjection: createEnvironmentRpcQueryAtomFamily(runtime, {
59 label: "environment-data:orchestration-v2:thread-projection",
60 tag: ORCHESTRATION_V2_WS_METHODS.getThreadProjection,
61 staleTimeMs: 0,
62 idleTtlMs: 0,
63 }),
64 shell: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
65 label: "environment-data:orchestration-v2:shell",
66 tag: ORCHESTRATION_V2_WS_METHODS.subscribeShell,
67 }),
68 thread: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
69 label: "environment-data:orchestration-v2:thread",
70 tag: ORCHESTRATION_V2_WS_METHODS.subscribeThread,
71 idleTtlMs: 0,
72 }),
73 },
74 turnDiff: createEnvironmentRpcQueryAtomFamily(runtime, {
75 label: "environment-data:orchestration:turn-diff",
76 tag: ORCHESTRATION_V2_WS_METHODS.getTurnDiff,
77 }),
78 workflowScript: createEnvironmentRpcQueryAtomFamily(runtime, {
79 label: "environment-data:orchestration:workflow-script",
80 tag: ORCHESTRATION_V2_WS_METHODS.getWorkflowScript,
81 // Scripts are immutable per run: cache generously.
82 staleTimeMs: 300_000,
83 idleTtlMs: 300_000,
84 }),
85 stopWorkflow: createEnvironmentRpcCommand(runtime, {
86 label: "environment-data:orchestration:stop-workflow",
87 tag: ORCHESTRATION_V2_WS_METHODS.stopWorkflow,
88 }),
89 // Keyed by the item revision, so a live row refetches as its output grows.
90 turnItem: createEnvironmentRpcQueryAtomFamily(runtime, {
91 label: "environment-data:orchestration:turn-item",
92 tag: ORCHESTRATION_V2_WS_METHODS.getTurnItem,
93 staleTimeMs: 60_000,
94 idleTtlMs: 60_000,
95 }),
96 fullThreadDiff: createEnvironmentRpcQueryAtomFamily(runtime, {
97 label: "environment-data:orchestration:full-thread-diff",
98 tag: ORCHESTRATION_V2_WS_METHODS.getFullThreadDiff,
99 }),
100 threadFind: createEnvironmentRpcQueryAtomFamily(runtime, {
101 label: "environment-data:orchestration:thread-find",
102 tag: ORCHESTRATION_V2_WS_METHODS.searchThread,
103 staleTimeMs: 0,
104 idleTtlMs: 0,
105 }),
106 threadFindProgressive: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
107 label: "environment-data:orchestration:thread-find-progressive",
108 tag: ORCHESTRATION_V2_WS_METHODS.searchThreadStream,
109 completeWhen: (result) => result.complete !== false,
110 idleTtlMs: 0,
111 }),
112 threadSearch: createEnvironmentRpcQueryAtomFamily(runtime, {
113 label: "environment-data:orchestration:thread-search",
114 tag: ORCHESTRATION_V2_WS_METHODS.searchThreads,
115 staleTimeMs: 30_000,
116 idleTtlMs: 60_000,
117 }),
118 archivedShellSnapshot: createEnvironmentQueryAtomFamily(runtime, {
119 label: "environment-data:orchestration:archived-shell-snapshot",
120 execute: (
121 input: EnvironmentRpcInput<typeof ORCHESTRATION_V2_WS_METHODS.getArchivedShellSnapshot>,
122 ) => request(ORCHESTRATION_V2_WS_METHODS.getArchivedShellSnapshot, input),
123 executeDisconnected: () =>
124 Effect.gen(function* () {
125 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
126 const unavailable = () =>
127 new EnvironmentRpcUnavailableError({
128 environmentId: supervisor.target.environmentId,
129 message: new ArchiveError({ reason: "unavailable" }).message,
130 });
131 const hub = yield* Effect.serviceOption(HubConnection.HubConnection);
132 if (Option.isNone(hub)) return yield* unavailable();
133 yield* hub.value.refreshInbox.pipe(Effect.ignore);
134 const entry = (yield* SubscriptionRef.get(hub.value.state)).inbox.environments.find(
135 (entry) => entry.environmentId === supervisor.target.environmentId,
136 );
137 if (!entry) return yield* unavailable();
138 return {
139 schemaVersion: entry.snapshot.schemaVersion,
140 snapshotSequence: entry.snapshot.snapshotSequence,
141 projects: entry.snapshot.projects,
142 threads: entry.snapshot.archivedThreads,
143 };
144 }),
145 }),
146 };
147}