feat: backup through agent

wip

Dockerfile

@@ -91,6 +91,7 @@ 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,33 @@
+import { afterEach, describe, expect, mock, test } from "bun:test";
+
+const loadAgentsManagerModule = async () => {
+	const moduleUrl = new URL("../agents-manager.ts", import.meta.url);
+	moduleUrl.searchParams.set("test", crypto.randomUUID());
+	return import(moduleUrl.href);
+};
+
+afterEach(() => {
+	delete (globalThis as Record<string, unknown>).__agentManager;
+	mock.restore();
+});
+
+describe("agents-manager module", () => {
+	test("reuses the existing global agent manager without stopping it", async () => {
+		const existingManager = {
+			start: mock(() => {}),
+			sendBackup: mock(() => false),
+			cancelBackup: mock(() => false),
+			setBackupEventHandlers: mock(() => {}),
+			getBackupEventHandlers: mock(() => ({})),
+			stop: mock(() => {}),
+		};
+
+		(globalThis as Record<string, unknown>).__agentManager = existingManager;
+
+		const { agentManager } = await loadAgentsManagerModule();
+
+		expect(agentManager).toBe(existingManager);
+		expect(existingManager.stop).not.toHaveBeenCalled();
+		expect((globalThis as Record<string, unknown>).__agentManager).toBe(existingManager);
+	});
+});

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

