#!/usr/bin/env bun
/**
 * dench fs daemon
 *
 * Two-way sync between the local Daytona volume mount (/workspace) and
 * Convex Storage. Lives inside every sandbox, started by
 * sandbox-bootstrap.ts on sandbox spin-up.
 *
 * Architecture:
 *   • chokidar watches /workspace for local writes. On change, we hash
 *     the file. If the hash differs from the last successful sync hash,
 *     we upload to Convex Storage via generateUploadUrl + commitFileUpload.
 *   • Convex realtime subscription on `fileTree` rows for this org.
 *     On remote change, if `contentHash` differs from our local
 *     `lastSyncedHash`, we download from Convex Storage to /workspace.
 *   • Per-path 300ms trailing debounce on uploads (batches tight write
 *     loops like editor saves).
 *   • Optimistic conflict via `expectedPreviousHash`. On 409, we stash the
 *     local copy at the next free `<name> (N).<ext>` sibling (reusing an
 *     existing sibling that already holds the same bytes), then re-pull the
 *     latest canonical version. Surfaced in UI.
 *   • Per-path circuit breaker: if a path syncs >5 times in 10s, pause
 *     sync on that path and log loudly to stderr (the live state of
 *     active breakers is exposed via `dench fs status`).
 *
 * FUSE / Mountpoint-for-S3 caveats:
 *   /workspace is a Daytona persistent volume, mounted via AWS
 *   Mountpoint for S3 (a FUSE driver). Two upstream limitations
 *   matter:
 *     • inotify events never fire on the mount
 *       (awslabs/mountpoint-s3#1290). We auto-detect FUSE backing via
 *       `/proc/self/mountinfo` and switch chokidar to `usePolling`
 *       mode in that case.
 *     • rename(2) returns ENOSYS on general-purpose buckets, so the
 *       daemon never relies on rename to detect moves; we just see
 *       add/unlink pairs through chokidar's polling backend.
 *   A periodic reconciliation walk also runs on a 5s tick (10s on
 *   non-FUSE) to catch anything chokidar missed.
 *
 * Live streaming subcommands (one-shot, not persistent):
 *   dench-fs-daemon stream-open <path>            -> prints token
 *   dench-fs-daemon stream-append <token> <chunk>
 *   dench-fs-daemon stream-close <token>
 *
 * Usage (daemon mode):
 *   dench-fs-daemon \
 *     --workspace /workspace \
 *     --hot-cache /tmp/dench-cache
 *
 * `--org-id` is optional: when omitted, the daemon resolves the organization
 * from DENCH_API_KEY via Convex before it starts watching files.
 */
import { createHash } from "node:crypto";
import {
  mkdir,
  readdir,
  readFile,
  rename,
  unlink,
  writeFile,
} from "node:fs/promises";
import { basename, dirname, extname, join, relative } from "node:path";
import { ConvexClient, ConvexHttpClient } from "convex/browser";
import { makeFunctionReference } from "convex/server";
import { decodeInlineTextOrUndefined } from "./fs-daemon-binary";
import {
  type DrainStatus,
  FLUSH_DEFAULT_TIMEOUT_MS,
  runFlushPolling,
} from "./fs-daemon-flush";
import {
  detectWorkspaceFsType,
  findMountForPath,
  isFuseFsType,
  type MountInfo,
  readMountInfo,
} from "./fs-daemon-mount";
import {
  createPullQueueDrain,
  createRemoteFileQueue,
} from "./fs-daemon-pull-queue";
import { cachedWorkspaceMatches, isDaemonHealthy } from "./fs-daemon-health";
import { HashTracker, type BreakerState } from "./fs-daemon-sync-tracker";
import {
  WorkspaceIo,
  shouldIgnore,
  type WorkspaceWorkerStatus,
} from "./fs-daemon-workspace";
import {
  downloadRemoteFile,
  parsePullRequest,
  pullTimeRemaining,
  withinPullDeadline,
  type PullRequest,
} from "./fs-daemon-pull";

let activeWorkspaceIo: WorkspaceIo | undefined;
let daemonMode = false;
function workspaceIo(workspace: string): WorkspaceIo {
  activeWorkspaceIo ??= new WorkspaceIo(workspace);
  if (activeWorkspaceIo.workspace !== workspace)
    throw new Error("workspace_io_invalid: root changed");
  return activeWorkspaceIo;
}

type FileTreeSyncRow = {
  path: string;
  contentHash?: string;
  isDir: boolean;
};

type Args = {
  orgId?: string;
  workspace: string;
  hotCache: string;
};

function parseArgs(argv: string[]): Args {
  const out: Partial<Args> = {};
  for (let i = 0; i < argv.length; i++) {
    const flag = argv[i];
    const value = argv[i + 1];
    switch (flag) {
      case "--org-id":
        out.orgId = value;
        i++;
        break;
      case "--workspace":
        out.workspace = value;
        i++;
        break;
      case "--hot-cache":
        out.hotCache = value;
        i++;
        break;
      default:
        // Ignore unknown flags so future bake updates don't crash the daemon.
        break;
    }
  }
  out.orgId ??= process.env.DENCH_ORG_ID?.trim() || undefined;
  out.workspace ??= process.env.DENCH_VOLUME_PATH?.trim() || "/workspace";
  out.hotCache ??=
    process.env.DENCH_HOT_CACHE_PATH?.trim() || "/tmp/dench-cache";
  return out as Args;
}

function sha256(bytes: Uint8Array): string {
  return createHash("sha256").update(bytes).digest("hex");
}

// Stash a conflicting local copy at a sibling path, using an incrementing
// suffix (`foo (1).txt`, `foo (2).txt`, …) instead of a timestamp so
// repeated conflicts produce a clean, stable series rather than an
// ever-growing pile of `.conflicted-<ts>` files. The file is written
// locally; the daemon's normal watcher path syncs it to Convex like any
// other file. Returns the local path written.
//
// Two properties keep the deterministic suffix safe:
//   • Content-aware reuse — if a `(N)` slot already holds these exact bytes,
//     the copy is already stashed, so we return it instead of minting yet
//     another sibling. This is what keeps repeated conflicts on the same
//     unsynced content from piling up (the whole point of the change), and
//     it also means an unrelated remote `(N)` file mirrored to local disk is
//     skipped rather than overwritten.
//   • Atomic claim — otherwise we write with O_EXCL (`wx`), so two handlers
//     racing for the same slot can't clobber each other; the loser gets
//     EEXIST and advances to the next N.
async function stashConflictCopy(args: {
  io: WorkspaceIo;
  absPath: string;
  bytes: Uint8Array;
  hash: string;
}): Promise<string> {
  const { absPath, bytes, hash } = args;
  const dir = dirname(absPath);
  const ext = extname(absPath);
  const stem = basename(absPath, ext);
  for (let i = 1; i <= 1000; i++) {
    const candidate = join(dir, `${stem} (${i})${ext}`);
    const existing = await args.io.readFile(candidate).catch((error) => {
      if (error.code === "ENOENT") return null;
      throw error;
    });
    if (existing) {
      // Already holds our bytes → nothing to do. Different content → this
      // slot belongs to something else, try the next one.
      if (sha256(existing) === hash) return candidate;
      continue;
    }
    try {
      await args.io.writeFile(candidate, bytes, { flag: "wx" });
      return candidate;
    } catch (error) {
      if ((error as NodeJS.ErrnoException).code === "EEXIST") continue;
      throw error;
    }
  }
  // Every increment up to the cap is taken — fall back to a timestamp, which
  // is effectively unique, so we still stash the bytes somewhere.
  const fallback = join(dir, `${stem} (${Date.now()})${ext}`);
  await args.io.writeFile(fallback, bytes);
  return fallback;
}

function normalizePosixWorkspacePath(input: string): string {
  let cleaned = input.replace(/\\/g, "/").replace(/\/+/g, "/");
  if (cleaned.startsWith("/workspace/")) {
    cleaned = cleaned.slice("/workspace".length);
  } else if (cleaned === "/workspace") {
    cleaned = "/";
  }
  if (!cleaned.startsWith("/")) cleaned = `/${cleaned}`;
  return cleaned.length > 1 && cleaned.endsWith("/")
    ? cleaned.slice(0, -1)
    : cleaned;
}

