deepseek-harness/dsh-workspace-doctor.mjs

#!/usr/bin/env node
/**
 * dsh-workspace-doctor — report, and optionally repair, DSH workspace grouping
 * in a container deployment whose state volume outlives the container itself.
 *
 * Why this exists
 * ---------------
 * dsh joins a session to a project ("Workspace") when the session is created in
 * that project's directory, plus a one-time history bootstrap. The workspace
 * record stores an absolute, realpath-canonicalized directory path, and a
 * session only keeps its membership while that path still resolves. Updating a
 * container recreates it: everything outside the mounted /workspace and state
 * volume — in particular a project directory created under $HOME — is gone.
 * Those sessions then drop out of their project and the chats collect under
 * "Ungrouped", permanently, because the history bootstrap never runs again.
 *
 * What this tool does
 * -------------------
 * It reads the state volume only. It never touches a session log, never deletes
 * anything, and never invents a project:
 *   --repair-dirs  recreate missing workspace directories (mkdir -p), so the
 *                  sessions already listed under them stay listed
 *   --reattach     re-account sessions whose recorded cwd matches an existing
 *                  workspace path but which no workspace lists, so those chats
 *                  reappear under their project
 * Any write is preceded by a timestamped backup of the workspace registry and
 * is written atomically, in the same format dsh itself writes.
 *
 * Usage: dsh-workspace-doctor [--check|--repair-dirs|--reattach|--all]
 *                             [--dry-run] [--state DIR] [--owner UID:GID] [--json]
 */

import { copyFile, chown, mkdir, open, readFile, readdir, rename, rm, stat } from "node:fs/promises";
import { randomUUID } from "node:crypto";
import { existsSync, realpathSync, readFileSync } from "node:fs";
import { dirname, join, resolve, sep } from "node:path";
import zlib from "node:zlib";

/** Workspace domain unit this tool knows how to write. */
const KNOWN_UNIT = { name: "workspace", version: 2 };
/** Directories dsh can count on surviving a container update. */
const PERSISTENT_ROOTS = ["/workspace"];
/** Sessions above this size are skipped rather than read whole into memory. */
const MAX_SESSION_BYTES = 64 * 1024 * 1024;

const USAGE = `Usage: dsh-workspace-doctor [options]

  --check        report only (default); exits 1 when it finds something to fix
  --repair-dirs  recreate missing workspace directories
  --reattach     re-account sessions that match a workspace but are not listed
  --all          both repairs
  --dry-run      print what a repair would change, write nothing
  --state DIR    dsh state volume (default: $DSH_HOME or /home/node/.dsh)
  --owner U:G    chown files this tool writes (default: $PUID:$PGID when run as root)
  --json         machine-readable report
  -h, --help     this text
`;

/** Parse argv; unknown flags are fatal so a typo never becomes a silent no-op. */
function parseArgs(argv) {
	const opts = {
		mode: "check",
		dryRun: false,
		json: false,
		state: process.env.DSH_HOME ?? "/home/node/.dsh",
		owner: undefined
	};
	const owner = process.env.PUID !== undefined && process.env.PGID !== undefined
		? `${process.env.PUID}:${process.env.PGID}`
		: undefined;
	for (let i = 0; i < argv.length; i += 1) {
		switch (argv[i]) {
			case "--check": opts.mode = "check"; break;
			case "--repair-dirs": opts.mode = "repair-dirs"; break;
			case "--reattach": opts.mode = "reattach"; break;
			case "--all": opts.mode = "all"; break;
			case "--dry-run": opts.dryRun = true; break;
			case "--json": opts.json = true; break;
			case "--state": opts.state = argv[++i] ?? die("--state needs a directory"); break;
			case "--owner": opts.owner = argv[++i] ?? die("--owner needs UID:GID"); break;
			case "-h":
			case "--help": process.stdout.write(USAGE); process.exit(0); break;
			default: die(`unknown option '${argv[i]}'`);
		}
	}
	if (opts.owner === undefined && process.getuid?.() === 0) opts.owner = owner;
	return opts;
}

function die(message) {
	process.stderr.write(`dsh-workspace-doctor: ${message}\n`);
	process.exit(2);
}

/** Read JSON, answering undefined for a missing file and throwing on corruption. */
async function readJson(path) {
	try {
		return JSON.parse(await readFile(path, "utf8"));
	} catch (error) {
		if (error.code === "ENOENT") return undefined;
		throw new Error(`cannot read ${path}: ${error.message}`);
	}
}

/** realpath when the path resolves, the normalized spelling otherwise. */
function canon(path) {
	try {
		return realpathSync(path);
	} catch {
		return resolve(path);
	}
}

