import { type Daytona } from '@daytonaio/sdk';
import type {
  PlayRunnerExecutionConfig,
  PlayRunnerResult,
} from '@shared_libs/play-runtime/protocol';
import { isIsolatedRuntimeSchedulerSchema } from '@shared_libs/play-runtime/runtime-scheduler-topology';
import { PLAY_RUNNER_TIMEOUT_SECONDS } from '@shared_libs/play-runtime/runtime-constants';
import {
  STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS,
  validatePlaySandboxRuntimeLimits,
} from '@shared_libs/play-runtime/sandbox-runtime-limits';
import type { PlayRunnerRuntimeLifecycleEvent } from '../types';
import { DAYTONA_PLAY_RUNNER_LABEL_SOURCE } from './daytona-labels';

const DAYTONA_CREATE_TIMEOUT_SECONDS = 10;
const DAYTONA_CREATE_RETRY_DELAYS_MS = [0, 500, 1_500] as const;
const DAYTONA_TIMED_OUT_CREATE_RECONCILE_DELAYS_MS = [
  0, 1_000, 3_000, 6_000, 12_000,
] as const;
// Explicit runner deadline + scheduler GC own the normal lifecycle. Daytona's
// inactivity stop is a wider crash backstop measured from sandbox creation, so
// setup time cannot consume the terminal-flush grace.
const DAYTONA_AUTO_ARCHIVE_INTERVAL_MINUTES = 15;
// Daytona's Deepline-managed default snapshot is the only admitted prebuilt.
// Resources are verified during acquisition, before customer code or billing.
export const DAYTONA_SANDBOX_CPU = STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS.cpu;
export const DAYTONA_SANDBOX_MEMORY_GIB =
  STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS.memoryGiB;
export const DAYTONA_SANDBOX_DISK_GIB =
  STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS.diskGiB;
export const DAYTONA_SANDBOX_GPU = 0;
const DAYTONA_NETWORK_ALLOW_LIST_ENV = 'DEEPLINE_DAYTONA_NETWORK_ALLOW_LIST';

export const DAYTONA_CANCELLED_ERROR = 'Daytona play runner cancelled';

export type DaytonaClient = Pick<Daytona, 'create'> &
  Partial<Pick<Daytona, 'get'>>;
export type DaytonaSandbox = Awaited<ReturnType<DaytonaClient['create']>>;
export type DaytonaExecutionContext = PlayRunnerExecutionConfig['context'];
export type DaytonaStageEmitter = (
  stage: string,
  extra?: Record<string, unknown>,
) => void;

type DaytonaCreateCallObserver = (
  event: PlayRunnerRuntimeLifecycleEvent,
) => Promise<void>;

export type AcquiredDaytonaSandbox = {
  sandbox: DaytonaSandbox;
  daytonaOrganizationId: string;
  billingStartedAt: number;
  billingEndedAt?: number;
  sandboxCapacityLeaseId?: string;
};

export type OneShotDaytonaSandboxLifecycle = {
  startedAt: number;
  acquiredSandboxPromise: Promise<AcquiredDaytonaSandbox>;
  createFreshSandbox: () => Promise<AcquiredDaytonaSandbox>;
  acquiredSandboxes: () => readonly AcquiredDaytonaSandbox[];
  settlePendingCreates: () => Promise<void>;
  dispose: () => Promise<void>;
};

export type DaytonaSandboxAcquisitionUnavailableReason =
  | 'daytona_total_cpu_limit_exceeded'
  | 'daytona_acquisition_rate_limited'
  | 'daytona_sandbox_start_timeout';

/**
 * A typed, pre-customer-code acquisition rejection. Only this error may move
 * managed placement to another provider.
 */
export class DaytonaSandboxAcquisitionUnavailableError extends Error {
  readonly reason: DaytonaSandboxAcquisitionUnavailableReason;

  constructor(
    reason: DaytonaSandboxAcquisitionUnavailableReason,
    message: string,
  ) {
    super(message);
    this.name = 'DaytonaSandboxAcquisitionUnavailableError';
    this.reason = reason;
  }
}