type PullResponse = {
  errors: Array<{ error: string; path: string }>;
  id: string;
  ok: boolean;
  pulled: Array<{ contentHash: string; path: string }>;
  respondedAt: number;
};

const SYNC_DEBOUNCE_MS = 300;
const MAX_INLINE_TEXT_BYTES = 100_000;
// Reconciliation interval. AWS documents Mountpoint-for-S3's metadata
// staleness window at ~1s; polling tighter than 2s wastes round-trips
// without catching anything new.
const RECONCILE_INTERVAL_FUSE_MS = 5_000;
const RECONCILE_INTERVAL_DEFAULT_MS = 30_000;
// Status / breaker state visible to `dench fs status` over a tiny
// JSON file the daemon refreshes after each tick.
const DAEMON_STATUS_PATH = "/tmp/dench-fs-daemon.status.json";
const PULL_REQUEST_DIR = "/tmp/dench-fs-daemon.pull";

// Mount detection helpers live in fs-daemon-mount.ts so they can be
// unit-tested without pulling chokidar / convex / etc. They are
// re-exported here so existing imports of `cli/fs-daemon` keep
// working in case anything outside this file relied on them.
export {
  detectWorkspaceFsType,
  findMountForPath,
  isFuseFsType,
  readMountInfo,
  type MountInfo,
};

// Convex function references — must match `convex/_generated/api.d.ts` paths.
const api = {
  files: {
    generateUploadUrl: makeFunctionReference<"mutation">(
      "functions/files:generateUploadUrl",
    ),
    commitFileUpload: makeFunctionReference<"mutation">(
      "functions/files:commitFileUpload",
    ),
    deleteFile: makeFunctionReference<"mutation">("functions/files:deleteFile"),
    listTree: makeFunctionReference<"query">("functions/files:listTree"),
    listTreeSnapshot: makeFunctionReference<"query">(
      "functions/files:listTreeSnapshot",
    ),
    resolveApiKeyOrganization: makeFunctionReference<"query">(
      "functions/files:resolveApiKeyOrganization",
    ),
    appendLiveBuffer: makeFunctionReference<"mutation">(
      "functions/files:appendLiveBuffer",
    ),
    finalizeLiveBuffer: makeFunctionReference<"mutation">(
      "functions/files:finalizeLiveBuffer",
    ),
    getFileSignedDownloadUrl: makeFunctionReference<"action">(
      "functions/files:getFileSignedDownloadUrl",
    ),
    reportDaemonHealth: makeFunctionReference<"mutation">(
      "functions/files:reportDaemonHealth",
    ),
  },
};

class DenchFileClient {
  private http: ConvexHttpClient;
  private realtime: ConvexClient | null = null;

  constructor(
    convexUrl: string,
    private readonly apiKey: string,
  ) {
    this.http = new ConvexHttpClient(convexUrl);
  }

  async resolveOrganizationId(): Promise<string> {
    const result = (await this.http.query(api.files.resolveApiKeyOrganization, {
      apiKey: this.apiKey,
    })) as { organizationId: string };
    return result.organizationId;
  }

  async generateUploadUrl(path: string): Promise<string> {
    return (await this.http.mutation(api.files.generateUploadUrl, {
      path,
      apiKey: this.apiKey,
    })) as string;
  }

  async putBytes(uploadUrl: string, bytes: Uint8Array): Promise<string> {
    const response = await fetch(uploadUrl, {
      method: "POST",
      headers: { "content-type": "application/octet-stream" },
      body: new Blob([bytes as BlobPart]),
    });
    if (!response.ok) {
      throw new Error(
        `Storage upload failed (${response.status}): ${await response
          .text()
          .catch(() => "")}`,
      );
    }
    const { storageId } = (await response.json()) as { storageId: string };
    return storageId;
  }

  async commitFileUpload(args: {
    path: string;
    storageId: string;
    contentHash: string;
    size: number;
    expectedPreviousHash?: string;
    text?: string;
    lastModifiedBy: string;
  }): Promise<{ versionConflict: boolean; currentHash?: string }> {
    try {
      await this.http.mutation(api.files.commitFileUpload, {
        ...args,
        apiKey: this.apiKey,
      });
      return { versionConflict: false };
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error);
      if (
        message.includes("version_mismatch") ||
        message.includes("Hash mismatch")
      ) {
        // Convex serializes ConvexError data into the message; pull
        // currentHash if present.
        const m = /currentHash["']?:["']?([a-f0-9]{64})/.exec(message);
        return { versionConflict: true, currentHash: m?.[1] };
      }
      throw error;
    }
  }

  async deleteFile(path: string): Promise<void> {
    await this.http.mutation(api.files.deleteFile, {
      path,
      apiKey: this.apiKey,
    });
  }

  /**
   * Read direct children of a Convex file-tree prefix. Used by `dench
   * fs status` to recursively crawl the canonical tree and diff
   * against the local FS snapshot.
   */
  async listTree(prefix: string): Promise<FileTreeSyncRow[]> {
    const rows = (await this.http.query(api.files.listTree, {
      prefix,
      apiKey: this.apiKey,
    })) as FileTreeSyncRow[];
    return Array.isArray(rows) ? rows : [];
  }

  async getFileTreeRow(path: string): Promise<FileTreeSyncRow | null> {
    const normalized = normalizePosixWorkspacePath(path);
    const parent = dirname(normalized) === "." ? "/" : dirname(normalized);
    const rows = await this.listTree(parent);
    return rows.find((row) => row.path === normalized) ?? null;
  }

  async getSignedDownloadUrl(path: string): Promise<string | null> {
    const url = (await this.http.action(api.files.getFileSignedDownloadUrl, {
      path,
      apiKey: this.apiKey,
    })) as string | null;
    return url ?? null;
  }

  async appendLiveBuffer(args: {
    path: string;
    runId: string;
    chunk: string;
  }): Promise<void> {
    await this.http.mutation(api.files.appendLiveBuffer, {
      ...args,
      apiKey: this.apiKey,
    } as never);
  }

  async finalizeLiveBuffer(path: string): Promise<void> {
    await this.http.mutation(api.files.finalizeLiveBuffer, {
      path,
      apiKey: this.apiKey,
    });
  }

  /**
   * Heartbeat the per-org `fileTreeHealth` row. Best-effort — a Convex
   * blip should never block the reconcile path.
   */
  async reportHealth(args: {
    pid: number;
    fsType: string;
    isFuse: boolean;
    workspace: string;
    lastReconcileFinishedAt: number;
    lastFlushAt?: number;
    onDiskCount: number;
    activeBreakers: number;
    pendingDrain: number;
  }): Promise<void> {
    await this.http.mutation(api.files.reportDaemonHealth, {
      ...args,
      apiKey: this.apiKey,
    } as never);
  }

  /**
   * Subscribe to fileTree updates for the org and invoke onChange whenever
   * a row's contentHash diverges from our local lastSyncedHash. The
   * download path is implemented in the watcher.
   */
  subscribeFileTree(
    convexUrl: string,
    onChange: (rows: FileTreeSyncRow[]) => void,
  ): () => void {
    if (!this.realtime) {
      this.realtime = new ConvexClient(convexUrl);
    }
    const unsubscribe = this.realtime.onUpdate(
      api.files.listTreeSnapshot,
      { apiKey: this.apiKey },
      (snapshot: unknown) => {
        // Remote writes may land deep in the tree (for example seeded
        // `/skills/<slug>/SKILL.md` files). A root-only `listTree`
        // subscription only sees the `/skills` directory row, so it
        // never pulls the actual file bytes down to the volume.
        const rows =
          snapshot &&
          typeof snapshot === "object" &&
          Array.isArray((snapshot as { rows?: unknown }).rows)
            ? (snapshot as { rows: FileTreeSyncRow[] }).rows
            : [];
        onChange(rows);
      },
    );
    return unsubscribe;
  }
}