/** Whether a path lives on a mount the container update preserves. */
function storageOf(path, state) {
	if (path === state || path.startsWith(state + sep)) return "state volume";
	if (PERSISTENT_ROOTS.some((root) => path === root || path.startsWith(root + sep))) return "workspace mount";
	return "container filesystem";
}

/** Version rank of a session log file, so a migrated log wins over its source. */
function logRank(name) {
	const match = /\.v(\d+)\.jsonl(\.zstd)?$/.exec(name);
	if (match !== null) return Number(match[1]);
	return /\.jsonl(\.zstd)?$/.test(name) ? -1 : Number.NEGATIVE_INFINITY;
}

/** First line of a session log — the header — decompressed when needed. */
function headerLine(file) {
	let text;
	if (file.endsWith(".zstd")) {
		if (typeof zlib.zstdDecompressSync !== "function") return undefined;
		const raw = readFileSync(file);
		try {
			text = zlib.zstdDecompressSync(raw).toString("utf8");
		} catch {
			return undefined;
		}
	} else {
		text = readFileSync(file, "utf8");
	}
	const end = text.indexOf("\n");
	return (end === -1 ? text : text.slice(0, end)).trim();
}

/** Project every stored session to its durable header. */
async function readSessions(sessionsRoot, warnings) {
	const sessions = [];
	if (!existsSync(sessionsRoot)) return sessions;
	for (const bucket of await readdir(sessionsRoot)) {
		const bucketDir = join(sessionsRoot, bucket);
		let entries;
		try {
			entries = await readdir(bucketDir, { withFileTypes: true });
		} catch {
			continue;
		}
		for (const entry of entries) {
			if (!entry.isDirectory()) continue;
			const dir = join(bucketDir, entry.name);
			const files = (await readdir(dir)).filter((name) => logRank(name) !== Number.NEGATIVE_INFINITY);
			if (files.length === 0) continue;
			const log = files.sort((left, right) => logRank(right) - logRank(left))[0];
			const file = join(dir, log);
			const size = (await stat(file)).size;
			if (size > MAX_SESSION_BYTES) {
				warnings.push(`session '${entry.name}': log is ${size} bytes; skipped`);
				continue;
			}
			const line = headerLine(file);
			if (line === undefined) {
				warnings.push(`session '${entry.name}': cannot read header from ${log}; skipped`);
				continue;
			}
			let header;
			try {
				header = JSON.parse(line);
			} catch {
				warnings.push(`session '${entry.name}': header is not JSON; skipped`);
				continue;
			}
			if (typeof header.cwd !== "string") {
				warnings.push(`session '${entry.name}': header carries no cwd; skipped`);
				continue;
			}
			sessions.push({
				id: typeof header.id === "string" ? header.id : entry.name,
				cwd: header.cwd,
				createdAt: typeof header.createdAt === "number" ? header.createdAt : undefined,
				origin: header.origin,
				storeDir: bucket
			});
		}
	}
	return sessions;
}

/** Apply ownership to what this tool wrote, when it runs as root. */
async function applyOwner(path, owner, warnings, recursive = false) {
	if (owner === undefined) return;
	const [uid, gid] = owner.split(":").map(Number);
	if (!Number.isInteger(uid) || !Number.isInteger(gid)) {
		warnings.push(`--owner '${owner}' is not UID:GID; leaving ownership untouched`);
		return;
	}
	try {
		await chown(path, uid, gid, recursive);
	} catch (error) {
		warnings.push(`cannot chown ${path}: ${error.message}`);
	}
}

/**
 * Durably replace the registry exactly the way dsh's own JSON storage backend
 * does: same-directory temp file, mode 0600, fsync, atomic rename, directory
 * fsync. A crash mid-write must never leave a half registry behind.
 */
async function writeAtomic(path, data) {
	const tmp = join(dirname(path), `.workspace-doctor-${randomUUID()}.tmp`);
	try {
		const handle = await open(tmp, "wx", 0o600);
		try {
			await handle.writeFile(data, "utf8");
			await handle.sync();
		} finally {
			await handle.close();
		}
		await rename(tmp, path);
	} catch (error) {
		await rm(tmp, { force: true });
		throw error;
	}
	try {
		const dir = await open(dirname(path), "r");
		try {
			await dir.sync();
		} finally {
			await dir.close();
		}
	} catch {
		// A directory fsync failure costs crash-durability only; the write stands.
	}
}