export function resolveDaytonaSandboxAcquisitionUnavailableReason(
  errors: readonly string[],
): DaytonaSandboxAcquisitionUnavailableReason | null {
  if (errors.length === 0) return null;
  const isCpuLimit = (error: string) =>
    /Total CPU limit exceeded\.\s*Maximum allowed:\s*\d+/i.test(error);
  const isRateLimit = (error: string) =>
    /ThrottlerException|Too Many Requests|(?:status code|HTTP)\s*429/i.test(
      error,
    );
  if (errors.every(isCpuLimit)) {
    return 'daytona_total_cpu_limit_exceeded';
  }
  if (errors.every((error) => isCpuLimit(error) || isRateLimit(error))) {
    return 'daytona_acquisition_rate_limited';
  }
  return null;
}

function isDaytonaSandboxStartTimeout(error: string): boolean {
  return /Failed to create and start sandbox within 10 seconds\. Operation timed out\./.test(
    error,
  );
}

function isDaytonaSandboxStateChangeConflict(error: unknown): boolean {
  const message = error instanceof Error ? error.message : String(error);
  return /Sandbox state change in progress/i.test(message);
}

type DaytonaCreateResult = {
  sandbox: DaytonaSandbox;
  attempt: number;
  attemptElapsedMs: number;
  sandboxCapacityLeaseId?: string;
};

function daytonaCreateErrorClass(
  error: unknown,
): 'timeout' | 'capacity' | 'rate_limit' | 'other' {
  const message = error instanceof Error ? error.message : String(error);
  if (/timed out|timeout/i.test(message)) return 'timeout';
  if (/total cpu limit|capacity/i.test(message)) return 'capacity';
  if (/too many requests|rate limit|429/i.test(message)) return 'rate_limit';
  return 'other';
}

async function rejectAcquiredSandbox(
  sandbox: DaytonaSandbox,
  reason: string,
  reservation?: { leaseId: string },
  releaseSandboxCapacity?: (leaseId: string) => Promise<void>,
): Promise<never> {
  try {
    await sandbox.delete(30);
  } catch (error) {
    const cleanupError = error instanceof Error ? error.message : String(error);
    throw new Error(
      `${reason} Defensive deletion of Daytona sandbox ${sandbox.id} also failed: ${cleanupError}`,
      { cause: error },
    );
  }
  if (reservation) {
    await releaseSandboxCapacity?.(reservation.leaseId);
  }
  throw new Error(reason);
}

function normalizeLabelValue(value: string | null | undefined): string | null {
  const trimmed = value?.trim();
  return trimmed ? trimmed.slice(0, 63) : null;
}

function daytonaOrgIdFromContext(
  context: DaytonaExecutionContext,
): string | null {
  return context.orgId?.trim() || null;
}

export function validateDaytonaExecutionContext(
  context: DaytonaExecutionContext,
): string {
  const orgId = daytonaOrgIdFromContext(context);
  if (!orgId) {
    throw new Error(
      'Missing required org context for Daytona execution. Refusing to start a shared sandbox without orgId.',
    );
  }
  return orgId;
}

/**
 * Outbound-network policy for a one-shot sandbox. The invariant this protects:
 * CUSTOMER code in PRODUCTION never runs with unrestricted egress — the
 * production worker must provision `DEEPLINE_DAYTONA_NETWORK_ALLOW_LIST`
 * (comma-separated CIDRs, per the Daytona `networkAllowList` contract) and
 * every prod sandbox is created with that allow-list, which Daytona treats as
 * "block all egress except these CIDRs".
 *
 * "Production" here is the RUN'S topology, not the build flag: a run whose
 * scheduler schema is isolated (per-PR preview CI, per-worktree dev — see
 * `isIsolatedRuntimeSchedulerSchema`) executes synthetic internal test plays
 * against dynamic per-PR hosts (Vercel preview app, per-PR Fly gateway) whose
 * IPs cannot be enumerated as stable CIDRs, so the allow-list requirement
 * cannot apply there. `NODE_ENV=production` alone is the wrong key — the Fly
 * worker image bakes it for BOTH prod and preview fleets (build optimization),
 * which made every preview sandbox create fail. Fail-closed: an absent schema
 * is the production topology, never "unknown".
 *
 * An explicitly provisioned allow-list is always honored (preview included),
 * so a future preview fleet with enumerable egress can opt in to full parity.
 */
/**
 * Keep only IPv4 CIDR entries from a comma-separated allow-list. Daytona's
 * `networkAllowList` is IPv4-only; an IPv6 entry fails every sandbox create.
 * Returns null when nothing valid remains so the production guard below still
 * fires loudly rather than sending an empty list.
 */
