import { randomUUID as nodeRandomUUID } from "node:crypto";
import {
  chmod,
  mkdir,
  mkdtemp,
  readdir,
  rename,
  rm,
  stat,
  unlink,
} from "node:fs/promises";
import { dirname, isAbsolute, join, normalize, resolve } from "node:path";
import { setTimeout as delay } from "node:timers/promises";

import {
  isNodeError,
  readWorkflowJson,
  sha256,
  syncWorkflowDirectory,
  writeWorkflowJson,
} from "./atomic-storage.js";

const OWNER_KEYS = [
  "schema",
  "kind",
  "stage",
  "generation",
  "workspaceId",
  "resourceId",
  "runId",
  "workerId",
  "attemptId",
  "workerAttemptGeneration",
  "supervisorGeneration",
  "sessionPath",
  "pid",
  "processStartIdentity",
  "bootIdentity",
  "executable",
  "hostname",
  "createdAt",
] as const;

const GUARD_KEYS = [
  "schema",
  "generation",
  "pid",
  "processStartIdentity",
  "bootIdentity",
  "executable",
  "hostname",
  "createdAt",
] as const;

export interface ProcessIdentity {
  readonly pid: number;
  readonly processStartIdentity: string;
  readonly bootIdentity: string | null;
  readonly executable: string;
  readonly hostname: string;
}

export type ProcessInspection =
  | { readonly kind: "live"; readonly identity: ProcessIdentity }
  | { readonly kind: "absent" }
  | { readonly kind: "unknown"; readonly diagnostic: string };

export interface ProcessIdentityInspector {
  current(): Promise<ProcessIdentity>;
  inspect(pid: number): Promise<ProcessInspection>;
}

export interface WorkflowLeaseOwnerV1 {
  readonly schema: 1;
  readonly kind: "supervisor" | "worker-session";
  readonly stage: "active" | "reserved";
  readonly generation: string;
  readonly workspaceId: string;
  readonly resourceId: string;
  readonly runId: string | null;
  readonly workerId: string | null;
  readonly attemptId: string | null;
  readonly workerAttemptGeneration: string | null;
  readonly supervisorGeneration: string | null;
  readonly sessionPath: string | null;
  readonly pid: number;
  readonly processStartIdentity: string;
  readonly bootIdentity: string | null;
  readonly executable: string;
  readonly hostname: string;
  readonly createdAt: string;
}

interface LifecycleGuardV1 {
  readonly schema: 1;
  readonly generation: string;
  readonly pid: number;
  readonly processStartIdentity: string;
  readonly bootIdentity: string | null;
  readonly executable: string;
  readonly hostname: string;
  readonly createdAt: string;
}

export interface WorkflowLeaseManagerOptions {
  readonly namespaceDirectory: string;
  readonly workspaceId: string;
  readonly inspector: ProcessIdentityInspector;
  readonly randomUUID?: () => string;
  readonly now?: () => Date;
  readonly guardWaitMs?: number;
}

export interface AcquireWorkerSessionInput {
  readonly sessionPath: string;
  readonly runId: string;
  readonly workerId: string;
  readonly attemptId: string;
  readonly workerAttemptGeneration: string;
  readonly supervisorGeneration: string;
  readonly processIdentity: ProcessIdentity;
}

export type WorkflowLeaseRefusalReason =
  | "remote-owner"
  | "unverifiable-owner"
  | "corrupt-owner"
  | "lifecycle-busy"
  | "lifecycle-unverifiable";

export type WorkflowLeaseAcquireResult =
  | {
      readonly kind: "acquired";
      readonly owner: WorkflowLeaseOwnerV1;
      readonly reclaimed?: {
        readonly owner: WorkflowLeaseOwnerV1;
        readonly quarantinePath: string;
      };
    }
  | { readonly kind: "held"; readonly owner: WorkflowLeaseOwnerV1 }
  | {
      readonly kind: "refused";
      readonly reason: WorkflowLeaseRefusalReason;
      readonly diagnostic: string;
    };

export type WorkflowLeaseReleaseResult =
  | { readonly kind: "released" }
  | { readonly kind: "not-owner"; readonly owner?: WorkflowLeaseOwnerV1 }
  | { readonly kind: "refused"; readonly diagnostic: string };