async function main() {
	const opts = parseArgs(process.argv.slice(2));
	const state = resolve(opts.state);
	const registryPath = join(state, "storages", "workspace.json");
	const warnings = [];

	const doc = await readJson(registryPath);
	if (doc === undefined) {
		process.stdout.write(`dsh-workspace-doctor: no workspace registry at ${registryPath} — nothing to do\n`);
		return 0;
	}
	const unit = doc.unit ?? {};
	const records = doc.tables?.workspaces;
	if (typeof records !== "object" || records === null) die(`${registryPath} has no workspaces table`);
	const knowFormat = unit.name === KNOWN_UNIT.name && unit.version === KNOWN_UNIT.version;

	// Membership as dsh stores it: one owner per session id.
	const accounted = new Map();
	for (const [workspaceId, record] of Object.entries(records)) {
		for (const sessionId of record.sessionIds ?? []) {
			if (accounted.has(sessionId)) warnings.push(`session '${sessionId}' is accounted by two workspaces; left untouched`);
			else accounted.set(sessionId, workspaceId);
		}
	}

	const sessions = await readSessions(join(state, "sessions"), warnings);
	const subagents = sessions.filter((session) => session.origin === "subagent").length;

	// Which sessions belong to a workspace by their own recorded cwd.
	const reattachable = new Map();
	for (const workspaceId of Object.keys(records)) reattachable.set(workspaceId, []);
	for (const session of sessions) {
		if (session.origin === "subagent") continue;
		if (accounted.has(session.id)) continue;
		const cwd = canon(session.cwd);
		for (const [workspaceId, record] of Object.entries(records)) {
			if (canon(record.path) !== cwd) continue;
			reattachable.get(workspaceId).push(session);
			break;
		}
	}
	for (const list of reattachable.values()) list.sort((left, right) => (right.createdAt ?? 0) - (left.createdAt ?? 0));

	// Directory health per workspace.
	const rows = [];
	for (const [workspaceId, record] of Object.entries(records)) {
		let missing = false;
		try {
			missing = !(await stat(record.path)).isDirectory();
		} catch {
			missing = true;
		}
		const wait = reattachable.get(workspaceId) ?? [];
		rows.push({
			workspaceId,
			path: record.path,
			title: record.title,
			dir: missing ? "MISSING" : "ok",
			missing,
			storage: storageOf(record.path, state),
			members: (record.sessionIds ?? []).length,
			reattach: wait.length,
			sessions: wait
		});
	}
	const missingDirs = rows.filter((row) => row.missing);
	const recoverable = rows.reduce((total, row) => total + row.reattach, 0);
	const ephemeral = rows.filter((row) => row.storage === "container filesystem");
	const doDirs = opts.mode === "repair-dirs" || opts.mode === "all";
	const doReattach = opts.mode === "reattach" || opts.mode === "all";

	const report = {
		state,
		registry: registryPath,
		format: `${unit.name ?? "?"} v${unit.version ?? "?"}`,
		workspaces: rows.length,
		sessions: sessions.length,
		subagents,
		missingDirs: missingDirs.map((row) => row.path),
		ephemeral: ephemeral.map((row) => row.path),
		reattachable: rows.flatMap((row) => row.sessions.map((session) => ({ workspace: row.path, sessionId: session.id, cwd: session.cwd, createdAt: session.createdAt }))),
		repairedDirs: [],
		reattached: [],
		dryRun: opts.dryRun,
		warnings
	};

	if (opts.json !== true) {
		process.stdout.write(`dsh-workspace-doctor: state=${state} registry=${report.format}\n`);
		process.stdout.write(`${rows.length} workspace(s), ${sessions.length} stored session(s) (${subagents} subagent)\n\n`);
		process.stdout.write(`${"PROJECT".padEnd(30)} ${"DIR".padEnd(8)} ${"STORAGE".padEnd(17)} ${"SESSIONS".padEnd(9)} REATTACH\n`);
		for (const row of rows) {
			process.stdout.write(`${row.path.slice(0, 29).padEnd(30)} ${row.dir.padEnd(8)} ${row.storage.padEnd(17)} ${String(row.members).padEnd(9)} ${row.reattach === 0 ? "-" : String(row.reattach)}\n`);
		}
	}

	// Repair 1: a workspace whose directory is gone cannot take new sessions, and
	// the sessions it lists lose their validity on its next write. Recreate it.
	const dirsChanged = [];
	let failed = 0;
	if (doDirs) {
		for (const row of missingDirs) {
			if (opts.dryRun) {
				dirsChanged.push(row.path);
				continue;
			}
			try {
				await mkdir(row.path, { recursive: true });
				// Everything under a path that did not exist was just created by us, so
				// it is ours to hand to the runtime user (root-run entrypoint, --owner).
				await applyOwner(row.path, opts.owner, warnings, true);
				dirsChanged.push(row.path);
			} catch (error) {
				failed += 1;
				warnings.push(`cannot create ${row.path}: ${error.message}`);
			}
		}
	}

	// Repair 2: sessions dsh recorded in a workspace directory but never joined to
	// it (created before the project existed, or after the directory was lost).
	const reattached = [];
	if (doReattach) {
		for (const row of rows) {
			if (row.sessions.length === 0) continue;
			const record = records[row.workspaceId];
			const additions = row.sessions.filter((session) => !(record.sessionIds ?? []).includes(session.id));
			if (additions.length === 0) continue;
			reattached.push({ path: row.path, sessionIds: additions.map((session) => session.id) });
			if (opts.dryRun) continue;
			record.sessionIds = [...(record.sessionIds ?? []), ...additions.map((session) => session.id)];
			record.updatedAt = new Date().toISOString();
		}
	}

	const changed = dirsChanged.length > 0 || reattached.length > 0;
	if (changed && opts.dryRun !== true) {
		if (!knowFormat) die(`refusing to write ${registryPath}: unknown registry format ${report.format} (expected ${KNOWN_UNIT.name} v${KNOWN_UNIT.version})`);
		const stamp = new Date().toISOString().replace(/[-:]/g, "").replace(/\..+$/, "");
		const backup = `${registryPath}.bak-${stamp}`;
		await copyFile(registryPath, backup);
		await writeAtomic(registryPath, `${JSON.stringify(doc, null, 2)}\n`);
		await applyOwner(registryPath, opts.owner, warnings);
		await applyOwner(backup, opts.owner, warnings);
		report.backup = backup;
	}
	report.repairedDirs = dirsChanged;
	report.reattached = reattached;

	// Exit contract: a report says whether there is anything to fix; a repair says
	// whether it managed to. An ephemeral-but-repaired workspace is not a failure,
	// or every container start would log one.
	const pending = missingDirs.length > 0 || recoverable > 0;
	const exitCode = opts.mode === "check" || opts.dryRun ? (pending ? 1 : 0) : failed > 0 ? 1 : 0;

	if (opts.json === true) {
		process.stdout.write(`${JSON.stringify(report, null, 2)}\n`);
		return exitCode;
	}

	if (changed) {
		process.stdout.write(`\n${opts.dryRun ? "would repair" : "repaired"}:\n`);
		for (const path of dirsChanged) process.stdout.write(`  created directory ${path}\n`);
		for (const entry of reattached) process.stdout.write(`  re-accounted ${entry.sessionIds.length} session(s) under ${entry.path}\n`);
		if (opts.dryRun) process.stdout.write("  (dry run: nothing was written)\n");
		else if (report.backup !== undefined) process.stdout.write(`  backup: ${report.backup}\n`);
	}

	if (recoverable > 0 && !doReattach) {
		process.stdout.write(`\n${recoverable} session(s) are recorded in a workspace directory but no workspace lists them:\n`);
		for (const row of rows) {
			for (const session of row.sessions) {
				const when = session.createdAt === undefined ? "unknown" : new Date(session.createdAt).toISOString().slice(0, 16);
				process.stdout.write(`  ${session.id}  ${when}  ${row.path}\n`);
			}
		}
	}
	if (missingDirs.length > 0 && !doDirs) {
		process.stdout.write(`\n${missingDirs.length} workspace directory(ies) are missing:\n`);
		for (const row of missingDirs) process.stdout.write(`  ${row.path}\n`);
	}

	const attention = [];
	if (missingDirs.length > 0 && !doDirs) attention.push(`${missingDirs.length} missing workspace director(ies)`);
	if (recoverable > 0 && !doReattach) attention.push(`${recoverable} recoverable ungrouped session(s)`);
	if (ephemeral.length > 0) attention.push(`${ephemeral.length} workspace(s) outside the persisted mounts`);

	if (attention.length > 0) {
		process.stdout.write(`\n${attention.join(", ")}.\n`);
		if (ephemeral.length > 0) {
			process.stdout.write("Workspace directories outside /workspace and the state volume do not survive an\n");
			process.stdout.write("update: only their names are recreated here, never their files. Keep project data\n");
			process.stdout.write("under /workspace.\n");
		}
		if (!changed && (missingDirs.length > 0 || recoverable > 0)) {
			process.stdout.write("Fix: dsh-workspace-doctor --all   (container: set DSH_WORKSPACE_DOCTOR=repair)\n");
		}
	} else {
		process.stdout.write("\nGrouping is intact: every workspace directory resolves and no session is orphaned.\n");
	}
	for (const warning of warnings) process.stdout.write(`warning: ${warning}\n`);
	return exitCode;
}

try {
	process.exitCode = await main();
} catch (error) {
	die(error.message);
}