export function filterIpv4CidrAllowList(raw: string | null): string | null {
  if (!raw) return raw;
  const entries = raw
    .split(',')
    .map((entry) => entry.trim())
    .filter(Boolean);
  const ipv4 = entries.filter((entry) => !entry.includes(':'));
  const dropped = entries.filter((entry) => entry.includes(':'));
  if (dropped.length > 0) {
    console.warn(
      `[daytona] dropping ${dropped.length} non-IPv4 CIDR(s) from ` +
        `${DAYTONA_NETWORK_ALLOW_LIST_ENV} (Daytona networkAllowList is ` +
        `IPv4-only): ${dropped.join(', ')}`,
    );
  }
  return ipv4.length > 0 ? ipv4.join(',') : null;
}

export function resolveDaytonaSandboxNetworkPolicy(input: {
  runtimeSchedulerSchema: string | null | undefined;
  env?: NodeJS.ProcessEnv;
}): { networkAllowList: string | null } {
  const env = input.env ?? process.env;
  // Daytona's `networkAllowList` accepts IPv4 CIDRs only — it rejects sandbox
  // create outright if any entry is IPv6 ("Invalid IP address ... Must be a
  // valid IPv4 address"). An operator-provisioned list can pick up AAAA-resolved
  // hosts, so drop IPv6 (colon-bearing) entries and keep the IPv4 CIDRs that
  // actually constrain egress. Warn loudly so a dropped entry is visible.
  const networkAllowList = filterIpv4CidrAllowList(
    env[DAYTONA_NETWORK_ALLOW_LIST_ENV]?.trim() || null,
  );
  if (
    !networkAllowList &&
    env.NODE_ENV === 'production' &&
    !isIsolatedRuntimeSchedulerSchema(input.runtimeSchedulerSchema)
  ) {
    throw new Error(
      `${DAYTONA_NETWORK_ALLOW_LIST_ENV} is required for production runs. ` +
        `Refusing to start customer code with unrestricted outbound network ` +
        `access. Provision a comma-separated CIDR allow-list on the production ` +
        `worker. (Isolated-schema runs — preview CI / worktrees — are exempt: ` +
        `they execute synthetic internal plays against dynamic per-PR hosts.)`,
    );
  }
  return { networkAllowList };
}

async function createOneShotDaytonaSandbox(input: {
  daytona: DaytonaClient;
  orgId: string;
  context: DaytonaExecutionContext;
  sandboxName: string;
}): Promise<DaytonaSandbox> {
  const limits = validatePlaySandboxRuntimeLimits(
    input.context.sandboxRuntimeLimits ?? {
      ...STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS,
    },
  );
  const orgId = normalizeLabelValue(input.orgId);
  const workflowId = normalizeLabelValue(input.context.workflowId);
  const playId = normalizeLabelValue(input.context.playId);
  const runId = normalizeLabelValue(input.context.runId);
  const labels: {
    source: string;
    orgId?: string;
    workflowId?: string;
    playId?: string;
    runId?: string;
  } = {
    source: DAYTONA_PLAY_RUNNER_LABEL_SOURCE,
  };
  if (orgId) labels.orgId = orgId;
  if (workflowId) labels.workflowId = workflowId;
  if (playId) labels.playId = playId;
  if (runId) labels.runId = runId;
  const { networkAllowList } = resolveDaytonaSandboxNetworkPolicy({
    runtimeSchedulerSchema: input.context.runtimeSchedulerSchema ?? null,
  });

  const commonParams = {
    name: input.sandboxName,
    labels,
    // This is also the provider-owned backstop for an HTTP create timeout
    // whose named sandbox never becomes lookup-addressable to this process:
    // Daytona deletes an ephemeral sandbox immediately when it stops.
    ephemeral: true,
    autoDeleteInterval: 0,
    autoStopInterval: Math.ceil(
      (limits.timeoutSeconds + PLAY_RUNNER_TIMEOUT_SECONDS - 30 * 60) / 60,
    ),
    autoArchiveInterval: DAYTONA_AUTO_ARCHIVE_INTERVAL_MINUTES,
    // A non-empty `networkAllowList` IS the "block all egress except these
    // CIDRs" control; Daytona rejects create when `networkBlockAll: true` is
    // combined with a non-empty allow-list ("networkBlockAll: true cannot be
    // combined with a non-empty networkAllowList or domainAllowList"). Pass
    // the allow-list alone so the egress restriction holds without the
    // contradictory flag.
    ...(networkAllowList ? { networkAllowList } : {}),
  };
  return input.daytona.create(commonParams, {
    timeout: DAYTONA_CREATE_TIMEOUT_SECONDS,
  });
}