export type WorkflowLeaseProjection =
  | { readonly kind: "empty" }
  | { readonly kind: "owned"; readonly owner: WorkflowLeaseOwnerV1 }
  | { readonly kind: "corrupt"; readonly diagnostic: string };

export interface WorkerSessionLeaseDiagnostic {
  readonly resourceId: string;
  readonly path: string;
  readonly projection: WorkflowLeaseProjection;
}

export class WorkflowLeaseFenceError extends Error {
  constructor(message: string) {
    super(message);
    this.name = "WorkflowLeaseFenceError";
  }
}

interface LeaseResource {
  readonly kind: "supervisor" | "worker-session";
  readonly resourceId: string;
  readonly lockPath: string;
  readonly guardPath: string;
  readonly quarantineStem: string;
}

type OwnerState =
  | { readonly kind: "live" }
  | { readonly kind: "stale"; readonly reason: string }
  | { readonly kind: "remote" }
  | { readonly kind: "unknown"; readonly diagnostic: string };

type GuardResult<T> =
  | { readonly ok: true; readonly value: T }
  | { readonly ok: false; readonly reason: WorkflowLeaseRefusalReason; readonly diagnostic: string };

export class WorkflowLeaseManager {
  readonly #namespaceDirectory: string;
  readonly #workspaceId: string;
  readonly #inspector: ProcessIdentityInspector;
  readonly #randomUUID: () => string;
  readonly #now: () => Date;
  readonly #guardWaitMs: number;

  constructor(options: WorkflowLeaseManagerOptions) {
    if (!isAbsolute(options.namespaceDirectory)) {
      throw new TypeError("Workflow namespace must be absolute");
    }
    this.#namespaceDirectory = normalize(options.namespaceDirectory);
    this.#workspaceId = requireNonEmpty(options.workspaceId, "workspace ID");
    this.#inspector = options.inspector;
    this.#randomUUID = options.randomUUID ?? nodeRandomUUID;
    this.#now = options.now ?? (() => new Date());
    this.#guardWaitMs = options.guardWaitMs ?? 1_000;
  }