async function pullDownRemoteFile(
  client: DenchFileClient,
  args: {
    workspace: string;
    posixPath: string;
    expectedHash: string;
    tracker: HashTracker;
    deadline?: number;
    shouldPublish?: () => boolean;
  },
): Promise<void> {
  const deadline = args.deadline ?? Date.now() + 30_000;
  const buffer = await downloadRemoteFile({
    resolveUrl: () => client.getSignedDownloadUrl(args.posixPath),
    expectedHash: args.expectedHash,
    deadline,
  });
  const localPath = `${args.workspace}${args.posixPath}`;
  if (args.shouldPublish && !args.shouldPublish()) return;
  const io = workspaceIo(args.workspace);
  await io.mkdir(
    dirname(localPath),
    Math.min(8_000, pullTimeRemaining(deadline)),
  );
  if (args.shouldPublish && !args.shouldPublish()) return;
  await io.writeFile(
    localPath,
    buffer,
    undefined,
    Math.min(8_000, pullTimeRemaining(deadline)),
  );
  pullTimeRemaining(deadline);
  args.tracker.set(localPath, args.expectedHash);
  console.log(
    `[dench-fs-daemon] pulled remote ${args.posixPath} (${buffer.byteLength}B, sha256=${args.expectedHash.slice(0, 8)})`,
  );
}

async function processPullRequests(args: {
  client: DenchFileClient;
  tracker: HashTracker;
  workspace: string;
}): Promise<void> {
  await mkdir(PULL_REQUEST_DIR, { recursive: true });
  const entries = await readdir(PULL_REQUEST_DIR).catch(() => []);
  const requestFiles = entries.filter(
    (entry) => entry.startsWith("request-") && entry.endsWith(".json"),
  );
  for (const entry of requestFiles) {
    const requestPath = `${PULL_REQUEST_DIR}/${entry}`;
    let request: PullRequest | null = null;
    try {
      request = parsePullRequest(
        JSON.parse(await readFile(requestPath, "utf-8")),
        entry,
      );
    } catch (error) {
      console.error(
        `[dench-fs-daemon] invalid pull request ${entry}: ${
          error instanceof Error ? error.message : String(error)
        }`,
      );
      await unlink(requestPath).catch(() => undefined);
      continue;
    }
    const response: PullResponse = {
      errors: [],
      id: request.id,
      ok: true,
      pulled: [],
      respondedAt: Date.now(),
    };
    const deadline = Math.min(
      request.deadlineAt ?? request.requestedAt + 10_000,
      Date.now() + 60_000,
    );
    for (const rawPath of request.paths) {
      const path = normalizePosixWorkspacePath(rawPath);
      try {
        if (path.split("/").includes("..") || path.includes("\0"))
          throw new Error("pull_invalid_path");
        const row = await withinPullDeadline(
          () => args.client.getFileTreeRow(path),
          deadline,
        );
        if (!row || row.isDir || !row.contentHash) {
          throw new Error("File is not present in the workspace file tree.");
        }
        await pullDownRemoteFile(args.client, {
          expectedHash: row.contentHash,
          posixPath: path,
          tracker: args.tracker,
          workspace: args.workspace,
          deadline,
        });
        response.pulled.push({ contentHash: row.contentHash, path });
      } catch (error) {
        response.ok = false;
        response.errors.push({
          error: error instanceof Error ? error.message : String(error),
          path,
        });
      }
    }
    response.respondedAt = Date.now();
    const responsePath = `${PULL_REQUEST_DIR}/response-${request.id}.json`;
    const temporaryResponsePath = `${responsePath}.${process.pid}.tmp`;
    await writeFile(temporaryResponsePath, JSON.stringify(response), "utf-8");
    await rename(temporaryResponsePath, responsePath);
    await unlink(requestPath).catch(() => undefined);
  }
}

/**
 * sysexits.h-style exit codes the supervisor (in
 * src/lib/daytona/sandbox-bootstrap.ts) recognizes as **permanent** —
 * keep restarting won't fix a missing env var or a missing dependency,
 * so the supervisor gives up instead of crash-looping. Anything else
 * (network blip, transient Convex error) exits 1 and gets retried.
 */
const EXIT_CONFIG_ERROR = 78;