async function reconcileAndDeleteTimedOutDaytonaSandbox(input: {
  daytona: DaytonaClient;
  sandboxName: string;
}): Promise<boolean> {
  if (!input.daytona.get) return false;
  let lastDeleteConflict: unknown = null;
  for (const delayMs of DAYTONA_TIMED_OUT_CREATE_RECONCILE_DELAYS_MS) {
    if (delayMs > 0) {
      await new Promise((resolve) => setTimeout(resolve, delayMs));
    }
    let sandbox: DaytonaSandbox;
    try {
      sandbox = await input.daytona.get(input.sandboxName);
    } catch {
      continue;
    }
    try {
      await sandbox.delete(30);
    } catch (error) {
      // Daytona can make the named sandbox lookup-addressable before its
      // timed-out create transition has settled. That exact conflict is not a
      // cleanup verdict: re-read the same provider identity after the bounded
      // reconcile delay instead of failing the run or launching a duplicate
      // sandbox while the first one is still becoming ready.
      if (isDaytonaSandboxStateChangeConflict(error)) {
        lastDeleteConflict = error;
        continue;
      }
      throw new Error(
        `Daytona timed-out sandbox ${sandbox.id} was reconciled by name but could not be deleted. Modal fallback suppressed.`,
        { cause: error },
      );
    }
    return true;
  }
  if (lastDeleteConflict) {
    throw new Error(
      `Daytona timed-out sandbox ${input.sandboxName} remained in a state transition across the bounded cleanup window. Modal fallback suppressed.`,
      { cause: lastDeleteConflict },
    );
  }
  return false;
}

