fix: restart agent after an unexpected exit

.dockerignore

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

Dockerfile

@@ -64,6 +64,8 @@ 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
 
@@ -86,11 +88,14 @@ 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__/agents-manager.test.ts

@@ -0,0 +1,82 @@
+import { EventEmitter } from "node:events";
+import * as childProcess from "node:child_process";
+import { PassThrough } from "node:stream";
+import { afterEach, expect, mock, test, vi } from "bun:test";
+
+const spawnMock = mock();
+
+await mock.module("node:child_process", () => ({
+	...childProcess,
+	spawn: spawnMock,
+}));
+
+import { spawnLocalAgent, stopLocalAgent } from "../agents-manager";
+
+const flushMicrotasks = async () => {
+	await Promise.resolve();
+};
+
+type FakeChildProcess = EventEmitter & {
+	stdout: PassThrough;
+	stderr: PassThrough;
+	exitCode: number | null;
+	signalCode: NodeJS.Signals | null;
+	kill: ReturnType<typeof mock>;
+};
+
+const createFakeChild = () => {
+	const child = new EventEmitter() as FakeChildProcess;
+
+	child.stdout = new PassThrough();
+	child.stderr = new PassThrough();
+	child.exitCode = null;
+	child.signalCode = null;
+	child.kill = mock(() => {
+		child.exitCode = 0;
+		child.emit("exit", 0, null);
+		return true;
+	});
+
+	return child;
+};
+
+afterEach(async () => {
+	await stopLocalAgent();
+	spawnMock.mockReset();
+	mock.restore();
+	vi.useRealTimers();
+});
+
+test("respawns the local agent after an unexpected exit", async () => {
+	vi.useFakeTimers();
+
+	const firstChild = createFakeChild();
+	const secondChild = createFakeChild();
+	spawnMock.mockReturnValueOnce(firstChild).mockReturnValueOnce(secondChild);
+
+	await spawnLocalAgent();
+
+	firstChild.exitCode = 1;
+	firstChild.emit("exit", 1, null);
+
+	vi.advanceTimersByTime(1_000);
+	await flushMicrotasks();
+
+	expect(spawnMock).toHaveBeenCalledTimes(2);
+});
+
+test("does not respawn the local agent after an intentional stop", async () => {
+	vi.useFakeTimers();
+
+	const child = createFakeChild();
+	spawnMock.mockReturnValue(child);
+
+	await spawnLocalAgent();
+	await stopLocalAgent();
+
+	vi.advanceTimersByTime(1_000);
+	await flushMicrotasks();
+
+	expect(spawnMock).toHaveBeenCalledTimes(1);
+	expect(child.kill).toHaveBeenCalledTimes(1);
+});

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

@@ -0,0 +1,169 @@
+import { expect, mock, test } from "bun:test";
+import waitForExpect from "wait-for-expect";
+import { createAgentMessage } from "@zerobyte/contracts/agent-protocol";
+import { createControllerAgentSession } from "../controller-agent-session";
+
+const createSocket = (send = mock(() => 1)) => {
+	const close = mock(() => undefined);
+
+	return {
+		data: { id: "connection-1", agentId: "local", organizationId: null, agentName: "Local Agent" },
+		send,
+		close,
+	};
+};
+
+test("close emits a synthetic backup.cancelled for a started backup", () => {
+	const onBackupCancelled = mock(() => undefined);
+	const session = createControllerAgentSession(
+		createSocket() as unknown as Parameters<typeof createControllerAgentSession>[0],
+		{
+			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() as unknown as Parameters<typeof createControllerAgentSession>[0],
+			{
+				onBackupCancelled,
+			},
+		);
+
+		session.handleMessage(
+			createAgentMessage("backup.started", {
+				jobId: testCase.jobId,
+				scheduleId: testCase.scheduleId,
+			}),
+		);
+		session.handleMessage(testCase.terminalMessage);
+		session.close();
+
+		expect(onBackupCancelled).toHaveBeenCalledTimes(testCase.expectedCancelledCalls);
+	}
+});
+
+test("close emits a synthetic backup.cancelled for a queued backup", () => {
+	const onBackupCancelled = mock(() => undefined);
+	const session = createControllerAgentSession(
+		createSocket() as unknown as Parameters<typeof createControllerAgentSession>[0],
+		{
+			onBackupCancelled,
+		},
+	);
+
+	session.sendBackup({
+		jobId: "job-queued",
+		scheduleId: "schedule-queued",
+		organizationId: "org-1",
+		sourcePath: "/tmp/source",
+		repositoryConfig: {
+			backend: "local",
+			path: "/tmp/repository",
+		},
+		options: {},
+		runtime: {
+			password: "password",
+			cacheDir: "/tmp/cache",
+			passFile: "/tmp/pass",
+			defaultExcludes: [],
+		},
+	});
+
+	session.close();
+
+	expect(onBackupCancelled).toHaveBeenCalledTimes(1);
+	expect(onBackupCancelled).toHaveBeenCalledWith({
+		jobId: "job-queued",
+		scheduleId: "schedule-queued",
+		message:
+			"The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes.",
+	});
+});
+
+test("a dropped backup.cancel closes the session and emits a synthetic backup.cancelled", async () => {
+	const send = mock(() => 0);
+	const socket = createSocket(send);
+	const onBackupCancelled = mock(() => undefined);
+	const session = createControllerAgentSession(
+		socket as unknown as Parameters<typeof createControllerAgentSession>[0],
+		{
+			onBackupCancelled,
+		},
+	);
+
+	session.handleMessage(
+		createAgentMessage("backup.started", {
+			jobId: "job-1",
+			scheduleId: "schedule-1",
+		}),
+	);
+	session.sendBackupCancel({
+		jobId: "job-1",
+		scheduleId: "schedule-1",
+	});
+
+	await waitForExpect(() => {
+		expect(send).toHaveBeenCalledTimes(1);
+		expect(socket.close).toHaveBeenCalledTimes(1);
+		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.",
+		});
+	});
+});

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

@@ -0,0 +1,12 @@
+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