async function runDaemon(daemonArgs: Args): Promise<void> {
  const convexUrl = process.env.CONVEX_URL?.trim();
  const apiKey = process.env.DENCH_API_KEY?.trim();
  if (!convexUrl || !apiKey) {
    console.error(
      "[dench-fs-daemon] CONVEX_URL or DENCH_API_KEY missing in env; refusing to start",
    );
    process.exit(EXIT_CONFIG_ERROR);
  }

  const client = new DenchFileClient(convexUrl, apiKey);
  const resolvedOrgId =
    daemonArgs.orgId ?? (await client.resolveOrganizationId());
  const mount = await detectWorkspaceFsType(daemonArgs.workspace);
  console.log(
    `[dench-fs-daemon] starting org=${resolvedOrgId} workspace=${daemonArgs.workspace} fsType=${mount.fsType}${mount.isFuse ? " (FUSE)" : ""}`,
  );
  if (mount.isFuse) {
    console.log(
      "[dench-fs-daemon] FUSE backing detected; enabling polling mode (inotify is not delivered through the mounted workspace)",
    );
  }
  daemonMode = true;
  const io = workspaceIo(daemonArgs.workspace);

  const tracker = new HashTracker();
  let retryRemote: (() => Promise<void>) | undefined;
  let previousWorkerState = io.snapshot().state;
  const status = new DaemonStatusWriter({
    workspace: daemonArgs.workspace,
    organizationId: resolvedOrgId,
    fsType: mount.fsType,
    isFuse: mount.isFuse,
  });
  // Hook the writer to the live tracker so every status flush
  // captures the current pending/in-flight counts. The flush
  // subcommand polls these via the status file.
  status.setDrainSnapshot(() => tracker.drainStatus());
  status.setWorkerSnapshot(() => io.snapshot());
  io.on("status", () => {
    void status.flush();
    const state = io.snapshot().state;
    if (state === "ready" && previousWorkerState !== "ready")
      void retryRemote?.();
    previousWorkerState = state;
  });
  const heartbeat = setInterval(() => {
    void status.flush();
    if (io.snapshot().state === "ready") void retryRemote?.();
  }, 5_000);
  tracker.setBreakerListener((breakerState) => {
    status.recordBreakerEvent(breakerState);
    void status.flush();
  });

  // Register the pull signal before publishing the daemon status. `fs pull`
  // treats that status as readiness; publishing first left a startup window
  // where SIGUSR2 was lost (or terminated the daemon) and the request timed
  // out. A queued rerun also prevents signals received mid-drain from being
  // coalesced away.
  const drainPullRequests = createPullQueueDrain({
    processRequests: () =>
      processPullRequests({
        client,
        tracker,
        workspace: daemonArgs.workspace,
      }),
    onError: (error) =>
      console.error(
        `[dench-fs-daemon] pull queue failed: ${
          error instanceof Error ? error.message : String(error)
        }`,
      ),
  });
  process.on("SIGUSR2", () => {
    console.log(`[dench-fs-daemon] SIGUSR2 received — processing pull queue`);
    void drainPullRequests();
  });
  await status.flush();
  // Recover requests left behind if a previous daemon exited after the
  // request file was written but before it could process the signal.
  void drainPullRequests();

  // The watcher, its polling stat calls, and every mount operation live in
  // the child. Do not wait for it before accepting requests or reporting health.
  const reconcileIntervalMs = mount.isFuse
    ? RECONCILE_INTERVAL_FUSE_MS
    : RECONCILE_INTERVAL_DEFAULT_MS;
  void io
    .watch(mount.isFuse)
    .catch((error) => console.error("[dench-fs-daemon] " + error.message));
  io.on("add", (absPath: string) => scheduleSync(absPath));
  io.on("change", (absPath: string) => scheduleSync(absPath));
  io.on("unlink", async (absPath: string) => {
    const rel = relative(daemonArgs.workspace, absPath);
    if (shouldIgnore(rel)) return;
    try {
      await client.deleteFile(`/${rel}`);
      tracker.forget(absPath);
      console.log(`[dench-fs-daemon] deleted /${rel}`);
    } catch (error) {
      console.error(
        `[dench-fs-daemon] delete failed for /${rel}: ${
          error instanceof Error ? error.message : String(error)
        }`,
      );
    }
  });

  function scheduleSync(absPath: string): void {
    const rel = relative(daemonArgs.workspace, absPath);
    if (shouldIgnore(rel)) return;
    if (tracker.recordAndCheckBreaker(absPath)) return;
    tracker.schedule(absPath, SYNC_DEBOUNCE_MS, async () => {
      const stats = await io.stat(absPath).catch((error) => {
        if (error.code === "ENOENT") return null;
        throw error;
      });
      if (!stats || !stats.isFile) return;
      const bytes = await io.readFile(absPath);
      const hash = sha256(bytes);
      const previous = tracker.get(absPath);
      if (previous === hash) return;
      const posixPath = `/${rel}`;
      console.log(
        `[dench-fs-daemon] sync ${posixPath} (${stats.size}B, sha256=${hash.slice(0, 8)})`,
      );
      const uploadUrl = await client.generateUploadUrl(posixPath);
      const storageId = await client.putBytes(uploadUrl, bytes);
      // Only inline text for SMALL files that look like real UTF-8.
      // Without the binary check, a 50KB PNG would get
      // `TextDecoder({ fatal: false }).decode(bytes)`'d into a stream
      // of U+FFFD replacement chars and persisted into
      // `fileContents.text`, corrupting every viewer that hits the
      // inline cache instead of the storage blob. See cli/fs-daemon-binary.ts.
      const inlineText =
        stats.size <= MAX_INLINE_TEXT_BYTES
          ? decodeInlineTextOrUndefined(bytes)
          : undefined;
      const result = await client.commitFileUpload({
        path: posixPath,
        storageId,
        contentHash: hash,
        size: stats.size,
        expectedPreviousHash: previous,
        text: inlineText,
        lastModifiedBy: "daemon",
      });
      if (result.versionConflict) {
        // Stash the local copy in a sibling file, then re-pull the remote
        // version into the canonical path.
        const conflicted = await stashConflictCopy({
          io,
          absPath,
          bytes,
          hash,
        });
        console.warn(
          `[dench-fs-daemon] version conflict for ${posixPath}; local copy stashed at ${conflicted}`,
        );
        if (result.currentHash) {
          await pullDownRemoteFile(client, {
            workspace: daemonArgs.workspace,
            posixPath,
            expectedHash: result.currentHash,
            tracker,
          });
        }
        return;
      }
      tracker.set(absPath, hash);
    });
  }

  // Subscribe to remote fileTree changes and download anything whose
  // contentHash differs from our local lastSyncedHash.
  const remoteQueue = createRemoteFileQueue({
    isReady: () => io.snapshot().state === "ready",
    getHash: (path) => tracker.get(`${daemonArgs.workspace}${path}`),
    pull: (row, shouldPublish) =>
      pullDownRemoteFile(client, {
        workspace: daemonArgs.workspace,
        posixPath: row.path,
        expectedHash: row.contentHash,
        tracker,
        shouldPublish,
      }),
    onError: (error, path) =>
      console.error(
        `[dench-fs-daemon] remote pull failed${path ? ` for ${path}` : ""}`,
        error instanceof Error ? error.message : "unknown",
      ),
  });
  retryRemote = remoteQueue.retry;
  const unsubscribe = client.subscribeFileTree(convexUrl, (rows) => {
    void remoteQueue.update(rows);
  });

  // Periodic walk + reconciliation. Catches changes chokidar missed
  // (any file write on a Mountpoint volume from a process other than
  // this sandbox, or from the Daytona SDK uploadFile path) and writes
  // a heartbeat to the status file so `dench fs status` can confirm
  // the daemon is alive and what it last did. Also mirrors the
  // heartbeat to Convex `fileTreeHealth` so the UI sidebar + chat-
  // turn end-of-turn check can surface drift loudly.
  const reconciler = startReconcileLoop({
    workspace: daemonArgs.workspace,
    intervalMs: reconcileIntervalMs,
    tracker,
    status,
    onLocalChange: scheduleSync,
    onLocalDelete: async (absPath) => {
      const rel = relative(daemonArgs.workspace, absPath);
      if (shouldIgnore(rel)) return;
      try {
        await client.deleteFile(`/${rel}`);
        tracker.forget(absPath);
        console.log(
          `[dench-fs-daemon] reconcile deleted /${rel} (chokidar missed unlink)`,
        );
      } catch (error) {
        console.error(
          `[dench-fs-daemon] reconcile delete failed for /${rel}: ${
            error instanceof Error ? error.message : String(error)
          }`,
        );
      }
    },
    reportHealth: async (snapshot) => {
      await client.reportHealth({
        pid: process.pid,
        fsType: mount.fsType,
        isFuse: mount.isFuse,
        workspace: daemonArgs.workspace,
        lastReconcileFinishedAt: snapshot.finishedAt,
        lastFlushAt: status.snapshot().lastFlushAt ?? undefined,
        onDiskCount: snapshot.observed,
        activeBreakers: snapshot.activeBreakers,
        pendingDrain: snapshot.pendingDrain,
      });
    },
  });

  // SIGUSR1 = "flush requested" from `dench-fs-daemon flush` (chat-turn
  // calls it after every execute_bash that touched /workspace). Run an
  // immediate reconcile tick + record the flush completion timestamp so
  // the polling caller can confirm the signal was honored. Errors are
  // swallowed loudly into the daemon log; the flush command itself is
  // responsible for surfacing timeouts.
  process.on("SIGUSR1", () => {
    const requestedAt = Date.now();
    console.log(
      `[dench-fs-daemon] SIGUSR1 received — running reconcile tick on demand`,
    );
    void reconciler
      .runOneTick()
      .then(() => {
        status.recordFlushCompleted(requestedAt);
        return status.flush();
      })
      .catch((error) =>
        console.error(
          `[dench-fs-daemon] flush tick failed: ${
            error instanceof Error ? error.message : String(error)
          }`,
        ),
      )
      .catch(() => undefined);
  });

  process.on("SIGTERM", () => {
    reconciler.stop();
    unsubscribe();
    clearInterval(heartbeat);
    io.close();
    process.exit(0);
  });

  console.log(
    `[dench-fs-daemon] watching ${daemonArgs.workspace} (reconcile every ${reconcileIntervalMs}ms)`,
  );
  setInterval(() => undefined, 60_000);
}

// ── Reconciliation loop ────────────────────────────────────────────────────
//
// Runs on a fixed tick and:
//   • Walks the workspace, collects (absPath, mtimeMs, size) for every
//     non-ignored file.
//   • Compares against the in-memory tracker (`tracker.knownPaths()`).
//     New paths or paths whose mtime/size moved → schedule a sync.
//   • Paths the tracker knew about that are now absent from disk →
//     enqueue a delete.
//   • Refreshes the daemon status file with the latest scan summary.

type FileSnapshot = {
  absPath: string;
  size: number;
  mtimeMs: number;
};

async function snapshotWorkspace(
  workspace: string,
): Promise<Map<string, FileSnapshot>> {
  const files = await workspaceIo(workspace).scan();
  return new Map(files.map((file) => [file.absPath, file]));
}

type ReconcileLoopHandle = {
  /** Tear down the periodic timers (called on SIGTERM). */
  stop: () => void;
  /**
   * Run the reconcile body once on demand. Returned so the SIGUSR1
   * handler can drain disk-only writes immediately without waiting
   * for the next 5–30s tick — that's how `dench-fs-daemon flush`
   * achieves its sub-second turnaround.
   */
  runOneTick: () => Promise<void>;
};