async function createRetriedOneShotDaytonaSandbox(input: {
  daytona: DaytonaClient;
  orgId: string;
  context: DaytonaExecutionContext;
  emitStage: DaytonaStageEmitter;
  observeCreateCall?: DaytonaCreateCallObserver;
  reserveSandboxCapacity?: () => Promise<{ leaseId: string }>;
  releaseSandboxCapacity?: (leaseId: string) => Promise<void>;
  nextProviderAttempt: () => number;
  startedAt: number;
}): Promise<DaytonaCreateResult> {
  const errors: string[] = [];
  for (const [index, delayMs] of DAYTONA_CREATE_RETRY_DELAYS_MS.entries()) {
    const attempt = input.nextProviderAttempt();
    const sandboxName = `dl-${crypto.randomUUID()}`;
    if (delayMs > 0) {
      input.emitStage('create:retry', {
        attempt: index,
        elapsedMs: Date.now() - input.startedAt,
        reason: 'bounded_backoff',
        delayMs,
      });
      await new Promise((resolve) => setTimeout(resolve, delayMs));
    }
    const attemptStartedAt = Date.now();
    await input.observeCreateCall?.({
      type: 'daytona_create_call_started',
      occurredAtMs: attemptStartedAt,
      providerAttempt: attempt,
    });
    const reservation = await input.reserveSandboxCapacity?.();
    let sandbox: DaytonaSandbox;
    try {
      sandbox = await createOneShotDaytonaSandbox({
        daytona: input.daytona,
        orgId: input.orgId,
        context: input.context,
        sandboxName,
      });
    } catch (error) {
      // A normal rejected create has no durable provider resource. A timeout
      // can be ambiguous, so retain its lease until the crash TTL rather than
      // admitting another sandbox against an unknown provider outcome.
      if (reservation && !isDaytonaSandboxStartTimeout(String(error))) {
        await input.releaseSandboxCapacity?.(reservation.leaseId);
      }
      await input.observeCreateCall?.({
        type: 'daytona_create_call_failed',
        occurredAtMs: Date.now(),
        providerAttempt: attempt,
        errorClass: daytonaCreateErrorClass(error),
      });
      const message = error instanceof Error ? error.message : String(error);
      errors.push(message);
      input.emitStage('create:attempt_failed', {
        attempt,
        elapsedMs: Date.now() - input.startedAt,
        attemptElapsedMs: Date.now() - attemptStartedAt,
        error: message,
      });
      console.warn('[play-runner.daytona.create_attempt_failed]', {
        workflowId: input.context.workflowId ?? null,
        runId: input.context.runId ?? null,
        attempt,
        error: message,
      });
      if (isDaytonaSandboxStartTimeout(message)) {
        const deleted = await reconcileAndDeleteTimedOutDaytonaSandbox({
          daytona: input.daytona,
          sandboxName,
        });
        if (deleted) {
          // The provider-confirmed reconciliation means this reservation can
          // no longer represent a live sandbox. Release before returning the
          // acquisition-unavailable result so Modal fallback is not blocked
          // behind the unbound crash TTL.
          if (reservation) {
            await input.releaseSandboxCapacity?.(reservation.leaseId);
          }
          throw new DaytonaSandboxAcquisitionUnavailableError(
            'daytona_sandbox_start_timeout',
            `${message} Timed-out Daytona sandbox was reconciled and deleted before fallback.`,
          );
        }
        throw new Error(
          `${message} Daytona did not return a cleanup-addressable sandbox identity; Modal fallback suppressed. The named ephemeral sandbox retains Daytona's auto-delete-on-stop backstop.`,
          { cause: error },
        );
      }
      continue;
    }
    // The outcome must be attributed immediately after Daytona acknowledges
    // creation, before resource-policy validation can reject it. Otherwise a
    // created-but-rejected sandbox is indistinguishable from an unknown
    // create result. Do not turn a failed *post-create* journal write into a
    // new provider create: the sandbox is already real. The missing durable
    // outcome is deliberately loud in worker logs and makes the capture gate
    // fail as incomplete telemetry, while the normal resource ledger still
    // records the known sandbox for cleanup below.
    const acquiredAt = Date.now();
    try {
      await input.observeCreateCall?.({
        type: 'daytona_create_call_succeeded',
        occurredAtMs: acquiredAt,
        providerAttempt: attempt,
        sandboxId: sandbox.id,
      });
    } catch {
      console.error('[play-runner.daytona.create_lifecycle_event_unrecorded]', {
        workflowId: input.context.workflowId ?? null,
        runId: input.context.runId ?? null,
        attempt,
        sandboxId: sandbox.id,
        eventType: 'daytona_create_call_succeeded',
      });
    }
    return {
      sandbox,
      attempt,
      attemptElapsedMs: acquiredAt - attemptStartedAt,
      ...(reservation ? { sandboxCapacityLeaseId: reservation.leaseId } : {}),
    };
  }
  const message = `Daytona sandbox create failed across ${errors.length} bounded attempts: ${errors.join('; ')}`;
  const fallbackReason =
    resolveDaytonaSandboxAcquisitionUnavailableReason(errors);
  if (fallbackReason) {
    throw new DaytonaSandboxAcquisitionUnavailableError(
      fallbackReason,
      message,
    );
  }
  throw new Error(message);
}

