.dockerignore

@@ -11,7 +11,6 @@
 !**/components.json
 
 !app/**
-!apps/agent/**
 !packages/**
 !public/**
 

Dockerfile

@@ -64,8 +64,6 @@ COPY --from=deps /deps/shoutrrr /usr/local/bin/shoutrrr
 
 COPY ./package.json ./bun.lock ./
 COPY ./packages/core/package.json ./packages/core/package.json
-COPY ./packages/contracts/package.json ./packages/contracts/package.json
-COPY ./apps/agent/package.json ./apps/agent/package.json
 
 RUN bun install --frozen-lockfile --ignore-scripts
 
@@ -88,14 +86,11 @@ WORKDIR /app
 
 COPY ./package.json ./bun.lock ./
 COPY ./packages/core/package.json ./packages/core/package.json
-COPY ./packages/contracts/package.json ./packages/contracts/package.json
-COPY ./apps/agent/package.json ./apps/agent/package.json
 RUN bun install --frozen-lockfile
 
 COPY . .
 
 RUN bun run build
-RUN bun build apps/agent/src/index.ts --outfile .output/agent/index.mjs --target bun
 
 FROM base AS production
 

app/server/modules/agents/__tests__/controller-agent-session.test.ts

@@ -1,86 +0,0 @@
-import { expect, mock, test } from "bun:test";
-import { createAgentMessage } from "@zerobyte/contracts/agent-protocol";
-import { createControllerAgentSession } from "../controller-agent-session";
-
-const createSocket = () => {
-	return {
-		data: { id: "connection-1", agentId: "local", organizationId: null, agentName: "Local Agent" },
-		send: mock(() => undefined),
-	} as unknown as Parameters<typeof createControllerAgentSession>[0];
-};
-
-test("close emits a synthetic backup.cancelled for a started backup", () => {
-	const onBackupCancelled = mock(() => undefined);
-	const session = createControllerAgentSession(createSocket(), {
-		onBackupCancelled,
-	});
-
-	session.handleMessage(
-		createAgentMessage("backup.started", {
-			jobId: "job-1",
-			scheduleId: "schedule-1",
-		}),
-	);
-
-	session.close();
-
-	expect(onBackupCancelled).toHaveBeenCalledTimes(1);
-	expect(onBackupCancelled).toHaveBeenCalledWith({
-		jobId: "job-1",
-		scheduleId: "schedule-1",
-		message:
-			"The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes.",
-	});
-});
-
-test("close does not emit a synthetic backup.cancelled after a terminal event", () => {
-	for (const testCase of [
-		{
-			jobId: "job-1",
-			scheduleId: "schedule-1",
-			terminalMessage: createAgentMessage("backup.completed", {
-				jobId: "job-1",
-				scheduleId: "schedule-1",
-				exitCode: 0,
-				result: null,
-			}),
-			expectedCancelledCalls: 0,
-		},
-		{
-			jobId: "job-2",
-			scheduleId: "schedule-2",
-			terminalMessage: createAgentMessage("backup.failed", {
-				jobId: "job-2",
-				scheduleId: "schedule-2",
-				error: "backup failed",
-			}),
-			expectedCancelledCalls: 0,
-		},
-		{
-			jobId: "job-3",
-			scheduleId: "schedule-3",
-			terminalMessage: createAgentMessage("backup.cancelled", {
-				jobId: "job-3",
-				scheduleId: "schedule-3",
-				message: "Backup was cancelled",
-			}),
-			expectedCancelledCalls: 1,
-		},
-	]) {
-		const onBackupCancelled = mock(() => undefined);
-		const session = createControllerAgentSession(createSocket(), {
-			onBackupCancelled,
-		});
-
-		session.handleMessage(
-			createAgentMessage("backup.started", {
-				jobId: testCase.jobId,
-				scheduleId: testCase.scheduleId,
-			}),
-		);
-		session.handleMessage(testCase.terminalMessage);
-		session.close();
-
-		expect(onBackupCancelled).toHaveBeenCalledTimes(testCase.expectedCancelledCalls);
-	}
-});

app/server/modules/agents/agent-protocol.ts

@@ -0,0 +1,98 @@
+import { z } from "zod";
+import { safeJsonParse } from "~/server/utils/json";
+
+const backupCommandSchema = z
+	.object({
+		type: z.literal("backup"),
+		payload: z.object({ scheduleId: z.string() }),
+	})
+	.strict();
+
+const agentReadySchema = z
+	.object({
+		type: z.literal("agent.ready"),
+		payload: z.object({ agentId: z.string() }),
+	})
+	.strict();
+
+const backupStartedSchema = z
+	.object({
+		type: z.literal("backup.started"),
+		payload: z.object({ scheduleId: z.string() }),
+	})
+	.strict();
+
+const controllerMessageSchema = z.discriminatedUnion("type", [backupCommandSchema]);
+const agentMessageSchema = z.discriminatedUnion("type", [agentReadySchema, backupStartedSchema]);
+
+export type BackupCommandPayload = z.infer<typeof backupCommandSchema>["payload"];
+export type ControllerMessage = z.infer<typeof controllerMessageSchema>;
+export type AgentMessage = z.infer<typeof agentMessageSchema>;
+
+type Brand<TValue, TBrand extends string> = TValue & {
+	readonly __brand: TBrand;
+};
+
+type MessageSender = {
+	send(message: string): unknown;
+};
+
+export type ControllerWireMessage = Brand<string, "ControllerWireMessage">;
+export type AgentWireMessage = Brand<string, "AgentWireMessage">;
+
+type PayloadForMessage<TMessage extends { type: string; payload: unknown }, TType extends TMessage["type"]> = Extract<
+	TMessage,
+	{ type: TType }
+>["payload"];
+
+const parseJsonMessage = (data: string) => safeJsonParse<unknown>(data);
+
+export const parseControllerMessage = (data: ControllerWireMessage) => {
+	const parsed = parseJsonMessage(data);
+	if (parsed === null) {
+		return null;
+	}
+
+	return controllerMessageSchema.safeParse(parsed);
+};
+
+export const parseAgentMessage = (data: string) => {
+	const parsed = parseJsonMessage(data);
+	if (parsed === null) {
+		return null;
+	}
+
+	return agentMessageSchema.safeParse(parsed);
+};
+
+export const createControllerMessage = <TType extends ControllerMessage["type"]>(
+	type: TType,
+	payload: PayloadForMessage<ControllerMessage, TType>,
+) =>
+	JSON.stringify(
+		controllerMessageSchema.parse({
+			type,
+			payload,
+		}),
+	) as ControllerWireMessage;
+
+export const createAgentMessage = <TType extends AgentMessage["type"]>(
+	type: TType,
+	payload: PayloadForMessage<AgentMessage, TType>,
+) =>
+	JSON.stringify(
+		agentMessageSchema.parse({
+			type,
+			payload,
+		}),
+	) as AgentWireMessage;
+
+export const sendControllerMessage = (target: MessageSender, message: ControllerWireMessage) => {
+	target.send(message);
+};
+
+export const sendAgentMessage = (target: MessageSender, message: AgentWireMessage) => {
+	target.send(message);
+};
+
+export type ControllerData = MessageEvent<ControllerWireMessage>;

app/server/modules/agents/agent-tokens.ts

@@ -1,12 +0,0 @@
-import { cryptoUtils } from "~/server/utils/crypto";
-
-export const deriveLocalAgentToken = async () => {
-	return cryptoUtils.deriveSecret("zerobyte:local-agent-token");
-};
-
-export const validateAgentToken = async (token: string) => {
-	const localToken = await deriveLocalAgentToken();
-	if (token === localToken) {
-		return { agentId: "local", organizationId: null, agentName: "local" };
-	}
-};

app/server/modules/agents/agents-manager.ts

@@ -1,348 +1,158 @@
-import { type ChildProcess, spawn } from "node:child_process";
+import { spawn } from "node:child_process";
-import { existsSync } from "node:fs";
 import path from "node:path";
-import { Effect, Exit, Ref, Scope } from "effect";
-import { logger } from "@zerobyte/core/node";
-import type {
-	BackupCancelPayload,
-	BackupCancelledPayload,
-	BackupCompletedPayload,
-	BackupFailedPayload,
-	BackupProgressPayload,
-	BackupRunPayload,
-	BackupStartedPayload,
-} from "@zerobyte/contracts/agent-protocol";
-import { config } from "../../core/config";
-import { validateAgentToken, deriveLocalAgentToken } from "./agent-tokens";
 import {
-	createControllerAgentSession,
+	createControllerMessage,
-	type AgentConnectionData,
+	parseAgentMessage,
-	type ControllerAgentSession,
+	sendControllerMessage,
-} from "./controller-agent-session";
+	type BackupCommandPayload,
+} from "./agent-protocol";
+import { logger } from "~/server/utils/logger";
 
-type AgentBackupEventContext = {
+type AgentConnectionData = {
-	agentId: string;
+	id: string;
-	agentName: string;
+	agentId?: string;
-	payload:
-		| BackupStartedPayload
-		| BackupProgressPayload
-		| BackupCompletedPayload
-		| BackupFailedPayload
-		| BackupCancelledPayload;
 };
 
-export type AgentBackupEventHandlers = {
+type AgentSocket = Bun.ServerWebSocket<AgentConnectionData>;
-	onBackupStarted?: (context: AgentBackupEventContext & { payload: BackupStartedPayload }) => void;
+type AgentServer = ReturnType<typeof Bun.serve<AgentConnectionData>>;
-	onBackupProgress?: (context: AgentBackupEventContext & { payload: BackupProgressPayload }) => void;
+const AGENT_SERVER_KEY = Symbol.for("zerobyte.agent-manager.server");
-	onBackupCompleted?: (context: AgentBackupEventContext & { payload: BackupCompletedPayload }) => void;
+const AGENT_SOCKETS_KEY = Symbol.for("zerobyte.agent-manager.sockets");
-	onBackupFailed?: (context: AgentBackupEventContext & { payload: BackupFailedPayload }) => void;
+
-	onBackupCancelled?: (context: AgentBackupEventContext & { payload: BackupCancelledPayload }) => void;
+const globalState = globalThis as typeof globalThis & {
+	[AGENT_SERVER_KEY]?: AgentServer;
+	[AGENT_SOCKETS_KEY]?: Map<string, AgentSocket>;
 };
 
-type AgentManagerRuntime = ReturnType<typeof createAgentManagerRuntime>;
+const getServer = () => globalState[AGENT_SERVER_KEY] ?? null;
-type AgentRuntimeState = {
+const getAgentSockets = () => {
-	agentManager: AgentManagerRuntime;
+	globalState[AGENT_SOCKETS_KEY] ??= new Map<string, AgentSocket>();
-	localAgent: ChildProcess | null;
+	return globalState[AGENT_SOCKETS_KEY];
+};
+const clearAgentSockets = () => {
+	getAgentSockets().clear();
 };
 
-type ProcessWithAgentRuntime = NodeJS.Process & {
+const setServer = (server: AgentServer | null) => {
-	__zerobyteAgentRuntime?: AgentRuntimeState;
+	if (server) {
-};
+		globalState[AGENT_SERVER_KEY] = server;
-
+		return;
-const getAgentRuntimeState = () => {
-	const runtimeProcess = process as ProcessWithAgentRuntime;
-	const existingRuntime = runtimeProcess.__zerobyteAgentRuntime;
-
-	if (existingRuntime) {
-		return existingRuntime;
 	}
 
-	const runtime = {
+	delete globalState[AGENT_SERVER_KEY];
-		agentManager: createAgentManagerRuntime(),
-		localAgent: null,
-	};
-
-	runtimeProcess.__zerobyteAgentRuntime = runtime;
-	return runtime;
 };
 
-const getAgentManagerRuntime = () => getAgentRuntimeState().agentManager;
+export const spawnLocalAgent = () => {
+	const wsUrl = `ws://localhost:3001`;
 
-export const spawnLocalAgent = async () => {
+	const agentEntryPoint = path.join(process.cwd(), "app", "server", "modules", "agents", "local-agent.ts");
-	await stopLocalAgent();
 
-	const sourceEntryPoint = path.join(process.cwd(), "apps", "agent", "src", "index.ts");
+	const localAgent = spawn("bun", ["run", agentEntryPoint], {
-	const productionEntryPoint = path.join(process.cwd(), ".output", "agent", "index.mjs");
-
-	if (config.__prod__ && !existsSync(productionEntryPoint)) {
-		throw new Error(`Local agent entrypoint not found at ${productionEntryPoint}`);
-	}
-
-	const agentEntryPoint = config.__prod__ ? productionEntryPoint : sourceEntryPoint;
-	const agentToken = await deriveLocalAgentToken();
-	const args = config.__prod__ ? ["run", agentEntryPoint] : ["run", "--watch", agentEntryPoint];
-
-	const runtime = getAgentRuntimeState();
-	const agentProcess = spawn("bun", args, {
 		env: {
 			...process.env,
-			ZEROBYTE_CONTROLLER_URL: "ws://localhost:3001",
+			ZEROBYTE_CONTROLLER_URL: wsUrl,
-			ZEROBYTE_AGENT_TOKEN: agentToken,
 		},
 		stdio: ["ignore", "pipe", "pipe"],
 	});
 
-	runtime.localAgent = agentProcess;
+	localAgent.stdout?.on("data", (data: Buffer) => {
-
-	agentProcess.stdout?.on("data", (data: Buffer) => {
 		const line = data.toString().trim();
 		if (line) logger.info(`[agent] ${line}`);
 	});
 
-	agentProcess.stderr?.on("data", (data: Buffer) => {
+	localAgent.stderr?.on("data", (data: Buffer) => {
 		const line = data.toString().trim();
 		if (line) logger.error(`[agent] ${line}`);
 	});
 
-	agentProcess.on("exit", (code, signal) => {
+	localAgent.on("exit", (code, signal) => {
-		if (runtime.localAgent === agentProcess) {
-			runtime.localAgent = null;
-		}
 		logger.info(`Agent process exited with code ${code} and signal ${signal}`);
 	});
 };
 
-export const stopLocalAgent = async () => {
+export const agentManager = {
-	const runtime = getAgentRuntimeState();
+	start: () => {
-	if (!runtime.localAgent) {
+		const existingServer = getServer();
-		return;
+		if (existingServer) {
-	}
+			existingServer.stop(true);
-
+			setServer(null);
-	const agentProcess = runtime.localAgent;
+			clearAgentSockets();
-	runtime.localAgent = null;
-
-	if (agentProcess.exitCode !== null || agentProcess.signalCode !== null) {
-		return;
-	}
-
-	const exited = new Promise<void>((resolve) => {
-		agentProcess.once("exit", () => {
-			resolve();
-		});
-	});
-
-	agentProcess.kill();
-	await exited;
-};
-
-const createAgentManagerRuntime = () => {
-	const sessionsRef = Effect.runSync(Ref.make<Map<string, ControllerAgentSession>>(new Map()));
-	const backupHandlersRef = Effect.runSync(Ref.make<AgentBackupEventHandlers>({}));
-	let runtimeScope: Scope.CloseableScope | null = null;
-
-	const getSessions = () => Effect.runSync(Ref.get(sessionsRef));
-	const getBackupHandlers = () => Effect.runSync(Ref.get(backupHandlersRef));
-	const setSessions = (sessions: Map<string, ControllerAgentSession>) => {
-		Effect.runSync(Ref.set(sessionsRef, sessions));
-	};
-	const setBackupHandlers = (handlers: AgentBackupEventHandlers) => {
-		Effect.runSync(Ref.set(backupHandlersRef, handlers));
-	};
-
-	const closeAllSessions = () => {
-		const sessions = getSessions();
-		for (const session of sessions.values()) {
-			session.close();
-		}
-		setSessions(new Map());
-	};
-
-	const getSession = (agentId: string) => getSessions().get(agentId);
-
-	const setSession = (agentId: string, session: ControllerAgentSession) => {
-		const existingSession = getSession(agentId);
-		if (existingSession) {
-			existingSession.close();
-		}
-
-		const nextSessions = new Map(getSessions());
-		nextSessions.set(agentId, session);
-		setSessions(nextSessions);
-	};
-
-	const removeSession = (agentId: string, connectionId: string) => {
-		const session = getSession(agentId);
-		if (!session || session.connectionId !== connectionId) {
-			return;
-		}
-
-		session.close();
-		const nextSessions = new Map(getSessions());
-		nextSessions.delete(agentId);
-		setSessions(nextSessions);
-	};
-
-	const handleBackupStarted = (ws: Bun.ServerWebSocket<AgentConnectionData>, payload: BackupStartedPayload) => {
-		getBackupHandlers().onBackupStarted?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload });
-	};
-
-	const handleBackupProgress = (ws: Bun.ServerWebSocket<AgentConnectionData>, payload: BackupProgressPayload) => {
-		getBackupHandlers().onBackupProgress?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload });
-	};
-
-	const handleBackupCompleted = (ws: Bun.ServerWebSocket<AgentConnectionData>, payload: BackupCompletedPayload) => {
-		getBackupHandlers().onBackupCompleted?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload });
-	};
-
-	const handleBackupFailed = (ws: Bun.ServerWebSocket<AgentConnectionData>, payload: BackupFailedPayload) => {
-		getBackupHandlers().onBackupFailed?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload });
-	};
-
-	const handleBackupCancelled = (ws: Bun.ServerWebSocket<AgentConnectionData>, payload: BackupCancelledPayload) => {
-		getBackupHandlers().onBackupCancelled?.({ agentId: ws.data.agentId, agentName: ws.data.agentName, payload });
-	};
-
-	const acquireServer = Effect.acquireRelease(
-		Effect.sync(() =>
-			Bun.serve<AgentConnectionData>({
-				port: 3001,
-				async fetch(req, srv) {
-					const url = new URL(req.url);
-					const token = url.searchParams.get("token");
-
-					if (!token) {
-						return new Response("Missing token", { status: 401 });
-					}
-
-					const result = await validateAgentToken(token);
-					if (!result) {
-						return new Response("Invalid or revoked token", { status: 401 });
-					}
-
-					const upgraded = srv.upgrade(req, {
-						data: {
-							id: Bun.randomUUIDv7(),
-							agentId: result.agentId,
-							organizationId: result.organizationId,
-							agentName: result.agentName,
-						},
-					});
-					if (upgraded) return undefined;
-					return new Response("WebSocket upgrade failed", { status: 400 });
-				},
-				websocket: {
-					open: (ws) => {
-						setSession(
-							ws.data.agentId,
-							createControllerAgentSession(ws, {
-								onBackupStarted: (payload) => handleBackupStarted(ws, payload),
-								onBackupProgress: (payload) => handleBackupProgress(ws, payload),
-								onBackupCompleted: (payload) => handleBackupCompleted(ws, payload),
-								onBackupFailed: (payload) => handleBackupFailed(ws, payload),
-								onBackupCancelled: (payload) => handleBackupCancelled(ws, payload),
-							}),
-						);
-						logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) connected on ${ws.data.id}`);
-					},
-					message: (ws, data) => {
-						if (typeof data !== "string") {
-							logger.warn(`Ignoring non-text message from agent ${ws.data.agentId}`);
-							return;
-						}
-
-						const session = getSession(ws.data.agentId);
-						if (!session || session.connectionId !== ws.data.id) {
-							logger.warn(`No active session for agent ${ws.data.agentId} on ${ws.data.id}`);
-							return;
-						}
-
-						session.handleMessage(data);
-					},
-					close: (ws) => {
-						removeSession(ws.data.agentId, ws.data.id);
-						logger.info(`Agent "${ws.data.agentName}" (${ws.data.agentId}) disconnected`);
-					},
-				},
-			}),
-		),
-		(server) =>
-			Effect.sync(() => {
-				closeAllSessions();
-				void server.stop(true);
-			}),
-	);
-
-	const stop = () => {
-		if (!runtimeScope) {
-			return;
-		}
-
-		logger.info("Stopping Agent Manager...");
-		const scope = runtimeScope;
-		runtimeScope = null;
-		Effect.runSync(Scope.close(scope, Exit.succeed(undefined)));
-	};
-
-	const start = () => {
-		if (runtimeScope) {
-			stop();
 		}
 
 		logger.info("Starting Agent Manager...");
-		const scope = Effect.runSync(Scope.make());
+		const server = Bun.serve<AgentConnectionData>({
+			port: 3001,
+			fetch(req, srv) {
+				const upgraded = srv.upgrade(req, { data: { id: Bun.randomUUIDv7() } });
+				if (upgraded) return undefined;
+				return new Response("Agent WebSocket endpoint", { status: 200 });
+			},
+			websocket: {
+				open: (ws) => logger.info(`WebSocket opened with id: ${ws.data.id}`),
+				message: (ws, data) => {
+					if (typeof data !== "string") {
+						logger.warn(`Ignoring non-text message from agent connection ${ws.data.id}`);
+						return;
+					}
 
-		try {
+					const parsed = parseAgentMessage(data);
-			const server = Effect.runSync(Scope.extend(acquireServer, scope));
+
-			runtimeScope = scope;
+					if (parsed === null) {
-			logger.info(`Agent Manager listening on port ${server.port}`);
+						logger.warn(`Invalid JSON from agent connection ${ws.data.id}`);
-		} catch (error) {
+						return;
-			Effect.runSync(Scope.close(scope, Exit.fail(error)));
+					}
-			throw error;
+
+					if (!parsed.success) {
+						logger.warn(`Invalid agent message on connection ${ws.data.id}: ${parsed.error.message}`);
+						return;
+					}
+
+					switch (parsed.data.type) {
+						case "agent.ready": {
+							ws.data.agentId = parsed.data.payload.agentId;
+							getAgentSockets().set(parsed.data.payload.agentId, ws);
+							logger.info(`Backup agent ${parsed.data.payload.agentId} is ready on connection ${ws.data.id}`);
+							break;
+						}
+						case "backup.started": {
+							logger.info(
+								`Backup started on agent ${ws.data.agentId ?? ws.data.id} for schedule ${parsed.data.payload.scheduleId}`,
+							);
+							break;
+						}
+					}
+				},
+				close: (ws) => {
+					if (ws.data.agentId && getAgentSockets().get(ws.data.agentId) === ws) {
+						getAgentSockets().delete(ws.data.agentId);
+					}
+
+					logger.info(`WebSocket closed for agent ${ws.data.agentId ?? ws.data.id}`);
+				},
+			},
+		});
+		setServer(server);
+
+		logger.info(`Agent Manager listening on port ${server.port}`);
+	},
+	sendBackup: (agentId: string, payload: BackupCommandPayload) => {
+		const agentSocket = getAgentSockets().get(agentId);
+
+		if (!agentSocket) {
+			logger.warn(`Cannot send backup command. Agent ${agentId} is not connected.`);
+			return false;
 		}
-	};
 
-	return {
+		sendControllerMessage(agentSocket, createControllerMessage("backup", payload));
-		start,
+		logger.info(`Sent backup command to agent ${agentId} for schedule ${payload.scheduleId}`);
-		sendBackup: (agentId: string, payload: BackupRunPayload) => {
+		return true;
-			const session = getSession(agentId);
+	},
+	stop: () => {
+		const server = getServer();
+		if (!server) return;
 
-			if (!session) {
+		logger.info("Stopping Agent Manager...");
-				logger.warn(`Cannot send backup command. Agent ${agentId} is not connected.`);
+		server.stop(true);
-				return false;
+		setServer(null);
-			}
+		clearAgentSockets();
-
+	},
-			if (!session.isReady()) {
-				logger.warn(`Cannot send backup command. Agent ${agentId} is not ready.`);
-				return false;
-			}
-
-			const jobId = session.sendBackup(payload);
-			logger.info(`Sent backup command ${jobId} to agent ${agentId} for schedule ${payload.scheduleId}`);
-			return true;
-		},
-		cancelBackup: (agentId: string, payload: BackupCancelPayload) => {
-			const session = getSession(agentId);
-
-			if (!session) {
-				logger.warn(`Cannot cancel backup command. Agent ${agentId} is not connected.`);
-				return false;
-			}
-
-			session.sendBackupCancel(payload);
-			logger.info(`Sent backup cancel for command ${payload.jobId} to agent ${agentId}`);
-			return true;
-		},
-		setBackupEventHandlers: (handlers: AgentBackupEventHandlers) => {
-			setBackupHandlers(handlers);
-		},
-		getBackupEventHandlers: () => getBackupHandlers(),
-		stop,
-	};
-};
-
-export const agentManager = getAgentManagerRuntime();
-
-export const stopAgentRuntime = async () => {
-	getAgentManagerRuntime().stop();
-	await stopLocalAgent();
 };

app/server/modules/agents/controller-agent-session.ts

@@ -1,210 +0,0 @@
-import { Effect, Fiber, Queue, Ref } from "effect";
-import {
-	createControllerMessage,
-	parseAgentMessage,
-	type AgentMessage,
-	type BackupCancelledPayload,
-	type BackupCompletedPayload,
-	type BackupFailedPayload,
-	type BackupProgressPayload,
-	type BackupRunPayload,
-	type BackupCancelPayload,
-	type BackupStartedPayload,
-	type ControllerWireMessage,
-} from "@zerobyte/contracts/agent-protocol";
-import { logger } from "@zerobyte/core/node";
-import { toMessage } from "@zerobyte/core/utils";
-
-export type AgentConnectionData = {
-	id: string;
-	agentId: string;
-	organizationId: string | null;
-	agentName: string;
-};
-
-type AgentSocket = Bun.ServerWebSocket<AgentConnectionData>;
-
-type SessionState = {
-	isReady: boolean;
-	lastSeenAt: number | null;
-	lastPongAt: number | null;
-};
-
-type ControllerAgentSessionHandlers = {
-	onBackupStarted?: (payload: BackupStartedPayload) => void;
-	onBackupProgress?: (payload: BackupProgressPayload) => void;
-	onBackupCompleted?: (payload: BackupCompletedPayload) => void;
-	onBackupFailed?: (payload: BackupFailedPayload) => void;
-	onBackupCancelled?: (payload: BackupCancelledPayload) => void;
-};
-
-export type ControllerAgentSession = {
-	readonly connectionId: string;
-	handleMessage: (data: string) => void;
-	sendBackup: (payload: BackupRunPayload) => string;
-	sendBackupCancel: (payload: BackupCancelPayload) => void;
-	isReady: () => boolean;
-	close: () => void;
-};
-
-export const createControllerAgentSession = (
-	socket: AgentSocket,
-	handlers: ControllerAgentSessionHandlers = {},
-): ControllerAgentSession => {
-	const outboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
-	const activeBackupJobs = Effect.runSync(Ref.make<Map<string, string>>(new Map()));
-	const state = Effect.runSync(
-		Ref.make<SessionState>({
-			isReady: false,
-			lastSeenAt: null,
-			lastPongAt: null,
-		}),
-	);
-
-	const offerOutbound = (message: ControllerWireMessage) => {
-		void Effect.runPromise(Queue.offer(outboundQueue, message)).catch((error) => {
-			logger.error(`Failed to queue outbound message for agent ${socket.data.agentId}: ${toMessage(error)}`);
-		});
-	};
-
-	const updateState = (update: (current: SessionState) => SessionState) => {
-		Effect.runSync(Ref.update(state, update));
-	};
-
-	const setActiveBackupJob = (jobId: string, scheduleId: string) => {
-		Effect.runSync(
-			Ref.update(activeBackupJobs, (current) => {
-				const next = new Map(current);
-				next.set(jobId, scheduleId);
-				return next;
-			}),
-		);
-	};
-
-	const deleteActiveBackupJob = (jobId: string) => {
-		Effect.runSync(
-			Ref.update(activeBackupJobs, (current) => {
-				const next = new Map(current);
-				next.delete(jobId);
-				return next;
-			}),
-		);
-	};
-
-	const writerFiber = Effect.runFork(
-		Effect.forever(
-			Effect.gen(function* () {
-				const message = yield* Queue.take(outboundQueue);
-				yield* Effect.sync(() => {
-					try {
-						socket.send(message);
-					} catch (error) {
-						logger.error(
-							`Failed to send message to agent ${socket.data.agentId} on ${socket.data.id}: ${toMessage(error)}`,
-						);
-					}
-				});
-			}),
-		),
-	);
-
-	const heartbeatFiber = Effect.runFork(
-		Effect.forever(
-			Effect.gen(function* () {
-				yield* Effect.sleep("15 seconds");
-				yield* Queue.offer(
-					outboundQueue,
-					createControllerMessage("heartbeat.ping", {
-						sentAt: Date.now(),
-					}),
-				);
-			}),
-		),
-	);
-
-	const handleAgentMessage = (message: AgentMessage) => {
-		updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
-
-		switch (message.type) {
-			case "agent.ready": {
-				updateState((current) => ({ ...current, isReady: true }));
-				logger.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`);
-				break;
-			}
-			case "backup.started": {
-				setActiveBackupJob(message.payload.jobId, message.payload.scheduleId);
-				logger.info(
-					`Backup ${message.payload.jobId} started on agent ${socket.data.agentId} for schedule ${message.payload.scheduleId}`,
-				);
-				handlers.onBackupStarted?.(message.payload);
-				break;
-			}
-			case "backup.progress": {
-				handlers.onBackupProgress?.(message.payload);
-				break;
-			}
-			case "backup.completed": {
-				deleteActiveBackupJob(message.payload.jobId);
-				handlers.onBackupCompleted?.(message.payload);
-				break;
-			}
-			case "backup.failed": {
-				deleteActiveBackupJob(message.payload.jobId);
-				handlers.onBackupFailed?.(message.payload);
-				break;
-			}
-			case "backup.cancelled": {
-				deleteActiveBackupJob(message.payload.jobId);
-				handlers.onBackupCancelled?.(message.payload);
-				break;
-			}
-			case "heartbeat.pong": {
-				updateState((current) => ({ ...current, lastPongAt: message.payload.sentAt }));
-				break;
-			}
-		}
-	};
-
-	return {
-		connectionId: socket.data.id,
-		handleMessage: (data: string) => {
-			const parsed = parseAgentMessage(data);
-
-			if (parsed === null) {
-				logger.warn(`Invalid JSON from agent ${socket.data.agentId}`);
-				return;
-			}
-
-			if (!parsed.success) {
-				logger.warn(`Invalid agent message from ${socket.data.agentId}: ${parsed.error.message}`);
-				return;
-			}
-
-			handleAgentMessage(parsed.data);
-		},
-		sendBackup: (payload) => {
-			offerOutbound(createControllerMessage("backup.run", payload));
-			return payload.jobId;
-		},
-		sendBackupCancel: (payload) => {
-			offerOutbound(createControllerMessage("backup.cancel", payload));
-		},
-		isReady: () => Effect.runSync(Ref.get(state)).isReady,
-		close: () => {
-			updateState((current) => ({ ...current, isReady: false }));
-			const pendingJobs = Effect.runSync(Ref.get(activeBackupJobs));
-			Effect.runSync(Ref.set(activeBackupJobs, new Map()));
-			for (const [jobId, scheduleId] of pendingJobs) {
-				handlers.onBackupCancelled?.({
-					jobId,
-					scheduleId,
-					message:
-						"The connection to the backup agent was lost while this backup was running. Restart the backup to ensure it completes.",
-				});
-			}
-			void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {});
-			void Effect.runPromise(Fiber.interrupt(heartbeatFiber)).catch(() => {});
-			void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {});
-		},
-	};
-};

app/server/modules/agents/local-agent.ts

@@ -0,0 +1,61 @@
+import { logger } from "~/server/utils/logger";
+import { createAgentMessage, parseControllerMessage, sendAgentMessage } from "./agent-protocol";
+
+const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL;
+
+class Agent {
+	private ws: WebSocket | null = null;
+
+	constructor(public id: string) {
+		this.connect();
+	}
+
+	private connect() {
+		if (!controllerUrl) {
+			throw new Error("Env variable ZEROBYTE_CONTROLLER_URL is not set");
+		}
+
+		this.ws = new WebSocket(controllerUrl);
+		this.ws.onopen = () => {
+			logger.info(`Agent ${this.id} connected to controller`);
+
+			if (this.ws) {
+				sendAgentMessage(this.ws, createAgentMessage("agent.ready", { agentId: this.id }));
+			}
+		};
+
+		this.ws.onmessage = (event) => {
+			const parsed = parseControllerMessage(event.data);
+
+			if (parsed === null) {
+				console.error(`Agent ${this.id} received invalid JSON`);
+				return;
+			}
+
+			if (!parsed.success) {
+				console.error(`Agent ${this.id} received an invalid message: ${parsed.error.message}`);
+				return;
+			}
+
+			switch (parsed.data.type) {
+				case "backup":
+					logger.info(`Agent ${this.id} starting backup for schedule ${parsed.data.payload.scheduleId}`);
+					if (this.ws) {
+						sendAgentMessage(
+							this.ws,
+							createAgentMessage("backup.started", { scheduleId: parsed.data.payload.scheduleId }),
+						);
+					}
+					break;
+			}
+		};
+		this.ws.onclose = () => {
+			logger.info(`Agent ${this.id} disconnected from controller`);
+		};
+		this.ws.onerror = (error) => {
+			logger.error(`Agent ${this.id} encountered an error:`, error);
+		};
+	}
+}
+
+new Agent(Bun.randomUUIDv7());

app/server/modules/backups/__tests__/backups.execution.test.ts

@@ -1,6 +1,7 @@
 import waitForExpect from "wait-for-expect";
 import { afterEach, describe, expect, test, vi } from "vitest";
 import { backupsService } from "../backups.service";
+import { backupsExecutionService } from "../backups.execution";
 import { createTestVolume } from "~/test/helpers/volume";
 import { createTestBackupSchedule } from "~/test/helpers/backup";
 import { createTestRepository } from "~/test/helpers/repository";
@@ -14,11 +15,6 @@ import { restic } from "~/server/core/restic";
 import { NotFoundError, BadRequestError } from "http-errors-enhanced";
 import { repositoriesService } from "~/server/modules/repositories/repositories.service";
 import { repoMutex } from "~/server/core/repository-mutex";
-import { agentManager } from "~/server/modules/agents/agents-manager";
-import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
-import { fromAny } from "@total-typescript/shoehorn";
-import { scheduleQueries } from "../backups.queries";
-import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups";
 
 const setup = () => {
 	const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
@@ -26,7 +22,6 @@ const setup = () => {
 	);
 	const resticForgetMock = vi.fn(() => Promise.resolve({ success: true, data: null }));
 	const resticCopyMock = vi.fn(() => Promise.resolve({ success: true, output: "" }));
-	const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
 	const refreshStatsMock = vi.fn(() =>
 		Promise.resolve({
 			total_size: 0,
@@ -43,15 +38,11 @@ const setup = () => {
 	vi.spyOn(restic, "copy").mockImplementation(resticCopyMock);
 	vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
-	vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock);
-	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 
 	return {
 		resticBackupMock,
 		resticForgetMock,
 		resticCopyMock,
-		sendBackupMock,
-		cancelBackupMock,
 		refreshStatsMock,
 	};
 };
@@ -72,7 +63,7 @@ describe("backup execution - validation failures", () => {
 		});
 
 		// act
-		const result = await backupsService.validateBackupExecution(schedule.id);
+		const result = await backupsExecutionService.validateBackupExecution(schedule.id);
 
 		// assert
 		expect(result.type).toBe("failure");
@@ -83,71 +74,10 @@ describe("backup execution - validation failures", () => {
 		expect(resticBackupMock).not.toHaveBeenCalled();
 	});
 
-	test("should fail backup when volume does not exist", async () => {
-		// arrange
-		setup();
-		const volume = await createTestVolume();
-		const repository = await createTestRepository();
-		const schedule = await createTestBackupSchedule({
-			volumeId: volume.id,
-			repositoryId: repository.id,
-		});
-
-		const hydratedSchedule = await scheduleQueries.findById(schedule.id, TEST_ORG_ID);
-		expect(hydratedSchedule).toBeDefined();
-		const scheduleWithoutVolume = {
-			...hydratedSchedule,
-			volume: null,
-		};
-		vi.spyOn(scheduleQueries, "findById").mockResolvedValueOnce(fromAny(scheduleWithoutVolume));
-
-		// act
-		const result = await backupsService.validateBackupExecution(schedule.id);
-
-		// assert
-		expect(result.type).toBe("failure");
-		if (result.type === "failure") {
-			expect(result.error).toBeInstanceOf(NotFoundError);
-			expect(result.error.message).toBe("Volume not found");
-			expect(result.partialContext?.schedule).toBeDefined();
-		}
-	});
-
-	test("should fail backup when repository does not exist", async () => {
-		// arrange
-		setup();
-		const volume = await createTestVolume();
-		const repository = await createTestRepository();
-		const schedule = await createTestBackupSchedule({
-			volumeId: volume.id,
-			repositoryId: repository.id,
-		});
-
-		const hydratedSchedule = await scheduleQueries.findById(schedule.id, TEST_ORG_ID);
-		expect(hydratedSchedule).toBeDefined();
-		const scheduleWithoutRepository = {
-			...hydratedSchedule,
-			repository: null,
-		};
-		vi.spyOn(scheduleQueries, "findById").mockResolvedValueOnce(fromAny(scheduleWithoutRepository));
-
-		// act
-		const result = await backupsService.validateBackupExecution(schedule.id);
-
-		// assert
-		expect(result.type).toBe("failure");
-		if (result.type === "failure") {
-			expect(result.error).toBeInstanceOf(NotFoundError);
-			expect(result.error.message).toBe("Repository not found");
-			expect(result.partialContext?.schedule).toBeDefined();
-			expect(result.partialContext?.volume).toBeDefined();
-		}
-	});
-
 	test("should fail backup when schedule does not exist", async () => {
 		setup();
 		// act
-		const result = await backupsService.validateBackupExecution(99999);
+		const result = await backupsExecutionService.validateBackupExecution(99999);
 
 		// assert
 		expect(result.type).toBe("failure");
@@ -178,9 +108,9 @@ describe("stop backup", () => {
 			});
 		});
 
-		await backupsService.executeBackup(schedule.id);
+		await backupsExecutionService.executeBackup(schedule.id);
 
-		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("error: open /mnt/data/private.db: permission denied");
 	});
@@ -206,63 +136,15 @@ describe("stop backup", () => {
 			});
 		});
 
-		await backupsService.executeBackup(schedule.id);
+		await backupsExecutionService.executeBackup(schedule.id);
 
-		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("error");
 		expect(updatedSchedule.lastBackupError).toBe(
 			"Permissions 0755 for '/tmp/zerobyte-ssh-key' are too open.\nThis private key will be ignored.",
 		);
 	});
 
-	test("should block forget on the same repository until the active backup completes", async () => {
-		const { resticBackupMock, resticForgetMock, sendBackupMock } = setup();
-		const volume = await createTestVolume();
-		const repository = await createTestRepository();
-		const schedule = await createTestBackupSchedule({
-			volumeId: volume.id,
-			repositoryId: repository.id,
-			retentionPolicy: { keepHourly: 24 },
-		});
-
-		let completeBackup: (() => void) | undefined;
-		resticBackupMock.mockImplementationOnce(
-			() =>
-				new Promise((resolve) => {
-					completeBackup = () => resolve({ exitCode: 0, summary: generateBackupOutput(), error: "" });
-				}),
-		);
-
-		const backupPromise = backupsService.executeBackup(schedule.id);
-
-		await waitForExpect(() => {
-			expect(sendBackupMock).toHaveBeenCalledTimes(1);
-		});
-
-		let forgetFinished = false;
-		const forgetPromise = backupsService.runForget(schedule.id).finally(() => {
-			forgetFinished = true;
-		});
-
-		await new Promise((resolve) => setTimeout(resolve, 50));
-
-		expect(resticForgetMock).not.toHaveBeenCalled();
-		expect(forgetFinished).toBe(false);
-
-		expect(completeBackup).toBeDefined();
-		completeBackup?.();
-
-		await backupPromise;
-		await forgetPromise;
-
-		expect(resticForgetMock).toHaveBeenCalled();
-		expect(resticForgetMock).toHaveBeenCalledWith(
-			repository.config,
-			expect.objectContaining({ keepHourly: 24 }),
-			expect.objectContaining({ tag: schedule.shortId, organizationId: TEST_ORG_ID }),
-		);
-	});
-
 	test("should stop a running backup", async () => {
 		// arrange
 		const { resticBackupMock } = setup();
@@ -290,19 +172,19 @@ describe("stop backup", () => {
 			});
 		});
 
-		const executePromise = backupsService.executeBackup(schedule.id);
+		const executePromise = backupsExecutionService.executeBackup(schedule.id);
 
 		await waitForExpect(async () => {
-			const runningSchedule = await getScheduleByIdOrShortId(schedule.id);
+			const runningSchedule = await backupsService.getScheduleById(schedule.id);
 			expect(runningSchedule.lastBackupStatus).toBe("in_progress");
 		});
 
 		// act
-		await backupsService.stopBackup(schedule.id);
+		await backupsExecutionService.stopBackup(schedule.id);
 		await executePromise;
 
 		// assert
-		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 	});
@@ -333,7 +215,7 @@ describe("stop backup", () => {
 			});
 		});
 
-		const executePromise = backupsService.executeBackup(schedule.id);
+		const executePromise = backupsExecutionService.executeBackup(schedule.id);
 
 		await waitForExpect(async () => {
 			const queuedSchedule = await backupsService.getScheduleById(schedule.id);
@@ -342,7 +224,7 @@ describe("stop backup", () => {
 
 		expect(resticBackupMock).not.toHaveBeenCalled();
 
-		await backupsService.stopBackup(schedule.id);
+		await backupsExecutionService.stopBackup(schedule.id);
 		await executePromise;
 
 		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
@@ -365,7 +247,7 @@ describe("stop backup", () => {
 		});
 
 		// act & assert
-		await expect(backupsService.stopBackup(schedule.id)).rejects.toThrow(
+		await expect(backupsExecutionService.stopBackup(schedule.id)).rejects.toThrow(
 			"No backup is currently running for this schedule",
 		);
 
@@ -375,30 +257,10 @@ describe("stop backup", () => {
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 	});
 
-	test("should reset a stuck in_progress status even when no backup is running", async () => {
-		// arrange
-		setup();
-		const volume = await createTestVolume();
-		const repository = await createTestRepository();
-		const schedule = await createTestBackupSchedule({
-			volumeId: volume.id,
-			repositoryId: repository.id,
-			lastBackupStatus: "in_progress",
-		});
-
-		// act
-		await backupsService.stopBackup(schedule.id).catch(() => {});
-
-		// assert
-		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
-		expect(updatedSchedule.lastBackupStatus).toBe("warning");
-		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
-	});
-
 	test("should throw NotFoundError when schedule does not exist", async () => {
 		setup();
 		// act & assert
-		await expect(backupsService.stopBackup(99999)).rejects.toThrow("Backup schedule not found");
+		await expect(backupsExecutionService.stopBackup(99999)).rejects.toThrow("Backup schedule not found");
 	});
 });
 
@@ -419,7 +281,7 @@ describe("retention policy - runForget", () => {
 		});
 
 		// act
-		await backupsService.runForget(schedule.id);
+		await backupsExecutionService.runForget(schedule.id);
 
 		// assert
 		expect(resticForgetMock).toHaveBeenCalledWith(
@@ -448,7 +310,7 @@ describe("retention policy - runForget", () => {
 		});
 
 		// act & assert
-		await expect(backupsService.runForget(schedule.id)).rejects.toThrow(
+		await expect(backupsExecutionService.runForget(schedule.id)).rejects.toThrow(
 			"No retention policy configured for this schedule",
 		);
 	});
@@ -456,7 +318,7 @@ describe("retention policy - runForget", () => {
 	test("should throw NotFoundError when schedule does not exist", async () => {
 		setup();
 		// act & assert
-		await expect(backupsService.runForget(99999)).rejects.toThrow("Backup schedule not found");
+		await expect(backupsExecutionService.runForget(99999)).rejects.toThrow("Backup schedule not found");
 	});
 
 	test("should throw NotFoundError when repository does not exist", async () => {
@@ -469,7 +331,9 @@ describe("retention policy - runForget", () => {
 		});
 
 		// act & assert
-		await expect(backupsService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow("Repository not found");
+		await expect(backupsExecutionService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow(
+			"Repository not found",
+		);
 	});
 });
 
@@ -488,7 +352,7 @@ describe("mirror operations", () => {
 		await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
 
 		// act
-		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		expect(resticCopyMock).toHaveBeenCalledWith(
@@ -515,7 +379,7 @@ describe("mirror operations", () => {
 		await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id, { enabled: false });
 
 		// act
-		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		expect(resticCopyMock).not.toHaveBeenCalled();
@@ -535,7 +399,7 @@ describe("mirror operations", () => {
 		const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
 
 		// act
-		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		const mirrors = await backupsService.getMirrors(schedule.id);
@@ -566,7 +430,7 @@ describe("mirror operations", () => {
 		});
 
 		// act
-		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		const mirrors = await backupsService.getMirrors(schedule.id);
@@ -593,7 +457,7 @@ describe("mirror operations", () => {
 		resticCopyMock.mockImplementationOnce(() => Promise.reject(new Error("Copy failed")));
 
 		// act
-		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		const mirrors = await backupsService.getMirrors(schedule.id);
@@ -621,7 +485,7 @@ describe("mirror operations", () => {
 		resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" }));
 
 		// act
-		await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
+		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
 
 		await waitForExpect(() => {
 			expect(resticCopyMock).toHaveBeenCalled();
@@ -652,7 +516,7 @@ describe("mirror operations", () => {
 		resticForgetMock.mockClear();
 
 		// act
-		await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
+		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
 
 		await waitForExpect(() => {
 			expect(resticCopyMock).toHaveBeenCalled();

app/server/modules/backups/__tests__/backups.service.test.ts

@@ -12,14 +12,11 @@ import { db } from "~/server/db/db";
 import { backupScheduleMirrorsTable, repositoriesTable, volumesTable } from "~/server/db/schema";
 import { TEST_ORG_ID } from "~/test/helpers/organization";
 import * as context from "~/server/core/request-context";
+import { backupsExecutionService } from "../backups.execution";
 import { repositoriesService } from "~/server/modules/repositories/repositories.service";
-import { agentManager } from "~/server/modules/agents/agents-manager";
-import { createAgentBackupMocks } from "~/test/helpers/agent-mock";
-import { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups";
 
 const setup = () => {
-	const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
+	const resticBackupMock = vi.fn(() => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
-	const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
 	const refreshStatsMock = vi.fn(() =>
 		Promise.resolve({
 			total_size: 0,
@@ -33,13 +30,9 @@ const setup = () => {
 	vi.spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock);
 	vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
-	vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock);
-	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 
 	return {
 		resticBackupMock,
-		sendBackupMock,
-		cancelBackupMock,
 		refreshStatsMock,
 	};
 };
@@ -66,10 +59,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsService.executeBackup(schedule.id);
+		await backupsExecutionService.executeBackup(schedule.id);
 
 		// assert
-		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
 		expect(updatedSchedule.nextBackupAt).not.toBeNull();
 
 		const nextBackupAt = new Date(updatedSchedule.nextBackupAt ?? 0);
@@ -91,7 +84,7 @@ describe("execute backup", () => {
 		});
 
 		// act
-		await backupsService.executeBackup(schedule.id);
+		await backupsExecutionService.executeBackup(schedule.id);
 
 		// assert
 		expect(resticBackupMock).not.toHaveBeenCalled();
@@ -113,7 +106,7 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsService.executeBackup(schedule.id, true);
+		await backupsExecutionService.executeBackup(schedule.id, true);
 
 		// assert
 		expect(resticBackupMock).toHaveBeenCalled();
@@ -139,7 +132,7 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsService.executeBackup(schedule.id, true);
+		await backupsExecutionService.executeBackup(schedule.id, true);
 
 		// assert
 		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
@@ -162,13 +155,13 @@ describe("execute backup", () => {
 		});
 
 		// act
-		void backupsService.executeBackup(schedule.id);
+		void backupsExecutionService.executeBackup(schedule.id);
 
 		await waitForExpect(() => {
 			expect(resticBackupMock).toHaveBeenCalledTimes(1);
 		});
 
-		await backupsService.executeBackup(schedule.id);
+		await backupsExecutionService.executeBackup(schedule.id);
 
 		// assert
 		expect(resticBackupMock).toHaveBeenCalledTimes(1);
@@ -189,10 +182,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsService.executeBackup(schedule.id);
+		await backupsExecutionService.executeBackup(schedule.id);
 
 		// assert
-		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 	});
 
@@ -211,10 +204,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsService.executeBackup(schedule.id);
+		await backupsExecutionService.executeBackup(schedule.id);
 
 		// assert
-		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
+		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("error");
 	});
 });
@@ -236,7 +229,7 @@ describe("getSchedulesToExecute", () => {
 		});
 
 		// act
-		const schedulesToExecute = await backupsService.getSchedulesToExecute();
+		const schedulesToExecute = await backupsExecutionService.getSchedulesToExecute();
 
 		// assert
 		expect(schedulesToExecute).toContain(schedule.id);
@@ -253,7 +246,7 @@ describe("getScheduleByIdOrShortId", () => {
 			repositoryId: repository.id,
 		});
 
-		const found = await getScheduleByIdOrShortId(String(schedule.id));
+		const found = await backupsService.getScheduleByIdOrShortId(String(schedule.id));
 
 		expect(found.id).toBe(schedule.id);
 		expect(found.shortId).toBe(schedule.shortId);
@@ -268,7 +261,7 @@ describe("getScheduleByIdOrShortId", () => {
 			repositoryId: repository.id,
 		});
 
-		const found = await getScheduleByIdOrShortId(schedule.shortId);
+		const found = await backupsService.getScheduleByIdOrShortId(schedule.shortId);
 
 		expect(found.id).toBe(schedule.id);
 		expect(found.shortId).toBe(schedule.shortId);
@@ -281,8 +274,10 @@ describe("getScheduleByIdOrShortId", () => {
 			organizationId: otherOrgId,
 		});
 
-		await expect(getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow("Backup schedule not found");
+		await expect(backupsService.getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow(
-		await expect(getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found");
+			"Backup schedule not found",
+		);
+		await expect(backupsService.getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found");
 	});
 });
 

app/server/modules/backups/backup-executor.ts

@@ -1,18 +1,11 @@
-import { logger } from "@zerobyte/core/node";
+import { restic } from "../../core/restic";
-import type { BackupProgressPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
-import type { ResticBackupOutputDto } from "@zerobyte/core/restic";
-import { resticDeps } from "../../core/restic";
 import type { BackupSchedule, Repository, Volume } from "../../db/schema";
-import { agentManager } from "../agents/agents-manager";
+import type { ResticBackupOutputDto, ResticBackupProgressDto } from "@zerobyte/core/restic";
-import { decryptRepositoryConfig } from "../repositories/repository-config-secrets";
-import { getVolumePath } from "../volumes/helpers";
 import { createBackupOptions } from "./backup.helpers";
-
+import { getVolumePath } from "../volumes/helpers";
-const LOCAL_AGENT_ID = "local";
 
 type BackupExecutionRequest = {
 	scheduleId: number;
-	jobId: string;
 	schedule: BackupSchedule;
 	volume: Volume;
 	repository: Repository;
@@ -21,14 +14,7 @@ type BackupExecutionRequest = {
 	onProgress: (progress: BackupExecutionProgress) => void;
 };
 
-type ActiveBackupExecution = {
+export type BackupExecutionProgress = ResticBackupProgressDto;
-	scheduleId: number;
-	scheduleShortId: string;
-	onProgress: (progress: BackupExecutionProgress) => void;
-	resolve: (result: BackupExecutionResult) => void;
-};
-
-export type BackupExecutionProgress = BackupProgressPayload["progress"];
 
 export type BackupExecutionResult =
 	| {
@@ -43,228 +29,59 @@ export type BackupExecutionResult =
 	  }
 	| {
 			status: "failed";
-			error: string;
+			error: unknown;
 	  }
 	| {
 			status: "cancelled";
 			message?: string;
 	  };
 
-const trackedAbortControllersByScheduleId = new Map<number, AbortController>();
+const activeControllersByScheduleId = new Map<number, AbortController>();
-const activeExecutionsByJobId = new Map<string, ActiveBackupExecution>();
-const activeExecutionJobIdsByScheduleId = new Map<number, string>();
-const requestedCancellationsByScheduleId = new Set<number>();
-
-const getCancellationError = (signal: AbortSignal, message?: string) =>
-	signal.reason instanceof Error ? signal.reason : new Error(message ?? "Backup was stopped by the user");
-
-const createBackupRunPayload = async ({
-	jobId,
-	schedule,
-	volume,
-	repository,
-	organizationId,
-}: BackupExecutionRequest): Promise<BackupRunPayload> => {
-	const sourcePath = getVolumePath(volume);
-	const { signal: _ignoredSignal, ...options } = createBackupOptions(schedule, sourcePath);
-	const repositoryConfig = await decryptRepositoryConfig(repository.config);
-	const encryptedResticPassword = await resticDeps.getOrganizationResticPassword(organizationId);
-	const resticPassword = await resticDeps.resolveSecret(encryptedResticPassword);
-
-	return {
-		jobId,
-		scheduleId: schedule.shortId,
-		organizationId,
-		sourcePath,
-		repositoryConfig,
-		options: {
-			...options,
-			compressionMode: repository.compressionMode ?? "auto",
-		},
-		runtime: {
-			password: resticPassword,
-			cacheDir: resticDeps.resticCacheDir,
-			passFile: resticDeps.resticPassFile,
-			defaultExcludes: resticDeps.defaultExcludes,
-			hostname: resticDeps.hostname,
-		},
-	};
-};
-
-const clearActiveExecution = (jobId: string) => {
-	const activeExecution = activeExecutionsByJobId.get(jobId);
-	if (!activeExecution) {
-		return null;
-	}
-
-	activeExecutionsByJobId.delete(jobId);
-	activeExecutionJobIdsByScheduleId.delete(activeExecution.scheduleId);
-	return activeExecution;
-};
-
-const getActiveExecution = (jobId: string, scheduleId: string, eventName: string, executorId: string) => {
-	const activeExecution = activeExecutionsByJobId.get(jobId);
-	if (!activeExecution) {
-		logger.warn(`Received ${eventName} for unknown job ${jobId} from executor ${executorId}`);
-		return null;
-	}
-
-	if (activeExecution.scheduleShortId !== scheduleId) {
-		logger.warn(
-			`Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from executor ${executorId}`,
-		);
-		return null;
-	}
-
-	return activeExecution;
-};
-
-agentManager.setBackupEventHandlers({
-	onBackupStarted: ({ agentId, payload }) => {
-		getActiveExecution(payload.jobId, payload.scheduleId, "backup.started", agentId);
-	},
-	onBackupProgress: ({ agentId, payload }) => {
-		const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.progress", agentId);
-		if (!activeExecution) {
-			return;
-		}
-
-		activeExecution.onProgress(payload.progress);
-	},
-	onBackupCompleted: ({ agentId, payload }) => {
-		const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.completed", agentId);
-		if (!activeExecution) {
-			return;
-		}
-
-		requestedCancellationsByScheduleId.delete(activeExecution.scheduleId);
-		clearActiveExecution(payload.jobId);
-		activeExecution.resolve({
-			status: "completed",
-			exitCode: payload.exitCode,
-			result: payload.result,
-			warningDetails: payload.warningDetails ?? null,
-		});
-	},
-	onBackupFailed: ({ agentId, payload }) => {
-		const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.failed", agentId);
-		if (!activeExecution) {
-			return;
-		}
-
-		requestedCancellationsByScheduleId.delete(activeExecution.scheduleId);
-		clearActiveExecution(payload.jobId);
-		activeExecution.resolve({
-			status: "failed",
-			error: payload.errorDetails ?? payload.error,
-		});
-	},
-	onBackupCancelled: ({ agentId, payload }) => {
-		const activeExecution = getActiveExecution(payload.jobId, payload.scheduleId, "backup.cancelled", agentId);
-		if (!activeExecution) {
-			return;
-		}
-
-		const wasRequested = requestedCancellationsByScheduleId.has(activeExecution.scheduleId);
-		requestedCancellationsByScheduleId.delete(activeExecution.scheduleId);
-		clearActiveExecution(payload.jobId);
-		activeExecution.resolve({
-			status: "cancelled",
-			message: wasRequested ? undefined : payload.message,
-		});
-	},
-});
 
 export const backupExecutor = {
 	track: (scheduleId: number) => {
 		const abortController = new AbortController();
-		trackedAbortControllersByScheduleId.set(scheduleId, abortController);
+		activeControllersByScheduleId.set(scheduleId, abortController);
 		return abortController;
 	},
 	untrack: (scheduleId: number, abortController: AbortController) => {
-		if (trackedAbortControllersByScheduleId.get(scheduleId) === abortController) {
+		if (activeControllersByScheduleId.get(scheduleId) === abortController) {
-			trackedAbortControllersByScheduleId.delete(scheduleId);
+			activeControllersByScheduleId.delete(scheduleId);
 		}
-		requestedCancellationsByScheduleId.delete(scheduleId);
 	},
-	execute: async (request: Omit<BackupExecutionRequest, "jobId">) => {
+	execute: async (params: BackupExecutionRequest): Promise<BackupExecutionResult> => {
-		if (request.signal.aborted) {
+		const { schedule, volume, repository, organizationId, signal, onProgress } = params;
-			throw getCancellationError(request.signal);
-		}
-
-		const jobId = Bun.randomUUIDv7();
-		const payload = await createBackupRunPayload({ ...request, jobId });
-
-		if (request.signal.aborted) {
-			throw getCancellationError(request.signal);
-		}
-
-		const completion = new Promise<BackupExecutionResult>((resolve) => {
-			activeExecutionsByJobId.set(jobId, {
-				scheduleId: request.scheduleId,
-				scheduleShortId: request.schedule.shortId,
-				onProgress: request.onProgress,
-				resolve,
-			});
-			activeExecutionJobIdsByScheduleId.set(request.scheduleId, jobId);
-		});
-
-		let dispatched = false;
-		const handleAbort = () => {
-			const activeExecution = activeExecutionsByJobId.get(jobId);
-			if (!activeExecution) {
-				return;
-			}
-
-			if (!dispatched) {
-				clearActiveExecution(jobId);
-				activeExecution.resolve({ status: "cancelled" });
-				return;
-			}
-
-			requestedCancellationsByScheduleId.add(request.scheduleId);
-			if (
-				!agentManager.cancelBackup(LOCAL_AGENT_ID, {
-					jobId,
-					scheduleId: activeExecution.scheduleShortId,
-				})
-			) {
-				requestedCancellationsByScheduleId.delete(request.scheduleId);
-				clearActiveExecution(jobId);
-				activeExecution.resolve({ status: "cancelled" });
-			}
-		};
-
-		request.signal.addEventListener("abort", handleAbort, { once: true });
-
-		if (request.signal.aborted) {
-			return completion;
-		}
-
-		dispatched = agentManager.sendBackup(LOCAL_AGENT_ID, payload);
-		if (!dispatched) {
-			request.signal.removeEventListener("abort", handleAbort);
-			requestedCancellationsByScheduleId.delete(request.scheduleId);
-			clearActiveExecution(jobId);
-			return {
-				status: "unavailable",
-				error: new Error("Local backup agent is not connected"),
-			} satisfies BackupExecutionResult;
-		}
-
 		try {
-			return await completion;
+			const volumePath = getVolumePath(volume);
-		} finally {
+			const backupOptions = createBackupOptions(schedule, volumePath, signal);
-			request.signal.removeEventListener("abort", handleAbort);
+
+			const result = await restic.backup(repository.config, volumePath, {
+				...backupOptions,
+				compressionMode: repository.compressionMode ?? "auto",
+				organizationId,
+				onProgress,
+			});
+
+			return {
+				status: "completed",
+				exitCode: result.exitCode,
+				result: result.result,
+				warningDetails: result.warningDetails,
+			} satisfies BackupExecutionResult;
+		} catch (error) {
+			return {
+				status: "failed",
+				error,
+			} satisfies BackupExecutionResult;
 		}
 	},
 	cancel: (scheduleId: number) => {
-		const abortController = trackedAbortControllersByScheduleId.get(scheduleId);
+		const abortController = activeControllersByScheduleId.get(scheduleId);
 		if (!abortController) {
 			return false;
 		}
 
-		abortController.abort(new Error("Backup was stopped by the user"));
+		abortController.abort();
 		return true;
 	},
 };

app/server/modules/backups/backups.execution.ts

@@ -0,0 +1,11 @@
+import { backupsService } from "./backups.service";
+
+export const backupsExecutionService = {
+	executeBackup: backupsService.executeBackup,
+	validateBackupExecution: backupsService.validateBackupExecution,
+	getSchedulesToExecute: backupsService.getSchedulesToExecute,
+	stopBackup: backupsService.stopBackup,
+	runForget: backupsService.runForget,
+	copyToMirrors: backupsService.copyToMirrors,
+	getBackupProgress: backupsService.getBackupProgress,
+};

app/server/modules/backups/backups.service.ts

@@ -44,6 +44,7 @@ const getScheduleById = async (scheduleId: number) => {
 const getScheduleByShortId = async (shortId: ShortId) => {
 	return getScheduleByIdOrShortId(shortId);
 };
+
 const createSchedule = async (data: CreateBackupScheduleBody) => {
 	const organizationId = getOrganizationId();
 	if (data.cronExpression && !isValidCron(data.cronExpression)) {
@@ -433,10 +434,6 @@ const executeBackup = async (scheduleId: number, manual = false) => {
 				case "failed":
 					return handleBackupFailure(scheduleId, ctx.organizationId, executionResult.error, ctx);
 				case "cancelled":
-					if (abortController.signal.aborted) {
-						return;
-					}
-
 					return handleBackupCancellation(scheduleId, ctx.organizationId, executionResult.message);
 			}
 		} finally {

app/server/modules/lifecycle/__tests__/shutdown.test.ts

@@ -1,58 +0,0 @@
-import { afterEach, describe, expect, mock, spyOn, test } from "bun:test";
-import { Scheduler } from "../../../core/scheduler";
-import * as agentsManagerModule from "../../agents/agents-manager";
-import * as backendModule from "../../backends/backend";
-import type { VolumeBackend } from "../../backends/backend";
-import { createTestVolume } from "~/test/helpers/volume";
-
-const loadShutdownModule = async () => {
-	const moduleUrl = new URL("../shutdown.ts", import.meta.url);
-	moduleUrl.searchParams.set("test", crypto.randomUUID());
-	return import(moduleUrl.href);
-};
-
-afterEach(() => {
-	mock.restore();
-});
-
-describe("shutdown", () => {
-	test("stops the agent runtime before unmounting mounted volumes", async () => {
-		const events: string[] = [];
-		const stopScheduler = mock(async () => {
-			events.push("scheduler.stop");
-		});
-		const stopAgentRuntime = mock(async () => {
-			events.push("agents.stop");
-		});
-		const unmountVolume = mock(async () => {
-			events.push("backend.unmount");
-			return { status: "unmounted" as const };
-		});
-
-		await createTestVolume({
-			name: "Shutdown test volume",
-			config: {
-				backend: "directory",
-				path: "/Applications",
-			},
-			status: "mounted",
-		});
-
-		spyOn(Scheduler, "stop").mockImplementation(stopScheduler);
-		spyOn(agentsManagerModule, "stopAgentRuntime").mockImplementation(stopAgentRuntime);
-		spyOn(backendModule, "createVolumeBackend").mockImplementation(
-			() =>
-				({
-					mount: async () => ({ status: "mounted" as const }),
-					unmount: unmountVolume,
-					checkHealth: async () => ({ status: "mounted" as const }),
-				}) satisfies VolumeBackend,
-		);
-
-		const { shutdown } = await loadShutdownModule();
-
-		await shutdown();
-
-		expect(events).toEqual(["scheduler.stop", "agents.stop", "backend.unmount"]);
-	});
-});

app/server/modules/lifecycle/bootstrap.ts

@@ -8,9 +8,9 @@ let bootstrapPromise: Promise<void> | undefined;
 const runBootstrap = async () => {
 	await runDbMigrations();
 	await runMigrations();
-	agentManager.start();
-	await spawnLocalAgent();
 	await startup();
+	agentManager.start();
+	spawnLocalAgent();
 };
 
 export const bootstrapApplication = async () => {

app/server/modules/lifecycle/shutdown.ts

@@ -2,11 +2,9 @@ import { Scheduler } from "../../core/scheduler";
 import { db } from "../../db/db";
 import { logger } from "@zerobyte/core/node";
 import { createVolumeBackend } from "../backends/backend";
-import { stopAgentRuntime } from "../agents/agents-manager";
 
 export const shutdown = async () => {
 	await Scheduler.stop();
-	await stopAgentRuntime();
 
 	const volumes = await db.query.volumesTable.findMany({
 		where: { status: "mounted" },

app/server/plugins/bootstrap.ts

@@ -2,20 +2,8 @@ import { definePlugin } from "nitro";
 import { bootstrapApplication } from "../modules/lifecycle/bootstrap";
 import { logger } from "@zerobyte/core/node";
 import { toMessage } from "../utils/errors";
-import { stopAgentRuntime } from "../modules/agents/agents-manager";
-
-type ProcessWithAgentCloseHook = NodeJS.Process & {
-	__zerobyteAgentRuntimeCloseHookRegistered?: boolean;
-};
-
-export default definePlugin(async (nitroApp) => {
-	const runtimeProcess = process as ProcessWithAgentCloseHook;
-
-	if (!runtimeProcess.__zerobyteAgentRuntimeCloseHookRegistered) {
-		nitroApp.hooks.hook("close", stopAgentRuntime);
-		runtimeProcess.__zerobyteAgentRuntimeCloseHookRegistered = true;
-	}
 
+export default definePlugin(async () => {
 	await bootstrapApplication().catch((err) => {
 		logger.error(`Bootstrap failed: ${toMessage(err)}`);
 		process.exit(1);

app/test/helpers/agent-mock.ts

@@ -1,105 +0,0 @@
-import { mock } from "bun:test";
-import { fromAny } from "@total-typescript/shoehorn";
-import { agentManager } from "~/server/modules/agents/agents-manager";
-
-export const createAgentBackupMocks = (
-	resticBackupMock: (params: never) => Promise<{
-		exitCode: number;
-		summary: string;
-		error: string;
-		stderr?: string;
-	}>,
-) => {
-	const runningJobs = new Map<string, { scheduleId: string; cancelled: boolean }>();
-
-	const sendBackupMock = mock((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
-		const handlers = agentManager.getBackupEventHandlers();
-
-		runningJobs.set(payload.jobId, { scheduleId: payload.scheduleId, cancelled: false });
-
-		handlers.onBackupStarted?.({
-			agentId: "local",
-			agentName: "local",
-			payload: { jobId: payload.jobId, scheduleId: payload.scheduleId },
-		});
-
-		void (async () => {
-			const stderrLines: string[] = [];
-			const result = await resticBackupMock(
-				fromAny({
-					onStderr: (line: string) => {
-						stderrLines.push(line);
-					},
-				}),
-			);
-			const running = runningJobs.get(payload.jobId);
-			if (!running || running.cancelled) {
-				return;
-			}
-
-			if (result.exitCode === 0 || result.exitCode === 3) {
-				let parsedResult: Record<string, unknown> | null = null;
-				if (result.summary) {
-					try {
-						parsedResult = JSON.parse(result.summary) as Record<string, unknown>;
-					} catch {
-						parsedResult = null;
-					}
-				}
-
-				handlers.onBackupCompleted?.({
-					agentId: "local",
-					agentName: "local",
-					payload: {
-						jobId: payload.jobId,
-						scheduleId: payload.scheduleId,
-						exitCode: result.exitCode,
-						result: fromAny(parsedResult),
-						warningDetails: stderrLines.join("\n") || undefined,
-					},
-				});
-			} else {
-				const resultWithStderr = result as typeof result & { stderr?: string };
-				const errorDetails = stderrLines.join("\n") || resultWithStderr.stderr || result.error;
-
-				handlers.onBackupFailed?.({
-					agentId: "local",
-					agentName: "local",
-					payload: {
-						jobId: payload.jobId,
-						scheduleId: payload.scheduleId,
-						error: result.error || `Backup failed with code ${result.exitCode}`,
-						errorDetails,
-					},
-				});
-			}
-
-			runningJobs.delete(payload.jobId);
-		})().catch(() => {});
-
-		return true;
-	});
-
-	const cancelBackupMock = mock((_agentId: string, payload: { jobId: string; scheduleId: string }) => {
-		const running = runningJobs.get(payload.jobId);
-		if (!running) {
-			return false;
-		}
-
-		running.cancelled = true;
-		const handlers = agentManager.getBackupEventHandlers();
-		handlers.onBackupCancelled?.({
-			agentId: "local",
-			agentName: "local",
-			payload: {
-				jobId: payload.jobId,
-				scheduleId: payload.scheduleId,
-				message: "Backup was stopped by user",
-			},
-		});
-		runningJobs.delete(payload.jobId);
-		return true;
-	});
-
-	return { sendBackupMock, cancelBackupMock };
-};

apps/agent/.gitignore

@@ -1,34 +0,0 @@
-# dependencies (bun install)
-node_modules
-
-# output
-out
-dist
-*.tgz
-
-# code coverage
-coverage
-*.lcov
-
-# logs
-logs
-_.log
-report.[0-9]_.[0-9]_.[0-9]_.[0-9]_.json
-
-# dotenv environment variable files
-.env
-.env.development.local
-.env.test.local
-.env.production.local
-.env.local
-
-# caches
-.eslintcache
-.cache
-*.tsbuildinfo
-
-# IntelliJ based IDEs
-.idea
-
-# Finder (MacOS) folder config
-.DS_Store

apps/agent/package.json

@@ -1,17 +0,0 @@
-{
-	"name": "agent",
-	"private": true,
-	"type": "module",
-	"module": "index.ts",
-	"dependencies": {
-		"@zerobyte/contracts": "workspace:*",
-		"@zerobyte/core": "workspace:*",
-		"effect": "^3.18.4"
-	},
-	"devDependencies": {
-		"@types/bun": "latest"
-	},
-	"peerDependencies": {
-		"typescript": "^5"
-	}
-}

apps/agent/src/commands/backup-cancel.ts

@@ -1,20 +0,0 @@
-import { Effect } from "effect";
-import { type BackupCancelPayload } from "@zerobyte/contracts/agent-protocol";
-import { logger } from "@zerobyte/core/node";
-import type { ControllerCommandContext } from "../context";
-
-export const handleBackupCancelCommand = (context: ControllerCommandContext, payload: BackupCancelPayload) =>
-	Effect.sync(() => {
-		const running = context.getRunningJob(payload.jobId);
-		if (!running) {
-			logger.warn(`Backup ${payload.jobId} is not running`);
-			return;
-		}
-
-		if (running.scheduleId !== payload.scheduleId) {
-			logger.warn(`Ignoring cancel for backup ${payload.jobId} due to schedule mismatch ${payload.scheduleId}`);
-			return;
-		}
-
-		running.abortController.abort();
-	});

apps/agent/src/commands/backup-run.ts

@@ -1,108 +0,0 @@
-import { Effect } from "effect";
-import { createAgentMessage, type BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
-import { logger } from "@zerobyte/core/node";
-import { type ResticDeps } from "@zerobyte/core/restic";
-import { createRestic } from "@zerobyte/core/restic/server";
-import { toErrorDetails, toMessage } from "@zerobyte/core/utils";
-import type { ControllerCommandContext } from "../context";
-
-export const handleBackupRunCommand = (context: ControllerCommandContext, payload: BackupRunPayload) =>
-	Effect.fork(
-		Effect.gen(function* () {
-			const existing = context.getRunningJob(payload.jobId);
-			if (existing) {
-				context.offerOutbound(
-					createAgentMessage("backup.failed", {
-						jobId: payload.jobId,
-						scheduleId: payload.scheduleId,
-						error: "Backup job is already running",
-					}),
-				);
-				return;
-			}
-
-			logger.info(`Starting backup ${payload.jobId} for schedule ${payload.scheduleId}`);
-			const abortController = new AbortController();
-			context.setRunningJob(payload.jobId, { scheduleId: payload.scheduleId, abortController });
-
-			context.offerOutbound(
-				createAgentMessage("backup.started", {
-					jobId: payload.jobId,
-					scheduleId: payload.scheduleId,
-				}),
-			);
-
-			const deps: ResticDeps = {
-				resolveSecret: async (encrypted) => encrypted,
-				getOrganizationResticPassword: async () => payload.runtime.password,
-				resticCacheDir: payload.runtime.cacheDir,
-				resticPassFile: payload.runtime.passFile,
-				defaultExcludes: payload.runtime.defaultExcludes,
-				hostname: payload.runtime.hostname,
-			};
-
-			const restic = createRestic(deps);
-
-			try {
-				const result = yield* Effect.tryPromise(() =>
-					restic.backup(payload.repositoryConfig, payload.sourcePath, {
-						organizationId: payload.organizationId,
-						...payload.options,
-						signal: abortController.signal,
-						onProgress: (progress) => {
-							context.offerOutbound(
-								createAgentMessage("backup.progress", {
-									jobId: payload.jobId,
-									scheduleId: payload.scheduleId,
-									progress,
-								}),
-							);
-						},
-					}),
-				);
-
-				if (abortController.signal.aborted) {
-					context.offerOutbound(
-						createAgentMessage("backup.cancelled", {
-							jobId: payload.jobId,
-							scheduleId: payload.scheduleId,
-							message: "Backup was cancelled",
-						}),
-					);
-					return;
-				}
-
-				context.offerOutbound(
-					createAgentMessage("backup.completed", {
-						jobId: payload.jobId,
-						scheduleId: payload.scheduleId,
-						exitCode: result.exitCode,
-						result: result.result,
-						warningDetails: result.warningDetails ?? undefined,
-					}),
-				);
-			} catch (error) {
-				if (abortController.signal.aborted) {
-					context.offerOutbound(
-						createAgentMessage("backup.cancelled", {
-							jobId: payload.jobId,
-							scheduleId: payload.scheduleId,
-							message: "Backup was cancelled",
-						}),
-					);
-					return;
-				}
-
-				context.offerOutbound(
-					createAgentMessage("backup.failed", {
-						jobId: payload.jobId,
-						scheduleId: payload.scheduleId,
-						error: toMessage(error),
-						errorDetails: toErrorDetails(error),
-					}),
-				);
-			} finally {
-				context.deleteRunningJob(payload.jobId);
-			}
-		}),
-	).pipe(Effect.asVoid);

apps/agent/src/commands/heartbeat-ping.ts

@@ -1,14 +0,0 @@
-import { Effect } from "effect";
-import { createAgentMessage, type ControllerMessage } from "@zerobyte/contracts/agent-protocol";
-import type { ControllerCommandContext } from "../context";
-
-type HeartbeatPingPayload = Extract<ControllerMessage, { type: "heartbeat.ping" }>["payload"];
-
-export const handleHeartbeatPingCommand = (context: ControllerCommandContext, payload: HeartbeatPingPayload) =>
-	Effect.sync(() => {
-		context.offerOutbound(
-			createAgentMessage("heartbeat.pong", {
-				sentAt: payload.sentAt,
-			}),
-		);
-	});

apps/agent/src/commands/index.ts

@@ -1,19 +0,0 @@
-import type { ControllerMessage } from "@zerobyte/contracts/agent-protocol";
-import { handleBackupCancelCommand } from "./backup-cancel";
-import { handleBackupRunCommand } from "./backup-run";
-import type { ControllerCommandContext } from "../context";
-import { handleHeartbeatPingCommand } from "./heartbeat-ping";
-
-export const handleControllerCommand = (context: ControllerCommandContext, message: ControllerMessage) => {
-	switch (message.type) {
-		case "backup.run": {
-			return handleBackupRunCommand(context, message.payload);
-		}
-		case "backup.cancel": {
-			return handleBackupCancelCommand(context, message.payload);
-		}
-		case "heartbeat.ping": {
-			return handleHeartbeatPingCommand(context, message.payload);
-		}
-	}
-};

apps/agent/src/context.ts

@@ -1,13 +0,0 @@
-import type { AgentWireMessage } from "@zerobyte/contracts/agent-protocol";
-
-export type RunningJob = {
-	scheduleId: string;
-	abortController: AbortController;
-};
-
-export type ControllerCommandContext = {
-	getRunningJob: (jobId: string) => RunningJob | undefined;
-	setRunningJob: (jobId: string, job: RunningJob) => void;
-	deleteRunningJob: (jobId: string) => void;
-	offerOutbound: (message: AgentWireMessage) => void;
-};

apps/agent/src/controller-session.ts

@@ -1,126 +0,0 @@
-import { Effect, Fiber, Queue, Ref } from "effect";
-import {
-	createAgentMessage,
-	parseControllerMessage,
-	type AgentWireMessage,
-	type ControllerWireMessage,
-} from "@zerobyte/contracts/agent-protocol";
-import { logger } from "@zerobyte/core/node";
-import { toMessage } from "@zerobyte/core/utils";
-import { handleControllerCommand } from "./commands";
-
-export type ControllerSession = {
-	onOpen: () => void;
-	onMessage: (data: unknown) => void;
-	close: () => void;
-};
-
-export const createControllerSession = (ws: WebSocket): ControllerSession => {
-	const outboundQueue = Effect.runSync(Queue.bounded<AgentWireMessage>(64));
-	const inboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
-	const runningJobsRef = Effect.runSync(
-		Ref.make<Map<string, { scheduleId: string; abortController: AbortController }>>(new Map()),
-	);
-
-	const getRunningJob = (jobId: string) => Effect.runSync(Ref.get(runningJobsRef)).get(jobId);
-
-	const setRunningJob = (jobId: string, job: { scheduleId: string; abortController: AbortController }) => {
-		Effect.runSync(
-			Ref.update(runningJobsRef, (current) => {
-				const next = new Map(current);
-				next.set(jobId, job);
-				return next;
-			}),
-		);
-	};
-
-	const deleteRunningJob = (jobId: string) => {
-		Effect.runSync(
-			Ref.update(runningJobsRef, (current) => {
-				const next = new Map(current);
-				next.delete(jobId);
-				return next;
-			}),
-		);
-	};
-
-	const offerOutbound = (message: AgentWireMessage) => {
-		void Effect.runPromise(Queue.offer(outboundQueue, message)).catch((error) => {
-			logger.error(`Failed to queue outbound controller message: ${toMessage(error)}`);
-		});
-	};
-
-	const offerInbound = (message: ControllerWireMessage) => {
-		void Effect.runPromise(Queue.offer(inboundQueue, message)).catch((error) => {
-			logger.error(`Failed to queue inbound controller message: ${toMessage(error)}`);
-		});
-	};
-
-	const commandContext = {
-		getRunningJob,
-		setRunningJob,
-		deleteRunningJob,
-		offerOutbound,
-	};
-
-	const writerFiber = Effect.runFork(
-		Effect.forever(
-			Effect.gen(function* () {
-				const message = yield* Queue.take(outboundQueue);
-				yield* Effect.sync(() => {
-					try {
-						ws.send(message);
-					} catch (error) {
-						logger.error(`Failed to send controller message: ${toMessage(error)}`);
-					}
-				});
-			}),
-		),
-	);
-
-	const processorFiber = Effect.runFork(
-		Effect.forever(
-			Effect.gen(function* () {
-				const data = yield* Queue.take(inboundQueue);
-				const parsed = parseControllerMessage(data);
-
-				if (parsed === null) {
-					logger.warn("Agent received invalid JSON");
-					return;
-				}
-
-				if (!parsed.success) {
-					logger.warn(`Agent received an invalid message: ${parsed.error.message}`);
-					return;
-				}
-
-				yield* handleControllerCommand(commandContext, parsed.data);
-			}),
-		),
-	);
-
-	return {
-		onOpen: () => {
-			offerOutbound(createAgentMessage("agent.ready", { agentId: "" }));
-		},
-		onMessage: (data) => {
-			if (typeof data !== "string") {
-				logger.warn("Agent received a non-text message");
-				return;
-			}
-
-			offerInbound(data as ControllerWireMessage);
-		},
-		close: () => {
-			const runningJobs = Effect.runSync(Ref.get(runningJobsRef));
-			for (const running of runningJobs.values()) {
-				running.abortController.abort();
-			}
-			Effect.runSync(Ref.set(runningJobsRef, new Map()));
-			void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {});
-			void Effect.runPromise(Fiber.interrupt(processorFiber)).catch(() => {});
-			void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {});
-			void Effect.runPromise(Queue.shutdown(inboundQueue)).catch(() => {});
-		},
-	};
-};

apps/agent/src/index.ts

@@ -1,47 +0,0 @@
-import { logger } from "@zerobyte/core/node";
-import { createControllerSession, type ControllerSession } from "./controller-session";
-
-const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL;
-const agentToken = process.env.ZEROBYTE_AGENT_TOKEN;
-
-class Agent {
-	private ws: WebSocket | null = null;
-	private controllerSession: ControllerSession | null = null;
-
-	connect() {
-		if (!controllerUrl) {
-			throw new Error("Env variable ZEROBYTE_CONTROLLER_URL is not set");
-		}
-
-		if (!agentToken) {
-			throw new Error("Env variable ZEROBYTE_AGENT_TOKEN is not set");
-		}
-
-		const url = new URL(controllerUrl);
-		url.searchParams.set("token", agentToken);
-
-		this.ws = new WebSocket(url.toString());
-		this.controllerSession = createControllerSession(this.ws);
-
-		this.ws.onopen = () => {
-			logger.info("Agent connected to controller");
-			this.controllerSession?.onOpen();
-		};
-
-		this.ws.onmessage = (event) => {
-			this.controllerSession?.onMessage(event.data);
-		};
-		this.ws.onclose = () => {
-			this.controllerSession?.close();
-			this.controllerSession = null;
-			this.ws = null;
-			logger.info("Agent disconnected from controller");
-		};
-		this.ws.onerror = (error) => {
-			logger.error("Agent encountered an error:", error);
-		};
-	}
-}
-
-const agent = new Agent();
-agent.connect();

apps/agent/tsconfig.json

@@ -1,29 +0,0 @@
-{
-	"compilerOptions": {
-		// Environment setup & latest features
-		"lib": ["ESNext"],
-		"target": "ESNext",
-		"module": "Preserve",
-		"moduleDetection": "force",
-		"jsx": "react-jsx",
-		"allowJs": true,
-
-		// Bundler mode
-		"moduleResolution": "bundler",
-		"allowImportingTsExtensions": true,
-		"verbatimModuleSyntax": true,
-		"noEmit": true,
-
-		// Best practices
-		"strict": true,
-		"skipLibCheck": true,
-		"noFallthroughCasesInSwitch": true,
-		"noUncheckedIndexedAccess": true,
-		"noImplicitOverride": true,
-
-		// Some stricter flags (disabled by default)
-		"noUnusedLocals": false,
-		"noUnusedParameters": false,
-		"noPropertyAccessFromIndexSignature": false
-	}
-}

bun.lock

@@ -34,7 +34,6 @@
         "@tanstack/react-router": "^1.168.1",
         "@tanstack/react-router-ssr-query": "^1.166.10",
         "@tanstack/react-start": "^1.167.1",
-        "@zerobyte/contracts": "workspace:*",
         "@zerobyte/core": "workspace:*",
         "better-auth": "^1.5.5",
         "class-variance-authority": "^0.7.1",
@@ -47,7 +46,6 @@
         "dither-plugin": "^1.1.1",
         "dotenv": "^17.3.1",
         "drizzle-orm": "^1.0.0-beta.16-ea816b6",
-        "effect": "^3.18.4",
         "es-toolkit": "^1.45.1",
         "hono": "^4.12.8",
         "hono-openapi": "^1.3.0",
@@ -116,36 +114,10 @@
         "wait-for-expect": "^4.0.0",
       },
     },
-    "apps/agent": {
-      "name": "agent",
-      "dependencies": {
-        "@zerobyte/contracts": "workspace:*",
-        "@zerobyte/core": "workspace:*",
-        "effect": "^3.18.4",
-      },
-      "devDependencies": {
-        "@types/bun": "latest",
-      },
-      "peerDependencies": {
-        "typescript": "^5",
-      },
-    },
-    "packages/contracts": {
-      "name": "@zerobyte/contracts",
-      "dependencies": {
-        "@zerobyte/core": "workspace:*",
-      },
-      "devDependencies": {
-        "@types/bun": "latest",
-      },
-      "peerDependencies": {
-        "typescript": "^5",
-      },
-    },
     "packages/core": {
       "name": "@zerobyte/core",
       "devDependencies": {
-        "@types/bun": "^1.3.11",
+        "@types/bun": "latest",
       },
       "peerDependencies": {
         "typescript": "^5",
@@ -1072,16 +1044,12 @@
 
     "@xmldom/xmldom": ["@xmldom/xmldom@0.8.11", "", {}, "sha512-cQzWCtO6C8TQiYl1ruKNn2U6Ao4o4WBBcbL61yJl84x+j5sOWWFU9X7DpND8XZG3daDppSsigMdfAIl2upQBRw=="],
 
-    "@zerobyte/contracts": ["@zerobyte/contracts@workspace:packages/contracts"],
-
     "@zerobyte/core": ["@zerobyte/core@workspace:packages/core"],
 
     "abort-controller": ["abort-controller@3.0.0", "", { "dependencies": { "event-target-shim": "^5.0.0" } }, "sha512-h8lQ8tacZYnR3vNQTgibj+tODHI5/+l06Au2Pcriv/Gmet0eaj4TwWH41sO9wnHDiQsEj19q0drzdWdeAHtweg=="],
 
     "acorn": ["acorn@8.16.0", "", { "bin": { "acorn": "bin/acorn" } }, "sha512-UVJyE9MttOsBQIDKw1skb9nAwQuR5wuGD3+82K6JgJlm/Y+KI92oNsMNGZCYdDsVtRHSak0pcV5Dno5+4jh9sw=="],
 
-    "agent": ["agent@workspace:apps/agent"],
-
     "agent-base": ["agent-base@7.1.4", "", {}, "sha512-MnA+YT8fwfJPgBx3m60MNqakm30XOkyIoH1y6huTQvC0PwZG7ki8NacLBcrPbNoo8vEZy7Jpuk7+jMO+CUovTQ=="],
 
     "ansi-colors": ["ansi-colors@4.1.3", "", {}, "sha512-/6w/C21Pm1A7aZitlI5Ni/2J6FFQN8i1Cvz3kHABAAbw93v/NlvKdVOqz7CCWz/3iv/JplRSEEZ83XION15ovw=="],

package.json

@@ -59,7 +59,6 @@
 		"@tanstack/react-router": "^1.168.1",
 		"@tanstack/react-router-ssr-query": "^1.166.10",
 		"@tanstack/react-start": "^1.167.1",
-		"@zerobyte/contracts": "workspace:*",
 		"@zerobyte/core": "workspace:*",
 		"better-auth": "^1.5.5",
 		"class-variance-authority": "^0.7.1",
@@ -72,7 +71,6 @@
 		"dither-plugin": "^1.1.1",
 		"dotenv": "^17.3.1",
 		"drizzle-orm": "^1.0.0-beta.16-ea816b6",
-		"effect": "^3.18.4",
 		"es-toolkit": "^1.45.1",
 		"hono": "^4.12.8",
 		"hono-openapi": "^1.3.0",

packages/contracts/.gitignore

@@ -1,34 +0,0 @@
-# dependencies (bun install)
-node_modules
-
-# output
-out
-dist
-*.tgz
-
-# code coverage
-coverage
-*.lcov
-
-# logs
-logs
-_.log
-report.[0-9]_.[0-9]_.[0-9]_.[0-9]_.json
-
-# dotenv environment variable files
-.env
-.env.development.local
-.env.test.local
-.env.production.local
-.env.local
-
-# caches
-.eslintcache
-.cache
-*.tsbuildinfo
-
-# IntelliJ based IDEs
-.idea
-
-# Finder (MacOS) folder config
-.DS_Store

packages/contracts/README.md

@@ -1,15 +0,0 @@
-# contracts
-
-To install dependencies:
-
-```bash
-bun install
-```
-
-To run:
-
-```bash
-bun run index.ts
-```
-
-This project was created using `bun init` in bun v1.3.10. [Bun](https://bun.com) is a fast all-in-one JavaScript runtime.

packages/contracts/package.json

@@ -1,24 +0,0 @@
-{
-	"name": "@zerobyte/contracts",
-	"private": true,
-	"type": "module",
-	"exports": {
-		"./agent-protocol": {
-			"types": "./src/agent-protocol.ts",
-			"import": "./src/agent-protocol.ts",
-			"default": "./src/agent-protocol.ts"
-		}
-	},
-	"scripts": {
-		"tsc": "tsc --noEmit"
-	},
-	"dependencies": {
-		"@zerobyte/core": "workspace:*"
-	},
-	"devDependencies": {
-		"@types/bun": "latest"
-	},
-	"peerDependencies": {
-		"typescript": "^5"
-	}
-}

packages/contracts/src/agent-protocol.ts

@@ -1,219 +0,0 @@
-import { z } from "zod";
-import { safeJsonParse } from "@zerobyte/core/utils";
-import {
-	repositoryConfigSchema,
-	resticBackupOutputSchema,
-	resticBackupProgressSchema,
-	type CompressionMode,
-} from "@zerobyte/core/restic";
-
-const compressionModeSchema = z.enum(["off", "auto", "max"]) satisfies z.ZodType<CompressionMode>;
-
-const backupExecutionOptionsSchema = z
-	.object({
-		tags: z.array(z.string()).optional(),
-		oneFileSystem: z.boolean().optional(),
-		exclude: z.array(z.string()).optional(),
-		excludeIfPresent: z.array(z.string()).optional(),
-		includePaths: z.array(z.string()).optional(),
-		includePatterns: z.array(z.string()).optional(),
-		customResticParams: z.array(z.string()).optional(),
-		compressionMode: compressionModeSchema.optional(),
-	})
-	.strict();
-
-const backupRuntimeSchema = z
-	.object({
-		password: z.string(),
-		cacheDir: z.string(),
-		passFile: z.string(),
-		defaultExcludes: z.array(z.string()),
-		hostname: z.string().optional(),
-	})
-	.strict();
-
-const backupRunSchema = z
-	.object({
-		type: z.literal("backup.run"),
-		payload: z
-			.object({
-				jobId: z.string(),
-				scheduleId: z.string(),
-				organizationId: z.string(),
-				sourcePath: z.string(),
-				repositoryConfig: repositoryConfigSchema,
-				options: backupExecutionOptionsSchema,
-				runtime: backupRuntimeSchema,
-			})
-			.strict(),
-	})
-	.strict();
-
-const backupCancelSchema = z
-	.object({
-		type: z.literal("backup.cancel"),
-		payload: z.object({ jobId: z.string(), scheduleId: z.string() }).strict(),
-	})
-	.strict();
-
-const heartbeatPingSchema = z
-	.object({
-		type: z.literal("heartbeat.ping"),
-		payload: z.object({ sentAt: z.number() }),
-	})
-	.strict();
-
-const agentReadySchema = z
-	.object({
-		type: z.literal("agent.ready"),
-		payload: z.object({ agentId: z.string() }),
-	})
-	.strict();
-
-const backupStartedSchema = z
-	.object({
-		type: z.literal("backup.started"),
-		payload: z.object({ jobId: z.string(), scheduleId: z.string() }),
-	})
-	.strict();
-
-const backupProgressSchema = z
-	.object({
-		type: z.literal("backup.progress"),
-		payload: z
-			.object({
-				jobId: z.string(),
-				scheduleId: z.string(),
-				progress: resticBackupProgressSchema,
-			})
-			.strict(),
-	})
-	.strict();
-
-const backupCompletedSchema = z
-	.object({
-		type: z.literal("backup.completed"),
-		payload: z
-			.object({
-				jobId: z.string(),
-				scheduleId: z.string(),
-				exitCode: z.number(),
-				result: resticBackupOutputSchema.nullable(),
-				warningDetails: z.string().optional(),
-			})
-			.strict(),
-	})
-	.strict();
-
-const backupFailedSchema = z
-	.object({
-		type: z.literal("backup.failed"),
-		payload: z
-			.object({
-				jobId: z.string(),
-				scheduleId: z.string(),
-				error: z.string(),
-				errorDetails: z.string().optional(),
-			})
-			.strict(),
-	})
-	.strict();
-
-const backupCancelledSchema = z
-	.object({
-		type: z.literal("backup.cancelled"),
-		payload: z
-			.object({
-				jobId: z.string(),
-				scheduleId: z.string(),
-				message: z.string().optional(),
-			})
-			.strict(),
-	})
-	.strict();
-
-const heartbeatPongSchema = z
-	.object({
-		type: z.literal("heartbeat.pong"),
-		payload: z.object({ sentAt: z.number() }),
-	})
-	.strict();
-
-const controllerMessageSchema = z.discriminatedUnion("type", [
-	backupRunSchema,
-	backupCancelSchema,
-	heartbeatPingSchema,
-]);
-const agentMessageSchema = z.discriminatedUnion("type", [
-	agentReadySchema,
-	backupStartedSchema,
-	backupProgressSchema,
-	backupCompletedSchema,
-	backupFailedSchema,
-	backupCancelledSchema,
-	heartbeatPongSchema,
-]);
-
-export type BackupRunPayload = z.infer<typeof backupRunSchema>["payload"];
-export type BackupCancelPayload = z.infer<typeof backupCancelSchema>["payload"];
-export type BackupStartedPayload = z.infer<typeof backupStartedSchema>["payload"];
-export type BackupProgressPayload = z.infer<typeof backupProgressSchema>["payload"];
-export type BackupCompletedPayload = z.infer<typeof backupCompletedSchema>["payload"];
-export type BackupFailedPayload = z.infer<typeof backupFailedSchema>["payload"];
-export type BackupCancelledPayload = z.infer<typeof backupCancelledSchema>["payload"];
-export type ControllerMessage = z.infer<typeof controllerMessageSchema>;
-export type AgentMessage = z.infer<typeof agentMessageSchema>;
-
-type Brand<TValue, TBrand extends string> = TValue & {
-	readonly __brand: TBrand;
-};
-
-export type ControllerWireMessage = Brand<string, "ControllerWireMessage">;
-export type AgentWireMessage = Brand<string, "AgentWireMessage">;
-
-type PayloadForMessage<TMessage extends { type: string; payload: unknown }, TType extends TMessage["type"]> = Extract<
-	TMessage,
-	{ type: TType }
->["payload"];
-
-const parseJsonMessage = (data: string) => safeJsonParse<unknown>(data);
-
-export const parseControllerMessage = (data: ControllerWireMessage) => {
-	const parsed = parseJsonMessage(data);
-	if (parsed === null) {
-		return null;
-	}
-
-	return controllerMessageSchema.safeParse(parsed);
-};
-
-export const parseAgentMessage = (data: string) => {
-	const parsed = parseJsonMessage(data);
-	if (parsed === null) {
-		return null;
-	}
-
-	return agentMessageSchema.safeParse(parsed);
-};
-
-export const createControllerMessage = <TType extends ControllerMessage["type"]>(
-	type: TType,
-	payload: PayloadForMessage<ControllerMessage, TType>,
-) =>
-	JSON.stringify(
-		controllerMessageSchema.parse({
-			type,
-			payload,
-		}),
-	) as ControllerWireMessage;
-
-export const createAgentMessage = <TType extends AgentMessage["type"]>(
-	type: TType,
-	payload: PayloadForMessage<AgentMessage, TType>,
-) =>
-	JSON.stringify(
-		agentMessageSchema.parse({
-			type,
-			payload,
-		}),
-	) as AgentWireMessage;

packages/contracts/tsconfig.json

@@ -1,29 +0,0 @@
-{
-	"compilerOptions": {
-		// Environment setup & latest features
-		"lib": ["ESNext"],
-		"target": "ESNext",
-		"module": "Preserve",
-		"moduleDetection": "force",
-		"jsx": "react-jsx",
-		"allowJs": true,
-
-		// Bundler mode
-		"moduleResolution": "bundler",
-		"allowImportingTsExtensions": true,
-		"verbatimModuleSyntax": true,
-		"noEmit": true,
-
-		// Best practices
-		"strict": true,
-		"skipLibCheck": true,
-		"noFallthroughCasesInSwitch": true,
-		"noUncheckedIndexedAccess": true,
-		"noImplicitOverride": true,
-
-		// Some stricter flags (disabled by default)
-		"noUnusedLocals": false,
-		"noUnusedParameters": false,
-		"noPropertyAccessFromIndexSignature": false
-	}
-}

packages/core/package.json

@@ -29,7 +29,7 @@
 		"test": "bunx --bun vitest run --config ./vitest.config.ts"
 	},
 	"devDependencies": {
-		"@types/bun": "^1.3.11"
+		"@types/bun": "latest"
 	},
 	"peerDependencies": {
 		"typescript": "^5"