function startReconcileLoop(args: {
  workspace: string;
  intervalMs: number;
  tracker: HashTracker;
  status: DaemonStatusWriter;
  onLocalChange: (absPath: string) => void;
  onLocalDelete: (absPath: string) => Promise<void> | void;
  /**
   * Called once per tick (after the status file is flushed) to mirror
   * the daemon health into Convex `fileTreeHealth`. Best-effort —
   * Convex blips never block the reconcile path, but persistent
   * failures get logged.
   */
  reportHealth?: (snapshot: {
    finishedAt: number;
    observed: number;
    activeBreakers: number;
    pendingDrain: number;
  }) => Promise<void> | void;
}): ReconcileLoopHandle {
  const last = new Map<string, FileSnapshot>();

  let stopped = false;
  // Serialize concurrent ticks (SIGUSR1 + scheduled timer can race).
  // The flush-driven tick can arrive milliseconds before the scheduled
  // one; without serialization we'd double-walk the volume and the
  // status file's `lastReconcile.finishedAt` would jitter
  // backwards if writes interleave.
  let inFlightTick: Promise<void> | null = null;
  const tick = async () => {
    if (stopped) return;
    if (inFlightTick) {
      await inFlightTick;
      return;
    }
    const start = Date.now();
    let scanCompleted = false;
    let observed = 0;
    let scheduled = 0;
    let deleted = 0;
    inFlightTick = (async () => {
      try {
        const snapshot = await snapshotWorkspace(args.workspace);
        observed = snapshot.size;
        for (const [absPath, info] of snapshot) {
          const prev = last.get(absPath);
          if (
            !prev ||
            args.tracker.needsRetry(absPath) ||
            prev.size !== info.size ||
            prev.mtimeMs !== info.mtimeMs
          ) {
            args.onLocalChange(absPath);
            scheduled++;
          }
        }
        for (const absPath of last.keys()) {
          if (!snapshot.has(absPath)) {
            await args.onLocalDelete(absPath);
            deleted++;
          }
        }
        last.clear();
        for (const path of args.tracker.retryPaths()) {
          if (!snapshot.has(path)) args.tracker.forget(path);
        }
        for (const [absPath, info] of snapshot) {
          last.set(absPath, info);
        }
        scanCompleted = true;
      } catch (error) {
        console.error(
          `[dench-fs-daemon] reconcile scan failed: ${
            error instanceof Error ? error.message : String(error)
          }`,
        );
        throw error;
      } finally {
        if (scanCompleted) {
          const finishedAt = Date.now();
          args.status.recordReconcile({
            finishedAt,
            durationMs: finishedAt - start,
            observed,
            scheduledForSync: scheduled,
            deleted,
            breakers: args.tracker.activeBreakers(),
          });
          await args.status.flush();
          if (args.reportHealth) {
            // Best-effort heartbeat into Convex `fileTreeHealth`. The
            // chat-turn end-of-turn check + sidebar pill key off this
            // row to surface "sync degraded" without polling the
            // sandbox FS directly.
            const drain = args.tracker.drainStatus();
            try {
              await args.reportHealth({
                finishedAt,
                observed,
                activeBreakers: args.tracker.activeBreakers().length,
                pendingDrain: drain.pendingPaths + drain.inFlightUploads,
              });
            } catch (error) {
              console.error(
                `[dench-fs-daemon] reportHealth failed: ${
                  error instanceof Error ? error.message : String(error)
                }`,
              );
            }
          }
        } else {
          await args.status.flush();
        }
      }
    })();
    try {
      await inFlightTick;
    } finally {
      inFlightTick = null;
    }
  };

  // Stagger the first tick so we don't race chokidar's initial add
  // floods on daemon startup.
  const initial = setTimeout(
    () => {
      void tick().catch(() => undefined);
    },
    Math.min(args.intervalMs, 5_000),
  );
  const interval = setInterval(() => {
    void tick().catch(() => undefined);
  }, args.intervalMs);

  return {
    stop: () => {
      stopped = true;
      clearTimeout(initial);
      clearInterval(interval);
    },
    runOneTick: tick,
  };
}

// ── Status file ────────────────────────────────────────────────────────────
//
// `dench fs status` reads /tmp/dench-fs-daemon.status.json so the agent
// (and humans) can quickly see whether the daemon is alive, what
// filesystem it thinks /workspace is on, and how many files it just
// scanned. Lives on tmpfs so it never lands in the volume itself.

type DaemonStatus = {
  mountWorker?: WorkspaceWorkerStatus;
  pid: number;
  startedAt: number;
  lastUpdatedAt: number;
  workspace: string;
  organizationId: string;
  fsType: string;
  isFuse: boolean;
  lastReconcile: {
    finishedAt: number;
    durationMs: number;
    observed: number;
    scheduledForSync: number;
    deleted: number;
  } | null;
  /**
   * Wall-clock of the most recent SIGUSR1-driven flush request that
   * the daemon completed. Used by `dench-fs-daemon flush` to confirm
   * the daemon actually picked up the signal (not just that the
   * scheduled reconcile tick happened to land afterwards).
   */
  lastFlushAt: number | null;
  /**
   * Mirror of `HashTracker.drainStatus()` taken at the same moment as
   * `lastReconcile.finishedAt`. The flush poller waits for both
   * counts to be zero AND `lastReconcile.finishedAt > t0` before
   * declaring success.
   */
  pendingDrain: DrainStatus;
  activeBreakers: BreakerState[];
  recentBreakerEvents: BreakerState[];
};

class DaemonStatusWriter {
  private state: DaemonStatus;
  private writing: Promise<void> | null = null;
  // Pulled at flush() time so the on-disk drain count tracks the live
  // tracker without forcing every recordX() caller to plumb it through.
  private drainSnapshotFn: (() => DrainStatus) | null = null;
  private workerSnapshotFn: (() => WorkspaceWorkerStatus) | null = null;
  setWorkerSnapshot(fn: () => WorkspaceWorkerStatus): void {
    this.workerSnapshotFn = fn;
  }

  constructor(args: {
    workspace: string;
    organizationId: string;
    fsType: string;
    isFuse: boolean;
  }) {
    const now = Date.now();
    this.state = {
      pid: process.pid,
      startedAt: now,
      lastUpdatedAt: now,
      workspace: args.workspace,
      organizationId: args.organizationId,
      fsType: args.fsType,
      isFuse: args.isFuse,
      lastReconcile: null,
      lastFlushAt: null,
      pendingDrain: { pendingPaths: 0, inFlightUploads: 0 },
      activeBreakers: [],
      recentBreakerEvents: [],
    };
  }

  /**
   * Wire up a callback the writer invokes inside `flush()` to grab
   * the latest drain numbers off the live HashTracker. Kept as a
   * snapshot fn (not a stored counter) so recordReconcile/breaker
   * events don't have to thread the tracker through.
   */
  setDrainSnapshot(fn: () => DrainStatus): void {
    this.drainSnapshotFn = fn;
  }

  recordReconcile(args: {
    finishedAt: number;
    durationMs: number;
    observed: number;
    scheduledForSync: number;
    deleted: number;
    breakers: BreakerState[];
  }): void {
    this.state.lastReconcile = {
      finishedAt: args.finishedAt,
      durationMs: args.durationMs,
      observed: args.observed,
      scheduledForSync: args.scheduledForSync,
      deleted: args.deleted,
    };
    this.state.activeBreakers = args.breakers;
  }

  /**
   * Mark a SIGUSR1 flush request as completed. Independent of
   * recordReconcile because the same tick may be triggered by either
   * the timer OR a flush signal — `lastFlushAt` proves the signal
   * specifically was honored.
   */
  recordFlushCompleted(at: number): void {
    this.state.lastFlushAt = at;
  }

  recordBreakerEvent(state: BreakerState): void {
    const recent = [...this.state.recentBreakerEvents, state];
    // Cap at last 10 events so the file stays small.
    while (recent.length > 10) recent.shift();
    this.state.recentBreakerEvents = recent;
  }