async function acquireOneShotDaytonaSandbox(input: {
  daytona: DaytonaClient;
  orgId: string;
  context: DaytonaExecutionContext;
  emitStage: DaytonaStageEmitter;
  observeCreateCall?: DaytonaCreateCallObserver;
  reserveSandboxCapacity?: () => Promise<{ leaseId: string }>;
  releaseSandboxCapacity?: (leaseId: string) => Promise<void>;
  nextProviderAttempt: () => number;
  startedAt: number;
}): Promise<AcquiredDaytonaSandbox> {
  const limits = validatePlaySandboxRuntimeLimits(
    input.context.sandboxRuntimeLimits ?? {
      ...STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS,
    },
  );
  input.emitStage('create:start');
  const result = await createRetriedOneShotDaytonaSandbox(input);
  const granted = {
    cpu: result.sandbox.cpu,
    memoryGiB: result.sandbox.memory,
    diskGiB: result.sandbox.disk,
    gpu: result.sandbox.gpu ?? 0,
  };
  if (
    granted.cpu !== limits.cpu ||
    granted.memoryGiB !== limits.memoryGiB ||
    granted.diskGiB !== limits.diskGiB ||
    granted.gpu !== DAYTONA_SANDBOX_GPU
  ) {
    await rejectAcquiredSandbox(
      result.sandbox,
      `Daytona sandbox resource boundary mismatch: expected cpu=${limits.cpu} memoryGiB=${limits.memoryGiB} diskGiB=${limits.diskGiB} gpu=${DAYTONA_SANDBOX_GPU}, granted cpu=${granted.cpu} memoryGiB=${granted.memoryGiB} diskGiB=${granted.diskGiB} gpu=${granted.gpu}`,
      result.sandboxCapacityLeaseId
        ? { leaseId: result.sandboxCapacityLeaseId }
        : undefined,
      input.releaseSandboxCapacity,
    );
  }
  const configuredOrganizationId =
    process.env.DAYTONA_ORGANIZATION_ID?.trim() || null;
  const observedOrganizationId = result.sandbox.organizationId?.trim() || null;
  if (
    configuredOrganizationId &&
    observedOrganizationId &&
    configuredOrganizationId !== observedOrganizationId
  ) {
    await rejectAcquiredSandbox(
      result.sandbox,
      'Daytona sandbox organization routing mismatch. Refusing to run customer code in a sandbox whose observed organization differs from the configured organization.',
      result.sandboxCapacityLeaseId
        ? { leaseId: result.sandboxCapacityLeaseId }
        : undefined,
      input.releaseSandboxCapacity,
    );
  }
  let lookupOrganizationId: string | null = null;
  if (
    !observedOrganizationId &&
    !configuredOrganizationId &&
    input.daytona.get
  ) {
    try {
      const lookedUpSandbox = await input.daytona.get(result.sandbox.id);
      lookupOrganizationId = lookedUpSandbox.organizationId?.trim() || null;
    } catch {
      // The failure below is intentionally about the invariant, not the
      // provider response. The newly created sandbox is still deleted.
    }
  }
  const daytonaOrganizationId =
    observedOrganizationId ?? configuredOrganizationId ?? lookupOrganizationId;
  if (!daytonaOrganizationId) {
    return await rejectAcquiredSandbox(
      result.sandbox,
      'Daytona sandbox organization routing identity is missing. Refusing to run customer code without a durable cleanup routing domain.',
      result.sandboxCapacityLeaseId
        ? { leaseId: result.sandboxCapacityLeaseId }
        : undefined,
      input.releaseSandboxCapacity,
    );
  }
  const billingStartedAt = Date.now();
  const sandbox = result.sandbox;
  input.emitStage('create:done', {
    sandboxId: sandbox.id,
    attempt: result.attempt,
    elapsedMs: billingStartedAt - input.startedAt,
    attemptElapsedMs: result.attemptElapsedMs,
    // Daytona's default snapshot is intentionally provider-managed. Record the
    // resolved immutable snapshot identifier so a slow acquisition can be
    // distinguished from a bad warm-pool/default-snapshot selection without
    // logging provider credentials or customer-code details.
    snapshot: sandbox.snapshot?.trim() || null,
    cpu: granted.cpu,
    memoryGiB: granted.memoryGiB,
    diskGiB: granted.diskGiB,
  });
  return {
    sandbox,
    daytonaOrganizationId,
    billingStartedAt,
    ...(result.sandboxCapacityLeaseId
      ? { sandboxCapacityLeaseId: result.sandboxCapacityLeaseId }
      : {}),
  };
}

export function createOneShotDaytonaSandboxLifecycle(input: {
  daytona: DaytonaClient;
  context: DaytonaExecutionContext;
  emitStage: DaytonaStageEmitter;
  observeCreateCall?: DaytonaCreateCallObserver;
  reserveSandboxCapacity?: () => Promise<{ leaseId: string }>;
  releaseSandboxCapacity?: (leaseId: string) => Promise<void>;
  startedAt?: number;
}): OneShotDaytonaSandboxLifecycle {
  const orgId = validateDaytonaExecutionContext(input.context);
  const startedAt = input.startedAt ?? Date.now();
  let disposed = false;
  let providerAttempt = 0;
  const acquiredSandboxes = new Map<string, AcquiredDaytonaSandbox>();
  let latestAcquiredSandboxPromise: Promise<AcquiredDaytonaSandbox>;
  const createFreshSandbox = () => {
    latestAcquiredSandboxPromise = acquireOneShotDaytonaSandbox({
      daytona: input.daytona,
      orgId,
      context: input.context,
      emitStage: input.emitStage,
      observeCreateCall: input.observeCreateCall,
      reserveSandboxCapacity: input.reserveSandboxCapacity,
      releaseSandboxCapacity: input.releaseSandboxCapacity,
      nextProviderAttempt: () => {
        providerAttempt += 1;
        return providerAttempt;
      },
      startedAt,
    }).then((acquired) => {
      acquiredSandboxes.set(acquired.sandbox.id, acquired);
      return acquired;
    });
    return latestAcquiredSandboxPromise;
  };
  const acquiredSandboxPromise = createFreshSandbox();

  return {
    startedAt,
    acquiredSandboxPromise,
    createFreshSandbox,
    acquiredSandboxes: () => [...acquiredSandboxes.values()],
    settlePendingCreates: async () => {
      await latestAcquiredSandboxPromise.then(
        () => undefined,
        () => undefined,
      );
    },
    dispose: async () => {
      if (disposed) return;
      disposed = true;
      try {
        const acquired = await latestAcquiredSandboxPromise;
        await acquired.sandbox.delete(30);
        // `dispose` has provider-confirmed that the sandbox is gone. This is
        // normally the narrow cancellation window before the scheduler has
        // durably recorded the resource, so no cleanup job will release an
        // unbound lease on our behalf.
        if (acquired.sandboxCapacityLeaseId) {
          await input.releaseSandboxCapacity?.(acquired.sandboxCapacityLeaseId);
        }
      } catch (error) {
        console.warn('[play-runner.daytona.dispose_failed_before_acquire]', {
          error: error instanceof Error ? error.message : String(error),
        });
      }
    },
  };
}