@@ -0,0 +1,372 @@
+import { type ChildProcess, 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,
+	type AgentConnectionData,
+	type ControllerAgentSession,
+} from "./controller-agent-session";
+
+type AgentBackupEventContext = {
+	agentId: string;
+	agentName: string;
+	payload:
+		| BackupStartedPayload
+		| BackupProgressPayload
+		| BackupCompletedPayload
+		| BackupFailedPayload
+		| BackupCancelledPayload;
+};
+
+export type AgentBackupEventHandlers = {
+	onBackupStarted?: (context: AgentBackupEventContext & { payload: BackupStartedPayload }) => void;
+	onBackupProgress?: (context: AgentBackupEventContext & { payload: BackupProgressPayload }) => void;
+	onBackupCompleted?: (context: AgentBackupEventContext & { payload: BackupCompletedPayload }) => void;
+	onBackupFailed?: (context: AgentBackupEventContext & { payload: BackupFailedPayload }) => void;
+	onBackupCancelled?: (context: AgentBackupEventContext & { payload: BackupCancelledPayload }) => void;
+};
+
+type AgentManagerRuntime = ReturnType<typeof createAgentManagerRuntime>;
+type AgentRuntimeState = {
+	agentManager: AgentManagerRuntime;
+	localAgent: ChildProcess | null;
+	isStoppingLocalAgent: boolean;
+	localAgentRestartTimeout: ReturnType<typeof setTimeout> | null;
+};
+
+type ProcessWithAgentRuntime = NodeJS.Process & {
+	__zerobyteAgentRuntime?: AgentRuntimeState;
+};
+
+const getAgentRuntimeState = () => {
+	const runtimeProcess = process as ProcessWithAgentRuntime;
+	const existingRuntime = runtimeProcess.__zerobyteAgentRuntime;
+
+	if (existingRuntime) {
+		return existingRuntime;
+	}
+
+	const runtime = {
+		agentManager: createAgentManagerRuntime(),
+		localAgent: null,
+		isStoppingLocalAgent: false,
+		localAgentRestartTimeout: null,
+	};
+
+	runtimeProcess.__zerobyteAgentRuntime = runtime;
+	return runtime;
+};
+
+const getAgentManagerRuntime = () => getAgentRuntimeState().agentManager;
+
+export const spawnLocalAgent = async () => {
+	await stopLocalAgent();
+
+	const sourceEntryPoint = path.join(process.cwd(), "apps", "agent", "src", "index.ts");
+	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: {
+			PATH: process.env.PATH,
+			ZEROBYTE_CONTROLLER_URL: "ws://localhost:3001",
+			ZEROBYTE_AGENT_TOKEN: agentToken,
+		},
+		stdio: ["ignore", "pipe", "pipe"],
+	});
+
+	runtime.localAgent = agentProcess;
+
+	agentProcess.stdout?.on("data", (data: Buffer) => {
+		const line = data.toString().trim();
+		if (line) logger.info(`[agent] ${line}`);
+	});
+
+	agentProcess.stderr?.on("data", (data: Buffer) => {
+		const line = data.toString().trim();
+		if (line) logger.error(`[agent] ${line}`);
+	});
+
+	agentProcess.on("exit", (code, signal) => {
+		const shouldRestart = runtime.localAgent === agentProcess && !runtime.isStoppingLocalAgent;
+		if (runtime.localAgent === agentProcess) {
+			runtime.localAgent = null;
+		}
+		logger.info(`Agent process exited with code ${code} and signal ${signal}`);
+
+		if (!shouldRestart) {
+			return;
+		}
+
+		runtime.localAgentRestartTimeout = setTimeout(() => {
+			runtime.localAgentRestartTimeout = null;
+			void spawnLocalAgent().catch((error) => {
+				logger.error(`Failed to restart local agent: ${error instanceof Error ? error.message : String(error)}`);
+			});
+		}, 1_000);
+	});
+};
+
+export const stopLocalAgent = async () => {
+	const runtime = getAgentRuntimeState();
+	if (runtime.localAgentRestartTimeout) {
+		clearTimeout(runtime.localAgentRestartTimeout);
+		runtime.localAgentRestartTimeout = null;
+	}
+
+	if (!runtime.localAgent) {
+		return;
+	}
+
+	const agentProcess = runtime.localAgent;
+	runtime.localAgent = null;
+	runtime.isStoppingLocalAgent = true;
+
+	if (agentProcess.exitCode !== null || agentProcess.signalCode !== null) {
+		runtime.isStoppingLocalAgent = false;
+		return;
+	}
+
+	const exited = new Promise<void>((resolve) => {
+		agentProcess.once("exit", () => {
+			runtime.isStoppingLocalAgent = false;
+			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());
+
+		try {
+			const server = Effect.runSync(Scope.extend(acquireServer, scope));
+			runtimeScope = scope;
+			logger.info(`Agent Manager listening on port ${server.port}`);
+		} catch (error) {
+			Effect.runSync(Scope.close(scope, Exit.fail(error)));
+			throw error;
+		}
+	};
+
+	return {
+		start,
+		sendBackup: (agentId: string, payload: BackupRunPayload) => {
+			const session = getSession(agentId);
+
+			if (!session) {
+				logger.warn(`Cannot send backup command. Agent ${agentId} is not connected.`);
+				return false;
+			}
+
+			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

@@ -0,0 +1,255 @@
+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 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 => {
+	let isClosed = false;
+	const outboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
+	const activeBackupJobs = Effect.runSync(Ref.make<Map<string, string>>(new Map()));
+	const pendingBackupJobs = Effect.runSync(Ref.make<Map<string, string>>(new Map()));
+	const isReadyRef = Effect.runSync(Ref.make(false));
+
+	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 setIsReady = (isReady: boolean) => {
+		Effect.runSync(Ref.set(isReadyRef, isReady));
+	};
+
+	const setActiveBackupJob = (jobId: string, scheduleId: string) => {
+		Effect.runSync(
+			Ref.update(activeBackupJobs, (current) => {
+				const next = new Map(current);
+				next.set(jobId, scheduleId);
+				return next;
+			}),
+		);
+	};
+
+	const setPendingBackupJob = (jobId: string, scheduleId: string) => {
+		Effect.runSync(
+			Ref.update(pendingBackupJobs, (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 deletePendingBackupJob = (jobId: string) => {
+		Effect.runSync(
+			Ref.update(pendingBackupJobs, (current) => {
+				const next = new Map(current);
+				next.delete(jobId);
+				return next;
+			}),
+		);
+	};
+
+	const clearTrackedBackupJob = (jobId: string) => {
+		deletePendingBackupJob(jobId);
+		deleteActiveBackupJob(jobId);
+	};
+
+	const cancelTrackedJobs = (jobs: Map<string, string>, message: string) => {
+		for (const [jobId, scheduleId] of jobs) {
+			handlers.onBackupCancelled?.({
+				jobId,
+				scheduleId,
+				message,
+			});
+		}
+	};
+
+	const closeSession = () => {
+		if (isClosed) {
+			return;
+		}
+
+		isClosed = true;
+		setIsReady(false);
+		const pendingJobs = Effect.runSync(Ref.get(pendingBackupJobs));
+		Effect.runSync(Ref.set(pendingBackupJobs, new Map()));
+		cancelTrackedJobs(
+			pendingJobs,
+			"The connection to the backup agent was lost before this backup started. Restart the backup to ensure it completes.",
+		);
+
+		const activeJobs = Effect.runSync(Ref.get(activeBackupJobs));
+		Effect.runSync(Ref.set(activeBackupJobs, new Map()));
+		cancelTrackedJobs(
+			activeJobs,
+			"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(() => {});
+	};
+
+	const handleSendFailure = (reason: string) => {
+		logger.error(
+			`Closing session for agent ${socket.data.agentId} on ${socket.data.id} after an outbound websocket send failed: ${reason}`,
+		);
+
+		try {
+			socket.close();
+		} catch (error) {
+			logger.error(`Failed to close socket for agent ${socket.data.agentId} on ${socket.data.id}: ${toMessage(error)}`);
+		}
+
+		closeSession();
+	};
+
+	const writerFiber = Effect.runFork(
+		Effect.forever(
+			Effect.gen(function* () {
+				const message = yield* Queue.take(outboundQueue);
+				yield* Effect.sync(() => {
+					const sendResult = socket.send(message);
+					if (sendResult <= 0) {
+						handleSendFailure(sendResult === 0 ? "connection issue" : "backpressure");
+					}
+				});
+			}),
+		),
+	);
+
+	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) => {
+		switch (message.type) {
+			case "agent.ready": {
+				setIsReady(true);
+				logger.info(`Agent "${socket.data.agentName}" (${socket.data.agentId}) is ready`);
+				break;
+			}
+			case "backup.started": {
+				deletePendingBackupJob(message.payload.jobId);
+				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": {
+				clearTrackedBackupJob(message.payload.jobId);
+				handlers.onBackupCompleted?.(message.payload);
+				break;
+			}
+			case "backup.failed": {
+				clearTrackedBackupJob(message.payload.jobId);
+				handlers.onBackupFailed?.(message.payload);
+				break;
+			}
+			case "backup.cancelled": {
+				clearTrackedBackupJob(message.payload.jobId);
+				handlers.onBackupCancelled?.(message.payload);
+				break;
+			}
+			case "heartbeat.pong": {
+				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) => {
+			setPendingBackupJob(payload.jobId, payload.scheduleId);
+			offerOutbound(createControllerMessage("backup.run", payload));
+			return payload.jobId;
+		},
+		sendBackupCancel: (payload) => {
+			offerOutbound(createControllerMessage("backup.cancel", payload));
+		},
+		isReady: () => Effect.runSync(Ref.get(isReadyRef)),
+		close: closeSession,
+	};
+};

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

@@ -1,7 +1,6 @@
 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";
@@ -13,8 +12,13 @@ import * as spawnModule from "@zerobyte/core/node";
 import type { SafeSpawnParams } from "@zerobyte/core/node";
 import { restic } from "~/server/core/restic";
 import { NotFoundError, BadRequestError } from "http-errors-enhanced";
+import { fromAny } from "@total-typescript/shoehorn";
+import { scheduleQueries } from "../backups.queries";
 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 { getScheduleByIdOrShortId } from "../helpers/backup-schedule-lookups";
 
 const setup = () => {
 	const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
@@ -22,6 +26,7 @@ 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,
@@ -37,12 +42,16 @@ const setup = () => {
 	vi.spyOn(restic, "forget").mockImplementation(resticForgetMock);
 	vi.spyOn(restic, "copy").mockImplementation(resticCopyMock);
 	vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock);
+	vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock);
+	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
 
 	return {
 		resticBackupMock,
 		resticForgetMock,
 		resticCopyMock,
+		sendBackupMock,
+		cancelBackupMock,
 		refreshStatsMock,
 	};
 };
@@ -63,7 +72,7 @@ describe("backup execution - validation failures", () => {
 		});
 
 		// act
-		const result = await backupsExecutionService.validateBackupExecution(schedule.id);
+		const result = await backupsService.validateBackupExecution(schedule.id);
 
 		// assert
 		expect(result.type).toBe("failure");
@@ -74,10 +83,70 @@ 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 backupsExecutionService.validateBackupExecution(99999);
+		const result = await backupsService.validateBackupExecution(99999);
 
 		// assert
 		expect(result.type).toBe("failure");
@@ -108,9 +177,9 @@ describe("stop backup", () => {
 			});
 		});
 
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("error: open /mnt/data/private.db: permission denied");
 	});
@@ -136,15 +205,63 @@ describe("stop backup", () => {
 			});
 		});
 
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(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();
@@ -172,19 +289,19 @@ describe("stop backup", () => {
 			});
 		});
 
-		const executePromise = backupsExecutionService.executeBackup(schedule.id);
+		const executePromise = backupsService.executeBackup(schedule.id);
 
 		await waitForExpect(async () => {
-			const runningSchedule = await backupsService.getScheduleById(schedule.id);
+			const runningSchedule = await getScheduleByIdOrShortId(schedule.id);
 			expect(runningSchedule.lastBackupStatus).toBe("in_progress");
 		});
 
 		// act
-		await backupsExecutionService.stopBackup(schedule.id);
+		await backupsService.stopBackup(schedule.id);
 		await executePromise;
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 	});
@@ -198,36 +315,25 @@ describe("stop backup", () => {
 			repositoryId: repository.id,
 		});
 