  async flush(): Promise<void> {
    if (this.workerSnapshotFn) this.state.mountWorker = this.workerSnapshotFn();
    this.state.lastUpdatedAt = Date.now();
    if (this.drainSnapshotFn) {
      this.state.pendingDrain = this.drainSnapshotFn();
    }
    const body = `${JSON.stringify(this.state, null, 2)}\n`;
    // Serialize so concurrent flushes don't interleave writes.
    const previous = this.writing ?? Promise.resolve();
    this.writing = previous
      .catch(() => undefined)
      .then(() =>
        writeFile(DAEMON_STATUS_PATH, body, { encoding: "utf-8" }).catch(
          () => undefined,
        ),
      );
    await this.writing;
  }

  /**
   * Snapshot the current status state. Read-only — used by the
   * reconcile loop's per-tick health heartbeat to Convex without
   * forking a stale copy back into the writer.
   */
  snapshot(): Readonly<DaemonStatus> {
    return this.state;
  }
}

// ── Live streaming subcommands ─────────────────────────────────────────────
//
// Tokens are stored in /tmp/dench-stream-tokens.json so subsequent stream-
// append / stream-close calls can resolve back to the path + run id.

const TOKEN_STORE_PATH = "/tmp/dench-stream-tokens.json";

async function readTokenStore(): Promise<
  Record<string, { path: string; runId: string }>
> {
  try {
    const text = await readFile(TOKEN_STORE_PATH, "utf-8");
    return JSON.parse(text) as Record<string, { path: string; runId: string }>;
  } catch {
    return {};
  }
}

async function writeTokenStore(
  store: Record<string, { path: string; runId: string }>,
): Promise<void> {
  await writeFile(TOKEN_STORE_PATH, JSON.stringify(store), "utf-8");
}

function newToken(): string {
  return Math.random().toString(36).slice(2, 10) + Date.now().toString(36);
}

async function runStreamOpen(args: string[]): Promise<void> {
  const path = args[0];
  if (!path) {
    throw new Error("dench-fs-daemon stream-open requires <path>");
  }
  const runId = process.env.DENCH_RUN_ID?.trim() ?? `manual-${Date.now()}`;
  const token = newToken();
  const store = await readTokenStore();
  store[token] = { path, runId };
  await writeTokenStore(store);
  console.log(token);
}

async function runStreamAppend(args: string[]): Promise<void> {
  const token = args[0];
  const chunk = args.slice(1).join(" ");
  if (!token || chunk === undefined) {
    throw new Error("dench-fs-daemon stream-append requires <token> <chunk>");
  }
  const store = await readTokenStore();
  const entry = store[token];
  if (!entry) throw new Error(`Unknown stream token: ${token}`);
  const convexUrl = process.env.CONVEX_URL?.trim();
  const apiKey = process.env.DENCH_API_KEY?.trim();
  if (!convexUrl || !apiKey) {
    throw new Error(
      "stream-append requires CONVEX_URL and DENCH_API_KEY in env",
    );
  }
  const client = new DenchFileClient(convexUrl, apiKey);
  await client.appendLiveBuffer({
    path: entry.path,
    runId: entry.runId,
    chunk,
  });
}

async function runStreamClose(args: string[]): Promise<void> {
  const token = args[0];
  if (!token) {
    throw new Error("dench-fs-daemon stream-close requires <token>");
  }
  const store = await readTokenStore();
  const entry = store[token];
  if (!entry) throw new Error(`Unknown stream token: ${token}`);
  const convexUrl = process.env.CONVEX_URL?.trim();
  const apiKey = process.env.DENCH_API_KEY?.trim();
  if (!convexUrl || !apiKey) {
    throw new Error(
      "stream-close requires CONVEX_URL and DENCH_API_KEY in env",
    );
  }
  const client = new DenchFileClient(convexUrl, apiKey);
  await client.finalizeLiveBuffer(entry.path);
  delete store[token];
  await writeTokenStore(store);
}

// ── Flush subcommand ──────────────────────────────────────────────────────
//
// `dench-fs-daemon flush [--timeout-ms 5000]` blocks until the live
// daemon has finished a reconcile tick AND drained every pending sync
// timer + in-flight upload. Used by chat-turn after each `execute_bash`
// that touched /workspace so the model sees Convex tree state that
// matches what the bash command produced — no more 5s polling jitter
// where files appear in the sidebar after the chat turn already
// finished.
//
// Protocol:
//   1. Read pid from /tmp/dench-fs-daemon.status.json.
//   2. Send SIGUSR1 (the daemon's runDaemon registers a handler).
//   3. Poll the status file every POLL_INTERVAL_MS until both
//      a) `lastReconcile.finishedAt > t0`, AND
//      b) `pendingDrain.pendingPaths === 0 && pendingDrain.inFlightUploads === 0`.
//   4. Time out cleanly with exit 2 + the last-seen drain count so
//      callers (chat-turn) can decide whether to surface drift.
//
// Exit codes: 0 ok, 1 no daemon, 2 timeout, 3 signal failed.
//
// The polling logic itself lives in `cli/fs-daemon-flush.ts` so the
// unit tests can exercise it without booting the daemon.

async function runFlush(argv: string[]): Promise<void> {
  const timeoutIndex = argv.indexOf("--timeout-ms");
  const parsedTimeout =
    timeoutIndex === -1
      ? FLUSH_DEFAULT_TIMEOUT_MS
      : Number.parseInt(
          argv[timeoutIndex + 1] ?? `${FLUSH_DEFAULT_TIMEOUT_MS}`,
          10,
        );
  const timeoutMs = Number.isFinite(parsedTimeout)
    ? parsedTimeout
    : FLUSH_DEFAULT_TIMEOUT_MS;
  const json = argv.includes("--json");

  const outcome = await runFlushPolling({
    timeoutMs,
    readStatus: async () => (await readStatusFile()).state,
    isPidAlive,
    signal: (pid) => {
      // process.kill with signal 'SIGUSR1' on Linux. On Darwin/local-dev
      // the same call works for any node process; Windows lacks SIGUSR1
      // but the daemon never runs there.
      process.kill(pid, "SIGUSR1");
    },
  });

  if (json) {
    process.stdout.write(`${JSON.stringify(outcome)}\n`);
  } else if (outcome.ok) {
    process.stdout.write(
      `DENCH_FS_FLUSH=ok pid=${outcome.pid} duration=${outcome.durationMs}ms\n`,
    );
  } else if (outcome.reason === "timeout") {
    process.stderr.write(
      `DENCH_FS_FLUSH=timeout pid=${outcome.pid} duration=${outcome.durationMs}ms pending=${outcome.drain.pendingPaths} inflight=${outcome.drain.inFlightUploads}\n`,
    );
  } else if (outcome.reason === "signal_failed") {
    process.stderr.write(
      `DENCH_FS_FLUSH=signal_failed pid=${outcome.pid} error=${outcome.error}\n`,
    );
  } else {
    process.stderr.write(`DENCH_FS_FLUSH=no_daemon\n`);
  }

  if (outcome.ok) return;
  process.exitCode =
    outcome.reason === "no_daemon" ? 1 : outcome.reason === "timeout" ? 2 : 3;
}

