Merge pull request #327 from PetalNet/refactor/console-effect-ser-idiom

refactor(console): extract terminal domain effect + call domain effects in-process (drop internal fetch, getRequestEvent, runRemote)

Author: lishaduck

apps/console/src/lib/command-palette.remote.ts

@@ -1,6 +1,8 @@
-import { getRequestEvent } from "$app/server";
 import { publicConfig } from "$lib/config";
-import { searchMockPalette, type PaletteSearchResponse } from "$lib/data/palette";
+import { searchMockPalette } from "$lib/data/palette";
+import { searchPalette } from "$lib/server/domain/palette/service";
+import { currentPrincipal } from "$lib/server/domain/principal";
+import { ConsoleDomain } from "$lib/server/domain/service";
 import { Effect, Schema } from "effect";
 import { Query } from "svelte-effect-runtime";
 
@@ -13,21 +15,11 @@ const input = Schema.Struct({
  * request; SvelteKit owns transport, validation, serialization, and cancellation semantics.
  */
 export const searchCommandPalette = Query(input, ({ query: text }) =>
-	Effect.promise(async () => {
+	Effect.gen(function* () {
 		if (publicConfig.dataMode === "mock") return searchMockPalette(text);
-
-		const event = getRequestEvent();
-		const headers = new Headers({ accept: "application/json", origin: event.url.origin });
-		for (const name of ["authorization", "cookie"] as const) {
-			const value = event.request.headers.get(name);
-			if (value) headers.set(name, value);
-		}
-		const base = publicConfig.consoleApiBase ?? `${event.url.origin}/api/v1`;
-		const response = await event.fetch(
-			`${base}/palette/search?q=${encodeURIComponent(text)}&limit=24`,
-			{ headers },
-		);
-		if (!response.ok) throw new Error(`Palette search returned ${String(response.status)}`);
-		return (await response.json()) as PaletteSearchResponse;
+		const domain = yield* ConsoleDomain;
+		const services = yield* domain.services;
+		const principal = yield* currentPrincipal;
+		return yield* searchPalette(services, principal, text, 24);
 	}),
 );

apps/console/src/lib/components/CommandPalette.svelte

@@ -1,9 +1,9 @@
 <script lang="ts">
 	import { goto } from "$app/navigation";
 	import { searchCommandPalette } from "$lib/command-palette.remote";
-	import { runRemote } from "$lib/rpc/browser";
 	import type { PaletteItem, PaletteKind, PaletteSearchResponse } from "$lib/data/palette";
 	import { visibleNav } from "$lib/nav";
+	import { Effect } from "effect";
 	import Icon from "./Icon.svelte";
 	import ModalSurface from "./ModalSurface.svelte";
 
@@ -107,7 +107,7 @@
 		const current = ++requestId;
 		timer = setTimeout(() => void (async () => {
 			try {
-				const result = await runRemote(searchCommandPalette({ query: queryText }));
+				const result = await Effect.runPromise(searchCommandPalette({ query: queryText }));
 				if (current === requestId) remote = result;
 			} catch {
 				if (current === requestId) failed = true;

apps/console/src/lib/data/cost.remote.ts

@@ -23,10 +23,12 @@ export const compareCost = Query(costComparisonRequestSchema, (request) =>
 		const domain = yield* ConsoleDomain;
 		const services = yield* domain.services;
 		const principal = yield* currentPrincipal;
-		return yield* Effect.tryPromise({
-			try: () => compareCostPair(services.db.app, principal.scopes, request, services.costMeter),
-			catch: (cause) => cause,
-		}).pipe(
+		return yield* compareCostPair(
+			services.db.app,
+			principal.scopes,
+			request,
+			services.costMeter,
+		).pipe(
 			Effect.catch((cause) =>
 				cause instanceof CostComparisonUnavailableError
 					? HttpError("ServiceUnavailable", cause.message)

apps/console/src/lib/data/library.ts

@@ -393,16 +393,22 @@ function mapLibraryItem(item: ApiLibraryItem, hold?: string): LibraryItemView {
 	};
 }
 
+type LiveLibraryResults = readonly [
+	PromiseSettledResult<ApiEnvelope<ApiLibraryItem>>,
+	PromiseSettledResult<ApiEnvelope<ApiLibraryLink>>,
+	PromiseSettledResult<ApiEnvelope<ApiLibraryHold>>,
+	PromiseSettledResult<ApiEnvelope<ApiLibraryCuration>>,
+	PromiseSettledResult<ApiEnvelope<ApiLibraryCapability>>,
+];
+
 /** Map the scope-filtered Rev3 read surface; optional sources fail independently and honestly. */
-export async function readLiveLibrary(fetchFn: typeof fetch = fetch): Promise<LibraryData> {
-	const [itemsResult, linksResult, holdsResult, curationResult, capabilitiesResult] =
-		await Promise.allSettled([
-			readAllPages<ApiLibraryItem>("/library/items?limit=1000", fetchFn),
-			readAllPages<ApiLibraryLink>("/library/links?limit=1000", fetchFn),
-			readAllPages<ApiLibraryHold>("/library/holds?limit=1000", fetchFn),
-			readAllPages<ApiLibraryCuration>("/library/curation?limit=1000", fetchFn),
-			readAllPages<ApiLibraryCapability>("/library/capabilities?limit=1000", fetchFn),
-		]);
+export function assembleLiveLibrary([
+	itemsResult,
+	linksResult,
+	holdsResult,
+	curationResult,
+	capabilitiesResult,
+]: LiveLibraryResults): LibraryData {
 	if (itemsResult.status === "rejected") return liveEmptyLibrary;
 	const holds = holdsResult.status === "fulfilled" ? holdsResult.value.items : [];
 	const holdByItem = new Map(
@@ -481,3 +487,14 @@ export async function readLiveLibrary(fetchFn: typeof fetch = fetch): Promise<Li
 		},
 	};
 }
+
+export async function readLiveLibrary(fetchFn: typeof fetch = fetch): Promise<LibraryData> {
+	const settled = await Promise.allSettled([
+		readAllPages<ApiLibraryItem>("/library/items?limit=1000", fetchFn),
+		readAllPages<ApiLibraryLink>("/library/links?limit=1000", fetchFn),
+		readAllPages<ApiLibraryHold>("/library/holds?limit=1000", fetchFn),
+		readAllPages<ApiLibraryCuration>("/library/curation?limit=1000", fetchFn),
+		readAllPages<ApiLibraryCapability>("/library/capabilities?limit=1000", fetchFn),
+	]);
+	return assembleLiveLibrary(settled);
+}

apps/console/src/lib/data/work-settlement.remote.ts

@@ -1,32 +1,23 @@
-import { getRequestEvent, query } from "$app/server";
-import type { WorkSettlementSnapshot } from "$lib/api/types";
 import { publicConfig } from "$lib/config";
 import { mockWorkSettlement } from "$lib/data/work-settlement";
-import { error } from "@sveltejs/kit";
+import { currentPrincipal } from "$lib/server/domain/principal";
+import { readWorkSettlement } from "$lib/server/domain/reads/work-settlement";
+import { ConsoleDomain } from "$lib/server/domain/service";
+import { Effect } from "effect";
+import { Query } from "svelte-effect-runtime";
 
 function isMock(): boolean {
 	return publicConfig.dataMode === "mock";
 }
 
-function forwardedHeaders(): Headers {
-	const incoming = getRequestEvent().request.headers;
-	const headers = new Headers({ accept: "application/json" });
-	for (const name of ["authorization", "cookie", "x-dev-principal"]) {
-		const value = incoming.get(name);
-		if (value) headers.set(name, value);
-	}
-	return headers;
-}
-
 /** One caller-scoped RPC powers Work's settle strip and Library's task-history projection. */
-export const getWorkSettlement = query(async (): Promise<WorkSettlementSnapshot> => {
-	if (isMock()) return mockWorkSettlement();
-	const event = getRequestEvent();
-	const base = publicConfig.consoleApiBase ?? `${event.url.origin}/api/v1`;
-	const response = await event.fetch(`${base}/work/settlement`, {
-		headers: forwardedHeaders(),
-	});
-	if (!response.ok)
-		error(response.status, `Work settlement source returned ${String(response.status)}`);
-	return (await response.json()) as WorkSettlementSnapshot;
-});
+export const getWorkSettlement = Query(
+	Effect.gen(function* () {
+		if (isMock()) return mockWorkSettlement();
+		const domain = yield* ConsoleDomain;
+		const services = yield* domain.services;
+		const principal = yield* currentPrincipal;
+		if (!services.tracker) return yield* Effect.die(new Error("Tracker is unavailable"));
+		return yield* readWorkSettlement(services.tracker, principal.scopes);
+	}),
+);

apps/console/src/lib/operations.remote.ts

@@ -22,7 +22,7 @@ import type {
 	TaskItem,
 	WorkerItem,
 } from "$lib/api/types";
-import { executeOpPlane } from "$lib/server/api/console-api";
+import { executeOpPlane } from "$lib/server/domain/commands/op-plane";
 import { listDashboards } from "$lib/server/domain/dashboard/store";
 import { consumeOpRateLimit } from "$lib/server/domain/op-rate-limit";
 import { currentPrincipal } from "$lib/server/domain/principal";
@@ -37,6 +37,7 @@ import {
 	readTasks as readTasksCore,
 } from "$lib/server/domain/reads/tracker-reads";
 import { ConsoleDomain } from "$lib/server/domain/service";
+import { readTerminalAccess as readTerminalAccessCore } from "$lib/server/domain/terminal/service";
 import { Effect } from "effect";
 import { Command, Query } from "svelte-effect-runtime";
 
@@ -108,7 +109,8 @@ export const readPlaneRemote = Query("unchecked", (plane: ReadPlane) =>
 				zookie: principal.zookie,
 			} satisfies Me;
 		if (plane === "health")
-			return yield* Effect.tryPromise(async () => {
+			// The bus-health probe is a single scoped lake read; a fault is a defect, not a caller error.
+			return yield* Effect.promise(async () => {
 				const rows = await services.db.admin<{ bus_heartbeat_at: string | null }[]>`
 					select max(received_at)::text as bus_heartbeat_at
 					from events where type = 'console.bus.health'`;
@@ -120,33 +122,31 @@ export const readPlaneRemote = Query("unchecked", (plane: ReadPlane) =>
 				};
 			});
 		if (plane === "roster") {
-			const result = yield* Effect.tryPromise(() =>
-				readRosterCore(services.db.app, services.tracker, principal.scopes),
-			);
-			return { ...result, items: result.items.map((item) => flattenRosterItem(item as never)) };
+			const result = yield* readRosterCore(services.db.app, services.tracker, principal.scopes);
+			return { ...result, items: result.items.map((item) => flattenRosterItem(item)) };
 		}
-		if (plane === "executors")
-			return yield* Effect.tryPromise(() => readExecutorsCore(services.db.app, principal.scopes));
+		if (plane === "executors") return yield* readExecutorsCore(services.db.app, principal.scopes);
 		if (plane === "tasks" || plane === "leases") {
 			if (!services.tracker)
 				return yield* Effect.die(new Error("Tracker read adapter is unavailable"));
-			return plane === "tasks"
+			return yield* plane === "tasks"
 				? readTasksCore(services.tracker, principal.scopes)
 				: readLeasesCore(services.tracker, principal.scopes);
 		}
 		if (plane === "dashboards")
-			return yield* Effect.tryPromise(() =>
+			// The canonical plane reads dashboards with a fixed limit and no cursor, so the only modeled
+			// DashboardError (a bad cursor) is unreachable here — treat it as a defect, keeping this
+			// shared read's channel empty for every consumer.
+			return yield* Effect.orDie(
 				listDashboards(services.db.app, principal.scopes, services.cursorSecret, { limit: 100 }),
 			);
 		if (plane === "catalog")
-			return yield* Effect.tryPromise(() =>
-				readEntity(services.db.app, principal.scopes, "registry", { limit: 1_000 }),
-			);
+			return yield* readEntity(services.db.app, principal.scopes, "registry", { limit: 1_000 });
 		const kind = projectedKinds[plane];
 		if (kind === "attention") {
-			const envelope = yield* Effect.tryPromise(() =>
-				readTypedEntity(services.db.app, principal.scopes, kind, { limit: 1_000 }),
-			);
+			const envelope = yield* readTypedEntity(services.db.app, principal.scopes, kind, {
+				limit: 1_000,
+			});
 			// Attention items carry an operating lane; a caller only sees items for lanes it holds.
 			return {
 				...envelope,
@@ -158,11 +158,9 @@ export const readPlaneRemote = Query("unchecked", (plane: ReadPlane) =>
 			};
 		}
 
-		return yield* Effect.tryPromise(() =>
-			kind === "subscription"
-				? readTypedEntity(services.db.app, principal.scopes, kind, { limit: 1_000 })
-				: readEntity(services.db.app, principal.scopes, kind, { limit: 1_000 }),
-		);
+		return yield* kind === "subscription"
+			? readTypedEntity(services.db.app, principal.scopes, kind, { limit: 1_000 })
+			: readEntity(services.db.app, principal.scopes, kind, { limit: 1_000 });
 	}),
 );
 
@@ -171,9 +169,16 @@ export const runStructuredQuery = Query("unchecked", (request: StructuredQuery)
 		const domain = yield* ConsoleDomain;
 		const services = yield* domain.services;
 		const principal = yield* currentPrincipal;
-		return yield* Effect.tryPromise(() =>
-			runStructured(services.db.app, principal.scopes, request),
-		);
+		return yield* runStructured(services.db.app, principal.scopes, request);
+	}),
+);
+
+export const readTerminalAccessRemote = Query(
+	Effect.gen(function* () {
+		const domain = yield* ConsoleDomain;
+		const services = yield* domain.services;
+		const principal = yield* currentPrincipal;
+		return yield* readTerminalAccessCore(services, principal);
 	}),
 );
 
@@ -203,25 +208,26 @@ export const executeNamedOp = Command(
 						retryable: true,
 					},
 				} satisfies OpResult;
-			// The one authoritative command plane: catalog lookup, arg validation, authz, proposal
-			// posture, audit intent/outcome, and internal adapters — identical to POST /api/v1/op.
-			const { body } = yield* Effect.tryPromise(() =>
-				executeOpPlane(
-					services,
-					services.monitor,
-					{
-						schema_version: 1,
-						id: input.id ?? crypto.randomUUID(),
-						op: input.op,
-						args: input.args,
-						...(input.reason !== undefined ? { reason: input.reason } : {}),
-						...(input.task_id !== undefined ? { task_id: input.task_id } : {}),
-						dry_run: input.dry_run ?? false,
-					},
-					principal,
-				),
+			// The one authoritative command plane, composed in process from the domain: catalog lookup,
+			// arg validation, authz, proposal posture, audit intent/outcome, and internal adapters —
+			// identical to POST /api/v1/op, which now runs the SAME domain effect at its HTTP edge.
+			const { body } = yield* executeOpPlane(
+				services,
+				services.monitor,
+				{
+					schema_version: 1,
+					id: input.id ?? crypto.randomUUID(),
+					op: input.op,
+					args: input.args,
+					...(input.reason !== undefined ? { reason: input.reason } : {}),
+					...(input.task_id !== undefined ? { task_id: input.task_id } : {}),
+					dry_run: input.dry_run ?? false,
+				},
+				principal,
 			);
-			return body as unknown as OpResult;
+			// The command plane returns the loosely-typed HTTP op envelope; narrow it to its documented
+			// OpResult shape (a genuine, unavoidable narrowing of the shared envelope, not a mock cast).
+			return body as OpResult;
 		}),
 );