  async acquireSupervisor(): Promise<WorkflowLeaseAcquireResult> {
    const current = await this.#inspector.current();
    validateProcessIdentity(current, "current process identity");
    return this.#acquire(this.#supervisorResource(), ownerFromIdentity({
      kind: "supervisor",
      stage: "active",
      generation: this.#randomUUID(),
      workspaceId: this.#workspaceId,
      resourceId: this.#workspaceId,
      identity: current,
      now: this.#now(),
    }));
  }

  async releaseSupervisor(generation: string): Promise<WorkflowLeaseReleaseResult> {
    requireNonEmpty(generation, "supervisor generation");
    const current = await this.#inspector.current();
    return this.#release(this.#supervisorResource(), generation, (owner) =>
      owner.kind === "supervisor" && ownerIsProcess(owner, current));
  }

  async withCurrentSupervisor<T>(
    generation: string,
    operation: () => Promise<T>,
  ): Promise<T> {
    requireNonEmpty(generation, "supervisor generation");
    const resource = this.#supervisorResource();
    const guarded = await this.#withGuard(resource, async (current) => {
      const projection = await this.#inspect(resource);
      if (projection.kind !== "owned"
        || projection.owner.kind !== "supervisor"
        || projection.owner.generation !== generation
        || projection.owner.workspaceId !== this.#workspaceId
        || !ownerIsProcess(projection.owner, current)) {
        throw new WorkflowLeaseFenceError("Workflow supervisor generation is not current");
      }
      return operation();
    });
    if (!guarded.ok) throw new WorkflowLeaseFenceError(guarded.diagnostic);
    return guarded.value;
  }

  async inspectSupervisor(): Promise<WorkflowLeaseProjection> {
    await this.#initialize();
    return this.#inspect(this.#supervisorResource());
  }

  async acquireWorkerSession(
    input: AcquireWorkerSessionInput,
  ): Promise<WorkflowLeaseAcquireResult> {
    const sessionPath = canonicalSessionPath(input.sessionPath);
    validateProcessIdentity(input.processIdentity, "worker attempt process identity");
    requireNonEmpty(input.runId, "run ID");
    requireNonEmpty(input.workerId, "worker ID");
    requireNonEmpty(input.attemptId, "attempt ID");
    requireNonEmpty(input.workerAttemptGeneration, "worker attempt generation");
    return this.withCurrentSupervisor(input.supervisorGeneration, async () => {
      const resource = this.#workerResource(sessionPath);
      return this.#acquire(resource, ownerFromIdentity({
        kind: "worker-session",
        stage: "active",
        generation: this.#randomUUID(),
        workspaceId: this.#workspaceId,
        resourceId: resource.resourceId,
        runId: input.runId,
        workerId: input.workerId,
        attemptId: input.attemptId,
        workerAttemptGeneration: input.workerAttemptGeneration,
        supervisorGeneration: input.supervisorGeneration,
        sessionPath,
        identity: input.processIdentity,
        now: this.#now(),
      }));
    });
  }

  async reserveWorkerSession(
    input: Omit<AcquireWorkerSessionInput, "processIdentity">,
  ): Promise<WorkflowLeaseAcquireResult> {
    const sessionPath = canonicalSessionPath(input.sessionPath);
    requireNonEmpty(input.runId, "run ID");
    requireNonEmpty(input.workerId, "worker ID");
    requireNonEmpty(input.attemptId, "attempt ID");
    requireNonEmpty(input.workerAttemptGeneration, "worker attempt generation");
    return this.withCurrentSupervisor(input.supervisorGeneration, async () => {
      const current = await this.#inspector.current();
      const resource = this.#workerResource(sessionPath);
      return this.#acquire(resource, ownerFromIdentity({
        kind: "worker-session",
        stage: "reserved",
        generation: this.#randomUUID(),
        workspaceId: this.#workspaceId,
        resourceId: resource.resourceId,
        runId: input.runId,
        workerId: input.workerId,
        attemptId: input.attemptId,
        workerAttemptGeneration: input.workerAttemptGeneration,
        supervisorGeneration: input.supervisorGeneration,
        sessionPath,
        identity: current,
        now: this.#now(),
      }));
    });
  }

  async activateWorkerSession(input: {
    readonly sessionPath: string;
    readonly generation: string;
    readonly supervisorGeneration: string;
    readonly workerAttemptGeneration: string;
    readonly processIdentity: ProcessIdentity;
  }): Promise<WorkflowLeaseOwnerV1> {
    const sessionPath = canonicalSessionPath(input.sessionPath);
    validateProcessIdentity(input.processIdentity, "worker attempt process identity");
    return this.withCurrentSupervisor(input.supervisorGeneration, async () => {
      const resource = this.#workerResource(sessionPath);
      const guarded = await this.#withGuard(resource, async () => {
        const projection = await this.#inspect(resource);
        if (projection.kind !== "owned"
          || projection.owner.kind !== "worker-session"
          || projection.owner.stage !== "reserved"
          || projection.owner.generation !== input.generation
          || projection.owner.supervisorGeneration !== input.supervisorGeneration
          || projection.owner.workerAttemptGeneration !== input.workerAttemptGeneration
          || projection.owner.sessionPath !== sessionPath) {
          throw new WorkflowLeaseFenceError("Worker-session reservation is not current");
        }
        const owner: WorkflowLeaseOwnerV1 = {
          ...projection.owner,
          stage: "active",
          pid: input.processIdentity.pid,
          processStartIdentity: input.processIdentity.processStartIdentity,
          bootIdentity: input.processIdentity.bootIdentity,
          executable: input.processIdentity.executable,
          hostname: input.processIdentity.hostname,
        };
        await writeWorkflowJson(join(resource.lockPath, "owner.json"), owner);
        return owner;
      });
      if (!guarded.ok) throw new WorkflowLeaseFenceError(guarded.diagnostic);
      return guarded.value;
    });
  }

  async releaseWorkerSession(input: {
    readonly sessionPath: string;
    readonly generation: string;
    readonly supervisorGeneration: string;
  }): Promise<WorkflowLeaseReleaseResult> {
    const sessionPath = canonicalSessionPath(input.sessionPath);
    return this.withCurrentSupervisor(input.supervisorGeneration, () =>
      this.#release(this.#workerResource(sessionPath), input.generation, (owner) =>
        owner.kind === "worker-session"
          && owner.sessionPath === sessionPath
          && owner.supervisorGeneration === input.supervisorGeneration));
  }

  async inspectWorkerSession(sessionPath: string): Promise<WorkflowLeaseProjection> {
    await this.#initialize();
    return this.#inspect(this.#workerResource(canonicalSessionPath(sessionPath)));
  }

  async diagnoseWorkerSessionLeases(): Promise<readonly WorkerSessionLeaseDiagnostic[]> {
    const parent = join(this.#namespaceDirectory, "worker-session-leases");
    let entries: string[];
    try {
      entries = await readdir(parent);
    } catch (error) {
      if (isNodeError(error) && error.code === "ENOENT") return [];
      throw error;
    }
    const records: WorkerSessionLeaseDiagnostic[] = [];
    for (const entry of entries.sort()) {
      if (!entry.endsWith(".lock") || entry.startsWith(".")) continue;
      const resourceId = entry.slice(0, -".lock".length);
      const path = join(parent, entry);
      records.push({
        resourceId,
        path,
        projection: await this.#inspect({
          kind: "worker-session",
          resourceId,
          lockPath: path,
          guardPath: join(parent, `${resourceId}.lifecycle.guard.json`),
          quarantineStem: `worker-session.${resourceId}`,
        }),
      });
    }
    return records;
  }

  async #acquire(
    resource: LeaseResource,
    candidate: WorkflowLeaseOwnerV1,
  ): Promise<WorkflowLeaseAcquireResult> {
    await this.#initialize();
    const guarded = await this.#withGuard(resource, async (current) => {
      const projection = await this.#inspect(resource);
      if (projection.kind === "corrupt") {
        return {
          kind: "refused" as const,
          reason: "corrupt-owner" as const,
          diagnostic: projection.diagnostic,
        };
      }
      let reclaimed: { readonly owner: WorkflowLeaseOwnerV1; readonly quarantinePath: string } | undefined;
      if (projection.kind === "owned") {
        const state = await classifyProcessRecord(projection.owner, current, this.#inspector);
        if (state.kind === "live") return { kind: "held" as const, owner: projection.owner };
        if (state.kind === "remote") {
          return {
            kind: "refused" as const,
            reason: "remote-owner" as const,
            diagnostic: `Workflow lease owner is on remote host ${projection.owner.hostname}`,
          };
        }
        if (state.kind === "unknown") {
          return {
            kind: "refused" as const,
            reason: "unverifiable-owner" as const,
            diagnostic: state.diagnostic,
          };
        }
        const quarantinePath = join(
          this.#namespaceDirectory,
          "quarantine",
          `${resource.quarantineStem}.${safePathPart(projection.owner.generation)}.${this.#now().getTime()}.${safePathPart(this.#randomUUID())}`,
        );
        await rename(resource.lockPath, quarantinePath);
        await syncWorkflowDirectory(dirname(resource.lockPath));
        await syncWorkflowDirectory(this.#namespaceDirectory);
        reclaimed = { owner: projection.owner, quarantinePath };
      }
      await this.#createLease(resource, candidate);
      return {
        kind: "acquired" as const,
        owner: candidate,
        ...(reclaimed ? { reclaimed } : {}),
      };
    });
    return guarded.ok
      ? guarded.value
      : { kind: "refused", reason: guarded.reason, diagnostic: guarded.diagnostic };
  }

  async #release(
    resource: LeaseResource,
    generation: string,
    ownerMatchesAuthority: (owner: WorkflowLeaseOwnerV1) => boolean,
  ): Promise<WorkflowLeaseReleaseResult> {
    requireNonEmpty(generation, "lease generation");
    await this.#initialize();
    const guarded = await this.#withGuard(resource, async () => {
      const projection = await this.#inspect(resource);
      if (projection.kind === "empty") return { kind: "not-owner" as const };
      if (projection.kind === "corrupt") {
        return { kind: "refused" as const, diagnostic: projection.diagnostic };
      }
      if (projection.owner.generation !== generation || !ownerMatchesAuthority(projection.owner)) {
        return { kind: "not-owner" as const, owner: projection.owner };
      }
      const releasedPath = join(
        this.#namespaceDirectory,
        `.${resource.quarantineStem}.released-${safePathPart(generation)}-${safePathPart(this.#randomUUID())}`,
      );
      await rename(resource.lockPath, releasedPath);
      await syncWorkflowDirectory(dirname(resource.lockPath));
      await syncWorkflowDirectory(this.#namespaceDirectory);
      await rm(releasedPath, { recursive: true });
      await syncWorkflowDirectory(this.#namespaceDirectory);
      return { kind: "released" as const };
    });
    return guarded.ok
      ? guarded.value
      : { kind: "refused", diagnostic: guarded.diagnostic };
  }

  async #createLease(resource: LeaseResource, owner: WorkflowLeaseOwnerV1): Promise<void> {
    const parent = dirname(resource.lockPath);
    await mkdir(parent, { recursive: true, mode: 0o700 });
    const prepared = await mkdtemp(join(parent, ".lease.prepare-"));
    await chmod(prepared, 0o700);
    let installed = false;
    try {
      await writeWorkflowJson(join(prepared, "owner.json"), owner, { createOnly: true });
      await syncWorkflowDirectory(prepared);
      await rename(prepared, resource.lockPath);
      installed = true;
      await syncWorkflowDirectory(parent);
    } finally {
      if (!installed) await rm(prepared, { recursive: true, force: true });
    }
  }

  async #inspect(resource: LeaseResource): Promise<WorkflowLeaseProjection> {
    let raw: unknown;
    try {
      raw = await readWorkflowJson(join(resource.lockPath, "owner.json"));
    } catch (error) {
      return { kind: "corrupt", diagnostic: `Workflow lease JSON is invalid: ${toError(error).message}` };
    }
    if (raw === undefined) {
      try {
        await stat(resource.lockPath);
      } catch (error) {
        if (isNodeError(error) && error.code === "ENOENT") return { kind: "empty" };
        throw error;
      }
      return { kind: "corrupt", diagnostic: "Workflow lease exists without owner.json" };
    }
    try {
      const owner = parseWorkflowLeaseOwnerV1(raw);
      if (owner.workspaceId !== this.#workspaceId
        || owner.kind !== resource.kind
        || owner.resourceId !== resource.resourceId) {
        throw new Error("Workflow lease resource identity does not match its namespace");
      }
      if (owner.kind === "worker-session"
        && owner.sessionPath !== null
        && this.#workerResource(owner.sessionPath).resourceId !== resource.resourceId) {
        throw new Error("Worker-session lease path does not match its resource hash");
      }
      return { kind: "owned", owner };
    } catch (error) {
      return { kind: "corrupt", diagnostic: toError(error).message };
    }
  }

  async #withGuard<T>(
    resource: LeaseResource,
    operation: (current: ProcessIdentity) => Promise<T>,
  ): Promise<GuardResult<T>> {
    await this.#initialize();
    await mkdir(join(resource.guardPath, ".."), { recursive: true, mode: 0o700 });
    const current = await this.#inspector.current();
    validateProcessIdentity(current, "current process identity");
    const deadline = Date.now() + this.#guardWaitMs;

    while (true) {
      const guard: LifecycleGuardV1 = {
        schema: 1,
        generation: this.#randomUUID(),
        pid: current.pid,
        processStartIdentity: current.processStartIdentity,
        bootIdentity: current.bootIdentity,
        executable: current.executable,
        hostname: current.hostname,
        createdAt: this.#now().toISOString(),
      };
      try {
        await writeWorkflowJson(resource.guardPath, guard, { createOnly: true });
      } catch (error) {
        if (!isNodeError(error) || error.code !== "EEXIST") throw error;
        const existing = await this.#readGuard(resource.guardPath);
        if (existing.kind === "corrupt") {
          return { ok: false, reason: "lifecycle-unverifiable", diagnostic: existing.diagnostic };
        }
        if (existing.kind === "empty") continue;
        const state = await classifyProcessRecord(existing.guard, current, this.#inspector);
        if (state.kind === "stale") {
          const quarantinePath = join(
            this.#namespaceDirectory,
            "quarantine",
            `guard.${resource.quarantineStem}.${safePathPart(existing.guard.generation)}.${safePathPart(this.#randomUUID())}.json`,
          );
          try {
            await rename(resource.guardPath, quarantinePath);
            await syncWorkflowDirectory(dirname(resource.guardPath));
            await syncWorkflowDirectory(this.#namespaceDirectory);
          } catch (renameError) {
            if (!isNodeError(renameError) || renameError.code !== "ENOENT") throw renameError;
          }
          continue;
        }
        if (state.kind === "remote" || state.kind === "unknown") {
          return {
            ok: false,
            reason: "lifecycle-unverifiable",
            diagnostic: state.kind === "remote"
              ? `Workflow lifecycle guard belongs to remote host ${existing.guard.hostname}`
              : state.diagnostic,
          };
        }
        if (Date.now() >= deadline) {
          return {
            ok: false,
            reason: "lifecycle-busy",
            diagnostic: "Workflow lease lifecycle operation is busy; no owner was displaced",
          };
        }
        await delay(5);
        continue;
      }

      let value: T | undefined;
      let operationError: unknown;
      try {
        value = await operation(current);
      } catch (error) {
        operationError = error;
      }
      let releaseError: unknown;
      try {
        await this.#releaseGuard(resource.guardPath, guard.generation);
      } catch (error) {
        releaseError = error;
      }
      if (operationError !== undefined && releaseError !== undefined) {
        throw new AggregateError(
          [operationError, releaseError],
          "Workflow lease operation and guard release both failed",
        );
      }
      if (operationError !== undefined) throw operationError;
      if (releaseError !== undefined) throw releaseError;
      return { ok: true, value: value as T };
    }
  }

  async #readGuard(path: string): Promise<
    | { readonly kind: "empty" }
    | { readonly kind: "owned"; readonly guard: LifecycleGuardV1 }
    | { readonly kind: "corrupt"; readonly diagnostic: string }
  > {
    let raw: unknown;
    try {
      raw = await readWorkflowJson(path);
    } catch (error) {
      return { kind: "corrupt", diagnostic: `Workflow guard JSON is invalid: ${toError(error).message}` };
    }
    if (raw === undefined) return { kind: "empty" };
    try {
      return { kind: "owned", guard: parseLifecycleGuardV1(raw) };
    } catch (error) {
      return { kind: "corrupt", diagnostic: toError(error).message };
    }
  }

  async #releaseGuard(path: string, generation: string): Promise<void> {
    const existing = await this.#readGuard(path);
    if (existing.kind !== "owned" || existing.guard.generation !== generation) {
      throw new WorkflowLeaseFenceError("Workflow lifecycle guard generation changed before release");
    }
    await unlink(path);
    await syncWorkflowDirectory(dirname(path));
  }

  async #initialize(): Promise<void> {
    await mkdir(this.#namespaceDirectory, { recursive: true, mode: 0o700 });
    await mkdir(join(this.#namespaceDirectory, "worker-session-leases"), { recursive: true, mode: 0o700 });
    await mkdir(join(this.#namespaceDirectory, "quarantine"), { recursive: true, mode: 0o700 });
  }

  #supervisorResource(): LeaseResource {
    return {
      kind: "supervisor",
      resourceId: this.#workspaceId,
      lockPath: join(this.#namespaceDirectory, "supervisor.lock"),
      guardPath: join(this.#namespaceDirectory, "supervisor.lifecycle.guard.json"),
      quarantineStem: "supervisor",
    };
  }

  #workerResource(sessionPath: string): LeaseResource {
    const resourceId = sha256(sessionPath).slice("sha256:".length);
    const parent = join(this.#namespaceDirectory, "worker-session-leases");
    return {
      kind: "worker-session",
      resourceId,
      lockPath: join(parent, `${resourceId}.lock`),
      guardPath: join(parent, `${resourceId}.lifecycle.guard.json`),
      quarantineStem: `worker-session.${resourceId}`,
    };
  }
}

export function parseWorkflowLeaseOwnerV1(value: unknown): WorkflowLeaseOwnerV1 {
  const object = strictRecord(value, OWNER_KEYS, "workflow lease owner");
  if (numeric(object, "schema") !== 1) throw new Error("workflow lease owner schema must be 1");
  const kind = enumString(object, "kind", ["supervisor", "worker-session"] as const);
  const owner: WorkflowLeaseOwnerV1 = {
    schema: 1,
    kind,
    stage: enumString(object, "stage", ["active", "reserved"] as const),
    generation: nonEmptyString(object, "generation"),
    workspaceId: nonEmptyString(object, "workspaceId"),
    resourceId: nonEmptyString(object, "resourceId"),
    runId: nullableString(object, "runId"),
    workerId: nullableString(object, "workerId"),
    attemptId: nullableString(object, "attemptId"),
    workerAttemptGeneration: nullableString(object, "workerAttemptGeneration"),
    supervisorGeneration: nullableString(object, "supervisorGeneration"),
    sessionPath: nullableAbsolutePath(object, "sessionPath"),
    pid: positiveInteger(object, "pid"),
    processStartIdentity: nonEmptyString(object, "processStartIdentity"),
    bootIdentity: nullableString(object, "bootIdentity"),
    executable: absolutePath(object, "executable"),
    hostname: nonEmptyString(object, "hostname"),
    createdAt: isoTimestamp(object, "createdAt"),
  };
  const workerFields = [
    owner.runId,
    owner.workerId,
    owner.attemptId,
    owner.workerAttemptGeneration,
    owner.supervisorGeneration,
    owner.sessionPath,
  ];
  if (kind === "supervisor" && workerFields.some((field) => field !== null)) {
    throw new Error("Supervisor lease cannot contain worker-session fields");
  }
  if (kind === "supervisor" && owner.stage !== "active") {
    throw new Error("Supervisor lease must be active");
  }
  if (kind === "worker-session" && workerFields.some((field) => field === null)) {
    throw new Error("Worker-session lease requires run, worker, attempt, generation, and session fields");
  }
  return owner;
}

function parseLifecycleGuardV1(value: unknown): LifecycleGuardV1 {
  const object = strictRecord(value, GUARD_KEYS, "workflow lifecycle guard");
  if (numeric(object, "schema") !== 1) throw new Error("workflow guard schema must be 1");
  return {
    schema: 1,
    generation: nonEmptyString(object, "generation"),
    pid: positiveInteger(object, "pid"),
    processStartIdentity: nonEmptyString(object, "processStartIdentity"),
    bootIdentity: nullableString(object, "bootIdentity"),
    executable: absolutePath(object, "executable"),
    hostname: nonEmptyString(object, "hostname"),
    createdAt: isoTimestamp(object, "createdAt"),
  };
}

function ownerFromIdentity(input: {
  readonly kind: "supervisor" | "worker-session";
  readonly stage: "active" | "reserved";
  readonly generation: string;
  readonly workspaceId: string;
  readonly resourceId: string;
  readonly runId?: string;
  readonly workerId?: string;
  readonly attemptId?: string;
  readonly workerAttemptGeneration?: string;
  readonly supervisorGeneration?: string;
  readonly sessionPath?: string;
  readonly identity: ProcessIdentity;
  readonly now: Date;
}): WorkflowLeaseOwnerV1 {
  return {
    schema: 1,
    kind: input.kind,
    stage: input.stage,
    generation: requireNonEmpty(input.generation, "lease generation"),
    workspaceId: input.workspaceId,
    resourceId: input.resourceId,
    runId: input.runId ?? null,
    workerId: input.workerId ?? null,
    attemptId: input.attemptId ?? null,
    workerAttemptGeneration: input.workerAttemptGeneration ?? null,
    supervisorGeneration: input.supervisorGeneration ?? null,
    sessionPath: input.sessionPath ?? null,
    pid: input.identity.pid,
    processStartIdentity: input.identity.processStartIdentity,
    bootIdentity: input.identity.bootIdentity,
    executable: input.identity.executable,
    hostname: input.identity.hostname,
    createdAt: input.now.toISOString(),
  };
}

async function classifyProcessRecord(
  record: Pick<WorkflowLeaseOwnerV1, "hostname" | "pid" | "processStartIdentity" | "bootIdentity" | "executable">,
  current: ProcessIdentity,
  inspector: ProcessIdentityInspector,
): Promise<OwnerState> {
  if (record.hostname !== current.hostname) return { kind: "remote" };
  const inspection = await inspector.inspect(record.pid);
  if (inspection.kind === "absent") return { kind: "stale", reason: "PID is absent" };
  if (inspection.kind === "unknown") return inspection;
  const observed = inspection.identity;
  if (record.processStartIdentity !== observed.processStartIdentity) {
    return { kind: "stale", reason: "PID start identity changed" };
  }
  if (normalize(record.executable) !== normalize(observed.executable)) {
    return { kind: "stale", reason: "PID executable changed" };
  }
  if (record.bootIdentity !== null
    && observed.bootIdentity !== null
    && record.bootIdentity !== observed.bootIdentity) {
    return { kind: "stale", reason: "host boot identity changed" };
  }
  return { kind: "live" };
}

function ownerIsProcess(owner: WorkflowLeaseOwnerV1, identity: ProcessIdentity): boolean {
  return owner.hostname === identity.hostname
    && owner.pid === identity.pid
    && owner.processStartIdentity === identity.processStartIdentity
    && normalize(owner.executable) === normalize(identity.executable)
    && (owner.bootIdentity === null
      || identity.bootIdentity === null
      || owner.bootIdentity === identity.bootIdentity);
}

function canonicalSessionPath(path: string): string {
  if (!isAbsolute(path)) throw new TypeError("Worker session path must be absolute");
  return normalize(resolve(path));
}

function validateProcessIdentity(identity: ProcessIdentity, name: string): void {
  if (!Number.isSafeInteger(identity.pid) || identity.pid <= 0) throw new Error(`${name} PID is invalid`);
  requireNonEmpty(identity.processStartIdentity, `${name} start identity`);
  if (!isAbsolute(identity.executable)) throw new Error(`${name} executable must be absolute`);
  requireNonEmpty(identity.hostname, `${name} hostname`);
}

function strictRecord(
  value: unknown,
  allowed: readonly string[],
  name: string,
): Record<string, unknown> {
  if (typeof value !== "object" || value === null || Array.isArray(value)) {
    throw new Error(`${name} must be an object`);
  }
  const object = value as Record<string, unknown>;
  const unknown = Object.keys(object).filter((key) => !allowed.includes(key));
  if (unknown.length > 0) throw new Error(`${name} contains unknown fields: ${unknown.join(", ")}`);
  return object;
}

function enumString<T extends string>(
  object: Record<string, unknown>,
  key: string,
  values: readonly T[],
): T {
  const value = nonEmptyString(object, key);
  if (!values.includes(value as T)) throw new Error(`${key} is invalid`);
  return value as T;
}

function nonEmptyString(object: Record<string, unknown>, key: string): string {
  const value = object[key];
  if (typeof value !== "string" || value.length === 0) throw new Error(`${key} must be a non-empty string`);
  return value;
}

function nullableString(object: Record<string, unknown>, key: string): string | null {
  const value = object[key];
  if (value === null) return null;
  if (typeof value !== "string" || value.length === 0) throw new Error(`${key} must be null or a non-empty string`);
  return value;
}

function nullableAbsolutePath(object: Record<string, unknown>, key: string): string | null {
  const value = nullableString(object, key);
  if (value === null) return null;
  if (!isAbsolute(value)) throw new Error(`${key} must be absolute`);
  return normalize(value);
}

function absolutePath(object: Record<string, unknown>, key: string): string {
  const value = nonEmptyString(object, key);
  if (!isAbsolute(value)) throw new Error(`${key} must be absolute`);
  return normalize(value);
}

function positiveInteger(object: Record<string, unknown>, key: string): number {
  const value = numeric(object, key);
  if (!Number.isSafeInteger(value) || value <= 0) throw new Error(`${key} must be a positive integer`);
  return value;
}

function numeric(object: Record<string, unknown>, key: string): number {
  const value = object[key];
  if (typeof value !== "number" || !Number.isFinite(value)) throw new Error(`${key} must be a number`);
  return value;
}

function isoTimestamp(object: Record<string, unknown>, key: string): string {
  const value = nonEmptyString(object, key);
  if (!Number.isFinite(Date.parse(value))) throw new Error(`${key} must be an ISO timestamp`);
  return value;
}

function requireNonEmpty(value: string, name: string): string {
  if (value.length === 0) throw new TypeError(`${name} is required`);
  return value;
}

function safePathPart(value: string): string {
  return value.replace(/[^A-Za-z0-9._-]/g, "_");
}

function toError(error: unknown): Error {
  return error instanceof Error ? error : new Error(String(error));
}