async function runPull(argv: string[]): Promise<void> {
  const json = argv.includes("--json");
  const wait = argv.includes("--wait");
  const timeoutIndex = argv.indexOf("--timeout-ms");
  const timeoutMs =
    timeoutIndex === -1
      ? 10_000
      : Number.parseInt(argv[timeoutIndex + 1] ?? "10000", 10);
  const paths: string[] = [];
  for (let i = 0; i < argv.length; i += 1) {
    const arg = argv[i];
    if (arg === "--json" || arg === "--wait") continue;
    if (arg === "--timeout-ms") {
      i += 1;
      continue;
    }
    paths.push(normalizePosixWorkspacePath(arg));
  }
  if (paths.length === 0) {
    throw new Error("dench-fs-daemon pull requires at least one path");
  }
  const status = (await readStatusFile()).state;
  if (!status || !isPidAlive(status.pid)) {
    const response = { ok: false, reason: "no_daemon" };
    if (json) process.stdout.write(`${JSON.stringify(response)}\n`);
    else process.stderr.write("DENCH_FS_PULL=no_daemon\n");
    process.exitCode = 1;
    return;
  }
  await mkdir(PULL_REQUEST_DIR, { recursive: true });
  const id = `${Date.now()}-${Math.random().toString(36).slice(2)}`;
  const requestedAt = Date.now();
  const budget = Number.isFinite(timeoutMs)
    ? Math.max(1, Math.min(60_000, timeoutMs))
    : 10_000;
  const request: PullRequest = {
    id,
    paths,
    requestedAt,
    deadlineAt: requestedAt + budget,
  };
  const requestPath = `${PULL_REQUEST_DIR}/request-${id}.json`;
  const temporaryRequestPath = `${requestPath}.${process.pid}.tmp`;
  try {
    await writeFile(temporaryRequestPath, JSON.stringify(request), "utf-8");
    await rename(temporaryRequestPath, requestPath);
  } catch (error) {
    await unlink(temporaryRequestPath).catch(() => undefined);
    throw error;
  }
  process.kill(status.pid, "SIGUSR2");
  if (!wait) {
    const response = { ok: true, id, queued: paths };
    if (json) process.stdout.write(`${JSON.stringify(response)}\n`);
    else process.stdout.write(`DENCH_FS_PULL=queued id=${id}\n`);
    return;
  }

  const responsePath = `${PULL_REQUEST_DIR}/response-${id}.json`;
  const deadline = requestedAt + budget;
  while (Date.now() < deadline) {
    try {
      const response = JSON.parse(
        await readFile(responsePath, "utf-8"),
      ) as PullResponse;
      await unlink(responsePath).catch(() => undefined);
      if (json) process.stdout.write(`${JSON.stringify(response)}\n`);
      else if (response.ok) {
        process.stdout.write(
          `DENCH_FS_PULL=ok pulled=${response.pulled.length}\n`,
        );
      } else {
        process.stderr.write(
          `DENCH_FS_PULL=failed errors=${response.errors.length}\n`,
        );
      }
      if (!response.ok) process.exitCode = 3;
      return;
    } catch {
      await new Promise((resolve) => setTimeout(resolve, 100));
    }
  }
  const timeoutResponse = { ok: false, id, paths, reason: "timeout" };
  if (json) process.stdout.write(`${JSON.stringify(timeoutResponse)}\n`);
  else process.stderr.write(`DENCH_FS_PULL=timeout id=${id}\n`);
  process.exitCode = 2;
}

async function main(): Promise<void> {
  const argv = process.argv.slice(2);
  const subcommand = argv[0];
  // Subcommand mode (one-shot, used by `dench fs stream-*` + sync).
  if (subcommand === "stream-open") {
    await runStreamOpen(argv.slice(1));
    return;
  }
  if (subcommand === "stream-append") {
    await runStreamAppend(argv.slice(1));
    return;
  }
  if (subcommand === "stream-close") {
    await runStreamClose(argv.slice(1));
    return;
  }
  if (subcommand === "sync") {
    await runInitialSync(argv.slice(1));
    return;
  }
  if (subcommand === "flush") {
    await runFlush(argv.slice(1));
    return;
  }
  if (subcommand === "pull") {
    await runPull(argv.slice(1));
    return;
  }
  if (subcommand === "status") {
    await runStatus(argv.slice(1));
    return;
  }
  // Daemon mode (long-running, started by sandbox-bootstrap).
  await runDaemon(parseArgs(argv));
}

// ── Status subcommand ─────────────────────────────────────────────────────
//
// `dench-fs-daemon status [--json] [--workspace /workspace]` prints a
// cached health readout: process up?, worker progress, last successful
// reconcile, and active circuit breakers. It never scans the mount or
// calls Convex, so repeated health probes cannot create stuck workers.

const FS_DAEMON_LOG_PATH = "/var/log/dench/fs-daemon.log";

type DriftItem =
  | { kind: "only_disk"; path: string; size: number; mtimeMs: number }
  | { kind: "only_convex"; path: string }
  | {
      kind: "hash_mismatch";
      path: string;
      diskSize: number;
      diskHashShort: string;
      convexHashShort: string;
    };

type StatusReport = {
  daemon: {
    healthy: boolean;
    workspaceMatches: boolean;
    pidAlive: boolean;
    statusFilePath: string;
    statusFileExists: boolean;
    statusFileAgeMs: number | null;
    state: DaemonStatus | null;
  };
  filesystem: {
    workspace: string;
    fsType: string;
    isFuse: boolean;
    source: string;
  };
  counts: {
    onDisk: number | null;
    inConvex: number | null;
    onlyOnDisk: number | null;
    onlyInConvex: number | null;
    hashMismatches: number | null;
  };
  drift: DriftItem[];
  driftStatus: "not_checked";
  recentLogTail: string[];
};

async function runStatus(argv: string[]): Promise<void> {
  const json = argv.includes("--json");
  const workspaceIndex = argv.indexOf("--workspace");
  const workspace =
    workspaceIndex === -1
      ? process.env.DENCH_VOLUME_PATH?.trim() || "/workspace"
      : (argv[workspaceIndex + 1] ?? "/workspace");
  // Legacy --no-hash / --drift-limit flags are accepted but status no
  // longer scans files. Unknown counts stay null, never a healthy zero.
  const report = await collectStatus({ workspace });
  if (!report.daemon.healthy) process.exitCode = 1;
  if (json) {
    process.stdout.write(`${JSON.stringify(report, null, 2)}\n`);
    return;
  }
  printStatusHuman(report);
}

async function collectStatus(args: {
  workspace: string;
}): Promise<StatusReport> {
  const [statusFile, mount, logTail] = await Promise.all([
    readStatusFile(),
    detectWorkspaceFsType(args.workspace),
    readLogTail(20),
  ]);
  // A status command must never spawn its own mount worker. Repeated probes
  // against D-state FUSE would otherwise accumulate orphaned children outside
  // the daemon supervisor's single-worker limit. Report cached health honestly.
  const pidAlive = statusFile.state ? isPidAlive(statusFile.state.pid) : false;
  const workspaceMatches = cachedWorkspaceMatches(
    statusFile.state?.workspace,
    args.workspace,
  );
  return {
    daemon: {
      healthy: workspaceMatches && isDaemonHealthy(statusFile.state, pidAlive),
      workspaceMatches,
      pidAlive,
      statusFilePath: DAEMON_STATUS_PATH,
      statusFileExists: statusFile.exists,
      statusFileAgeMs: statusFile.state
        ? Date.now() - statusFile.state.lastUpdatedAt
        : null,
      state: statusFile.state,
    },
    filesystem: {
      workspace: args.workspace,
      fsType: mount.fsType,
      isFuse: mount.isFuse,
      source: mount.source,
    },
    counts: {
      onDisk: workspaceMatches
        ? (statusFile.state?.lastReconcile?.observed ?? null)
        : null,
      inConvex: null,
      onlyOnDisk: null,
      onlyInConvex: null,
      hashMismatches: null,
    },
    drift: [],
    driftStatus: "not_checked",
    recentLogTail: logTail,
  };
}

async function readStatusFile(): Promise<{
  exists: boolean;
  state: DaemonStatus | null;
}> {
  try {
    const text = await readFile(DAEMON_STATUS_PATH, "utf-8");
    return { exists: true, state: JSON.parse(text) as DaemonStatus };
  } catch {
    return { exists: false, state: null };
  }
}

function isPidAlive(pid: number): boolean {
  if (!Number.isInteger(pid) || pid <= 0) return false;
  try {
    process.kill(pid, 0);
    return true;
  } catch {
    return false;
  }
}

async function readLogTail(lines: number): Promise<string[]> {
  try {
    const text = await readFile(FS_DAEMON_LOG_PATH, "utf-8");
    const all = text.split("\n").filter((line) => line.length > 0);
    return all.slice(-lines);
  } catch {
    return [];
  }
}