@@ -1,14 +1,20 @@
 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 {
+import { createControllerAgentSession, type ControllerAgentSession } from "./controller-agent-session";
-	createControllerAgentSession,
-	type BackupDispatchPayload,
-	type ControllerAgentSession,
-} from "./controller-agent-session";
 
 type AgentConnectionData = {
 	id: string;
@@ -17,19 +23,45 @@ type AgentConnectionData = {
 	agentName: string;
 };
 
+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;
+};
+
 export const spawnLocalAgent = async () => {
 	const previousAgent = (globalThis as Record<string, unknown>).__localAgent as ChildProcess | undefined;
 	if (previousAgent) {
 		previousAgent.kill();
 	}
 
-	const agentEntryPoint = path.join(process.cwd(), "apps", "agent", "src", "index.ts");
+	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 localAgent = spawn("bun", args, {
 		env: {
-			PATH: process.env.PATH,
+			...process.env,
 			ZEROBYTE_CONTROLLER_URL: "ws://localhost:3001",
 			ZEROBYTE_AGENT_TOKEN: agentToken,
 		},
@@ -55,12 +87,17 @@ export const spawnLocalAgent = async () => {
 
 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();
@@ -95,6 +132,26 @@ const createAgentManagerRuntime = () => {
 		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>({
@@ -125,7 +182,16 @@ const createAgentManagerRuntime = () => {
 				},
 				websocket: {
 					open: (ws) => {
-						setSession(ws.data.agentId, createControllerAgentSession(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) => {
@@ -187,7 +253,7 @@ const createAgentManagerRuntime = () => {
 
 	return {
 		start,
-		sendBackup: (agentId: string, payload: BackupDispatchPayload) => {
+		sendBackup: (agentId: string, payload: BackupRunPayload) => {
 			const session = getSession(agentId);
 
 			if (!session) {
@@ -195,10 +261,31 @@ const createAgentManagerRuntime = () => {
 				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,
 	};
 };
@@ -207,7 +294,5 @@ const previous = (globalThis as Record<string, unknown>).__agentManager as
 	| ReturnType<typeof createAgentManagerRuntime>
 	| undefined;
 
-previous?.stop();
+export const agentManager = previous ?? createAgentManagerRuntime();
-
-export const agentManager = createAgentManagerRuntime();
 (globalThis as Record<string, unknown>).__agentManager = agentManager;

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

@@ -3,6 +3,13 @@ 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";
@@ -30,20 +37,27 @@ const toMessage = (error: unknown) => {
 	return String(error);
 };
 
-export type BackupDispatchPayload = {
+type ControllerAgentSessionHandlers = {
-	scheduleId: string;
+	onBackupStarted?: (payload: BackupStartedPayload) => void;
-	jobId?: string;
+	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: BackupDispatchPayload) => string;
+	sendBackup: (payload: BackupRunPayload) => string;
+	sendBackupCancel: (payload: BackupCancelPayload) => void;
 	isReady: () => boolean;
 	close: () => void;
 };
 
-export const createControllerAgentSession = (socket: AgentSocket): ControllerAgentSession => {
+export const createControllerAgentSession = (
+	socket: AgentSocket,
+	handlers: ControllerAgentSessionHandlers = {},
+): ControllerAgentSession => {
 	const outboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
 	const state = Effect.runSync(
 		Ref.make<SessionState>({
@@ -106,6 +120,27 @@ export const createControllerAgentSession = (socket: AgentSocket): ControllerAge
 				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": {
+				updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
+				handlers.onBackupProgress?.(message.payload);
+				break;
+			}
+			case "backup.completed": {
+				updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
+				handlers.onBackupCompleted?.(message.payload);
+				break;
+			}
+			case "backup.failed": {
+				updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
+				handlers.onBackupFailed?.(message.payload);
+				break;
+			}
+			case "backup.cancelled": {
+				updateState((current) => ({ ...current, lastSeenAt: Date.now() }));
+				handlers.onBackupCancelled?.(message.payload);
 				break;
 			}
 			case "heartbeat.pong": {
@@ -137,14 +172,19 @@ export const createControllerAgentSession = (socket: AgentSocket): ControllerAge
 			handleAgentMessage(parsed.data);
 		},
 		sendBackup: (payload) => {
-			const jobId = payload.jobId ?? Bun.randomUUIDv7();
 			offerOutbound(
 				createControllerMessage("backup.run", {
-					jobId,
+					...payload,
-					scheduleId: payload.scheduleId,
+				}),
+			);
+			return payload.jobId;
+		},
+		sendBackupCancel: (payload) => {
+			offerOutbound(
+				createControllerMessage("backup.cancel", {
+					...payload,
 				}),
 			);
-			return jobId;
 		},
 		isReady: () => Effect.runSync(Ref.get(state)).isReady,
 		close: () => {

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

@@ -15,6 +15,8 @@ import { restic } from "~/server/core/restic";
 import { NotFoundError, BadRequestError } from "http-errors-enhanced";
 import { repositoriesService } from "~/server/modules/repositories/repositories.service";
 import { repoMutex } from "~/server/core/repository-mutex";
+import { agentManager } from "~/server/modules/agents/agents-manager";
+import { fromAny } from "@total-typescript/shoehorn";
 
 const setup = () => {
 	const resticBackupMock = vi.fn((_: SafeSpawnParams) =>
@@ -22,6 +24,94 @@ const setup = () => {
 	);
 	const resticForgetMock = vi.fn(() => Promise.resolve({ success: true, data: null }));
 	const resticCopyMock = vi.fn(() => Promise.resolve({ success: true, output: "" }));
+	const runningJobs = new Map<string, { scheduleId: string; cancelled: boolean }>();
+	const sendBackupMock = vi.fn((_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);
+					},
+				} satisfies Partial<SafeSpawnParams>),
+			);
+			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 = vi.fn((_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;
+	});
 	const refreshStatsMock = vi.fn(() =>
 		Promise.resolve({
 			total_size: 0,
@@ -38,11 +128,15 @@ const setup = () => {
 	vi.spyOn(restic, "copy").mockImplementation(resticCopyMock);
 	vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
+	vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock);
+	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 
 	return {
 		resticBackupMock,
 		resticForgetMock,
 		resticCopyMock,
+		sendBackupMock,
+		cancelBackupMock,
 		refreshStatsMock,
 	};
 };

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

@@ -14,9 +14,99 @@ 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 { fromAny } from "@total-typescript/shoehorn";
 
 const setup = () => {
-	const resticBackupMock = vi.fn(() => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
+	const resticBackupMock = vi.fn((_: unknown) => Promise.resolve({ exitCode: 0, summary: "", error: "" }));
+	const runningJobs = new Map<string, { scheduleId: string; cancelled: boolean }>();
+	const sendBackupMock = vi.fn((_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 = vi.fn((_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;
+	});
 	const refreshStatsMock = vi.fn(() =>
 		Promise.resolve({
 			total_size: 0,
@@ -30,9 +120,13 @@ const setup = () => {
 	vi.spyOn(spawnModule, "safeSpawn").mockImplementation(resticBackupMock);
 	vi.spyOn(repositoriesService, "refreshRepositoryStats").mockImplementation(refreshStatsMock);
 	vi.spyOn(context, "getOrganizationId").mockReturnValue(TEST_ORG_ID);
+	vi.spyOn(agentManager, "sendBackup").mockImplementation(sendBackupMock);
+	vi.spyOn(agentManager, "cancelBackup").mockImplementation(cancelBackupMock);
 
 	return {
 		resticBackupMock,
+		sendBackupMock,
+		cancelBackupMock,
 		refreshStatsMock,
 	};
 };

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

@@ -1,8 +1,11 @@
-import { restic } from "../../core/restic";
+import type { BackupRunPayload } from "@zerobyte/contracts/agent-protocol";
+import { agentManager } from "../agents/agents-manager";
+import { resticDeps } from "../../core/restic";
 import type { BackupSchedule, Repository, Volume } from "../../db/schema";
 import type { ResticBackupOutputDto, ResticBackupProgressDto } from "@zerobyte/core/restic";
 import { createBackupOptions } from "./backup.helpers";
 import { getVolumePath } from "../volumes/helpers";
+import { decryptRepositoryConfig } from "../repositories/repository-config-secrets";
 
 type BackupExecutionRequest = {
 	scheduleId: number;
@@ -36,52 +39,239 @@ export type BackupExecutionResult =
 			message?: string;
 	  };
 
-const activeControllersByScheduleId = new Map<number, AbortController>();
+type ActiveBackupExecution = {
+	abortController: AbortController;
+	jobId?: string;
+	scheduleShortId?: string;
+	onProgress?: (progress: BackupExecutionProgress) => void;
+	resolve?: (result: BackupExecutionResult) => void;
+	reject?: (error: unknown) => void;
+	settled: boolean;
+};
+
+const LOCAL_AGENT_ID = "local";
+
+const activeBackupsByScheduleId = new Map<number, ActiveBackupExecution>();
+const activeScheduleIdsByJobId = new Map<string, number>();
+
+const resetPendingExecution = (activeBackup: ActiveBackupExecution) => {
+	activeBackup.jobId = undefined;
+	activeBackup.scheduleShortId = undefined;
+	activeBackup.onProgress = undefined;
+	activeBackup.resolve = undefined;
+	activeBackup.reject = undefined;
+	activeBackup.settled = false;
+};
+
+const getActiveBackupByJobId = (jobId: string) => {
+	const scheduleId = activeScheduleIdsByJobId.get(jobId);
+	if (scheduleId === undefined) {
+		return null;
+	}
+
+	const activeBackup = activeBackupsByScheduleId.get(scheduleId);
+	if (!activeBackup || activeBackup.jobId !== jobId) {
+		activeScheduleIdsByJobId.delete(jobId);
+		return null;
+	}
+
+	return { scheduleId, activeBackup };
+};
+
+const resolveActiveBackup = (activeBackup: ActiveBackupExecution, result: BackupExecutionResult) => {
+	if (activeBackup.settled) {
+		return;
+	}
+
+	activeBackup.settled = true;
+	activeBackup.resolve?.(result);
+};
+
+const rejectActiveBackup = (activeBackup: ActiveBackupExecution, error: unknown) => {
+	if (activeBackup.settled) {
+		return;
+	}
+
+	activeBackup.settled = true;
+	activeBackup.reject?.(error);
+};
+
+const getCancellationError = (signal: AbortSignal, message?: string) =>
+	signal.reason instanceof Error ? signal.reason : new Error(message ?? "Backup was stopped by the user");
+
+const buildAgentBackupPayload = async (params: BackupExecutionRequest, jobId: string): Promise<BackupRunPayload> => {
+	const { schedule, volume, repository, organizationId } = params;
+	const sourcePath = getVolumePath(volume);
+	const { signal: _ignoredSignal, ...options } = createBackupOptions(schedule, sourcePath);
+	const repositoryConfig = await decryptRepositoryConfig(repository.config);
+	const encryptedResticPassword = await resticDeps.getOrganizationResticPassword(organizationId);
+	const resticPassword = await resticDeps.resolveSecret(encryptedResticPassword);
+
+	return {
+		jobId,
+		scheduleId: schedule.shortId,
+		organizationId,
+		sourcePath,
+		repositoryConfig,
+		options: {
+			tags: options.tags,
+			oneFileSystem: options.oneFileSystem,
+			exclude: options.exclude,
+			excludeIfPresent: options.excludeIfPresent,
+			includePaths: options.includePaths,
+			includePatterns: options.includePatterns,
+			customResticParams: options.customResticParams,
+			compressionMode: repository.compressionMode ?? "auto",
+		},
+		runtime: {
+			password: resticPassword,
+			cacheDir: resticDeps.resticCacheDir,
+			passFile: resticDeps.resticPassFile,
+			defaultExcludes: resticDeps.defaultExcludes,
+			hostname: resticDeps.hostname,
+		},
+	};
+};
+
+agentManager.setBackupEventHandlers({
+	onBackupProgress: ({ payload }) => {
+		const running = getActiveBackupByJobId(payload.jobId);
+		if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) {
+			return;
+		}
+
+		running.activeBackup.onProgress?.(payload.progress);
+	},
+	onBackupCompleted: ({ payload }) => {
+		const running = getActiveBackupByJobId(payload.jobId);
+		if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) {
+			return;
+		}
+
+		activeScheduleIdsByJobId.delete(payload.jobId);
+		resolveActiveBackup(running.activeBackup, {
+			status: "completed",
+			exitCode: payload.exitCode,
+			result: payload.result,
+			warningDetails: payload.warningDetails ?? null,
+		});
+	},
+	onBackupFailed: ({ payload }) => {
+		const running = getActiveBackupByJobId(payload.jobId);
+		if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) {
+			return;
+		}
+
+		activeScheduleIdsByJobId.delete(payload.jobId);
+		resolveActiveBackup(running.activeBackup, {
+			status: "failed",
+			error: payload.errorDetails ?? payload.error,
+		});
+	},
+	onBackupCancelled: ({ payload }) => {
+		const running = getActiveBackupByJobId(payload.jobId);
+		if (!running || running.activeBackup.scheduleShortId !== payload.scheduleId) {
+			return;
+		}
+
+		activeScheduleIdsByJobId.delete(payload.jobId);
+
+		if (running.activeBackup.abortController.signal.aborted) {
+			rejectActiveBackup(
+				running.activeBackup,
+				getCancellationError(running.activeBackup.abortController.signal, payload.message),
+			);
+			return;
+		}
+
+		resolveActiveBackup(running.activeBackup, {
+			status: "cancelled",
+			message: payload.message,
+		});
+	},
+});
 
 export const backupExecutor = {
 	track: (scheduleId: number) => {
 		const abortController = new AbortController();
-		activeControllersByScheduleId.set(scheduleId, abortController);
+		activeBackupsByScheduleId.set(scheduleId, {
+			abortController,
+			settled: false,
+		});
 		return abortController;
 	},
 	untrack: (scheduleId: number, abortController: AbortController) => {
-		if (activeControllersByScheduleId.get(scheduleId) === abortController) {
+		const activeBackup = activeBackupsByScheduleId.get(scheduleId);
-			activeControllersByScheduleId.delete(scheduleId);
+		if (activeBackup?.abortController === abortController) {
+			if (activeBackup.jobId) {
+				activeScheduleIdsByJobId.delete(activeBackup.jobId);
+			}
+			activeBackupsByScheduleId.delete(scheduleId);
 		}
 	},
 	execute: async (params: BackupExecutionRequest): Promise<BackupExecutionResult> => {
-		const { schedule, volume, repository, organizationId, signal, onProgress } = params;
+		const activeBackup = activeBackupsByScheduleId.get(params.scheduleId);
-		try {
+		if (!activeBackup) {
-			const volumePath = getVolumePath(volume);
+			throw new Error(`Backup ${params.scheduleId} is not tracked`);
-			const backupOptions = createBackupOptions(schedule, volumePath, signal);
+		}
 
-			const result = await restic.backup(repository.config, volumePath, {
+		if (params.signal.aborted) {
-				...backupOptions,
+			throw getCancellationError(params.signal);
-				compressionMode: repository.compressionMode ?? "auto",
+		}
-				organizationId,
+
-				onProgress,
+		const jobId = Bun.randomUUIDv7();
+		const payload = await buildAgentBackupPayload(params, jobId);
+
+		if (params.signal.aborted) {
+			throw getCancellationError(params.signal);
+		}
+
+		activeBackup.jobId = jobId;
+		activeBackup.scheduleShortId = params.schedule.shortId;
+		activeBackup.onProgress = params.onProgress;
+		activeBackup.settled = false;
+
+		const completion = new Promise<BackupExecutionResult>((resolve, reject) => {
+			activeBackup.resolve = resolve;
+			activeBackup.reject = reject;
+		});
+
+		const handleAbort = () => {
+			activeScheduleIdsByJobId.delete(jobId);
+			agentManager.cancelBackup(LOCAL_AGENT_ID, {
+				jobId,
+				scheduleId: params.schedule.shortId,
 			});
+			rejectActiveBackup(activeBackup, getCancellationError(params.signal));
+		};
 
+		params.signal.addEventListener("abort", handleAbort, { once: true });
+		activeScheduleIdsByJobId.set(jobId, params.scheduleId);
+
+		const dispatched = agentManager.sendBackup(LOCAL_AGENT_ID, payload);
+		if (!dispatched) {
+			params.signal.removeEventListener("abort", handleAbort);
+			activeScheduleIdsByJobId.delete(jobId);
+			resetPendingExecution(activeBackup);
 			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;
-		} catch (error) {
-			return {
-				status: "failed",
-				error,
 			} satisfies BackupExecutionResult;
 		}
+
+		try {
+			return await completion;
+		} finally {
+			params.signal.removeEventListener("abort", handleAbort);
+		}
 	},
 	cancel: (scheduleId: number) => {
-		const abortController = activeControllersByScheduleId.get(scheduleId);
+		const activeBackup = activeBackupsByScheduleId.get(scheduleId);
-		if (!abortController) {
+		if (!activeBackup) {
 			return false;
 		}
 
-		abortController.abort();
+		activeBackup.abortController.abort(new Error("Backup was stopped by the user"));
 		return true;
 	},
 };

app/server/modules/lifecycle/bootstrap.ts

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

apps/agent/src/controller-session.ts

@@ -1,11 +1,15 @@
-import { Effect, Fiber, Queue } from "effect";
+import { Effect, Fiber, Queue, Ref } from "effect";
 import {
 	createAgentMessage,
 	parseControllerMessage,
+	type BackupRunPayload,
 	type AgentWireMessage,
+	type BackupCancelPayload,
 	type ControllerWireMessage,
 } from "@zerobyte/contracts/agent-protocol";
 import { logger } from "@zerobyte/core/node";
+import { ResticError, type ResticDeps } from "@zerobyte/core/restic";
+import { createRestic } from "@zerobyte/core/restic/server";
 
 const toMessage = (error: unknown) => {
 	if (error instanceof Error) {
@@ -15,6 +19,14 @@ const toMessage = (error: unknown) => {
 	return String(error);
 };
 
+const toErrorDetails = (error: unknown) => {
+	if (error instanceof ResticError) {
+		return error.details || error.summary;
+	}
+
+	return toMessage(error);
+};
+
 export type ControllerSession = {
 	onOpen: () => void;
 	onMessage: (data: unknown) => void;
@@ -24,6 +36,31 @@ export type ControllerSession = {
 export const createControllerSession = (ws: WebSocket): ControllerSession => {
 	const outboundQueue = Effect.runSync(Queue.bounded<AgentWireMessage>(64));
 	const inboundQueue = Effect.runSync(Queue.bounded<ControllerWireMessage>(64));
+	const runningJobsRef = Effect.runSync(
+		Ref.make<Map<string, { scheduleId: string; abortController: AbortController }>>(new Map()),
+	);
+
+	const getRunningJob = (jobId: string) => Effect.runSync(Ref.get(runningJobsRef)).get(jobId);
+
+	const setRunningJob = (jobId: string, job: { scheduleId: string; abortController: AbortController }) => {
+		Effect.runSync(
+			Ref.update(runningJobsRef, (current) => {
+				const next = new Map(current);
+				next.set(jobId, job);
+				return next;
+			}),
+		);
+	};
+
+	const deleteRunningJob = (jobId: string) => {
+		Effect.runSync(
+			Ref.update(runningJobsRef, (current) => {
+				const next = new Map(current);
+				next.delete(jobId);
+				return next;
+			}),
+		);
+	};
 
 	const offerOutbound = (message: AgentWireMessage) => {
 		void Effect.runPromise(Queue.offer(outboundQueue, message)).catch((error) => {
@@ -70,14 +107,131 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => {
 
 				switch (parsed.data.type) {
 					case "backup.run": {
-						logger.info(`Starting backup ${parsed.data.payload.jobId} for schedule ${parsed.data.payload.scheduleId}`);
+						const runBackup = async (payload: BackupRunPayload) => {
-						yield* Queue.offer(
+							const existing = getRunningJob(payload.jobId);
-							outboundQueue,
+							if (existing) {
-							createAgentMessage("backup.started", {
+								offerOutbound(
-								jobId: parsed.data.payload.jobId,
+									createAgentMessage("backup.failed", {
-								scheduleId: parsed.data.payload.scheduleId,
+										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();
+							setRunningJob(payload.jobId, { scheduleId: payload.scheduleId, abortController });
+
+							offerOutbound(
+								createAgentMessage("backup.started", {
+									jobId: payload.jobId,
+									scheduleId: payload.scheduleId,
+								}),
+							);
+
+							const deps: ResticDeps = {
+								resolveSecret: async (encrypted) => encrypted,
+								getOrganizationResticPassword: async () => payload.runtime.password,
+								resticCacheDir: payload.runtime.cacheDir,
+								resticPassFile: payload.runtime.passFile,
+								defaultExcludes: payload.runtime.defaultExcludes,
+								hostname: payload.runtime.hostname,
+							};
+
+							const restic = createRestic(deps);
+
+							try {
+								const result = await restic.backup(payload.repositoryConfig, payload.sourcePath, {
+									organizationId: payload.organizationId,
+									tags: payload.options.tags,
+									oneFileSystem: payload.options.oneFileSystem,
+									exclude: payload.options.exclude,
+									excludeIfPresent: payload.options.excludeIfPresent,
+									includePaths: payload.options.includePaths,
+									includePatterns: payload.options.includePatterns,
+									customResticParams: payload.options.customResticParams,
+									compressionMode: payload.options.compressionMode,
+									signal: abortController.signal,
+									onProgress: (progress) => {
+										offerOutbound(
+											createAgentMessage("backup.progress", {
+												jobId: payload.jobId,
+												scheduleId: payload.scheduleId,
+												progress,
+											}),
+										);
+									},
+								});
+
+								if (abortController.signal.aborted) {
+									offerOutbound(
+										createAgentMessage("backup.cancelled", {
+											jobId: payload.jobId,
+											scheduleId: payload.scheduleId,
+											message: "Backup was cancelled",
+										}),
+									);
+									return;
+								}
+
+								offerOutbound(
+									createAgentMessage("backup.completed", {
+										jobId: payload.jobId,
+										scheduleId: payload.scheduleId,
+										exitCode: result.exitCode,
+										result: result.result,
+										warningDetails: result.warningDetails ?? undefined,
+									}),
+								);
+							} catch (error) {
+								if (abortController.signal.aborted) {
+									offerOutbound(
+										createAgentMessage("backup.cancelled", {
+											jobId: payload.jobId,
+											scheduleId: payload.scheduleId,
+											message: "Backup was cancelled",
+										}),
+									);
+									return;
+								}
+
+								offerOutbound(
+									createAgentMessage("backup.failed", {
+										jobId: payload.jobId,
+										scheduleId: payload.scheduleId,
+										error: toMessage(error),
+										errorDetails: toErrorDetails(error),
+									}),
+								);
+							} finally {
+								deleteRunningJob(payload.jobId);
+							}
+						};
+
+						void runBackup(parsed.data.payload);
+						break;
+					}
+					case "backup.cancel": {
+						const cancelBackup = (payload: BackupCancelPayload) => {
+							const running = 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();
+						};
+
+						cancelBackup(parsed.data.payload);
 						break;
 					}
 					case "heartbeat.ping": {
@@ -107,6 +261,11 @@ export const createControllerSession = (ws: WebSocket): ControllerSession => {
 			offerInbound(data as ControllerWireMessage);
 		},
 		close: () => {
+			const runningJobs = Effect.runSync(Ref.get(runningJobsRef));
+			for (const running of runningJobs.values()) {
+				running.abortController.abort();
+			}
+			Effect.runSync(Ref.set(runningJobsRef, new Map()));
 			void Effect.runPromise(Fiber.interrupt(writerFiber)).catch(() => {});
 			void Effect.runPromise(Fiber.interrupt(processorFiber)).catch(() => {});
 			void Effect.runPromise(Queue.shutdown(outboundQueue)).catch(() => {});

packages/contracts/src/agent-protocol.ts

@@ -1,10 +1,58 @@
 import { z } from "zod";
 import { safeJsonParse } from "@zerobyte/core/utils";
+import {
+	repositoryConfigSchema,
+	resticBackupOutputSchema,
+	resticBackupProgressSchema,
+	type CompressionMode,
+} from "@zerobyte/core/restic";
 
-const backupCommandSchema = z
+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() }),
+		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();
 
@@ -29,6 +77,61 @@ const backupStartedSchema = z
 	})
 	.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"),
@@ -36,10 +139,28 @@ const heartbeatPongSchema = z
 	})
 	.strict();
 
-const controllerMessageSchema = z.discriminatedUnion("type", [backupCommandSchema, heartbeatPingSchema]);
+const controllerMessageSchema = z.discriminatedUnion("type", [
-const agentMessageSchema = z.discriminatedUnion("type", [agentReadySchema, backupStartedSchema, heartbeatPongSchema]);
+	backupRunSchema,
+	backupCancelSchema,
+	heartbeatPingSchema,
+]);
+const agentMessageSchema = z.discriminatedUnion("type", [
+	agentReadySchema,
+	backupStartedSchema,
+	backupProgressSchema,
+	backupCompletedSchema,
+	backupFailedSchema,
+	backupCancelledSchema,
+	heartbeatPongSchema,
+]);
 
-export type BackupCommandPayload = z.infer<typeof backupCommandSchema>["payload"];
+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>;