authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-06-19 21:09:22-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-06-19 21:09:22-07:00
log5c1ac922574340be4560eacf928237929497d11d
treeabd561916532dc5a36af7fd548811c2e98868d66
parent8bfdacf206608c0e14b1c5471dfa1f482a452acc
signature Signed by PGP key B5690EEEBB952194

[codex] normalize server process and preview Effect services (#3191)

Co-authored-by: codex <codex@users.noreply.github.com>

8 files changed, 297 insertions(+), 272 deletions(-)

apps/server/src/environment/Layers/ServerEnvironmentLabel.test.ts+5-5
......@@ -5,15 +5,15 @@ import * as Layer from "effect/Layer";
55import { HostProcessHostname, HostProcessPlatform } from "@t3tools/shared/hostProcess";
66import { vi } from "vite-plus/test";
77
8import { ProcessRunner, ProcessSpawnError, type ProcessRunnerShape } from "../../processRunner.ts";
8import * as ProcessRunner from "../../processRunner.ts";
99import { resolveServerEnvironmentLabel } from "./ServerEnvironmentLabel.ts";
1010import { ChildProcessSpawner } from "effect/unstable/process";
1111
12const runMock = vi.fn<ProcessRunnerShape["run"]>();
12const runMock = vi.fn<ProcessRunner.ProcessRunner["Service"]["run"]>();
1313
1414const ProcessRunnerTest = Layer.succeed(
15 ProcessRunner,
16 ProcessRunner.of({
15 ProcessRunner.ProcessRunner,
16 ProcessRunner.ProcessRunner.of({
1717 run: (input) => runMock(input),
1818 }),
1919);
......@@ -136,7 +136,7 @@ describe("resolveServerEnvironmentLabel", () => {
136136 Effect.gen(function* () {
137137 runMock.mockReturnValueOnce(
138138 Effect.fail(
139 new ProcessSpawnError({
139 new ProcessRunner.ProcessSpawnError({
140140 command: "scutil",
141141 args: ["--get", "ComputerName"],
142142 cause: new Error("spawn scutil ENOENT"),
apps/server/src/mcp/PreviewAutomationBroker.test.ts+8-8
......@@ -37,7 +37,7 @@ const makeOwner = (overrides: Partial<PreviewAutomationOwner> = {}): PreviewAuto
3737it.effect("atomically registers a connected owner and correlates its response", () =>
3838 Effect.scoped(
3939 Effect.gen(function* () {
40 const broker = yield* PreviewAutomationBroker.__testing.make;
40 const broker = yield* PreviewAutomationBroker.make;
4141 const requests = yield* broker.connect(makeOwner());
4242 yield* Stream.runForEach(requests, (request) =>
4343 broker.respond({
......@@ -61,7 +61,7 @@ it.effect("atomically registers a connected owner and correlates its response",
6161
6262it.effect("rejects calls when no focused owner exists", () =>
6363 Effect.gen(function* () {
64 const broker = yield* PreviewAutomationBroker.__testing.make;
64 const broker = yield* PreviewAutomationBroker.make;
6565 const error = yield* broker
6666 .invoke<void>({ scope, operation: "status", input: {} })
6767 .pipe(Effect.flip);
......@@ -72,7 +72,7 @@ it.effect("rejects calls when no focused owner exists", () =>
7272it.effect("routes interactive commands to a hidden durable browser host", () =>
7373 Effect.scoped(
7474 Effect.gen(function* () {
75 const broker = yield* PreviewAutomationBroker.__testing.make;
75 const broker = yield* PreviewAutomationBroker.make;
7676 const requests = yield* broker.connect(
7777 makeOwner({ clientId: "client-hidden", tabId: "tab-hidden" }),
7878 );
......@@ -89,7 +89,7 @@ it.effect("routes interactive commands to a hidden durable browser host", () =>
8989it.effect("lets the browser host resolve an active tab that has not been reported yet", () =>
9090 Effect.scoped(
9191 Effect.gen(function* () {
92 const broker = yield* PreviewAutomationBroker.__testing.make;
92 const broker = yield* PreviewAutomationBroker.make;
9393 const requests = yield* broker.connect(makeOwner({ tabId: null }));
9494 let routedTabId: string | undefined;
9595 yield* Stream.runForEach(requests, (request) => {
......@@ -108,7 +108,7 @@ it.effect("lets the browser host resolve an active tab that has not been reporte
108108it.effect("preserves current owner metadata when its request stream reconnects", () =>
109109 Effect.scoped(
110110 Effect.gen(function* () {
111 const broker = yield* PreviewAutomationBroker.__testing.make;
111 const broker = yield* PreviewAutomationBroker.make;
112112 const firstRequests = yield* broker.connect(makeOwner());
113113 yield* Stream.runDrain(firstRequests).pipe(Effect.forkScoped);
114114 yield* broker.reportOwner(makeOwner({ tabId: "tab-current", visible: true }));
......@@ -131,7 +131,7 @@ it.effect("preserves current owner metadata when its request stream reconnects",
131131it.effect("ignores stale owner cleanup after the client moves to another thread", () =>
132132 Effect.scoped(
133133 Effect.gen(function* () {
134 const broker = yield* PreviewAutomationBroker.__testing.make;
134 const broker = yield* PreviewAutomationBroker.make;
135135 const requests = yield* broker.connect(makeOwner());
136136 yield* Stream.runForEach(requests, (request) =>
137137 broker.respond({ requestId: request.requestId, ok: true }),
......@@ -152,7 +152,7 @@ it.effect("ignores stale owner cleanup after the client moves to another thread"
152152it.effect("fails requests assigned to a browser stream when that stream reconnects", () =>
153153 Effect.scoped(
154154 Effect.gen(function* () {
155 const broker = yield* PreviewAutomationBroker.__testing.make;
155 const broker = yield* PreviewAutomationBroker.make;
156156 const _requests = yield* broker.connect(makeOwner());
157157 const pending = yield* broker
158158 .invoke<void>({ scope, operation: "status", input: {} })
......@@ -170,7 +170,7 @@ it.effect("fails requests assigned to a browser stream when that stream reconnec
170170it.effect("falls back to an older connected owner when a newer report is not connected", () =>
171171 Effect.scoped(
172172 Effect.gen(function* () {
173 const broker = yield* PreviewAutomationBroker.__testing.make;
173 const broker = yield* PreviewAutomationBroker.make;
174174 const requests = yield* broker.connect(makeOwner({ clientId: "client-connected" }));
175175 yield* Stream.runForEach(requests, (request) =>
176176 broker.respond({ requestId: request.requestId, ok: true, result: "connected" }),
apps/server/src/mcp/PreviewAutomationBroker.ts+22-33
......@@ -35,34 +35,28 @@ export interface PreviewAutomationInvokeInput {
3535 readonly timeoutMs?: number;
3636}
3737
38export interface PreviewAutomationBrokerShape {
39 readonly connect: (
40 owner: PreviewAutomationOwner,
41 ) => Effect.Effect<Stream.Stream<PreviewAutomationRequest>>;
42 readonly reportOwner: (
43 owner: PreviewAutomationOwner,
44 ) => Effect.Effect<void, PreviewAutomationError>;
45 readonly clearOwner: (owner: PreviewAutomationOwnerIdentity) => Effect.Effect<void>;
46 readonly respond: (
47 response: PreviewAutomationResponse,
48 ) => Effect.Effect<void, PreviewAutomationError>;
49 readonly invoke: <A = unknown>(
50 request: PreviewAutomationInvokeInput,
51 ) => Effect.Effect<A, PreviewAutomationError>;
52}
53
5438export class PreviewAutomationBroker extends Context.Service<
5539 PreviewAutomationBroker,
56 PreviewAutomationBrokerShape
40 {
41 readonly connect: (
42 owner: PreviewAutomationOwner,
43 ) => Effect.Effect<Stream.Stream<PreviewAutomationRequest>>;
44 readonly reportOwner: (
45 owner: PreviewAutomationOwner,
46 ) => Effect.Effect<void, PreviewAutomationError>;
47 readonly clearOwner: (owner: PreviewAutomationOwnerIdentity) => Effect.Effect<void>;
48 readonly respond: (
49 response: PreviewAutomationResponse,
50 ) => Effect.Effect<void, PreviewAutomationError>;
51 readonly invoke: <A = unknown>(
52 request: PreviewAutomationInvokeInput,
53 ) => Effect.Effect<A, PreviewAutomationError>;
54 }
5755>()("t3/mcp/PreviewAutomationBroker") {}
5856
5957interface ClientConnection {
6058 readonly clientId: string;
61 readonly queue: Queue.Queue<
62 Parameters<PreviewAutomationBrokerShape["respond"]>[0] extends never
63 ? never
64 : import("@t3tools/contracts").PreviewAutomationRequest
65 >;
59 readonly queue: Queue.Queue<PreviewAutomationRequest>;
6660}
6761
6862interface PendingRequest {
......@@ -123,7 +117,7 @@ const makeResponseError = (
123117 }
124118};
125119
126const make = Effect.gen(function* PreviewAutomationBrokerMake() {
120export const make = Effect.gen(function* PreviewAutomationBrokerMake() {
127121 const state = yield* SynchronizedRef.make<BrokerState>({
128122 clients: new Map(),
129123 owners: new Map(),
......@@ -166,7 +160,7 @@ const make = Effect.gen(function* PreviewAutomationBrokerMake() {
166160 yield* Queue.shutdown(queue);
167161 });
168162
169 const connect: PreviewAutomationBrokerShape["connect"] = Effect.fn(
163 const connect: PreviewAutomationBroker["Service"]["connect"] = Effect.fn(
170164 "PreviewAutomationBroker.connect",
171165 )(function* (owner) {
172166 const clientId = owner.clientId;
......@@ -189,7 +183,7 @@ const make = Effect.gen(function* PreviewAutomationBrokerMake() {
189183 return Stream.fromQueue(queue).pipe(Stream.ensuring(disconnect(clientId, queue)));
190184 });
191185
192 const reportOwner: PreviewAutomationBrokerShape["reportOwner"] = Effect.fn(
186 const reportOwner: PreviewAutomationBroker["Service"]["reportOwner"] = Effect.fn(
193187 "PreviewAutomationBroker.reportOwner",
194188 )(function* (owner) {
195189 yield* SynchronizedRef.update(state, (current) => {
......@@ -199,7 +193,7 @@ const make = Effect.gen(function* PreviewAutomationBrokerMake() {
199193 });
200194 });
201195
202 const clearOwner: PreviewAutomationBrokerShape["clearOwner"] = Effect.fn(
196 const clearOwner: PreviewAutomationBroker["Service"]["clearOwner"] = Effect.fn(
203197 "PreviewAutomationBroker.clearOwner",
204198 )(function* (owner) {
205199 yield* SynchronizedRef.update(state, (current) => {
......@@ -217,7 +211,7 @@ const make = Effect.gen(function* PreviewAutomationBrokerMake() {
217211 });
218212 });
219213
220 const respond: PreviewAutomationBrokerShape["respond"] = Effect.fn(
214 const respond: PreviewAutomationBroker["Service"]["respond"] = Effect.fn(
221215 "PreviewAutomationBroker.respond",
222216 )(function* (response) {
223217 const pending = yield* SynchronizedRef.modify(state, (current) => {
......@@ -243,7 +237,7 @@ const make = Effect.gen(function* PreviewAutomationBrokerMake() {
243237 });
244238
245239 const invoke = Effect.fn("PreviewAutomationBroker.invoke")(function* <A = unknown>(
246 input: Parameters<PreviewAutomationBrokerShape["invoke"]>[0],
240 input: Parameters<PreviewAutomationBroker["Service"]["invoke"]>[0],
247241 ): Effect.fn.Return<A, PreviewAutomationError> {
248242 const current = yield* SynchronizedRef.get(state);
249243 const candidates = Array.from(current.owners.values())
......@@ -317,8 +311,3 @@ const make = Effect.gen(function* PreviewAutomationBrokerMake() {
317311}).pipe(Effect.withSpan("PreviewAutomationBroker.make"));
318312
319313export const layer = Layer.effect(PreviewAutomationBroker, make);
320
321/** Exposed for tests. */
322export const __testing = {
323 make,
324};
apps/server/src/preview/Manager.ts+98-95
......@@ -28,33 +28,30 @@ import {
2828 normalizePreviewUrl,
2929 PreviewUrlNormalizationError,
3030} from "@t3tools/shared/preview";
31import {
32 Context,
33 DateTime,
34 Effect,
35 Layer,
36 PubSub,
37 type Scope,
38 Stream,
39 SynchronizedRef,
40} from "effect";
41
42export interface PreviewManagerShape {
43 readonly open: (input: PreviewOpenInput) => Effect.Effect<PreviewSessionSnapshot, PreviewError>;
44 readonly navigate: (
45 input: PreviewNavigateInput,
46 ) => Effect.Effect<PreviewSessionSnapshot, PreviewError>;
47 readonly reportStatus: (input: PreviewReportStatusInput) => Effect.Effect<void, PreviewError>;
48 readonly refresh: (input: PreviewRefreshInput) => Effect.Effect<void, PreviewError>;
49 readonly close: (input: PreviewCloseInput) => Effect.Effect<void, PreviewError>;
50 readonly list: (input: PreviewListInput) => Effect.Effect<PreviewListResult>;
51 readonly events: Stream.Stream<PreviewEvent>;
52 readonly subscribeEvents: Effect.Effect<PubSub.Subscription<PreviewEvent>, never, Scope.Scope>;
53}
31import * as Context from "effect/Context";
32import * as DateTime from "effect/DateTime";
33import * as Effect from "effect/Effect";
34import * as Layer from "effect/Layer";
35import * as PubSub from "effect/PubSub";
36import * as Scope from "effect/Scope";
37import * as Stream from "effect/Stream";
38import * as SynchronizedRef from "effect/SynchronizedRef";
5439
55export class PreviewManager extends Context.Service<PreviewManager, PreviewManagerShape>()(
56 "t3/preview/Manager/PreviewManager",
57) {}
40export class PreviewManager extends Context.Service<
41 PreviewManager,
42 {
43 readonly open: (input: PreviewOpenInput) => Effect.Effect<PreviewSessionSnapshot, PreviewError>;
44 readonly navigate: (
45 input: PreviewNavigateInput,
46 ) => Effect.Effect<PreviewSessionSnapshot, PreviewError>;
47 readonly reportStatus: (input: PreviewReportStatusInput) => Effect.Effect<void, PreviewError>;
48 readonly refresh: (input: PreviewRefreshInput) => Effect.Effect<void, PreviewError>;
49 readonly close: (input: PreviewCloseInput) => Effect.Effect<void, PreviewError>;
50 readonly list: (input: PreviewListInput) => Effect.Effect<PreviewListResult>;
51 readonly events: Stream.Stream<PreviewEvent>;
52 readonly subscribeEvents: Effect.Effect<PubSub.Subscription<PreviewEvent>, never, Scope.Scope>;
53 }
54>()("t3/preview/Manager/PreviewManager") {}
5855
5956interface PreviewSessionState {
6057 readonly threadId: string;
......@@ -127,7 +124,7 @@ const buildIdleSnapshot = (input: {
127124 updatedAt: input.updatedAt,
128125});
129126
130const make = Effect.gen(function* PreviewManagerMake() {
127export const make = Effect.gen(function* PreviewManagerMake() {
131128 const stateRef = yield* SynchronizedRef.make<ManagerState>(initialState);
132129 // Unbounded PubSub is fine here — events are tiny and we don't want to
133130 // block publishers if a subscriber is slow. WS clients backpressure on
......@@ -184,38 +181,40 @@ const make = Effect.gen(function* PreviewManagerMake() {
184181 );
185182 };
186183
187 const open: PreviewManagerShape["open"] = Effect.fn("PreviewManager.open")(function* (input) {
188 const tabId = newPreviewTabId();
189 const updatedAt = yield* currentIsoTimestamp;
190 const snapshot = input.url
191 ? buildLoadingSnapshot({
184 const open: PreviewManager["Service"]["open"] = Effect.fn("PreviewManager.open")(
185 function* (input) {
186 const tabId = newPreviewTabId();
187 const updatedAt = yield* currentIsoTimestamp;
188 const snapshot = input.url
189 ? buildLoadingSnapshot({
190 threadId: input.threadId,
191 tabId,
192 url: yield* normalizeUrl(input.url),
193 title: "",
194 updatedAt,
195 })
196 : buildIdleSnapshot({ threadId: input.threadId, tabId, updatedAt });
197 yield* SynchronizedRef.update(stateRef, (state) => {
198 const sessions = new Map(state.sessions);
199 sessions.set(compositeKey(input.threadId, tabId), {
192200 threadId: input.threadId,
193201 tabId,
194 url: yield* normalizeUrl(input.url),
195 title: "",
196 updatedAt,
197 })
198 : buildIdleSnapshot({ threadId: input.threadId, tabId, updatedAt });
199 yield* SynchronizedRef.update(stateRef, (state) => {
200 const sessions = new Map(state.sessions);
201 sessions.set(compositeKey(input.threadId, tabId), {
202 snapshot,
203 });
204 return { sessions };
205 });
206 yield* PubSub.publish(eventsPubSub, {
207 type: "opened",
202208 threadId: input.threadId,
203209 tabId,
210 createdAt: snapshot.updatedAt,
204211 snapshot,
205212 });
206 return { sessions };
207 });
208 yield* PubSub.publish(eventsPubSub, {
209 type: "opened",
210 threadId: input.threadId,
211 tabId,
212 createdAt: snapshot.updatedAt,
213 snapshot,
214 });
215 return snapshot;
216 });
213 return snapshot;
214 },
215 );
217216
218 const navigate: PreviewManagerShape["navigate"] = Effect.fn("PreviewManager.navigate")(
217 const navigate: PreviewManager["Service"]["navigate"] = Effect.fn("PreviewManager.navigate")(
219218 function* (input) {
220219 const url = yield* normalizeUrl(input.url);
221220 return yield* mutateExistingSession(
......@@ -250,7 +249,7 @@ const make = Effect.gen(function* PreviewManagerMake() {
250249 },
251250 );
252251
253 const reportStatus: PreviewManagerShape["reportStatus"] = Effect.fn(
252 const reportStatus: PreviewManager["Service"]["reportStatus"] = Effect.fn(
254253 "PreviewManager.reportStatus",
255254 )(function* (input) {
256255 yield* mutateExistingSession(
......@@ -294,7 +293,7 @@ const make = Effect.gen(function* PreviewManagerMake() {
294293 );
295294 });
296295
297 const refresh: PreviewManagerShape["refresh"] = Effect.fn("PreviewManager.refresh")(
296 const refresh: PreviewManager["Service"]["refresh"] = Effect.fn("PreviewManager.refresh")(
298297 function* (input) {
299298 // Verify the session exists; the desktop bridge handles the actual reload
300299 // and will report progress back via `reportStatus`. No event emitted.
......@@ -304,50 +303,54 @@ const make = Effect.gen(function* PreviewManagerMake() {
304303 },
305304 );
306305
307 const close: PreviewManagerShape["close"] = Effect.fn("PreviewManager.close")(function* (input) {
308 const createdAt = yield* currentIsoTimestamp;
309 const events = yield* SynchronizedRef.modify(stateRef, (state) => {
310 const eventsToEmit: PreviewEvent[] = [];
311 const sessions = new Map(state.sessions);
312 const targets = input.tabId
313 ? [state.sessions.get(compositeKey(input.threadId, input.tabId))].filter(
314 (entry): entry is PreviewSessionState => entry !== undefined,
315 )
316 : sessionsForThread(state, input.threadId);
317 for (const target of targets) {
318 sessions.delete(compositeKey(target.threadId, target.tabId));
319 eventsToEmit.push({
320 type: "closed",
321 threadId: target.threadId,
322 tabId: target.tabId,
323 createdAt,
306 const close: PreviewManager["Service"]["close"] = Effect.fn("PreviewManager.close")(
307 function* (input) {
308 const createdAt = yield* currentIsoTimestamp;
309 const events = yield* SynchronizedRef.modify(stateRef, (state) => {
310 const eventsToEmit: PreviewEvent[] = [];
311 const sessions = new Map(state.sessions);
312 const targets = input.tabId
313 ? [state.sessions.get(compositeKey(input.threadId, input.tabId))].filter(
314 (entry): entry is PreviewSessionState => entry !== undefined,
315 )
316 : sessionsForThread(state, input.threadId);
317 for (const target of targets) {
318 sessions.delete(compositeKey(target.threadId, target.tabId));
319 eventsToEmit.push({
320 type: "closed",
321 threadId: target.threadId,
322 tabId: target.tabId,
323 createdAt,
324 });
325 }
326 if (eventsToEmit.length === 0) {
327 return [eventsToEmit, state] as const;
328 }
329 return [eventsToEmit, { sessions }] as const;
330 });
331 if (events.length > 0) {
332 yield* Effect.forEach(events, (event) => PubSub.publish(eventsPubSub, event), {
333 discard: true,
324334 });
325335 }
326 if (eventsToEmit.length === 0) {
327 return [eventsToEmit, state] as const;
328 }
329 return [eventsToEmit, { sessions }] as const;
330 });
331 if (events.length > 0) {
332 yield* Effect.forEach(events, (event) => PubSub.publish(eventsPubSub, event), {
333 discard: true,
334 });
335 }
336 });
336 },
337 );
337338
338 const list: PreviewManagerShape["list"] = Effect.fn("PreviewManager.list")(function* (input) {
339 return yield* SynchronizedRef.get(stateRef).pipe(
340 Effect.map(
341 (state): PreviewListResult => ({
342 sessions: sessionsForThread(state, input.threadId)
343 .map((s) => s.snapshot)
344 .toSorted((a, b) => a.updatedAt.localeCompare(b.updatedAt)),
345 }),
346 ),
347 );
348 });
339 const list: PreviewManager["Service"]["list"] = Effect.fn("PreviewManager.list")(
340 function* (input) {
341 return yield* SynchronizedRef.get(stateRef).pipe(
342 Effect.map(
343 (state): PreviewListResult => ({
344 sessions: sessionsForThread(state, input.threadId)
345 .map((s) => s.snapshot)
346 .toSorted((a, b) => a.updatedAt.localeCompare(b.updatedAt)),
347 }),
348 ),
349 );
350 },
351 );
349352
350 return {
353 return PreviewManager.of({
351354 open,
352355 navigate,
353356 reportStatus,
......@@ -356,7 +359,7 @@ const make = Effect.gen(function* PreviewManagerMake() {
356359 list,
357360 events,
358361 subscribeEvents: PubSub.subscribe(eventsPubSub),
359 } satisfies PreviewManagerShape;
362 });
360363}).pipe(Effect.withSpan("PreviewManager.make"));
361364
362365export const layer = Layer.effect(PreviewManager, make);
apps/server/src/preview/PortScanner.ts+54-49
......@@ -15,30 +15,36 @@ import { ThreadId, type DiscoveredLocalServer } from "@t3tools/contracts";
1515import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
1616import * as Net from "@t3tools/shared/Net";
1717import { LSOF_LOCAL_HOST_TOKENS } from "@t3tools/shared/preview";
18import { Cause, Context, Duration, Effect, Layer, Ref, Schedule, Scope } from "effect";
18import * as Cause from "effect/Cause";
19import * as Context from "effect/Context";
20import * as Duration from "effect/Duration";
21import * as Effect from "effect/Effect";
22import * as Layer from "effect/Layer";
23import * as Ref from "effect/Ref";
24import * as Schedule from "effect/Schedule";
25import * as Scope from "effect/Scope";
1926
20import { ProcessRunner } from "../processRunner.ts";
27import * as ProcessRunner from "../processRunner.ts";
2128
22export interface PortDiscoveryShape {
23 readonly scan: () => Effect.Effect<ReadonlyArray<DiscoveredLocalServer>>;
24 readonly subscribe: (
25 listener: (servers: ReadonlyArray<DiscoveredLocalServer>) => Effect.Effect<void>,
26 ) => Effect.Effect<void, never, Scope.Scope>;
27 readonly retain: Effect.Effect<void, never, Scope.Scope>;
28 readonly registerTerminalProcesses: (input: {
29 readonly threadId: string;
30 readonly terminalId: string;
31 readonly processIds: ReadonlyArray<number>;
32 }) => Effect.Effect<void>;
33 readonly unregisterTerminal: (input: {
34 readonly threadId: string;
35 readonly terminalId: string;
36 }) => Effect.Effect<void>;
37}
38
39export class PortDiscovery extends Context.Service<PortDiscovery, PortDiscoveryShape>()(
40 "t3/preview/PortScanner/PortDiscovery",
41) {}
29export class PortDiscovery extends Context.Service<
30 PortDiscovery,
31 {
32 readonly scan: () => Effect.Effect<ReadonlyArray<DiscoveredLocalServer>>;
33 readonly subscribe: (
34 listener: (servers: ReadonlyArray<DiscoveredLocalServer>) => Effect.Effect<void>,
35 ) => Effect.Effect<void, never, Scope.Scope>;
36 readonly retain: Effect.Effect<void, never, Scope.Scope>;
37 readonly registerTerminalProcesses: (input: {
38 readonly threadId: string;
39 readonly terminalId: string;
40 readonly processIds: ReadonlyArray<number>;
41 }) => Effect.Effect<void>;
42 readonly unregisterTerminal: (input: {
43 readonly threadId: string;
44 readonly terminalId: string;
45 }) => Effect.Effect<void>;
46 }
47>()("t3/preview/PortScanner/PortDiscovery") {}
4248
4349export const COMMON_DEV_PORTS: ReadonlyArray<number> = Object.freeze([
4450 3000, 3001, 3333, 4173, 4200, 4321, 5000, 5173, 5174, 5175, 5500, 8000, 8080, 8081, 8888, 9000,
......@@ -180,9 +186,9 @@ const serversEqual = (
180186 return true;
181187};
182188
183const make = Effect.gen(function* PortDiscoveryMake() {
189export const make = Effect.gen(function* PortDiscoveryMake() {
184190 const net = yield* Net.NetService;
185 const processRunner = yield* ProcessRunner;
191 const processRunner = yield* ProcessRunner.ProcessRunner;
186192 const hostPlatform = yield* HostProcessPlatform;
187193 const stateRef = yield* Ref.make<ScannerState>({
188194 lastSnapshot: [],
......@@ -296,14 +302,14 @@ const make = Effect.gen(function* PortDiscoveryMake() {
296302 }
297303 });
298304
299 const retain: PortDiscoveryShape["retain"] = Effect.acquireRelease(acquireRetention(), () =>
305 const retain: PortDiscovery["Service"]["retain"] = Effect.acquireRelease(acquireRetention(), () =>
300306 Ref.update(stateRef, (state) => ({
301307 ...state,
302308 retainCount: Math.max(0, state.retainCount - 1),
303309 })),
304310 );
305311
306 const subscribe: PortDiscoveryShape["subscribe"] = Effect.fn("PortDiscovery.subscribe")(
312 const subscribe: PortDiscovery["Service"]["subscribe"] = Effect.fn("PortDiscovery.subscribe")(
307313 (listener) =>
308314 Effect.acquireRelease(
309315 Ref.update(stateRef, (state) => ({
......@@ -319,29 +325,28 @@ const make = Effect.gen(function* PortDiscoveryMake() {
319325 ),
320326 );
321327
322 const registerTerminalProcesses: PortDiscoveryShape["registerTerminalProcesses"] = Effect.fn(
323 "PortDiscovery.registerTerminalProcesses",
324 )(function* (input) {
325 const owner = {
326 threadId: ThreadId.make(input.threadId),
327 terminalId: input.terminalId,
328 };
329 const processIds = new Set(
330 input.processIds.filter((processId) => Number.isInteger(processId) && processId > 0),
331 );
332 yield* Ref.update(stateRef, (state) => {
333 const terminalProcesses = new Map(state.terminalProcesses);
334 const key = terminalOwnerKey(owner);
335 if (processIds.size === 0) {
336 terminalProcesses.delete(key);
337 } else {
338 terminalProcesses.set(key, { owner, processIds });
339 }
340 return { ...state, terminalProcesses };
328 const registerTerminalProcesses: PortDiscovery["Service"]["registerTerminalProcesses"] =
329 Effect.fn("PortDiscovery.registerTerminalProcesses")(function* (input) {
330 const owner = {
331 threadId: ThreadId.make(input.threadId),
332 terminalId: input.terminalId,
333 };
334 const processIds = new Set(
335 input.processIds.filter((processId) => Number.isInteger(processId) && processId > 0),
336 );
337 yield* Ref.update(stateRef, (state) => {
338 const terminalProcesses = new Map(state.terminalProcesses);
339 const key = terminalOwnerKey(owner);
340 if (processIds.size === 0) {
341 terminalProcesses.delete(key);
342 } else {
343 terminalProcesses.set(key, { owner, processIds });
344 }
345 return { ...state, terminalProcesses };
346 });
341347 });
342 });
343348
344 const unregisterTerminal: PortDiscoveryShape["unregisterTerminal"] = Effect.fn(
349 const unregisterTerminal: PortDiscovery["Service"]["unregisterTerminal"] = Effect.fn(
345350 "PortDiscovery.unregisterTerminal",
346351 )(function* (input) {
347352 yield* Ref.update(stateRef, (state) => {
......@@ -351,13 +356,13 @@ const make = Effect.gen(function* PortDiscoveryMake() {
351356 });
352357 });
353358
354 return {
359 return PortDiscovery.of({
355360 scan: scanOnce,
356361 subscribe,
357362 retain,
358363 registerTerminalProcesses,
359364 unregisterTerminal,
360 } satisfies PortDiscoveryShape;
365 });
361366}).pipe(Effect.withSpan("PortDiscovery.make"));
362367
363368export const layer = Layer.effect(PortDiscovery, make);
apps/server/src/process/externalLauncher.ts+19-25
......@@ -22,7 +22,8 @@ import * as FileSystem from "effect/FileSystem";
2222import * as Layer from "effect/Layer";
2323import * as Option from "effect/Option";
2424import * as Path from "effect/Path";
25import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
25import * as ChildProcess from "effect/unstable/process/ChildProcess";
26import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
2627
2728// ==============================
2829// Definitions
......@@ -282,30 +283,23 @@ const resolveAvailableEditors = Effect.fn("externalLauncher.resolveAvailableEdit
282283 return yield* buildAvailableEditors(platform, env);
283284});
284285
285/**
286 * ExternalLauncherShape - Service API for browser and editor launch actions.
287 */
288export interface ExternalLauncherShape {
289 readonly resolveAvailableEditors: () => Effect.Effect<ReadonlyArray<EditorId>>;
290 /**
291 * Launch a URL target in the default browser.
292 */
293 readonly launchBrowser: (target: string) => Effect.Effect<void, ExternalLauncherError>;
294
295 /**
296 * Launch a workspace path in a selected editor integration.
297 *
298 * Launches the editor as a detached process so server startup is not blocked.
299 */
300 readonly launchEditor: (input: LaunchEditorInput) => Effect.Effect<void, ExternalLauncherError>;
301}
302
303286/**
304287 * ExternalLauncher - Service tag for browser/editor launch operations.
305288 */
306export class ExternalLauncher extends Context.Service<ExternalLauncher, ExternalLauncherShape>()(
307 "t3/process/externalLauncher",
308) {}
289export class ExternalLauncher extends Context.Service<
290 ExternalLauncher,
291 {
292 readonly resolveAvailableEditors: () => Effect.Effect<ReadonlyArray<EditorId>>;
293 /** Launch a URL target in the default browser. */
294 readonly launchBrowser: (target: string) => Effect.Effect<void, ExternalLauncherError>;
295 /**
296 * Launch a workspace path in a selected editor integration.
297 *
298 * Launches the editor as a detached process so server startup is not blocked.
299 */
300 readonly launchEditor: (input: LaunchEditorInput) => Effect.Effect<void, ExternalLauncherError>;
301 }
302>()("t3/process/externalLauncher") {}
309303
310304// ==============================
311305// Implementations
......@@ -397,7 +391,7 @@ const launchEditorProcess = Effect.fn("externalLauncher.launchEditorProcess")(fu
397391 );
398392});
399393
400const make = Effect.gen(function* () {
394export const make = Effect.gen(function* () {
401395 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
402396 const fileSystem = yield* FileSystem.FileSystem;
403397 const path = yield* Path.Path;
......@@ -410,7 +404,7 @@ const make = Effect.gen(function* () {
410404 Effect.provideService(Path.Path, path),
411405 );
412406
413 return {
407 return ExternalLauncher.of({
414408 resolveAvailableEditors: () => provideCommandResolutionServices(resolveAvailableEditors()),
415409 launchBrowser: (target) =>
416410 launchBrowser(target).pipe(
......@@ -424,7 +418,7 @@ const make = Effect.gen(function* () {
424418 ),
425419 ),
426420 ),
427 } satisfies ExternalLauncherShape;
421 });
428422});
429423
430424export const layer = Layer.effect(ExternalLauncher, make);
apps/server/src/processRunner.test.ts+11-17
......@@ -11,14 +11,7 @@ import { ChildProcessSpawner } from "effect/unstable/process";
1111import { HostProcessEnvironment, HostProcessPlatform } from "@t3tools/shared/hostProcess";
1212import { SpawnExecutableResolution } from "@t3tools/shared/shell";
1313
14import {
15 isWindowsCommandNotFound,
16 ProcessOutputLimitError,
17 ProcessRunner,
18 ProcessTimeoutError,
19 layer as ProcessRunnerLive,
20 type ProcessRunInput,
21} from "./processRunner.ts";
14import * as ProcessRunner from "./processRunner.ts";
2215
2316type ChildProcessCommand = {
2417 readonly command: string;
......@@ -68,15 +61,16 @@ function makeSpawner(
6861}
6962
7063const runWith =
71 (spawner: ChildProcessSpawner.ChildProcessSpawner["Service"]) => (input: ProcessRunInput) =>
72 Effect.service(ProcessRunner).pipe(
64 (spawner: ChildProcessSpawner.ChildProcessSpawner["Service"]) =>
65 (input: ProcessRunner.ProcessRunInput) =>
66 Effect.service(ProcessRunner.ProcessRunner).pipe(
7367 Effect.flatMap((runner) =>
7468 runner.run({
7569 ...input,
7670 }),
7771 ),
7872 Effect.provide(
79 ProcessRunnerLive.pipe(
73 ProcessRunner.layer.pipe(
8074 Layer.provide(Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner)),
8175 ),
8276 ),
......@@ -112,12 +106,12 @@ describe("runProcess", () => {
112106 return makeHandle({ stdout: "service ok" });
113107 }),
114108 );
115 const layer = ProcessRunnerLive.pipe(
109 const layer = ProcessRunner.layer.pipe(
116110 Layer.provide(Layer.succeed(ChildProcessSpawner.ChildProcessSpawner, spawner)),
117111 );
118112
119113 return Effect.gen(function* () {
120 const runner = yield* ProcessRunner;
114 const runner = yield* ProcessRunner.ProcessRunner;
121115 const result = yield* runner.run({
122116 command: "fake",
123117 args: ["--service"],
......@@ -175,7 +169,7 @@ describe("runProcess", () => {
175169 maxOutputBytes: 128,
176170 }).pipe(Effect.flip);
177171
178 expect(error).toBeInstanceOf(ProcessOutputLimitError);
172 expect(error).toBeInstanceOf(ProcessRunner.ProcessOutputLimitError);
179173 }),
180174 );
181175
......@@ -200,7 +194,7 @@ describe("runProcess", () => {
200194 timeout: "2 seconds",
201195 }).pipe(Effect.flip);
202196
203 expect(error).toBeInstanceOf(ProcessOutputLimitError);
197 expect(error).toBeInstanceOf(ProcessRunner.ProcessOutputLimitError);
204198 }),
205199 );
206200
......@@ -285,7 +279,7 @@ describe("runProcess", () => {
285279 yield* TestClock.adjust(Duration.millis(50));
286280 const error = yield* Fiber.join(errorFiber);
287281
288 expect(error).toBeInstanceOf(ProcessTimeoutError);
282 expect(error).toBeInstanceOf(ProcessRunner.ProcessTimeoutError);
289283 }),
290284 );
291285
......@@ -324,7 +318,7 @@ describe("runProcess", () => {
324318describe("isWindowsCommandNotFound", () => {
325319 it.effect("matches the localized German cmd.exe error text", () =>
326320 Effect.gen(function* () {
327 const isCommandNotFound = yield* isWindowsCommandNotFound(
321 const isCommandNotFound = yield* ProcessRunner.isWindowsCommandNotFound(
328322 1,
329323 "wird nicht als interner oder externer Befehl, betriebsfahiges Programm oder Batch-Datei erkannt",
330324 ).pipe(Effect.provideService(HostProcessPlatform, "win32"));
apps/server/src/processRunner.ts+80-40
......@@ -1,13 +1,14 @@
1import * as Data from "effect/Data";
21import * as Context from "effect/Context";
32import * as Duration from "effect/Duration";
43import * as Effect from "effect/Effect";
54import * as Layer from "effect/Layer";
65import * as Option from "effect/Option";
76import * as PlatformError from "effect/PlatformError";
7import * as Schema from "effect/Schema";
88import * as Scope from "effect/Scope";
99import * as Stream from "effect/Stream";
10import { ChildProcess, ChildProcessSpawner } from "effect/unstable/process";
10import * as ChildProcess from "effect/unstable/process/ChildProcess";
11import * as ChildProcessSpawner from "effect/unstable/process/ChildProcessSpawner";
1112import { HostProcessPlatform } from "@t3tools/shared/hostProcess";
1213import { resolveSpawnCommand } from "@t3tools/shared/shell";
1314import {
......@@ -42,42 +43,82 @@ export interface ProcessRunOutput {
4243 readonly stderrTruncated: boolean;
4344}
4445
45export class ProcessSpawnError extends Data.TaggedError("ProcessSpawnError")<{
46 readonly command: string;
47 readonly args: ReadonlyArray<string>;
48 readonly cwd?: string | undefined;
49 readonly cause: unknown;
50}> {}
46const ProcessInvocationFields = {
47 command: Schema.String,
48 args: Schema.Array(Schema.String),
49 cwd: Schema.optional(Schema.String),
50};
5151
52export class ProcessStdinError extends Data.TaggedError("ProcessStdinError")<{
52const formatProcessInvocation = (input: {
5353 readonly command: string;
5454 readonly args: ReadonlyArray<string>;
5555 readonly cwd?: string | undefined;
56 readonly cause: unknown;
57}> {}
56}): string => {
57 const command = [input.command, ...input.args].join(" ");
58 return input.cwd === undefined ? `'${command}'` : `'${command}' in '${input.cwd}'`;
59};
5860
59export class ProcessOutputLimitError extends Data.TaggedError("ProcessOutputLimitError")<{
60 readonly command: string;
61 readonly args: ReadonlyArray<string>;
62 readonly cwd?: string | undefined;
63 readonly stream: "stdout" | "stderr";
64 readonly maxBytes: number;
65}> {}
61export class ProcessSpawnError extends Schema.TaggedErrorClass<ProcessSpawnError>()(
62 "ProcessSpawnError",
63 {
64 ...ProcessInvocationFields,
65 cause: Schema.Defect(),
66 },
67) {
68 override get message(): string {
69 return `Failed to spawn process ${formatProcessInvocation(this)}`;
70 }
71}
6672
67export class ProcessReadError extends Data.TaggedError("ProcessReadError")<{
68 readonly command: string;
69 readonly args: ReadonlyArray<string>;
70 readonly cwd?: string | undefined;
71 readonly stream: "stdout" | "stderr" | "exitCode";
72 readonly cause: unknown;
73}> {}
73export class ProcessStdinError extends Schema.TaggedErrorClass<ProcessStdinError>()(
74 "ProcessStdinError",
75 {
76 ...ProcessInvocationFields,
77 cause: Schema.Defect(),
78 },
79) {
80 override get message(): string {
81 return `Failed to write stdin for process ${formatProcessInvocation(this)}`;
82 }
83}
7484
75export class ProcessTimeoutError extends Data.TaggedError("ProcessTimeoutError")<{
76 readonly command: string;
77 readonly args: ReadonlyArray<string>;
78 readonly cwd?: string | undefined;
79 readonly timeoutMs: number;
80}> {}
85export class ProcessOutputLimitError extends Schema.TaggedErrorClass<ProcessOutputLimitError>()(
86 "ProcessOutputLimitError",
87 {
88 ...ProcessInvocationFields,
89 stream: Schema.Literals(["stdout", "stderr"]),
90 maxBytes: Schema.Number,
91 },
92) {
93 override get message(): string {
94 return `Process ${formatProcessInvocation(this)} ${this.stream} exceeded ${this.maxBytes} bytes`;
95 }
96}
97
98export class ProcessReadError extends Schema.TaggedErrorClass<ProcessReadError>()(
99 "ProcessReadError",
100 {
101 ...ProcessInvocationFields,
102 stream: Schema.Literals(["stdout", "stderr", "exitCode"]),
103 cause: Schema.Defect(),
104 },
105) {
106 override get message(): string {
107 return `Failed to read ${this.stream} for process ${formatProcessInvocation(this)}`;
108 }
109}
110
111export class ProcessTimeoutError extends Schema.TaggedErrorClass<ProcessTimeoutError>()(
112 "ProcessTimeoutError",
113 {
114 ...ProcessInvocationFields,
115 timeoutMs: Schema.Number,
116 },
117) {
118 override get message(): string {
119 return `Process ${formatProcessInvocation(this)} timed out after ${this.timeoutMs}ms`;
120 }
121}
81122
82123export type ProcessRunError =
83124 | ProcessSpawnError
......@@ -86,13 +127,12 @@ export type ProcessRunError =
86127 | ProcessReadError
87128 | ProcessTimeoutError;
88129
89export interface ProcessRunnerShape {
90 readonly run: (input: ProcessRunInput) => Effect.Effect<ProcessRunOutput, ProcessRunError>;
91}
92
93export class ProcessRunner extends Context.Service<ProcessRunner, ProcessRunnerShape>()(
94 "t3/processRunner",
95) {}
130export class ProcessRunner extends Context.Service<
131 ProcessRunner,
132 {
133 readonly run: (input: ProcessRunInput) => Effect.Effect<ProcessRunOutput, ProcessRunError>;
134 }
135>()("t3/processRunner") {}
96136
97137const DEFAULT_TIMEOUT = "60 seconds";
98138const DEFAULT_MAX_OUTPUT_BYTES = 8 * 1024 * 1024;
......@@ -332,10 +372,10 @@ const runProcessCore = Effect.fn("processRunner.runProcessCore")(function* (
332372 } satisfies ProcessRunOutput;
333373});
334374
335export const make = Effect.fn("makeProcessRunner")(function* () {
375export const make = Effect.fn("ProcessRunner.make")(function* () {
336376 const spawner = yield* ChildProcessSpawner.ChildProcessSpawner;
337377
338 const run: ProcessRunnerShape["run"] = (input) =>
378 const run: ProcessRunner["Service"]["run"] = (input) =>
339379 finalizeRunProcess(runProcessCore(spawner, input), input);
340380
341381 return ProcessRunner.of({