-		vi.spyOn(repoMutex, "acquireShared").mockImplementation((_repositoryId, _operation, signal) => {
+		const releaseLock = await repoMutex.acquireExclusive(repository.id, "test");
-			return new Promise((_, reject) => {
+		const executePromise = backupsService.executeBackup(schedule.id);
-				if (signal?.aborted) {
-					reject(signal.reason instanceof Error ? signal.reason : new Error("Operation aborted"));
-					return;
-				}
-
-				signal?.addEventListener(
-					"abort",
-					() => {
-						reject(signal.reason instanceof Error ? signal.reason : new Error("Operation aborted"));
-					},
-					{ once: true },
-				);
-			});
-		});
-
-		const executePromise = backupsExecutionService.executeBackup(schedule.id);
 
+		try {
 			await waitForExpect(async () => {
-			const queuedSchedule = await backupsService.getScheduleById(schedule.id);
+				const queuedSchedule = await getScheduleByIdOrShortId(schedule.id);
 				expect(queuedSchedule.lastBackupStatus).toBe("in_progress");
 			});
 
 			expect(resticBackupMock).not.toHaveBeenCalled();
 
-		await backupsExecutionService.stopBackup(schedule.id);
+			await backupsService.stopBackup(schedule.id);
+		} finally {
+			releaseLock();
+		}
+
 		await executePromise;
 
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		expect(updatedSchedule.lastBackupError).toBe("Backup was stopped by the user");
 		expect(resticBackupMock).not.toHaveBeenCalled();
@@ -247,20 +353,40 @@ describe("stop backup", () => {
 		});
 
 		// act & assert
-		await expect(backupsExecutionService.stopBackup(schedule.id)).rejects.toThrow(
+		await expect(backupsService.stopBackup(schedule.id)).rejects.toThrow(
 			"No backup is currently running for this schedule",
 		);
 
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupAt).toBe(previousLastBackupAt);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 		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(backupsExecutionService.stopBackup(99999)).rejects.toThrow("Backup schedule not found");
+		await expect(backupsService.stopBackup(99999)).rejects.toThrow("Backup schedule not found");
 	});
 });
 
@@ -281,7 +407,7 @@ describe("retention policy - runForget", () => {
 		});
 
 		// act
-		await backupsExecutionService.runForget(schedule.id);
+		await backupsService.runForget(schedule.id);
 
 		// assert
 		expect(resticForgetMock).toHaveBeenCalledWith(
@@ -310,7 +436,7 @@ describe("retention policy - runForget", () => {
 		});
 
 		// act & assert
-		await expect(backupsExecutionService.runForget(schedule.id)).rejects.toThrow(
+		await expect(backupsService.runForget(schedule.id)).rejects.toThrow(
 			"No retention policy configured for this schedule",
 		);
 	});