function printStatusHuman(report: StatusReport): void {
  const lines: string[] = [];
  lines.push("dench-fs-daemon status");
  lines.push(
    "Cached daemon health only; live drift/hashes NOT CHECKED (null = unknown).",
  );
  lines.push("─".repeat(40));
  if (report.daemon.state) {
    const ageSec =
      report.daemon.statusFileAgeMs !== null
        ? Math.round(report.daemon.statusFileAgeMs / 1000)
        : null;
    lines.push(
      `daemon:          ${report.daemon.pidAlive ? "ALIVE" : "DEAD"} (pid ${report.daemon.state.pid})`,
    );
    lines.push(
      `sync health:     ${report.daemon.healthy ? "HEALTHY" : "DEGRADED"}`,
    );
    lines.push(
      `mount worker:    ${report.daemon.state.mountWorker?.state ?? "unknown"}`,
    );
    lines.push(
      `status file:     ${ageSec !== null ? `${ageSec}s old` : "missing"}`,
    );
    lines.push(`workspace:       ${report.filesystem.workspace}`);
    if (!report.daemon.workspaceMatches)
      lines.push(
        `cached workspace: ${report.daemon.state.workspace} (does not match requested root)`,
      );
    lines.push(
      `filesystem:      ${report.filesystem.fsType}${report.filesystem.isFuse ? " (FUSE — polling mode)" : ""}`,
    );
    if (report.daemon.state.lastReconcile) {
      const r = report.daemon.state.lastReconcile;
      const reconcileAge = Math.round((Date.now() - r.finishedAt) / 1000);
      lines.push(
        `last reconcile:  ${reconcileAge}s ago — observed=${r.observed} synced=${r.scheduledForSync} deleted=${r.deleted} (${r.durationMs}ms)`,
      );
    } else {
      lines.push("last reconcile:  never");
    }
    if (report.daemon.state.activeBreakers.length > 0) {
      lines.push(
        `active breakers: ${report.daemon.state.activeBreakers.length}`,
      );
      for (const breaker of report.daemon.state.activeBreakers) {
        const remain = Math.max(
          0,
          Math.round((breaker.resumesAt - Date.now()) / 1000),
        );
        lines.push(`  • ${breaker.path} (resumes in ${remain}s)`);
      }
    } else {
      lines.push("active breakers: none");
    }
  } else {
    lines.push(
      `daemon:          NOT RUNNING (status file missing at ${report.daemon.statusFilePath})`,
    );
    lines.push(`workspace:       ${report.filesystem.workspace}`);
    lines.push(
      `filesystem:      ${report.filesystem.fsType}${report.filesystem.isFuse ? " (FUSE — would use polling)" : ""}`,
    );
  }
  lines.push("");
  lines.push("file counts");
  lines.push("─".repeat(40));
  lines.push(`on disk:         ${report.counts.onDisk}`);
  lines.push(`in cloud:        ${report.counts.inConvex}`);
  lines.push(`only on disk:    ${report.counts.onlyOnDisk}`);
  lines.push(`only in cloud:   ${report.counts.onlyInConvex}`);
  lines.push(`hash mismatches: ${report.counts.hashMismatches}`);

  if (report.drift.length > 0) {
    lines.push("");
    lines.push(`drift (first ${report.drift.length})`);
    lines.push("─".repeat(40));
    for (const item of report.drift) {
      if (item.kind === "only_disk") {
        lines.push(`  + ${item.path}  ${item.size}B (only on disk)`);
      } else if (item.kind === "only_convex") {
        lines.push(`  - ${item.path}  (only in cloud)`);
      } else {
        lines.push(
          `  ! ${item.path}  disk=${item.diskHashShort}… cloud=${item.convexHashShort}…`,
        );
      }
    }
  }

  if (report.recentLogTail.length > 0) {
    lines.push("");
    lines.push("recent log tail");
    lines.push("─".repeat(40));
    for (const line of report.recentLogTail) lines.push(line);
  }

  process.stdout.write(`${lines.join("\n")}\n`);
}

/**
 * One-shot recursive sync. Walks the workspace + uploads every file to
 * Convex Storage / commitFileUpload. Used by the migration script
 * (`dench fs sync --initial`) when seeding a fresh Daytona volume from
 * an EC2 tarball.
 */
async function runInitialSync(argv: string[]): Promise<void> {
  const initial = argv.includes("--initial");
  const orgIdIndex = argv.indexOf("--org-id");
  const workspaceIndex = argv.indexOf("--workspace");
  const orgId =
    orgIdIndex === -1
      ? (process.env.DENCH_ORG_ID?.trim() ?? undefined)
      : argv[orgIdIndex + 1];
  const workspace =
    workspaceIndex === -1
      ? "/workspace"
      : (argv[workspaceIndex + 1] ?? "/workspace");
  if (!initial) {
    throw new Error(
      "dench fs sync --initial [--org-id <orgId>] [--workspace /workspace]",
    );
  }
  const convexUrl = process.env.CONVEX_URL?.trim();
  const apiKey = process.env.DENCH_API_KEY?.trim();
  if (!convexUrl || !apiKey) {
    throw new Error("dench fs sync requires CONVEX_URL + DENCH_API_KEY in env");
  }
  const client = new DenchFileClient(convexUrl, apiKey);
  const resolvedOrgId = orgId ?? (await client.resolveOrganizationId());
  process.stdout.write(
    `[dench-fs-daemon] initial sync org=${resolvedOrgId} workspace=${workspace}\n`,
  );
  const io = workspaceIo(workspace);

  let count = 0;
  let bytes = 0;
  const skip = 0;
  const files = await io.scan();
  for (const entry of files) {
    const abs = entry.absPath;
    const rel = relative(workspace, abs);
    const bytesBuf = await io.readFile(abs);
    const hash = sha256(bytesBuf);
    const posixPath = `/${rel.replace(/\\/g, "/")}`;
    const uploadUrl = await client.generateUploadUrl(posixPath);
    const storageId = await client.putBytes(uploadUrl, bytesBuf);
    await client.commitFileUpload({
      path: posixPath,
      storageId,
      contentHash: hash,
      size: bytesBuf.byteLength,
      // Same binary-aware inlining as scheduleSync — a fresh-volume
      // initial sync that includes screenshots / PDFs / .zip would
      // otherwise plant U+FFFD-corrupted text rows for every blob.
      text:
        bytesBuf.byteLength <= MAX_INLINE_TEXT_BYTES
          ? decodeInlineTextOrUndefined(bytesBuf)
          : undefined,
      lastModifiedBy: "daemon",
    });
    count += 1;
    bytes += bytesBuf.byteLength;
    if (count % 50 === 0) {
      process.stdout.write(`[dench-fs-daemon] synced ${count} files\n`);
    }
  }
  process.stdout.write(
    `[dench-fs-daemon] initial sync complete: ${count} files, ${bytes}B (skipped ${skip})\n`,
  );
}

// Always run main(). Earlier versions guarded this on an `isDirectEntrypoint`
// heuristic (`process.argv[1].endsWith("/fs-daemon")` etc.), but bun's global
// install symlinks the launcher as `/root/.bun/bin/dench-fs-daemon` — that
// path ends with `-fs-daemon`, not `/fs-daemon`, so the guard returned false
// and the script exited 0 silently. Nothing imports this file as a module any
// more (the test suite imports `fs-daemon-mount.ts` for the FUSE helpers), so
// the guard is unnecessary and matches the unconditional `main()` call in
// `cli/dench.ts`.
main()
  .finally(() => {
    if (!daemonMode) activeWorkspaceIo?.close();
  })
  .catch((error) => {
    console.error(
      `[dench-fs-daemon] fatal: ${
        error instanceof Error ? (error.stack ?? error.message) : String(error)
      }`,
    );
    // Always exit with a non-zero code so the supervisor (in
    // src/lib/daytona/sandbox-bootstrap.ts) can decide whether to restart.
    // Previously we parked the process on a noop interval, which made
    // `pgrep dench-fs-daemon` report "alive" for a daemon that had silently
    // given up syncing.
    activeWorkspaceIo?.close();
    process.exit(1);
  });