refactor(agent): harden local agent volume lifecycle (#863)

* refactor(agent): harden local agent volume lifecycle

* chore(test): remove un-used variable

* refactor(agent): create dedicated jobs for recurring tasks

* chore: pr feedbacks

* test: add missing fake agent controller

app/drizzle/20260505165117_early_purple_man/migration.sql

@@ -13,4 +13,4 @@ CREATE TABLE `agents_table` (
 );
 --> statement-breakpoint
 CREATE INDEX `agents_table_organization_id_idx` ON `agents_table` (`organization_id`);--> statement-breakpoint
-CREATE INDEX `agents_table_status_idx` ON `agents_table` (`status`);
+CREATE INDEX `agents_table_status_idx` ON `agents_table` (`status`);

app/server/jobs/__tests__/cleanup-dangling.test.ts

@@ -0,0 +1,18 @@
+import { afterEach, expect, test, vi } from "vitest";
+import { config } from "../../core/config";
+import { CleanupDanglingMountsJob } from "../cleanup-dangling";
+import * as mountinfo from "../../utils/mountinfo";
+
+afterEach(() => {
+	config.flags.enableLocalAgent = true;
+	vi.restoreAllMocks();
+});
+
+test("skips controller-local mount inspection when local volume execution is agent-owned", async () => {
+	config.flags.enableLocalAgent = true;
+	const readMountInfo = vi.spyOn(mountinfo, "readMountInfo");
+
+	await new CleanupDanglingMountsJob().run();
+
+	expect(readMountInfo).not.toHaveBeenCalled();
+});

app/server/jobs/cleanup-dangling.ts

@@ -10,9 +10,16 @@ import { toMessage } from "../utils/errors";
 import { VOLUME_MOUNT_BASE } from "../core/constants";
 import { db } from "../db/db";
 import { withContext } from "../core/request-context";
+import { config } from "../core/config";
+import { LOCAL_AGENT_ID } from "../modules/agents/constants";
 
 export class CleanupDanglingMountsJob extends Job {
 	async run() {
+		if (config.flags.enableLocalAgent) {
+			logger.debug("Skipping controller-local dangling mount cleanup because local volume execution is agent-owned.");
+			return { done: true, timestamp: new Date() };
+		}
+
 		const organizations = await db.query.organization.findMany({});
 		if (organizations.length === 0) {
 			logger.warn("No organizations found; skipping dangling mount cleanup to avoid false positives.");
@@ -22,7 +29,7 @@ export class CleanupDanglingMountsJob extends Job {
 		const allVolumes = [];
 		for (const org of organizations) {
 			const volumes = await withContext({ organizationId: org.id }, async () => volumeService.listVolumes());
-			allVolumes.push(...volumes);
+			allVolumes.push(...volumes.filter((volume) => volume.agentId === LOCAL_AGENT_ID));
 		}
 
 		const allSystemMounts = await readMountInfo();

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

@@ -1,6 +1,7 @@
 import { EventEmitter } from "node:events";
 import { PassThrough } from "node:stream";
 import { afterEach, beforeEach, expect, test, vi } from "vitest";
+import { fromAny } from "@total-typescript/shoehorn";
 import type { ProcessWithAgentRuntime } from "../helpers/runtime-state.dev";
 
 const spawnMock = vi.fn();
@@ -16,7 +17,7 @@ const processWithAgentRuntime = process as ProcessWithAgentRuntime;
 
 const setAgentRuntime = () => {
 	processWithAgentRuntime.__zerobyteAgentRuntime = {
-		agentManager: null,
+		agentManager: fromAny({ waitForAgentReady: vi.fn(async () => true) }),
 		localAgent: null,
 		isStoppingLocalAgent: false,
 		localAgentRestartTimeout: null,

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

@@ -3,14 +3,21 @@ import type { BackupRunPayload, VolumeCommand, VolumeCommandResult } from "@zero
 import { Effect } from "effect";
 import { config } from "../../core/config";
 import { createAgentManagerRuntime, type AgentManagerEvent } from "./controller/server";
+import { LOCAL_AGENT_ID } from "./constants";
 import { spawnLocalAgentProcess, stopLocalAgentProcess } from "./local/process";
-import type { BackupExecutionProgress, BackupExecutionResult } from "./helpers/runtime-state";
-import { createAgentRuntimeState } from "./helpers/runtime-state";
+import {
+	createAgentRuntimeState,
+	type AgentRuntimeState,
+	type BackupExecutionProgress,
+	type BackupExecutionResult,
+} from "./helpers/runtime-state";
 import { getDevAgentRuntimeState } from "./helpers/runtime-state.dev";
 export type { BackupExecutionProgress, BackupExecutionResult } from "./helpers/runtime-state";
 export type { ProcessWithAgentRuntime } from "./helpers/runtime-state.dev";
 
-const productionRuntimeState = createAgentRuntimeState();
+type ProcessWithProductionAgentRuntime = NodeJS.Process & {
+	__zerobyteProductionAgentRuntime?: AgentRuntimeState;
+};
 
 type AgentRunBackupRequest = {
 	scheduleId: number;
@@ -19,7 +26,18 @@ type AgentRunBackupRequest = {
 	onProgress: (progress: BackupExecutionProgress) => void;
 };
 
-const getAgentRuntimeState = () => (config.__prod__ ? productionRuntimeState : getDevAgentRuntimeState());
+const getProductionAgentRuntimeState = () => {
+	// Nitro production builds can bundle startup plugins and API handlers into separate chunks.
+	// Keep the live controller on process so both chunks see the same agent sessions.
+	const runtimeProcess = process as ProcessWithProductionAgentRuntime;
+	if (!runtimeProcess.__zerobyteProductionAgentRuntime) {
+		runtimeProcess.__zerobyteProductionAgentRuntime = createAgentRuntimeState();
+	}
+
+	return runtimeProcess.__zerobyteProductionAgentRuntime;
+};
+
+const getAgentRuntimeState = () => (config.__prod__ ? getProductionAgentRuntimeState() : getDevAgentRuntimeState());
 const getAgentManagerRuntime = () => getAgentRuntimeState().agentManager;
 const getActiveBackupsByScheduleId = () => getAgentRuntimeState().activeBackupsByScheduleId;
 const getActiveBackupScheduleIdsByJobId = () => getAgentRuntimeState().activeBackupScheduleIdsByJobId;
@@ -72,7 +90,9 @@ const getActiveBackupRun = (jobId: string, scheduleId: string, eventName: string
 	}
 
 	if (activeBackupRun.scheduleShortId !== scheduleId) {
-		logger.warn(`Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from agent ${agentId}`);
+		logger.warn(
+			`Ignoring ${eventName} for job ${jobId} due to schedule mismatch ${scheduleId} from agent ${agentId}`,
+		);
 		return null;
 	}
 
@@ -286,7 +306,18 @@ export const agentManager = {
 };
 
 export const startLocalAgent = async () => {
-	await spawnLocalAgentProcess(getAgentRuntimeState());
+	const runtime = getAgentRuntimeState();
+	await spawnLocalAgentProcess(runtime);
+
+	if (!runtime.agentManager) {
+		throw new Error(
+			`startLocalAgent spawned ${LOCAL_AGENT_ID} via spawnLocalAgentProcess, but runtime.agentManager is missing; waitForAgentReady cannot check readiness`,
+		);
+	}
+
+	if (!(await runtime.agentManager.waitForAgentReady(LOCAL_AGENT_ID))) {
+		throw new Error("Local agent did not become ready before startup");
+	}
 };
 
 // fallow-ignore-next-line unused-export

app/server/modules/agents/controller/server.ts

@@ -46,6 +46,7 @@ class StopAgentManagerServerError extends Data.TaggedError("StopAgentManagerServ
 export function createAgentManagerRuntime(onEvent: (event: AgentManagerEvent) => void) {
 	let sessions = new Map<string, ControllerAgentSessionHandle>();
 	let runtimeScope: Scope.CloseableScope | null = null;
+	const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms));
 
 	const closeSession = (sessionHandle: ControllerAgentSessionHandle) =>
 		Effect.gen(function* () {
@@ -79,6 +80,11 @@ export function createAgentManagerRuntime(onEvent: (event: AgentManagerEvent) =>
 	const getSessionHandle = (agentId: string) => sessions.get(agentId);
 	const getSession = (agentId: string) => getSessionHandle(agentId)?.session;
 
+	const isAgentReady = (agentId: string) => {
+		const session = getSession(agentId);
+		return !!session && Effect.runSync(session.isReady());
+	};
+
 	const handleSessionEvent = (params: { agentId: string; agentName: string; sessionId: string }) => {
 		const { agentId, agentName } = params;
 
@@ -290,6 +296,19 @@ export function createAgentManagerRuntime(onEvent: (event: AgentManagerEvent) =>
 
 	return {
 		start,
+		waitForAgentReady: async (agentId: string, timeoutMs = 10_000) => {
+			const deadline = Date.now() + timeoutMs;
+
+			while (Date.now() < deadline) {
+				if (isAgentReady(agentId)) {
+					return true;
+				}
+
+				await sleep(50);
+			}
+
+			return isAgentReady(agentId);
+		},
 		sendBackup: (agentId: string, payload: BackupRunPayload) =>
 			Effect.gen(function* () {
 				const session = getSession(agentId);

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

@@ -1,9 +1,10 @@
-import { afterEach, describe, expect, test, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import { Scheduler } from "../../../core/scheduler";
-import * as volumeHostModule from "../../../../../apps/agent/src/volume-host";
-import type { VolumeBackend } from "../../../../../apps/agent/src/volume-host";
 import * as bootstrapModule from "../bootstrap";
+import { agentManager } from "../../agents/agents-manager";
 import { createTestVolume } from "~/test/helpers/volume";
+import { config } from "~/server/core/config";
+import { db } from "~/server/db/db";
 
 const loadShutdownModule = async () => {
 	const moduleUrl = new URL("../shutdown.ts", import.meta.url);
@@ -11,12 +12,19 @@ const loadShutdownModule = async () => {
 	return import(moduleUrl.href);
 };
 
+let originalEnableLocalAgent: boolean;
+
+beforeEach(() => {
+	originalEnableLocalAgent = config.flags.enableLocalAgent;
+});
+
 afterEach(() => {
+	config.flags.enableLocalAgent = originalEnableLocalAgent;
 	vi.restoreAllMocks();
 });
 
 describe("shutdown", () => {
-	test("stops the agent runtime before unmounting mounted volumes", async () => {
+	test("does not unmount agent-owned volumes during controller shutdown", async () => {
 		const events: string[] = [];
 		const stopScheduler = vi.fn(async () => {
 			events.push("scheduler.stop");
@@ -24,35 +32,59 @@ describe("shutdown", () => {
 		const stopApplicationRuntime = vi.fn(async () => {
 			events.push("agents.stop");
 		});
-		const unmountVolume = vi.fn(async () => {
-			events.push("backend.unmount");
-			return { status: "unmounted" as const };
-		});
+		const runVolumeCommand = vi.spyOn(agentManager, "runVolumeCommand");
 
-		await createTestVolume({
+		const volume = await createTestVolume({
 			name: "Shutdown test volume",
 			config: {
 				backend: "directory",
 				path: "/Applications",
 			},
 			status: "mounted",
+			agentId: "agent-1",
 		});
 
 		vi.spyOn(Scheduler, "stop").mockImplementation(stopScheduler);
 		vi.spyOn(bootstrapModule, "stopApplicationRuntime").mockImplementation(stopApplicationRuntime);
-		vi.spyOn(volumeHostModule, "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"]);
+		expect(events).toEqual(["scheduler.stop", "agents.stop"]);
+		expect(runVolumeCommand).not.toHaveBeenCalled();
+		const updated = await db.query.volumesTable.findFirst({ where: { id: volume.id } });
+		expect(updated).toBeDefined();
+		expect(updated!.status).toBe("mounted");
+	});
+
+	test("keeps legacy controller-local fallback unmount on shutdown", async () => {
+		config.flags.enableLocalAgent = false;
+		const events: string[] = [];
+		vi.spyOn(Scheduler, "stop").mockImplementation(async () => {
+			events.push("scheduler.stop");
+		});
+		vi.spyOn(bootstrapModule, "stopApplicationRuntime").mockImplementation(async () => {
+			events.push("agents.stop");
+		});
+		const runVolumeCommand = vi.spyOn(agentManager, "runVolumeCommand").mockImplementation(async () => {
+			throw new Error("runVolumeCommand should not be called during fallback shutdown");
+		});
+
+		const volume = await createTestVolume({
+			name: "Fallback shutdown test volume",
+			config: { backend: "directory", path: "/Applications" },
+			status: "mounted",
+		});
+
+		const { shutdown } = await loadShutdownModule();
+
+		await shutdown();
+
+		const updated = await db.query.volumesTable.findFirst({ where: { id: volume.id } });
+		expect(events).toEqual(["scheduler.stop", "agents.stop"]);
+		expect(runVolumeCommand).not.toHaveBeenCalled();
+		expect(updated).toBeDefined();
+		expect(updated!.status).toBe("unmounted");
 	});
 });

app/server/modules/lifecycle/shutdown.ts

@@ -2,25 +2,28 @@ import { Scheduler } from "../../core/scheduler";
 import { db } from "../../db/db";
 import { logger } from "@zerobyte/core/node";
 import { stopApplicationRuntime } from "./bootstrap";
-import { decryptVolumeConfig } from "../volumes/volume-config-secrets";
-import { createVolumeBackend } from "../../../../apps/agent/src/volume-host";
+import { withContext } from "../../core/request-context";
+import { volumeService } from "../volumes/volume.service";
+import { toMessage } from "../../utils/errors";
+import { config } from "../../core/config";
+import { LOCAL_AGENT_ID } from "../agents/constants";
 
 export const shutdown = async () => {
 	await Scheduler.stop();
-	await stopApplicationRuntime();
 
-	const volumes = await db.query.volumesTable.findMany({
-		where: { status: "mounted" },
-	});
-
-	for (const volume of volumes) {
-		const backend = createVolumeBackend({
-			...volume,
-			config: await decryptVolumeConfig(volume.config),
-			provisioningId: volume.provisioningId ?? null,
+	if (!config.flags.enableLocalAgent) {
+		const volumes = await db.query.volumesTable.findMany({
+			where: { AND: [{ status: "mounted" }, { agentId: LOCAL_AGENT_ID }] },
 		});
-		const { status, error } = await backend.unmount();
 
-		logger.info(`Volume ${volume.name} unmount status: ${status}${error ? `, error: ${error}` : ""}`);
+		for (const volume of volumes) {
+			const { status, error } = await withContext({ organizationId: volume.organizationId }, () =>
+				volumeService.unmountVolume(volume.shortId),
+			).catch((error) => ({ status: "error" as const, error: toMessage(error) }));
+
+			logger.info(`Volume ${volume.name} unmount status: ${status}${error ? `, error: ${error}` : ""}`);
+		}
 	}
+
+	await stopApplicationRuntime();
 };

app/server/modules/lifecycle/startup.ts

@@ -17,6 +17,7 @@ import { backupsService } from "../backups/backups.service";
 import { config } from "~/server/core/config";
 import { syncProvisionedResources } from "../provisioning/provisioning";
 import { toMessage } from "~/server/utils/errors";
+import { LOCAL_AGENT_ID } from "../agents/constants";
 
 const ensureLatestConfigurationSchema = async () => {
 	const volumes = await db.query.volumesTable.findMany({});
@@ -71,23 +72,30 @@ export const startup = async () => {
 		logger.warn(`Removed ${deletedSchedules} orphaned backup schedule(s) during startup`);
 	}
 
-	const volumes = await db.query.volumesTable.findMany({
-		where: {
-			OR: [
-				{ status: "mounted" },
-				{
-					AND: [{ autoRemount: true }, { status: "error" }],
-				},
-			],
-		},
-	});
-
-	for (const volume of volumes) {
-		await withContext({ organizationId: volume.organizationId }, async () => {
-			await volumeService.mountVolume(volume.shortId).catch((err) => {
-				logger.error(`Error auto-remounting volume ${volume.name} on startup: ${err.message}`);
-			});
+	if (!config.flags.enableLocalAgent) {
+		const volumes = await db.query.volumesTable.findMany({
+			where: {
+				AND: [
+					{ agentId: LOCAL_AGENT_ID },
+					{
+						OR: [
+							{ status: "mounted" },
+							{
+								AND: [{ autoRemount: true }, { status: "error" }],
+							},
+						],
+					},
+				],
+			},
 		});
+
+		for (const volume of volumes) {
+			await withContext({ organizationId: volume.organizationId }, async () => {
+				await volumeService.mountVolume(volume.shortId).catch((err) => {
+					logger.error(`Error auto-remounting volume ${volume.name} on startup: ${err.message}`);
+				});
+			});
+		}
 	}
 
 	await db
@@ -102,7 +110,9 @@ export const startup = async () => {
 			logger.error(`Failed to update stuck backup schedules on startup: ${err.message}`);
 		});
 
-	Scheduler.build(CleanupDanglingMountsJob).schedule("0 * * * *");
+	if (!config.flags.enableLocalAgent) {
+		Scheduler.build(CleanupDanglingMountsJob).schedule("0 * * * *");
+	}
 	Scheduler.build(VolumeHealthCheckJob).schedule("*/30 * * * *");
 	Scheduler.build(RepositoryHealthCheckJob).schedule("50 12 * * *");
 	Scheduler.build(BackupExecutionJob).schedule("* * * * *");

app/server/modules/volumes/volume.service.ts

@@ -18,7 +18,12 @@ import { getOrganizationId } from "~/server/core/request-context";
 import { type ShortId } from "~/server/utils/branded";
 import { decryptVolumeConfig, encryptVolumeConfig } from "./volume-config-secrets";
 import type { VolumeCommand, VolumeCommandResult } from "@zerobyte/contracts/agent-protocol";
-import { createVolumeBackend, getStatFs, getVolumePath } from "../../../../apps/agent/src/volume-host";
+import {
+	createVolumeBackend,
+	getStatFs,
+	getVolumePath,
+	type AgentVolume,
+} from "../../../../apps/agent/src/volume-host";
 import {
 	browseFilesystem as browseHostFilesystem,
 	listVolumeFiles,
@@ -63,21 +68,24 @@ const volumeForAgent = async (volume: Volume): Promise<Volume> => ({
 	config: await decryptVolumeConfig(volume.config),
 });
 
-const volumeForHost = async (volume: Volume): Promise<Volume> => ({
+const volumeForHost = async (volume: Volume): Promise<AgentVolume> => ({
 	...volume,
-	shortId: volume.shortId,
 	config: await decryptVolumeConfig(volume.config),
 	provisioningId: volume.provisioningId ?? null,
 });
 
-// TODO(agent-rollout): Remove the local host execution branch once all installs run volume operations through agents.
+// Transitional fallback: older controller-only installs do not run the supervised local agent.
+// Keep all controller-local host execution behind this predicate so the fallback is easy to delete
+// once volume operations always go through the local agent.
 const shouldRunViaAgent = (volume: Volume) => volume.agentId !== LOCAL_AGENT_ID || config.flags.enableLocalAgent;
 
+const shouldUseControllerLocalVolumeFallback = (volume: Volume) => !shouldRunViaAgent(volume);
+
 const runVolumeBackendCommand = async (
 	volume: Volume,
 	name: "volume.mount" | "volume.unmount" | "volume.checkHealth",
 ) => {
-	if (!shouldRunViaAgent(volume)) {
+	if (shouldUseControllerLocalVolumeFallback(volume)) {
 		const backend = createVolumeBackend(await volumeForHost(volume));
 		switch (name) {
 			case "volume.mount":
@@ -376,7 +384,7 @@ const listFiles = async (shortId: ShortId, subPath?: string, offset: number = 0,
 	}
 
 	try {
-		if (!shouldRunViaAgent(volume)) {
+		if (shouldUseControllerLocalVolumeFallback(volume)) {
 			return await listVolumeFiles(await volumeForHost(volume), subPath, offset, limit);
 		}
 

apps/agent/src/commands/volume.test.ts

@@ -0,0 +1,127 @@
+import { afterEach, expect, test, vi } from "vitest";
+import { Effect } from "effect";
+import { fromPartial } from "@total-typescript/shoehorn";
+import {
+	parseAgentMessage,
+	type AgentWireMessage,
+	type VolumeCommandPayload,
+} from "@zerobyte/contracts/agent-protocol";
+import type { ControllerCommandContext } from "../context";
+
+const volumeHostMock = vi.hoisted(() => ({
+	createVolumeBackend: vi.fn(),
+	getStatFs: vi.fn(),
+	getVolumePath: vi.fn(),
+}));
+
+const operationsMock = vi.hoisted(() => ({
+	browseFilesystem: vi.fn(),
+	listVolumeFiles: vi.fn(),
+	testVolumeConnection: vi.fn(),
+}));
+
+vi.mock("../volume-host", () => volumeHostMock);
+vi.mock("../volume-host/operations", () => operationsMock);
+
+import { handleVolumeCommand } from "./volume";
+
+afterEach(() => {
+	vi.restoreAllMocks();
+	volumeHostMock.createVolumeBackend.mockReset();
+	volumeHostMock.getStatFs.mockReset();
+	volumeHostMock.getVolumePath.mockReset();
+	operationsMock.browseFilesystem.mockReset();
+	operationsMock.listVolumeFiles.mockReset();
+	operationsMock.testVolumeConnection.mockReset();
+});
+
+const runVolumeCommand = async (payload: VolumeCommandPayload) => {
+	const outboundMessages: AgentWireMessage[] = [];
+	const context = fromPartial<ControllerCommandContext>({
+		offerOutbound: (message: AgentWireMessage) =>
+			Effect.sync(() => {
+				outboundMessages.push(message);
+				return true;
+			}),
+	});
+
+	await Effect.runPromise(handleVolumeCommand(context, payload));
+	return outboundMessages.map((message) => parseAgentMessage(message));
+};
+
+test("runs backend-backed volume commands on the agent host", async () => {
+	const mount = vi.fn(async () => ({ status: "mounted" as const }));
+	volumeHostMock.createVolumeBackend.mockReturnValue({ mount });
+
+	const messages = await runVolumeCommand(
+		fromPartial<VolumeCommandPayload>({
+			commandId: "command-1",
+			command: {
+				name: "volume.mount",
+				volume: { id: 1, config: { backend: "directory", path: "/tmp/source" }, provisioningId: undefined },
+			},
+		}),
+	);
+
+	expect(volumeHostMock.createVolumeBackend).toHaveBeenCalledWith(
+		expect.objectContaining({ id: 1, config: { backend: "directory", path: "/tmp/source" }, provisioningId: null }),
+	);
+	expect(mount).toHaveBeenCalledOnce();
+	expect(messages[0]?.success).toBe(true);
+	if (messages[0]?.success && messages[0].data.type === "volume.commandResult") {
+		expect(messages[0].data.payload).toEqual({
+			commandId: "command-1",
+			status: "success",
+			command: { name: "volume.mount", result: { status: "mounted" } },
+		});
+	}
+});
+
+test("returns command errors without throwing", async () => {
+	operationsMock.browseFilesystem.mockRejectedValue(new Error("permission denied"));
+
+	const messages = await runVolumeCommand({
+		commandId: "command-2",
+		command: { name: "filesystem.browse", path: "/root" },
+	});
+
+	expect(messages[0]?.success).toBe(true);
+	if (messages[0]?.success && messages[0].data.type === "volume.commandResult") {
+		expect(messages[0].data.payload).toEqual({
+			commandId: "command-2",
+			status: "error",
+			error: "permission denied",
+		});
+	}
+});
+
+test("routes file listing commands to host operations", async () => {
+	operationsMock.listVolumeFiles.mockResolvedValue({
+		files: [],
+		path: "/logs",
+		offset: 0,
+		limit: 10,
+		total: 0,
+		hasMore: false,
+	});
+
+	await runVolumeCommand(
+		fromPartial<VolumeCommandPayload>({
+			commandId: "command-3",
+			command: {
+				name: "volume.listFiles",
+				volume: { id: 1, config: { backend: "directory", path: "/tmp/source" } },
+				subPath: "/logs",
+				offset: 0,
+				limit: 10,
+			},
+		}),
+	);
+
+	expect(operationsMock.listVolumeFiles).toHaveBeenCalledWith(
+		expect.objectContaining({ id: 1, config: { backend: "directory", path: "/tmp/source" } }),
+		"/logs",
+		0,
+		10,
+	);
+});

apps/agent/src/commands/volume.ts

@@ -47,7 +47,8 @@ const executeVolumeCommand = (command: VolumeCommand) =>
 			}
 			case "volume.listFiles": {
 				const result = yield* Effect.tryPromise({
-					try: () => listVolumeFiles(asVolume(command.volume), command.subPath, command.offset, command.limit),
+					try: () =>
+						listVolumeFiles(asVolume(command.volume), command.subPath, command.offset, command.limit),
 					catch: (error) => new VolumeCommandError({ cause: error }),
 				});
 
@@ -80,14 +81,14 @@ export const handleVolumeCommand = (context: ControllerCommandContext, payload:
 
 		return command;
 	}).pipe(
-		Effect.tapError((error) => {
-			return context.offerOutbound(
+		Effect.catchAll((error) =>
+			context.offerOutbound(
 				createAgentMessage("volume.commandResult", {
 					commandId: payload.commandId,
 					status: "error",
 					error: toMessage(error?.cause),
 				}),
-			);
-		}),
+			),
+		),
 	);
 };

apps/agent/src/index.ts

@@ -1,5 +1,7 @@
 import { logger } from "@zerobyte/core/node";
+import { Fiber } from "effect";
 import { createControllerSession, type ControllerSession } from "./controller-session";
+import { startAgentJobs } from "./jobs";
 
 const controllerUrl = process.env.ZEROBYTE_CONTROLLER_URL;
 const agentToken = process.env.ZEROBYTE_AGENT_TOKEN;
@@ -9,6 +11,15 @@ export class Agent {
 	private ws: WebSocket | null = null;
 	private controllerSession: ControllerSession | null = null;
 	private reconnectTimeout: ReturnType<typeof setTimeout> | null = null;
+	private jobFibers: Fiber.RuntimeFiber<never, never>[] | null = null;
+
+	private startJobs() {
+		if (this.jobFibers) {
+			return;
+		}
+
+		this.jobFibers = startAgentJobs();
+	}
 
 	private scheduleReconnect() {
 		if (this.reconnectTimeout) {
@@ -22,6 +33,8 @@ export class Agent {
 	}
 
 	connect() {
+		this.startJobs();
+
 		if (this.reconnectTimeout) {
 			clearTimeout(this.reconnectTimeout);
 			this.reconnectTimeout = null;

apps/agent/src/jobs/cleanup-dangling.ts

@@ -0,0 +1,21 @@
+import { logger } from "@zerobyte/core/node";
+import { toMessage } from "@zerobyte/core/utils";
+import { Effect } from "effect";
+import { cleanupDanglingVolumeMountDirectories } from "../volume-host/cleanup";
+import type { AgentJob } from "./types";
+
+const VOLUME_CLEANUP_INTERVAL_MS = 60 * 60 * 1000;
+
+export const CleanupDanglingVolumeMountsJob: AgentJob = {
+	name: "cleanup-dangling-volume-mounts",
+	intervalMs: VOLUME_CLEANUP_INTERVAL_MS,
+	run: () =>
+		Effect.tryPromise({
+			try: cleanupDanglingVolumeMountDirectories,
+			catch: (error) => error,
+		}).pipe(
+			Effect.catchAll((error) =>
+				Effect.sync(() => logger.warn(`Agent volume cleanup failed: ${toMessage(error)}`)),
+			),
+		),
+};

apps/agent/src/jobs/index.ts

@@ -0,0 +1,23 @@
+import { logger } from "@zerobyte/core/node";
+import { Effect, Fiber } from "effect";
+import { CleanupDanglingVolumeMountsJob } from "./cleanup-dangling";
+import type { AgentJob } from "./types";
+
+const agentJobs = [CleanupDanglingVolumeMountsJob];
+
+export const startAgentJobs = (jobs: readonly AgentJob[] = agentJobs) => {
+	return jobs.map((job) =>
+		Effect.runFork(
+			Effect.forever(
+				Effect.gen(function* () {
+					yield* job.run();
+					yield* Effect.sleep(job.intervalMs);
+				}),
+			).pipe(Effect.ensuring(logger.effect.debug(`Agent job stopped: ${job.name}`))),
+		),
+	);
+};
+
+export const stopAgentJobs = (fibers: readonly Fiber.RuntimeFiber<never, never>[]) => {
+	return Effect.forEach(fibers, (fiber) => Fiber.interrupt(fiber), { discard: true });
+};

apps/agent/src/jobs/types.ts

@@ -0,0 +1,7 @@
+import type { Effect } from "effect";
+
+export type AgentJob = {
+	name: string;
+	intervalMs: number;
+	run: () => Effect.Effect<void, never, never>;
+};

apps/agent/src/volume-host/__tests__/cleanup.test.ts

@@ -0,0 +1,49 @@
+import * as fs from "node:fs/promises";
+import * as os from "node:os";
+import * as path from "node:path";
+import { afterEach, expect, test, vi } from "vitest";
+
+let tempRoot: string | undefined;
+let mockMountPoints: string[] = [];
+
+afterEach(async () => {
+	vi.doUnmock("../constants");
+	vi.doUnmock("../fs");
+	vi.resetModules();
+	mockMountPoints = [];
+	if (tempRoot) {
+		await fs.rm(tempRoot, { recursive: true, force: true });
+		tempRoot = undefined;
+	}
+});
+
+const loadCleanup = async () => {
+	vi.resetModules();
+	tempRoot = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-agent-cleanup-"));
+	vi.doMock("../constants", () => ({ VOLUME_MOUNT_BASE: tempRoot }));
+	vi.doMock("../fs", () => ({
+		readMountInfo: async () => mockMountPoints.map((mountPoint) => ({ mountPoint, fstype: "fuse.sshfs" })),
+	}));
+
+	return import("../cleanup");
+};
+
+test("removes stale volume directories that are not mounted on the agent host", async () => {
+	const { cleanupDanglingVolumeMountDirectories } = await loadCleanup();
+	await fs.mkdir(path.join(tempRoot!, "stale-volume", "_data"), { recursive: true });
+
+	await cleanupDanglingVolumeMountDirectories();
+
+	await expect(fs.access(path.join(tempRoot!, "stale-volume"))).rejects.toThrow();
+});
+
+test("keeps volume directories that are still mounted on the agent host", async () => {
+	const { cleanupDanglingVolumeMountDirectories } = await loadCleanup();
+	const localVolumeDir = path.join(tempRoot!, "mounted-volume");
+	mockMountPoints = [path.join(localVolumeDir, "_data")];
+	await fs.mkdir(path.join(localVolumeDir, "_data"), { recursive: true });
+
+	await cleanupDanglingVolumeMountDirectories();
+
+	await expect(fs.access(localVolumeDir)).resolves.toBeNull();
+});

apps/agent/src/volume-host/__tests__/operations.test.ts

@@ -0,0 +1,66 @@
+import * as fs from "node:fs/promises";
+import * as os from "node:os";
+import * as path from "node:path";
+import { afterEach, expect, test } from "vitest";
+import { listVolumeFiles } from "../operations";
+import type { AgentVolume } from "../types";
+
+let tempRoot: string | undefined;
+
+afterEach(async () => {
+	if (tempRoot) {
+		await fs.rm(tempRoot, { recursive: true, force: true });
+		tempRoot = undefined;
+	}
+});
+
+const createDirectoryVolume = async (): Promise<AgentVolume> => {
+	tempRoot = await fs.mkdtemp(path.join(os.tmpdir(), "zerobyte-volume-ops-"));
+	return {
+		id: 1,
+		shortId: "volume-1",
+		name: "Test volume",
+		config: { backend: "directory", path: tempRoot },
+		createdAt: Date.now(),
+		updatedAt: Date.now(),
+		lastHealthCheck: Date.now(),
+		type: "directory",
+		status: "mounted",
+		lastError: null,
+		provisioningId: null,
+		autoRemount: true,
+		agentId: "local",
+		organizationId: "org-1",
+	};
+};
+
+test("listVolumeFiles returns sorted paginated entries inside the volume", async () => {
+	const volume = await createDirectoryVolume();
+	await fs.mkdir(path.join(tempRoot!, "z-dir"));
+	await fs.mkdir(path.join(tempRoot!, "a-dir"));
+	await fs.writeFile(path.join(tempRoot!, "b-file.txt"), "hello");
+
+	const result = await listVolumeFiles(volume, undefined, 1, 2);
+
+	expect(result).toMatchObject({
+		path: "/",
+		offset: 1,
+		limit: 2,
+		total: 3,
+		hasMore: false,
+	});
+	expect(result.files.map((entry) => entry.name)).toEqual(["z-dir", "b-file.txt"]);
+	expect(result.files[1]).toMatchObject({ path: "/b-file.txt", type: "file", size: 5 });
+});
+
+test("listVolumeFiles rejects traversal outside the volume", async () => {
+	const volume = await createDirectoryVolume();
+
+	await expect(listVolumeFiles(volume, "../outside", 0, 10)).rejects.toThrow("Invalid path");
+});
+
+test("listVolumeFiles reports missing directories consistently", async () => {
+	const volume = await createDirectoryVolume();
+
+	await expect(listVolumeFiles(volume, "missing", 0, 10)).rejects.toThrow("Directory not found");
+});

apps/agent/src/volume-host/cleanup.ts

@@ -0,0 +1,28 @@
+import * as fs from "node:fs/promises";
+import * as path from "node:path";
+import { logger } from "@zerobyte/core/node";
+import { toMessage } from "@zerobyte/core/utils";
+import { VOLUME_MOUNT_BASE } from "./constants";
+import { readMountInfo } from "./fs";
+
+export const cleanupDanglingVolumeMountDirectories = async () => {
+	const mounts = await readMountInfo().catch((error) => {
+		logger.warn(`Failed to read mount info for volume cleanup: ${toMessage(error)}`);
+		return [];
+	});
+	const mountedPaths = new Set(mounts.map((mount) => mount.mountPoint));
+	const volumeDirs = await fs.readdir(VOLUME_MOUNT_BASE).catch(() => []);
+
+	for (const dir of volumeDirs) {
+		const mountPath = path.join(VOLUME_MOUNT_BASE, dir, "_data");
+		if (mountedPaths.has(mountPath)) {
+			continue;
+		}
+
+		const fullPath = path.join(VOLUME_MOUNT_BASE, dir);
+		logger.info(`Removing stale volume mount directory at ${fullPath}`);
+		await fs.rm(fullPath, { recursive: true, force: true }).catch((error) => {
+			logger.warn(`Failed to remove stale volume mount directory ${fullPath}: ${toMessage(error)}`);
+		});
+	}
+};