@@ -318,7 +444,7 @@ describe("retention policy - runForget", () => {
 	test("should throw NotFoundError when schedule does not exist", async () => {
 		setup();
 		// act & assert
-		await expect(backupsExecutionService.runForget(99999)).rejects.toThrow("Backup schedule not found");
+		await expect(backupsService.runForget(99999)).rejects.toThrow("Backup schedule not found");
 	});
 
 	test("should throw NotFoundError when repository does not exist", async () => {
@@ -331,9 +457,7 @@ describe("retention policy - runForget", () => {
 		});
 
 		// act & assert
-		await expect(backupsExecutionService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow(
+		await expect(backupsService.runForget(schedule.id, "non-existent-repo")).rejects.toThrow("Repository not found");
-			"Repository not found",
-		);
 	});
 });
 
@@ -352,7 +476,7 @@ describe("mirror operations", () => {
 		await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		expect(resticCopyMock).toHaveBeenCalledWith(
@@ -379,7 +503,7 @@ describe("mirror operations", () => {
 		await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id, { enabled: false });
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		expect(resticCopyMock).not.toHaveBeenCalled();
@@ -399,7 +523,7 @@ describe("mirror operations", () => {
 		const mirror = await createTestBackupScheduleMirror(schedule.id, mirrorRepository.id);
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		const mirrors = await backupsService.getMirrors(schedule.id);
@@ -430,7 +554,7 @@ describe("mirror operations", () => {
 		});
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		const mirrors = await backupsService.getMirrors(schedule.id);
@@ -457,7 +581,7 @@ describe("mirror operations", () => {
 		resticCopyMock.mockImplementationOnce(() => Promise.reject(new Error("Copy failed")));
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, null);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, null);
 
 		// assert
 		const mirrors = await backupsService.getMirrors(schedule.id);
@@ -485,7 +609,7 @@ describe("mirror operations", () => {
 		resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" }));
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
 
 		await waitForExpect(() => {
 			expect(resticCopyMock).toHaveBeenCalled();
@@ -516,7 +640,7 @@ describe("mirror operations", () => {
 		resticForgetMock.mockClear();
 
 		// act
-		await backupsExecutionService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
+		await backupsService.copyToMirrors(schedule.id, sourceRepository, schedule.retentionPolicy);
 
 		await waitForExpect(() => {
 			expect(resticCopyMock).toHaveBeenCalled();
@@ -559,10 +683,10 @@ describe("mirror operations", () => {
 		);
 		resticCopyMock.mockImplementation(() => Promise.resolve({ success: true, output: "" }));
 
-		const firstCopyPromise = backupsExecutionService.copyToMirrors(firstSchedule.id, sourceRepository, null);
+		const firstCopyPromise = backupsService.copyToMirrors(firstSchedule.id, sourceRepository, null);
 		await firstCopyStarted;
 
-		const secondCopyPromise = backupsExecutionService.copyToMirrors(secondSchedule.id, sourceRepository, null);
+		const secondCopyPromise = backupsService.copyToMirrors(secondSchedule.id, sourceRepository, null);
 
 		try {
 			const secondCopyState = await Promise.race<"resolved" | "timeout">([

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

@@ -12,11 +12,14 @@ 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(() => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
+	const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
+	const { sendBackupMock, cancelBackupMock } = createAgentBackupMocks(resticBackupMock);
 	const refreshStatsMock = vi.fn(() =>
 		Promise.resolve({
 			total_size: 0,
@@ -29,10 +32,14 @@ const setup = () => {
 	);
 	vi.spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock);
 	vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock);
+	vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock);
+	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
 
 	return {
 		resticBackupMock,
+		sendBackupMock,
+		cancelBackupMock,
 		refreshStatsMock,
 	};
 };
@@ -59,10 +66,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.nextBackupAt).not.toBeNull();
 
 		const nextBackupAt = new Date(updatedSchedule.nextBackupAt ?? 0);
@@ -84,7 +91,7 @@ describe("execute backup", () => {
 		});
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
 		expect(resticBackupMock).not.toHaveBeenCalled();
@@ -106,11 +113,11 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id, true);
+		await backupsService.executeBackup(schedule.id, true);
 
 		// assert
 		expect(resticBackupMock).toHaveBeenCalled();
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("success");
 		expect(updatedSchedule.lastBackupAt).not.toBeNull();
 	});
@@ -132,10 +139,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id, true);
+		await backupsService.executeBackup(schedule.id, true);
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.nextBackupAt).toBeNull();
 	});
 
@@ -155,13 +162,13 @@ describe("execute backup", () => {
 		});
 
 		// act
-		void backupsExecutionService.executeBackup(schedule.id);
+		void backupsService.executeBackup(schedule.id);
 
 		await waitForExpect(() => {
 			expect(resticBackupMock).toHaveBeenCalledTimes(1);
 		});
 
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
 		expect(resticBackupMock).toHaveBeenCalledTimes(1);
@@ -182,10 +189,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("warning");
 	});
 
@@ -204,10 +211,10 @@ describe("execute backup", () => {
 		);
 
 		// act
-		await backupsExecutionService.executeBackup(schedule.id);
+		await backupsService.executeBackup(schedule.id);
 
 		// assert
-		const updatedSchedule = await backupsService.getScheduleById(schedule.id);
+		const updatedSchedule = await getScheduleByIdOrShortId(schedule.id);
 		expect(updatedSchedule.lastBackupStatus).toBe("error");
 	});
 });
@@ -229,7 +236,7 @@ describe("getSchedulesToExecute", () => {
 		});
 
 		// act
-		const schedulesToExecute = await backupsExecutionService.getSchedulesToExecute();
+		const schedulesToExecute = await backupsService.getSchedulesToExecute();
 
 		// assert
 		expect(schedulesToExecute).toContain(schedule.id);
@@ -246,7 +253,7 @@ describe("getScheduleByIdOrShortId", () => {
 			repositoryId: repository.id,
 		});
 
-		const found = await backupsService.getScheduleByIdOrShortId(String(schedule.id));
+		const found = await getScheduleByIdOrShortId(String(schedule.id));
 
 		expect(found.id).toBe(schedule.id);
 		expect(found.shortId).toBe(schedule.shortId);
@@ -261,7 +268,7 @@ describe("getScheduleByIdOrShortId", () => {
 			repositoryId: repository.id,
 		});
 
-		const found = await backupsService.getScheduleByIdOrShortId(schedule.shortId);
+		const found = await getScheduleByIdOrShortId(schedule.shortId);
 
 		expect(found.id).toBe(schedule.id);
 		expect(found.shortId).toBe(schedule.shortId);
@@ -274,10 +281,8 @@ describe("getScheduleByIdOrShortId", () => {
 			organizationId: otherOrgId,
 		});
 
-		await expect(backupsService.getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow(
+		await expect(getScheduleByIdOrShortId(schedule.shortId)).rejects.toThrow("Backup schedule not found");
-			"Backup schedule not found",
+		await expect(getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found");
-		);
-		await expect(backupsService.getScheduleByIdOrShortId(schedule.id)).rejects.toThrow("Backup schedule not found");
 	});
 });
 

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

@@ -1,11 +1,18 @@
-import { restic } from "../../core/restic";
+import type { BackupSchedule, Volume, Repository } from "../../db/schema";
-import type { BackupSchedule, Repository, Volume } from "../../db/schema";
+import { logger } from "@zerobyte/core/node";
-import type { ResticBackupOutputDto, ResticBackupProgressDto } from "@zerobyte/core/restic";
+import { resticDeps } from "../../core/restic";
-import { createBackupOptions } from "./backup.helpers";
+import type { ResticBackupOutputDto } from "@zerobyte/core/restic";
+import type { BackupProgressPayload, BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
+import { agentManager } from "../agents/agents-manager";
 import { getVolumePath } from "../volumes/helpers";
+import { decryptRepositoryConfig } from "../repositories/repository-config-secrets";
+import { createBackupOptions } from "./backup.helpers";
+
+const LOCAL_AGENT_ID = "local";
 
 type BackupExecutionRequest = {
 	scheduleId: number;
+	jobId: string;
 	schedule: BackupSchedule;
 	volume: Volume;
 	repository: Repository;
@@ -14,7 +21,14 @@ type BackupExecutionRequest = {
 	onProgress: (progress: BackupExecutionProgress) => void;
 };
 
-export type BackupExecutionProgress = ResticBackupProgressDto;
+type ActiveBackupExecution = {
+	scheduleId: number;
+	scheduleShortId: string;
+	onProgress: (progress: BackupExecutionProgress) => void;
+	resolve: (result: BackupExecutionResult) => void;
+};
+
+export type BackupExecutionProgress = BackupProgressPayload["progress"];
 
 export type BackupExecutionResult =
 	| {
@@ -29,15 +43,137 @@ export type BackupExecutionResult =
 	  }
 	| {
 			status: "failed";
-			error: unknown;
+			error: string;
 	  }
 	| {
 			status: "cancelled";
 			message?: string;
 	  };
 
+const activeExecutionsByJobId = new Map<string, ActiveBackupExecution>();
+const activeExecutionJobIdsByScheduleId = new Map<number, string>();
+const requestedCancellationsByScheduleId = new Set<number>();
 const activeControllersByScheduleId = new Map<number, AbortController>();
 
+const createBackupRunPayload = async ({
+	jobId,
+	schedule,
+	volume,
+	repository,
+	organizationId,
+}: BackupExecutionRequest): Promise<BackupRunPayload> => {
+	const sourcePath = getVolumePath(volume);
+	const { signal: _, ...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 clearExecutionState = (jobId: string, scheduleId: number) => {
+	requestedCancellationsByScheduleId.delete(scheduleId);
+	clearActiveExecution(jobId);
+};
+
+const resolveExecution = (jobId: string, activeExecution: ActiveBackupExecution, result: BackupExecutionResult) => {
+	clearExecutionState(jobId, activeExecution.scheduleId);
+	activeExecution.resolve(result);
+};
+
+const getActiveExecution = (jobId: string, scheduleId: string, eventName: string, agentId: string) => {
+	const activeExecution = activeExecutionsByJobId.get(jobId);
+	if (!activeExecution) {
+		logger.warn(`Received ${eventName} for unknown job ${jobId} from agent ${agentId}`);
+		return null;
+	}
+
+	if (activeExecution.scheduleShortId !== scheduleId) {
+		logger.warn(`Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from agent ${agentId}`);
+		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;
+		}
+
+		resolveExecution(payload.jobId, activeExecution, {
+			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;
+		}
+
+		resolveExecution(payload.jobId, activeExecution, {
+			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);
+		resolveExecution(payload.jobId, activeExecution, {
+			status: "cancelled",
+			message: wasRequested ? undefined : payload.message,
+		});
+	},
+});
+
 export const backupExecutor = {
 	track: (scheduleId: number) => {
 		const abortController = new AbortController();
@@ -49,39 +185,67 @@ export const backupExecutor = {
 			activeControllersByScheduleId.delete(scheduleId);
 		}
 	},
-	execute: async (params: BackupExecutionRequest): Promise<BackupExecutionResult> => {
+	execute: async (request: Omit<BackupExecutionRequest, "jobId">) => {
-		const { schedule, volume, repository, organizationId, signal, onProgress } = params;
+		const jobId = Bun.randomUUIDv7();
-		try {
+		const completion = new Promise<BackupExecutionResult>((resolve) => {
-			const volumePath = getVolumePath(volume);
+			activeExecutionsByJobId.set(jobId, {
-			const backupOptions = createBackupOptions(schedule, volumePath, signal);
+				scheduleId: request.scheduleId,
-
+				scheduleShortId: request.schedule.shortId,
-			const result = await restic.backup(repository.config, volumePath, {
+				onProgress: request.onProgress,
-				...backupOptions,
+				resolve,
-				compressionMode: repository.compressionMode ?? "auto",
+			});
-				organizationId,
+			activeExecutionJobIdsByScheduleId.set(request.scheduleId, jobId);
-				onProgress,
 		});
 
+		try {
+			if (request.signal.aborted) {
+				throw request.signal.reason || new Error("Operation aborted");
+			}
+
+			const payload = await createBackupRunPayload({ ...request, jobId });
+
+			if (request.signal.aborted) {
+				throw request.signal.reason || new Error("Operation aborted");
+			}
+
+			if (!agentManager.sendBackup(LOCAL_AGENT_ID, payload)) {
+				clearExecutionState(jobId, request.scheduleId);
 				return {
-				status: "completed",
+					status: "unavailable",
-				exitCode: result.exitCode,
+					error: new Error("Local backup agent is not connected"),
-				result: result.result,
-				warningDetails: result.warningDetails,
 				} satisfies BackupExecutionResult;
+			}
+
+			return completion;
 		} catch (error) {
-			return {
+			clearExecutionState(jobId, request.scheduleId);
-				status: "failed",
+			throw error;
-				error,
-			} satisfies BackupExecutionResult;
 		}
 	},
 	cancel: (scheduleId: number) => {
 		const abortController = activeControllersByScheduleId.get(scheduleId);
-		if (!abortController) {
+		if (abortController) {
+			abortController.abort();
+		}
+
+		const jobId = activeExecutionJobIdsByScheduleId.get(scheduleId);
+		if (!jobId) {
+			return abortController !== undefined;
+		}
+
+		const activeExecution = activeExecutionsByJobId.get(jobId);
+		if (!activeExecution) {
+			activeExecutionJobIdsByScheduleId.delete(scheduleId);
+			requestedCancellationsByScheduleId.delete(scheduleId);
 			return false;
 		}
 
-		abortController.abort();
+		requestedCancellationsByScheduleId.add(scheduleId);
+		agentManager.cancelBackup(LOCAL_AGENT_ID, {
+			jobId,
+			scheduleId: activeExecution.scheduleShortId,
+		});
+
 		return true;
 	},
 };

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

@@ -1,11 +0,0 @@
-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

@@ -37,14 +37,6 @@ const listSchedules = async () => {
 	return schedules.filter((schedule) => schedule.volume && schedule.repository);
 };
 
-const getScheduleById = async (scheduleId: number) => {
-	return getScheduleByIdOrShortId(scheduleId);
-};
-
-const getScheduleByShortId = async (shortId: ShortId) => {
-	return getScheduleByIdOrShortId(shortId);
-};
-
 const createSchedule = async (data: CreateBackupScheduleBody) => {
 	const organizationId = getOrganizationId();
 	if (data.cronExpression && !isValidCron(data.cronExpression)) {
@@ -477,9 +469,6 @@ const stopBackup = async (scheduleId: number) => {
 
 export const backupsService = {
 	listSchedules,
-	getScheduleById,
-	getScheduleByShortId,
-	getScheduleByIdOrShortId,
 	createSchedule,
 	updateSchedule,
 	deleteSchedule,

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

@@ -0,0 +1,58 @@
+import { afterEach, describe, expect, mock, spyOn, test } from "bun:test";
+import { Scheduler } from "../../../core/scheduler";
+import * as backendModule from "../../backends/backend";
+import type { VolumeBackend } from "../../backends/backend";
+import * as bootstrapModule from "../bootstrap";
+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 stopApplicationRuntime = 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(bootstrapModule, "stopApplicationRuntime").mockImplementation(stopApplicationRuntime);
+		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

@@ -1,4 +1,5 @@
 import { runDbMigrations } from "../../db/db";
+import { agentManager, spawnLocalAgent, stopAgentRuntime } from "../agents/agents-manager";
 import { runMigrations } from "./migrations";
 import { startup } from "./startup";
 
@@ -7,6 +8,8 @@ let bootstrapPromise: Promise<void> | undefined;
 const runBootstrap = async () => {
 	await runDbMigrations();
 	await runMigrations();
+	agentManager.start();
+	await spawnLocalAgent();
 	await startup();
 };
 
@@ -22,3 +25,11 @@ export const bootstrapApplication = async () => {
 		throw err;
 	}
 };
+
+export const stopApplicationRuntime = async () => {
+	try {
+		await stopAgentRuntime();
+	} finally {
+		bootstrapPromise = undefined;
+	}
+};

app/server/modules/lifecycle/shutdown.ts

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

app/server/plugins/bootstrap.ts

@@ -1,9 +1,11 @@
 import { definePlugin } from "nitro";
-import { bootstrapApplication } from "../modules/lifecycle/bootstrap";
+import { bootstrapApplication, stopApplicationRuntime } from "../modules/lifecycle/bootstrap";
 import { logger } from "@zerobyte/core/node";
 import { toMessage } from "../utils/errors";
 
-export default definePlugin(async () => {
+export default definePlugin(async (nitroApp) => {
+	nitroApp.hooks.hook("close", stopApplicationRuntime);
+
 	await bootstrapApplication().catch((err) => {
 		logger.error(`Bootstrap failed: ${toMessage(err)}`);
 		process.exit(1);

app/test/helpers/agent-mock.ts

@@ -0,0 +1,105 @@
+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

@@ -0,0 +1,34 @@
+# 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

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

apps/agent/src/__tests__/controller-session.test.ts

@@ -0,0 +1,93 @@
+import { afterEach, expect, mock, spyOn, test } from "bun:test";
+import { Effect } from "effect";
+import waitForExpect from "wait-for-expect";
+import { fromPartial } from "@total-typescript/shoehorn";
+import { createControllerMessage, parseAgentMessage } from "@zerobyte/contracts/agent-protocol";
+import * as resticServer from "@zerobyte/core/restic/server";
+import { createControllerSession } from "../controller-session";
+
+afterEach(() => {
+	mock.restore();
+});
+
+test("emits backup.failed when a backup command hits a restic error", async () => {
+	spyOn(resticServer, "createRestic").mockReturnValue(
+		fromPartial({
+			backup: () => Effect.fail("source path missing"),
+		}),
+	);
+
+	const outboundMessages: string[] = [];
+	const session = createControllerSession(
+		fromPartial({
+			send: (message: string) => {
+				outboundMessages.push(message);
+			},
+		}),
+	);
+
+	try {
+		session.onOpen();
+		session.onMessage(
+			createControllerMessage("backup.run", {
+				jobId: "job-1",
+				scheduleId: "schedule-1",
+				organizationId: "org-1",
+				sourcePath: "/tmp/missing-source",
+				repositoryConfig: {
+					backend: "local",
+					path: "/tmp/test-repository",
+				},
+				options: {},
+				runtime: {
+					password: "password",
+					cacheDir: "/tmp/restic-cache",
+					passFile: "/tmp/restic-pass",
+					defaultExcludes: [],
+				},
+			}),
+		);
+
+		await waitForExpect(() => {
+			const failedMessage = outboundMessages
+				.map((message) => parseAgentMessage(message))
+				.find((message) => message?.success && message.data.type === "backup.failed");
+
+			expect(failedMessage?.success).toBe(true);
+			if (!failedMessage || !failedMessage.success || failedMessage.data.type !== "backup.failed") {
+				return;
+			}
+
+			expect(failedMessage.data.payload).toEqual({
+				jobId: "job-1",
+				scheduleId: "schedule-1",
+				error: "source path missing",
+				errorDetails: "source path missing",
+			});
+		});
+	} finally {
+		session.close();
+	}
+});
+
+test("closes the websocket when an outbound send throws", async () => {
+	const close = mock(() => undefined);
+	const session = createControllerSession(
+		fromPartial({
+			send: () => {
+				throw new Error("socket write failed");
+			},
+			close,
+		}),
+	);
+
+	try {
+		session.onOpen();
+
+		await waitForExpect(() => {
+			expect(close).toHaveBeenCalledTimes(1);
+		});
+	} finally {
+		session.close();
+	}
+});

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

@@ -0,0 +1,20 @@
+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.gen(function* () {
+		const running = yield* 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

@@ -0,0 +1,112 @@
+import { Effect, Runtime } 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 = yield* context.getRunningJob(payload.jobId);
+			if (existing) {
+				yield* 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();
+			yield* context.setRunningJob(payload.jobId, { scheduleId: payload.scheduleId, abortController });
+
+			const sendCancelled = () => {
+				return context.offerOutbound(
+					createAgentMessage("backup.cancelled", {
+						jobId: payload.jobId,
+						scheduleId: payload.scheduleId,
+						message: "Backup was cancelled",
+					}),
+				);
+			};
+
+			yield* 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);
+			const runtime = yield* Effect.runtime<never>();
+
+			yield* restic
+				.backup(payload.repositoryConfig, payload.sourcePath, {
+					organizationId: payload.organizationId,
+					...payload.options,
+					signal: abortController.signal,
+					onProgress: (progress) => {
+						void Runtime.runPromise(
+							runtime,
+							context.offerOutbound(
+								createAgentMessage("backup.progress", {
+									jobId: payload.jobId,
+									scheduleId: payload.scheduleId,
+									progress,
+								}),
+							),
+						).catch((error) => {
+							logger.error(`Failed to send backup progress update: ${toMessage(error)}`);
+						});
+					},
+				})
+				.pipe(
+					Effect.matchEffect({
+						onSuccess: (result) => {
+							if (abortController.signal.aborted) {
+								return sendCancelled();
+							}
+
+							return context.offerOutbound(
+								createAgentMessage("backup.completed", {
+									jobId: payload.jobId,
+									scheduleId: payload.scheduleId,
+									exitCode: result.exitCode,
+									result: result.result,
+									warningDetails: result.warningDetails ?? undefined,
+								}),
+							);
+						},
+						onFailure: (error) => {
+							if (abortController.signal.aborted) {
+								return sendCancelled();
+							}
+
+							return context.offerOutbound(
+								createAgentMessage("backup.failed", {
+									jobId: payload.jobId,
+									scheduleId: payload.scheduleId,
+									error: toMessage(error),
+									errorDetails: toErrorDetails(error),
+								}),
+							);
+						},
+					}),
+					Effect.ensuring(context.deleteRunningJob(payload.jobId)),
+				);
+		}),
+	).pipe(Effect.asVoid);

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

@@ -0,0 +1,11 @@
+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) =>
+	context.offerOutbound(
+		createAgentMessage("heartbeat.pong", {
+			sentAt: payload.sentAt,
+		}),
+	);

apps/agent/src/commands/index.ts

@@ -0,0 +1,19 @@
+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

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

apps/agent/src/controller-session.ts

@@ -0,0 +1,146 @@
+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";
+import type { ControllerCommandContext, RunningJob } from "./context";
+
+export type ControllerSession = {
+	onOpen: () => void;
+	onMessage: (data: unknown) => void;
+	close: () => void;
+};
+
+export const createControllerSession = (ws: WebSocket): ControllerSession => {
+	let isClosed = false;
+	const outboundQueue = Effect.runSync(Queue.bounded<AgentWireMessage>(64));
+	const inboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
+	const runningJobsRef = Effect.runSync(Ref.make<Map<string, RunningJob>>(new Map()));
+
+	const getRunningJob = (jobId: string) => Ref.get(runningJobsRef).pipe(Effect.map((map) => map.get(jobId)));
+
+	const setRunningJob = (jobId: string, job: RunningJob) => {
+		return Ref.update(runningJobsRef, (current) => {
+			const next = new Map(current);
+			next.set(jobId, job);
+			return next;
+		});
+	};
+
+	const abortRunningJobs = Effect.gen(function* () {
+		const runningJobs = yield* Ref.modify(runningJobsRef, (current) => [current, new Map()]);
+		yield* Effect.sync(() => {
+			for (const runningJob of runningJobs.values()) {
+				runningJob.abortController.abort();
+			}
+		});
+	});
+
+	const deleteRunningJob = (jobId: string) => {
+		return Ref.update(runningJobsRef, (current) => {
+			const next = new Map(current);
+			next.delete(jobId);
+			return next;
+		});
+	};
+
+	const offerOutbound = (message: AgentWireMessage) => {
+		return Queue.offer(outboundQueue, message);
+	};
+
+	const offerInbound = (message: ControllerWireMessage) => {
+		return Queue.offer(inboundQueue, message);
+	};
+
+	const commandContext: ControllerCommandContext = {
+		getRunningJob,
+		setRunningJob,
+		deleteRunningJob,
+		offerOutbound,
+	};
+
+	const closeSession = () => {
+		if (isClosed) {
+			return;
+		}
+
+		isClosed = true;
+		void Effect.runPromise(abortRunningJobs).catch(() => {});
+		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(() => {});
+	};
+
+	const handleSendFailure = (reason: string) => {
+		logger.error(`Closing agent session after an outbound websocket send failed: ${reason}`);
+
+		try {
+			ws.close();
+		} catch (error) {
+			logger.error(`Failed to close controller websocket after send failure: ${toMessage(error)}`);
+		}
+
+		closeSession();
+	};
+
+	const writerFiber = Effect.runFork(
+		Effect.forever(
+			Effect.gen(function* () {
+				const message = yield* Queue.take(outboundQueue);
+				yield* Effect.sync(() => {
+					try {
+						ws.send(message);
+					} catch (error) {
+						handleSendFailure(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: () => {
+			void Effect.runPromise(offerOutbound(createAgentMessage("agent.ready", { agentId: "" }))).catch((error) => {
+				logger.error(`Failed to queue ready message: ${toMessage(error)}`);
+			});
+		},
+		onMessage: (data) => {
+			if (typeof data !== "string") {
+				logger.warn("Agent received a non-text message");
+				return;
+			}
+
+			void Effect.runPromise(offerInbound(data as ControllerWireMessage)).catch((error) => {
+				logger.error(`Failed to queue inbound message: ${toMessage(error)}`);
+			});
+		},
+		close: closeSession,
+	};
+};

apps/agent/src/index.ts

@@ -0,0 +1,67 @@
+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;
+const reconnectDelayMs = 1_000;
+
+export class Agent {
+	private ws: WebSocket | null = null;
+	private controllerSession: ControllerSession | null = null;
+	private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
+
+	connect() {
+		if (this.reconnectTimeout) {
+			clearTimeout(this.reconnectTimeout);
+			this.reconnectTimeout = null;
+		}
+
+		if (this.ws) {
+			return;
+		}
+
+		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");
+
+			if (!this.reconnectTimeout) {
+				this.reconnectTimeout = setTimeout(() => {
+					this.reconnectTimeout = null;
+					this.connect();
+				}, reconnectDelayMs);
+			}
+		};
+		this.ws.onerror = (error) => {
+			logger.error("Agent encountered an error:", error);
+		};
+	}
+}
+
+if (import.meta.main) {
+	const agent = new Agent();
+	agent.connect();
+}

apps/agent/tsconfig.json

@@ -0,0 +1,31 @@
+{
+	"compilerOptions": {
+		"plugins": [{ "name": "@effect/language-service" }],
+		// Environment setup & latest features
+		"lib": ["DOM", "ESNext"],
+		"target": "ESNext",
+		"module": "Preserve",
+		"moduleDetection": "force",
+		"jsx": "react-jsx",
+		"types": ["bun", "node"],
+		"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,6 +34,7 @@
         "@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",
@@ -46,6 +47,7 @@
         "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",
@@ -53,7 +55,7 @@
         "http-errors-enhanced": "^4.0.2",
         "input-otp": "^1.4.2",
         "isbot": "^5.1.36",
-        "lucide-react": "^1.0.1",
+        "lucide-react": "^0.577.0",
         "next-themes": "^0.4.6",
         "qrcode.react": "^4.2.0",
         "react": "^19.2.4",
@@ -73,6 +75,7 @@
       },
       "devDependencies": {
         "@babel/preset-typescript": "^7.28.5",
+        "@effect/language-service": "^0.84.2",
         "@faker-js/faker": "^10.3.0",
         "@happy-dom/global-registrator": "^20.8.4",
         "@hey-api/openapi-ts": "^0.94.4",
@@ -114,6 +117,32 @@
         "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": {
@@ -264,6 +293,8 @@
 
     "@drizzle-team/brocli": ["@drizzle-team/brocli@0.11.0", "", {}, "sha512-hD3pekGiPg0WPCCGAZmusBBJsDqGUR66Y452YgQsZOnkdQ7ViEPKuyP4huUGEZQefp8g34RRodXYmJ2TbCH+tg=="],
 
+    "@effect/language-service": ["@effect/language-service@0.84.2", "", { "bin": { "effect-language-service": "cli.js" } }, "sha512-l04qNxpiA8rY5yXWckRPJ7Mk5MNerXuNymSFf+IdflfI5i8jgL1bpBNLuP6ijg7wgjdHc/KmTnCj2kT0SCntuA=="],
+
     "@electric-sql/pglite": ["@electric-sql/pglite@0.3.15", "", {}, "sha512-Cj++n1Mekf9ETfdc16TlDi+cDDQF0W7EcbyRHYOAeZdsAe8M/FJg18itDTSwyHfar2WIezawM9o0EKaRGVKygQ=="],
 
     "@electric-sql/pglite-socket": ["@electric-sql/pglite-socket@0.0.20", "", { "peerDependencies": { "@electric-sql/pglite": "0.3.15" }, "bin": { "pglite-server": "dist/scripts/server.js" } }, "sha512-J5nLGsicnD9wJHnno9r+DGxfcZWh+YJMCe0q/aCgtG6XOm9Z7fKeite8IZSNXgZeGltSigM9U/vAWZQWdgcSFg=="],
@@ -1044,12 +1075,16 @@
 
     "@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=="],
@@ -1544,7 +1579,7 @@
 
     "lru.min": ["lru.min@1.1.4", "", {}, "sha512-DqC6n3QQ77zdFpCMASA1a3Jlb64Hv2N2DciFGkO/4L9+q/IpIAuRlKOvCXabtRW6cQf8usbmM6BE/TOPysCdIA=="],
 
-    "lucide-react": ["lucide-react@1.0.1", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-lih7tKEczCYOQjVEzpFuxEuNzlwf+1yhvlMlEkGWJM3va8Pugv8bYXc/pRtcjPncaP7k84X0Pt/71ufxvqEPtQ=="],
+    "lucide-react": ["lucide-react@0.577.0", "", { "peerDependencies": { "react": "^16.5.1 || ^17.0.0 || ^18.0.0 || ^19.0.0" } }, "sha512-4LjoFv2eEPwYDPg/CUdBJQSDfPyzXCRrVW1X7jrx/trgxnxkHFjnVZINbzvzxjN70dxychOfg+FTYwBiS3pQ5A=="],
 
     "luxon": ["luxon@3.7.2", "", {}, "sha512-vtEhXh/gNjI9Yg1u4jX/0YVPMvxzHuGgCm6tC5kZyb08yjGWGnqAjGJvcXbqQR2P3MyMEFnRbpcdFS6PBcLqew=="],
 

package.json

@@ -13,7 +13,7 @@
 		"start": "bun run .output/server/index.mjs",
 		"preview": "bunx --bun vite preview",
 		"cli:dev": "bun run app/server/cli/main.ts",
-		"cli": "ZEROBYTE_CLI=1 bun .output/server/_ssr/ssr.mjs",
+		"cli": "ZEROBYTE_CLI=1 bun .output/server/_ssr/index.mjs",
 		"tsc": "tsc --noEmit && turbo run tsc",
 		"start:dev": "docker compose down && docker compose up --build zerobyte-dev",
 		"start:prod": "docker compose down && docker compose up --build zerobyte-prod",
@@ -59,6 +59,7 @@
 		"@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",
@@ -71,6 +72,7 @@
 		"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",
@@ -78,7 +80,7 @@
 		"http-errors-enhanced": "^4.0.2",
 		"input-otp": "^1.4.2",
 		"isbot": "^5.1.36",
-		"lucide-react": "^1.0.1",
+		"lucide-react": "^0.577.0",
 		"next-themes": "^0.4.6",
 		"qrcode.react": "^4.2.0",
 		"react": "^19.2.4",
@@ -98,6 +100,7 @@
 	},
 	"devDependencies": {
 		"@babel/preset-typescript": "^7.28.5",
+		"@effect/language-service": "^0.84.2",
 		"@faker-js/faker": "^10.3.0",
 		"@happy-dom/global-registrator": "^20.8.4",
 		"@hey-api/openapi-ts": "^0.94.4",

packages/contracts/.gitignore

@@ -0,0 +1,34 @@
+# 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

@@ -0,0 +1,15 @@
+# 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

@@ -0,0 +1,24 @@
+{
+	"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

@@ -0,0 +1,219 @@
+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

@@ -0,0 +1,30 @@
+{
+	"compilerOptions": {
+		// Environment setup & latest features
+		"lib": ["ESNext"],
+		"target": "ESNext",
+		"module": "Preserve",
+		"moduleDetection": "force",
+		"jsx": "react-jsx",
+		"types": ["bun", "node"],
+		"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/src/restic/commands/__tests__/backup.test.ts

@@ -1,4 +1,5 @@
 import { afterEach, describe, expect, test, vi } from "vitest";
+import { Effect } from "effect";
 import * as cleanupModule from "../../helpers/cleanup-temporary-keys";
 import * as spawnModule from "../../../utils/spawn";
 import { ResticError } from "../../error";
@@ -62,7 +63,7 @@ const setup = ({ spawnResult = {}, onSpawnCall }: SetupOptions = {}) => {
 	let capturedArgs: string[] = [];
 
 	vi.spyOn(cleanupModule, "cleanupTemporaryKeys").mockImplementation(() => Promise.resolve());
-	vi.spyOn(spawnModule, "safeSpawn").mockImplementation((params) => {
+	vi.spyOn(spawnModule, "safeSpawn").mockImplementation((params: SafeSpawnParams) => {
 		capturedArgs = params.args;
 		return Promise.resolve(onSpawnCall?.(params)).then(() => ({
 			exitCode: 0,
@@ -87,6 +88,9 @@ const setup = ({ spawnResult = {}, onSpawnCall }: SetupOptions = {}) => {
 	};
 };
 
+const runBackup = (...args: Parameters<typeof backup>) => Effect.runPromise(backup(...args));
+const runBackupError = (...args: Parameters<typeof backup>) => Effect.runPromise(Effect.flip(backup(...args)));
+
 afterEach(() => {
 	vi.restoreAllMocks();
 });
@@ -95,7 +99,7 @@ describe("backup command", () => {
 	describe("argument construction", () => {
 		test("passes source path as positional arg when no include list is given", async () => {
 			const { getArgs, hasFlag } = setup();
-			await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
+			await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
 
 			expect(getArgs()).toContain("/mnt/data");
 			expect(hasFlag("--files-from")).toBe(false);
@@ -105,7 +109,7 @@ describe("backup command", () => {
 			const { getArgs } = setup();
 			const source = "--help";
 
-			await backup(config, source, { organizationId: "org-1" }, mockDeps);
+			await runBackup(config, source, { organizationId: "org-1" }, mockDeps);
 
 			const separatorIndex = getArgs().indexOf("--");
 			expect(separatorIndex).toBeGreaterThan(-1);
@@ -132,7 +136,7 @@ describe("backup command", () => {
 				},
 			});
 
-			await backup(
+			await runBackup(
 				config,
 				"/mnt/data",
 				{
@@ -152,7 +156,7 @@ describe("backup command", () => {
 
 		test("always includes DEFAULT_EXCLUDES as --exclude args", async () => {
 			const { getOptionValues } = setup();
-			await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
+			await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
 
 			expect(getOptionValues("--exclude").length).toBeGreaterThan(0);
 		});
@@ -161,7 +165,7 @@ describe("backup command", () => {
 	describe("exit code handling", () => {
 		test("returns parsed result on exit code 0", async () => {
 			setup();
-			const { result, exitCode } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
+			const { result, exitCode } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
 
 			expect(exitCode).toBe(0);
 			expect(result?.snapshot_id).toBe("abcd1234");
@@ -169,7 +173,7 @@ describe("backup command", () => {
 
 		test("returns result without throwing on exit code 3 (partial read errors)", async () => {
 			setup({ spawnResult: { exitCode: 3 } });
-			const { result, exitCode } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
+			const { result, exitCode } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
 
 			expect(exitCode).toBe(3);
 			expect(result).not.toBeNull();
@@ -178,15 +182,14 @@ describe("backup command", () => {
 		test("throws ResticError on non-zero, non-3 exit codes", async () => {
 			setup({ spawnResult: { exitCode: 1, summary: "", error: "fatal error" } });
 
-			await expect(backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps)).rejects.toBeInstanceOf(
+			const error = await runBackupError(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
-				ResticError,
+			expect(error).toBeInstanceOf(ResticError);
-			);
 		});
 
 		test("preserves the exit code inside the thrown ResticError", async () => {
 			setup({ spawnResult: { exitCode: 12, summary: "", error: "wrong password" } });
 
-			const error = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps).catch((e) => e);
+			const error = await runBackupError(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
 			expect(error).toBeInstanceOf(ResticError);
 			expect((error as ResticError).code).toBe(12);
 		});
@@ -204,7 +207,7 @@ describe("backup command", () => {
 				},
 			});
 
-			const error = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps).catch((e) => e);
+			const error = await runBackupError(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
 			expect(error).toBeInstanceOf(ResticError);
 			expect((error as ResticError).summary).toBe("Command failed: An error occurred while executing the command.");
 			expect((error as ResticError).details).toBe(
@@ -219,7 +222,7 @@ describe("backup command", () => {
 				spawnResult: { exitCode: 130, summary: "", error: "" },
 			});
 
-			const { result, exitCode, warningDetails } = await backup(
+			const { result, exitCode, warningDetails } = await runBackup(
 				config,
 				"/mnt/data",
 				{
@@ -238,7 +241,7 @@ describe("backup command", () => {
 	describe("output parsing", () => {
 		test("returns a fully parsed summary object on valid output", async () => {
 			setup();
-			const { result } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
+			const { result } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
 
 			expect(result).toMatchObject({
 				message_type: "summary",
@@ -249,14 +252,14 @@ describe("backup command", () => {
 
 		test("returns { result: null } when summary line is not valid JSON", async () => {
 			setup({ spawnResult: { summary: "not-json" } });
-			const { result } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
+			const { result } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
 
 			expect(result).toBeNull();
 		});
 
 		test("returns { result: null } when summary JSON does not satisfy the schema", async () => {
 			setup({ spawnResult: { summary: JSON.stringify({ message_type: "summary" }) } });
-			const { result } = await backup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
+			const { result } = await runBackup(config, "/mnt/data", { organizationId: "org-1" }, mockDeps);
 
 			expect(result).toBeNull();
 		});
@@ -267,7 +270,7 @@ describe("backup command", () => {
 			const progressUpdates: unknown[] = [];
 			setup({ onSpawnCall: (params) => params.onStdout?.(VALID_PROGRESS_LINE) });
 
-			await backup(
+			await runBackup(
 				config,
 				"/mnt/data",
 				{
@@ -294,7 +297,7 @@ describe("backup command", () => {
 			});
 
 			await expect(
-				backup(config, "/mnt/data", { organizationId: "org-1", onProgress: () => {} }, mockDeps),
+				runBackup(config, "/mnt/data", { organizationId: "org-1", onProgress: () => {} }, mockDeps),
 			).resolves.toBeDefined();
 		});
 
@@ -304,7 +307,7 @@ describe("backup command", () => {
 				onSpawnCall: (params) => params.onStdout?.(JSON.stringify({ message_type: "verbose_status", action: "scan" })),
 			});
 
-			await backup(
+			await runBackup(
 				config,
 				"/mnt/data",
 				{

packages/core/src/restic/commands/backup.ts

@@ -1,6 +1,7 @@
 import fs from "node:fs/promises";
 import os from "node:os";
 import path from "node:path";
+import { Effect } from "effect";
 import { throttle } from "es-toolkit";
 import type { CompressionMode, RepositoryConfig } from "../schemas";
 import { type ResticBackupProgressDto, resticBackupOutputSchema, resticBackupProgressSchema } from "../restic-dto";
@@ -13,7 +14,7 @@ import { ResticError } from "../error";
 import { logger, safeSpawn } from "../../node";
 import type { ResticDeps } from "../types";
 
-export const backup = async (
+export const backup = (
 	config: RepositoryConfig,
 	source: string,
 	options: {
@@ -30,7 +31,9 @@ export const backup = async (
 		customResticParams?: string[];
 	},
 	deps: ResticDeps,
-) => {
+) =>
+	Effect.tryPromise({
+		try: async () => {
 			const repoUrl = buildRepoUrl(config);
 			const env = await buildEnv(config, options.organizationId, deps);
 
@@ -210,4 +213,6 @@ export const backup = async (
 				exitCode: res.exitCode,
 				warningDetails: stderrLines.length > 0 ? stderrLines.join("\n") : null,
 			};
-};
+		},
+		catch: (error) => (error instanceof Error ? error : new Error(String(error))),
+	});

packages/core/tsconfig.json

@@ -1,5 +1,6 @@
 {
 	"compilerOptions": {
+		"plugins": [{ "name": "@effect/language-service" }],
 		// Environment setup & latest features
 		"lib": ["ESNext"],
 		"target": "ESNext",

tsconfig.json

@@ -1,6 +1,7 @@
 {
 	"include": ["app/**/*"],
 	"compilerOptions": {
+		"plugins": [{ "name": "@effect/language-service" }],
 		"lib": ["DOM", "DOM.Iterable", "ES2023"],
 		"types": ["bun", "node", "vite/client"],
 		"target": "ES2022",