/**
 * Deterministic sandbox delete on ALL backend exits (FIX 1). The old shape
 * attached `deferredRuntimeTasks` cleanup records that NOTHING consumed, gated
 * eager deletion behind a `DAYTONA_BACKGROUND_CLEANUP` env that was unset
 * everywhere — every run leaked paid idle compute until Daytona's autoStop
 * fired. Now every exit path deletes directly (fire-and-forget with structured
 * logs; a failed delete never throws into the result path). autoStop remains
 * only the backstop for a crashed worker process.
 *
 * Push execution note: a `detached_runner` suspended return deliberately
 * BYPASSES `withCleanup` — the runner is still executing in the sandbox; the
 * worker's wake leg owns that delete.
 */
export function createDaytonaSandboxCleanupManager(): {
  activate(acquired: AcquiredDaytonaSandbox): void;
  currentSandbox(): DaytonaSandbox | null;
  cleanupActiveSandboxForCancellation(): boolean;
  stashActiveSandboxForRetry(): void;
  withCleanup(
    result: PlayRunnerResult,
    input: { cancellationCleanupStarted: boolean },
  ): PlayRunnerResult;
} {
  let sandbox: DaytonaSandbox | null = null;
  let sandboxBillingStartedAt: number | null = null;
  const deletedSandboxIds = new Set<string>();

  const deleteSandboxOnce = (input: {
    reason: string;
    sandboxToDelete: DaytonaSandbox | null;
  }) => {
    const target = input.sandboxToDelete;
    if (!target || deletedSandboxIds.has(target.id)) {
      return false;
    }
    deletedSandboxIds.add(target.id);
    const billingStartedAt = sandboxBillingStartedAt;
    void target.delete(30).then(
      () => {
        console.info('[play-runner.daytona.cleanup_done]', {
          reason: input.reason,
          sandboxId: target.id,
          elapsedMs:
            billingStartedAt === null ? null : Date.now() - billingStartedAt,
        });
      },
      (error: unknown) => {
        console.warn('[play-runner.daytona.cleanup_failed]', {
          reason: input.reason,
          sandboxId: target.id,
          error: error instanceof Error ? error.message : String(error),
        });
      },
    );
    return true;
  };

  return {
    activate(acquired) {
      sandbox = acquired.sandbox;
      sandboxBillingStartedAt = acquired.billingStartedAt;
    },
    currentSandbox() {
      return sandbox;
    },
    cleanupActiveSandboxForCancellation() {
      return deleteSandboxOnce({
        reason: 'cancelled',
        sandboxToDelete: sandbox,
      });
    },
    stashActiveSandboxForRetry() {
      // The retired sandbox is dead — a fresh one replaces it for the retry.
      // Delete NOW instead of deferring to a consumer that never existed.
      deleteSandboxOnce({
        reason: 'retired_for_retry',
        sandboxToDelete: sandbox,
      });
      sandbox = null;
      sandboxBillingStartedAt = null;
    },
    withCleanup(result, input) {
      if (!input.cancellationCleanupStarted) {
        deleteSandboxOnce({ reason: 'terminal', sandboxToDelete: sandbox });
      }
      return result;
    },
  };
}