import { Daytona } from '@daytonaio/sdk';
import { randomUUID } from 'node:crypto';
import type {
  PlayRunnerBackend,
  PlayRunnerPreparedExecution,
  PlayRunnerPrepareInput,
} from '../types';
import {
  isRuntimeSandboxCapacityLimitError,
  RuntimeResourceFenceLostError,
} from '../types';
import { buildPlayRunnerBundle } from '../bundle';
import { findPlayRunnerResult, parsePlayRunnerEvents } from '../runner-events';
import type {
  PlayRunnerExecutionConfig,
  PlayRunnerRuntimeTiming,
  PlayRunnerResult,
} from '@shared_libs/play-runtime/protocol';
import {
  PLAY_RUNNER_TERMINAL_GRACE_SECONDS,
  STANDARD_PLAY_RUNTIME_LIMIT_SECONDS,
} from '@shared_libs/play-runtime/runtime-constants';
import {
  loadDaytonaRequiredConfig,
  loadDaytonaRunnerPathsConfig,
} from '@shared_libs/play-runtime/daytona-runtime-config';
import {
  DAYTONA_CANCELLED_ERROR,
  DaytonaSandboxAcquisitionUnavailableError,
  createDaytonaSandboxCleanupManager,
  createOneShotDaytonaSandboxLifecycle,
  type AcquiredDaytonaSandbox,
  type DaytonaClient,
  type DaytonaExecutionContext,
  type DaytonaSandbox,
  type OneShotDaytonaSandboxLifecycle,
  validateDaytonaExecutionContext,
} from './daytona-lifecycle';
import { stageDaytonaRunnerPayload } from './daytona-payload-transport';
import {
  DaytonaRunnerInitializationError,
  prepareDetachedDaytonaRunner,
} from './daytona-session-execution';
import { RUNTIME_RELIABILITY_POLICY } from '@shared_libs/play-runtime/runtime-reliability-policy';

const DAYTONA_COMMAND_RECOVERY_TIMEOUT_MS =
  RUNTIME_RELIABILITY_POLICY.sandbox.daytonaCommandRecoveryTimeoutMs;
const DAYTONA_COMMAND_RECOVERY_POLL_MS =
  RUNTIME_RELIABILITY_POLICY.sandbox.daytonaCommandRecoveryPollMs;
const DAYTONA_INFRASTRUCTURE_MAX_ATTEMPTS = 2;
const DAYTONA_UPLOAD_MAX_ATTEMPTS = 2;
const DAYTONA_UPLOAD_ATTEMPT_DEADLINE_MS = 90_000;
const DAYTONA_CRASH_DIAGNOSTIC_TAIL_MAX_BYTES = 64 * 1_024;
const DAYTONA_CRASH_DIAGNOSTIC_MAX_LINES = 40;
const DAYTONA_CRASH_DIAGNOSTIC_MAX_TEXT_BYTES = 12 * 1_024;

const RUNTIME_POSTGRES_CONNECT_RETRY_PATTERN =
  /\bRuntime Postgres\b.*\b(connection timed out|connect timeout|ETIMEDOUT|ECONNRESET|ECONNREFUSED|Connection terminated|Connection ended unexpectedly)\b/i;
const RUNTIME_RECEIPT_GATEWAY_TRANSPORT_RETRY_PATTERN =
  /AppRuntimeApiTransportError: (?:App runtime API|Runtime receipt gateway) transport exhausted action=(?:get|claim|mark|complete|fail|heartbeat|release|skip)_runtime_step_receipts?/i;
const RUNTIME_API_TRANSPORT_RETRY_PATTERN =
  /\bRuntime API request to .+ failed before receiving a response:\s*(?:fetch failed|connection (?:terminated|timed out|closed|reset)|ECONNRESET|ETIMEDOUT|ECONNREFUSED)\b/i;
const DAYTONA_INFRASTRUCTURE_RETRY_PATTERN =
  /\b(?:Request failed with status code (?:408|429|500|502|503|504)|ECONNRESET|ECONNREFUSED|ETIMEDOUT|UND_ERR_CONNECT_TIMEOUT|fetch failed|socket hang up|Daytona sandbox create failed across|Daytona sandbox create did not complete|Daytona readiness baseline)\b/i;

type DaytonaExecution = { exitCode: number | null; result: string };
export type DaytonaLookupFailure = {
  errorName: string;
  errorCode: string | null;
  httpStatus: number | null;
  requestId: string | null;
  detail: string;
};
export type DaytonaCrashDiagnostic = {
  schemaVersion: 1;
  collectionStatus: 'collected' | 'unavailable' | 'wrong_routing_domain';
  exitCodeFile: number | null;
  sessionExitCode: number | null;
  outputBytes: number | null;
  tailBytes: number;
  tailTruncated: boolean;
  signals: string[];
  unclassifiedLineCount: number;
  termination: DaytonaTerminationDiagnostic | null;
  collectionFailure: DaytonaLookupFailure | null;
};
export type DaytonaTerminationDiagnostic = {
  schemaVersion: 1;
  reason:
    | 'runtime_limit'
    | 'terminal_flush_limit'
    | 'startup_limit'
    | 'watchdog_signal'
    | 'child_spawn_error'
    | 'child_exit';
  runtimeStarted: boolean;
  runtimeCompleted: boolean;
  childExitCode?: number | null;
  childSignal?: string | null;
  watchdogSignal?: 'SIGTERM' | 'SIGINT';
  errorName?: string;
  errorCode?: string | null;
};

function parseDaytonaTerminationDiagnostic(
  value: unknown,
): DaytonaTerminationDiagnostic | null {
  if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
  const candidate = value as Record<string, unknown>;
  const reasons = new Set<DaytonaTerminationDiagnostic['reason']>([
    'runtime_limit',
    'terminal_flush_limit',
    'startup_limit',
    'watchdog_signal',
    'child_spawn_error',
    'child_exit',
  ]);
  if (
    candidate.schemaVersion !== 1 ||
    typeof candidate.reason !== 'string' ||
    !reasons.has(candidate.reason as DaytonaTerminationDiagnostic['reason']) ||
    typeof candidate.runtimeStarted !== 'boolean' ||
    typeof candidate.runtimeCompleted !== 'boolean'
  ) {
    return null;
  }
  const diagnostic: DaytonaTerminationDiagnostic = {
    schemaVersion: 1,
    reason: candidate.reason as DaytonaTerminationDiagnostic['reason'],
    runtimeStarted: candidate.runtimeStarted,
    runtimeCompleted: candidate.runtimeCompleted,
  };
  if (candidate.childExitCode === null) {
    diagnostic.childExitCode = null;
  } else if (
    typeof candidate.childExitCode === 'number' &&
    Number.isSafeInteger(candidate.childExitCode)
  ) {
    diagnostic.childExitCode = candidate.childExitCode;
  }
  if (candidate.childSignal === null) {
    diagnostic.childSignal = null;
  } else if (
    typeof candidate.childSignal === 'string' &&
    /^SIG[A-Z0-9]+$/.test(candidate.childSignal)
  ) {
    diagnostic.childSignal = candidate.childSignal;
  }
  if (
    candidate.watchdogSignal === 'SIGTERM' ||
    candidate.watchdogSignal === 'SIGINT'
  ) {
    diagnostic.watchdogSignal = candidate.watchdogSignal;
  }
  if (
    typeof candidate.errorName === 'string' &&
    /^[A-Za-z][A-Za-z0-9]{0,39}$/.test(candidate.errorName)
  ) {
    diagnostic.errorName = candidate.errorName;
  }
  if (
    candidate.errorCode === null ||
    (typeof candidate.errorCode === 'string' &&
      /^[A-Z0-9_]{1,40}$/.test(candidate.errorCode))
  ) {
    diagnostic.errorCode =
      typeof candidate.errorCode === 'string' ? candidate.errorCode : null;
  }
  return diagnostic;
}
type DaytonaUploadAttemptTiming = {
  attempt: number;
  sandboxId: string;
  acquireElapsedMs: number;
  uploadElapsedMs: number;
  attemptElapsedMs: number;
  totalElapsedMs: number;
};
type DaytonaPreparedExecution = PlayRunnerPreparedExecution & {
  kind: 'daytona';
  bundlePromise: Promise<string>;
  sandboxLifecycle: OneShotDaytonaSandboxLifecycle;
};

class DaytonaUploadDeadlineBreachError extends Error {
  readonly attemptTimings: DaytonaUploadAttemptTiming[];
  readonly deadlineMs: number;

  constructor(input: {
    deadlineMs: number;
    attemptTimings: DaytonaUploadAttemptTiming[];
  }) {
    super(
      `Daytona upload deadline breached after ${input.attemptTimings.length} attempts (deadlineMs=${input.deadlineMs}). ${formatDaytonaUploadAttemptTimings(input.attemptTimings)}`,
    );
    this.name = 'DaytonaUploadDeadlineBreachError';
    this.deadlineMs = input.deadlineMs;
    this.attemptTimings = input.attemptTimings;
  }
}

export const daytonaSdkClientFactory = {
  create(input: ConstructorParameters<typeof Daytona>[0]): DaytonaClient {
    return new Daytona(input);
  },
  /** Full client (get/delete by id) for best-effort GC of orphaned sandboxes. */
  createFull(input: ConstructorParameters<typeof Daytona>[0]): Daytona {
    return new Daytona(input);
  },
};

function stringProperty(value: unknown, key: string): string | null {
  if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
  const candidate = (value as Record<string, unknown>)[key];
  return typeof candidate === 'string' && candidate.trim()
    ? candidate.trim()
    : null;
}

function numberProperty(value: unknown, key: string): number | null {
  if (!value || typeof value !== 'object' || Array.isArray(value)) return null;
  const candidate = (value as Record<string, unknown>)[key];
  return typeof candidate === 'number' && Number.isFinite(candidate)
    ? candidate
    : null;
}

function redactDaytonaErrorDetail(value: string): string {
  return value
    .replace(
      /authorization\s*[=:]\s*bearer\s+[^\s,;]+/gi,
      'Authorization=Bearer [redacted]',
    )
    .replace(
      /(authorization\s*[=:]\s*)(?!bearer\s+\[redacted\])[^\s,;]+/gi,
      '$1[redacted]',
    )
    .replace(/(bearer\s+)[^\s,;]+/gi, '$1[redacted]')
    .replace(/((?:api[_-]?key|token)\s*[=:]\s*)[^\s,;]+/gi, '$1[redacted]')
    .replace(/\s+/g, ' ')
    .trim()
    .slice(0, 1_000);
}

/**
 * Preserve the actionable shape of a Daytona control-plane failure without
 * copying credentials or an arbitrarily large upstream response into logs.
 */
export function describeDaytonaLookupFailure(
  error: unknown,
): DaytonaLookupFailure {
  const record = error && typeof error === 'object' ? error : null;
  const response = record ? (record as Record<string, unknown>).response : null;
  const headers =
    response && typeof response === 'object'
      ? (response as Record<string, unknown>).headers
      : null;
  const message = error instanceof Error ? error.message : String(error);
  return {
    errorName: error instanceof Error ? error.name || 'Error' : 'NonErrorThrow',
    errorCode: stringProperty(record, 'code'),
    httpStatus:
      numberProperty(record, 'status') ??
      numberProperty(record, 'statusCode') ??
      numberProperty(response, 'status'),
    requestId:
      stringProperty(record, 'requestId') ??
      stringProperty(record, 'request_id') ??
      stringProperty(headers, 'x-request-id') ??
      stringProperty(headers, 'x-daytona-request-id'),
    detail: redactDaytonaErrorDetail(message),
  };
}

function shellQuoteDaytonaDiagnostic(value: string): string {
  return `'${value.replace(/'/g, `'\\''`)}'`;
}

const DAYTONA_CRASH_DIAGNOSTIC_LINE_PATTERN =
  /(?:\b(?:Type|Range|Reference|Syntax|Aggregate|Eval|URI)?Error(?:\s*\[[^\]]+\])?:|\bERR_[A-Z0-9_]+\b|\b(?:uncaught|unhandled(?:promiserejection)?)\b|\bFATAL ERROR\b|heap out of memory|\b(?:ENOMEM|EPIPE|ECONNRESET|ETIMEDOUT)\b|\bSIG(?:ABRT|BUS|FPE|ILL|KILL|SEGV|TERM)\b|^node:|^\s*at\s+(?:async\s+)?)/i;

function daytonaCrashDiagnosticSignature(line: string): string | null {
  const signals = new Set<string>();
  const errorName = line.match(
    /\b((?:Type|Range|Reference|Syntax|Aggregate|Eval|URI)?Error)\b/i,
  )?.[1];
  if (errorName) signals.add(`error:${errorName}`);

  if (/\buncaught\b/i.test(line)) signals.add('exception:uncaught');
  if (/\bunhandled(?:promiserejection)?\b/i.test(line))
    signals.add('exception:unhandled_rejection');
  if (/\bFATAL ERROR\b/i.test(line)) signals.add('runtime:fatal_error');
  if (/heap out of memory/i.test(line))
    signals.add('runtime:heap_out_of_memory');

  for (const code of line.matchAll(/\b(ENOMEM|EPIPE|ECONNRESET|ETIMEDOUT)\b/gi))
    signals.add(`transport:${code[1]!.toUpperCase()}`);
  for (const signal of line.matchAll(
    /\b(SIG(?:ABRT|BUS|FPE|ILL|KILL|SEGV|TERM))\b/gi,
  ))
    signals.add(`signal:${signal[1]!.toUpperCase()}`);

  const nodeErrorCode = line.match(/\bERR_[A-Z0-9_]+\b/i)?.[0];
  if (nodeErrorCode) signals.add('runtime:node_error_code');
  if (/^node:/i.test(line)) signals.add('runtime:node_error');
  if (/^\s*at\s+(?:async\s+)?/i.test(line)) signals.add('stack:frame');

  return signals.size > 0 ? [...signals].join(' ') : null;
}

/**
 * Reduce a captured runner-output tail to error and stack signals only.
 * Compact Daytona stdout contains JSON result envelopes which can contain
 * customer rows, so every JSON-looking line is excluded even when truncated.
 */
export function extractDaytonaCrashDiagnosticSignals(outputTail: string): {
  signals: string[];
  unclassifiedLineCount: number;
} {
  const candidates: string[] = [];
  let unclassifiedLineCount = 0;
  for (const rawLine of outputTail.split(/\r?\n/)) {
    const line = rawLine.trimEnd();
    if (!line.trim()) continue;
    if (/^\s*[\[{]/.test(line)) {
      unclassifiedLineCount += 1;
      continue;
    }
    if (!DAYTONA_CRASH_DIAGNOSTIC_LINE_PATTERN.test(line)) {
      unclassifiedLineCount += 1;
      continue;
    }
    const signature = daytonaCrashDiagnosticSignature(line);
    if (signature) candidates.push(signature);
  }

  const selected = candidates.slice(-DAYTONA_CRASH_DIAGNOSTIC_MAX_LINES);
  const signals: string[] = [];
  let retainedBytes = 0;
  for (const line of selected) {
    const lineBytes = Buffer.byteLength(line, 'utf8');
    if (
      retainedBytes + lineBytes + (signals.length > 0 ? 1 : 0) >
      DAYTONA_CRASH_DIAGNOSTIC_MAX_TEXT_BYTES
    ) {
      break;
    }
    signals.push(line);
    retainedBytes += lineBytes + (signals.length > 1 ? 1 : 0);
  }
  return { signals, unclassifiedLineCount };
}

/**
 * Read bounded crash evidence while the failed sandbox still exists. This is
 * operator telemetry only: it never becomes a run result or customer log.
 */
export async function readDetachedDaytonaCrashDiagnostic(input: {
  sandboxId: string;
  sessionId: string;
  cmdId: string;
  outputPath: string;
  exitCodePath: string;
  terminationDiagnosticPath?: string;
  expectedOrganizationId?: string | null;
}): Promise<DaytonaCrashDiagnostic> {
  const unavailable = (
    collectionFailure: DaytonaLookupFailure,
  ): DaytonaCrashDiagnostic => ({
    schemaVersion: 1,
    collectionStatus: 'unavailable',
    exitCodeFile: null,
    sessionExitCode: null,
    outputBytes: null,
    tailBytes: 0,
    tailTruncated: false,
    signals: [],
    unclassifiedLineCount: 0,
    termination: null,
    collectionFailure,
  });
  try {
    const { clientOptions } = loadDaytonaRequiredConfig();
    const sandbox = (await daytonaSdkClientFactory
      .createFull(clientOptions)
      .get(input.sandboxId)) as DaytonaSandbox;
    const expectedOrganizationId = input.expectedOrganizationId?.trim() || null;
    const observedOrganizationId = sandbox.organizationId?.trim() || null;
    if (
      expectedOrganizationId &&
      observedOrganizationId !== expectedOrganizationId
    ) {
      return {
        ...unavailable({
          errorName: 'DaytonaRoutingDomainMismatch',
          errorCode: 'wrong_routing_domain',
          httpStatus: null,
          requestId: null,
          detail: 'Sandbox belongs to a different Daytona organization.',
        }),
        collectionStatus: 'wrong_routing_domain',
      };
    }

    const script = `const fs=require('node:fs');const output=process.argv[1];const exit=process.argv[2];const max=Number(process.argv[3]);const terminationPath=process.argv[4]||'';const stat=fs.statSync(output);const start=Math.max(0,stat.size-max);const fd=fs.openSync(output,'r');const tail=Buffer.alloc(stat.size-start);fs.readSync(fd,tail,0,tail.length,start);fs.closeSync(fd);let exitCode=null;try{const value=Number.parseInt(fs.readFileSync(exit,'utf8').trim(),10);if(Number.isFinite(value))exitCode=value}catch{}let termination=null;try{const value=JSON.parse(fs.readFileSync(terminationPath,'utf8'));if(value&&typeof value==='object')termination=value}catch{}process.stdout.write(JSON.stringify({outputBytes:stat.size,tail:tail.toString('base64'),exitCode,termination}));`;
    const [snapshotResult, sessionResult] = await Promise.allSettled([
      sandbox.process.executeCommand(
        `node -e ${shellQuoteDaytonaDiagnostic(script)} ${shellQuoteDaytonaDiagnostic(input.outputPath)} ${shellQuoteDaytonaDiagnostic(input.exitCodePath)} ${DAYTONA_CRASH_DIAGNOSTIC_TAIL_MAX_BYTES} ${shellQuoteDaytonaDiagnostic(input.terminationDiagnosticPath ?? '')}`,
        undefined,
        {},
        8,
      ),
      sandbox.process.getSessionCommand(input.sessionId, input.cmdId),
    ]);
    if (snapshotResult.status !== 'fulfilled') {
      return unavailable(describeDaytonaLookupFailure(snapshotResult.reason));
    }
    const snapshot = JSON.parse(snapshotResult.value.result) as {
      outputBytes?: unknown;
      tail?: unknown;
      exitCode?: unknown;
      termination?: unknown;
    };
    const tailBuffer =
      typeof snapshot.tail === 'string'
        ? Buffer.from(snapshot.tail, 'base64')
        : Buffer.alloc(0);
    const tail = tailBuffer.toString('utf8');
    const outputBytes =
      typeof snapshot.outputBytes === 'number' &&
      Number.isFinite(snapshot.outputBytes)
        ? snapshot.outputBytes
        : null;
    const { signals, unclassifiedLineCount } =
      extractDaytonaCrashDiagnosticSignals(tail);
    const sessionExitCode =
      sessionResult.status === 'fulfilled' &&
      typeof sessionResult.value?.exitCode === 'number'
        ? sessionResult.value.exitCode
        : null;
    const termination = parseDaytonaTerminationDiagnostic(snapshot.termination);
    return {
      schemaVersion: 1,
      collectionStatus: 'collected',
      exitCodeFile:
        typeof snapshot.exitCode === 'number' &&
        Number.isFinite(snapshot.exitCode)
          ? snapshot.exitCode
          : null,
      sessionExitCode,
      outputBytes,
      tailBytes: tailBuffer.byteLength,
      tailTruncated:
        outputBytes !== null &&
        outputBytes > DAYTONA_CRASH_DIAGNOSTIC_TAIL_MAX_BYTES,
      signals,
      unclassifiedLineCount,
      termination,
      collectionFailure:
        sessionResult.status === 'rejected'
          ? describeDaytonaLookupFailure(sessionResult.reason)
          : null,
    };
  } catch (error) {
    return unavailable(describeDaytonaLookupFailure(error));
  }
}

/**
 * Best-effort deletion of an orphaned Daytona sandbox by id (push-execution B4).
 *
 * When the scheduler's expired-claim sweeper reclaims a run whose worker was
 * babysitting a sandbox, nothing tears that sandbox down — `ephemeral +
 * autoStop` is the only backstop, which leaks paid compute until it fires. The
 * worker records each acquired sandbox id under the fenced lease, so on the
 * reclaim / terminal-failure path it can look the sandbox up by id and delete
 * it. Failures are swallowed: the run already ended, and the autoStop backstop
 * still applies.
 */
/**
 * Timeout-wake verification for a detached push-execution runner (B2-final).
 *
 * When the parked attempt's ceiling timeout fires without a pushed terminal,
 * the wake leg calls this to distinguish "runner finished but its terminal
 * push was lost" from "runner died silently": read the session command's exit
 * code, then salvage the structured `result` event from the captured output
 * file. Everything is best-effort — a deleted/expired sandbox yields
 * `{ exitCode: null, result: null }` and the attempt fails through the normal
 * engine retry/fencing path.
 */
export async function inspectDetachedDaytonaRunner(input: {
  sandboxId: string;
  sessionId: string;
  cmdId: string;
  outputPath: string;
  exitCodePath: string;
}): Promise<{
  exitCode: number | null;
  result: PlayRunnerResult | null;
  diagnosis: {
    stage:
      | 'salvaged'
      | 'command_running'
      | 'result_missing'
      | 'command_inspection_failed'
      | 'sandbox_missing'
      | 'sandbox_lookup_failed';
    detail: string | null;
  };
}> {
  try {
    const { clientOptions } = loadDaytonaRequiredConfig();
    const daytona = daytonaSdkClientFactory.createFull(clientOptions);
    const sandbox = (await daytona.get(input.sandboxId)) as DaytonaSandbox;
    let exitCode: number | null = null;
    let commandFound = false;
    let commandInspectionError: string | null = null;
    try {
      const command = await sandbox.process.getSessionCommand(
        input.sessionId,
        input.cmdId,
      );
      commandFound = Boolean(command);
      exitCode =
        typeof command?.exitCode === 'number' ? command.exitCode : null;
    } catch (error) {
      commandInspectionError =
        error instanceof Error ? error.message : String(error);
      console.warn('[play-runner.daytona.detached_inspect_command_failed]', {
        sandboxId: input.sandboxId,
        cmdId: input.cmdId,
        error: error instanceof Error ? error.message : String(error),
      });
    }
    const recovered = await recoverDaytonaCommandExecution({
      sandbox,
      outputPath: input.outputPath,
      exitCodePath: input.exitCodePath,
      timeoutMs: 10_000,
    });
    if (!recovered.execution) {
      return {
        exitCode,
        result: null,
        diagnosis: commandInspectionError
          ? {
              stage: 'command_inspection_failed',
              detail: commandInspectionError,
            }
          : commandFound && exitCode === null
            ? { stage: 'command_running', detail: null }
            : { stage: 'result_missing', detail: null },
      };
    }
    const result =
      findPlayRunnerResult(parsePlayRunnerEvents(recovered.execution.result)) ??
      null;
    return {
      exitCode: exitCode ?? recovered.execution.exitCode,
      result,
      diagnosis: result
        ? { stage: 'salvaged', detail: null }
        : { stage: 'result_missing', detail: null },
    };
  } catch (error) {
    const failure = describeDaytonaLookupFailure(error);
    console.warn('[play-runner.daytona.detached_inspect_failed]', {
      sandboxId: input.sandboxId,
      cmdId: input.cmdId,
      failure,
    });
    return {
      exitCode: null,
      result: null,
      diagnosis: {
        stage:
          failure.httpStatus === 404
            ? 'sandbox_missing'
            : 'sandbox_lookup_failed',
        detail: failure.detail,
      },
    };
  }
}

/** Read the runner's exact customer-code completion fence before terminal GC. */
export async function readDetachedDaytonaRuntimeCompletion(input: {
  sandboxId: string;
  runtimeCompletedPath: string;
  expectedOrganizationId?: string | null;
}): Promise<number | null> {
  try {
    const { clientOptions } = loadDaytonaRequiredConfig();
    const sandbox = (await daytonaSdkClientFactory
      .createFull(clientOptions)
      .get(input.sandboxId)) as DaytonaSandbox;
    const expectedOrganizationId = input.expectedOrganizationId?.trim() || null;
    const observedOrganizationId = sandbox.organizationId?.trim() || null;
    if (
      expectedOrganizationId &&
      observedOrganizationId !== expectedOrganizationId
    ) {
      console.warn(
        '[play-runner.daytona.runtime_completion_wrong_routing_domain]',
        {
          sandboxId: input.sandboxId,
          expectedOrganizationId,
          observedOrganizationId,
        },
      );
      return null;
    }
    const marker = JSON.parse(
      (await sandbox.fs.downloadFile(input.runtimeCompletedPath, 5)).toString(
        'utf-8',
      ),
    ) as { at?: unknown };
    return typeof marker.at === 'number' && Number.isFinite(marker.at)
      ? marker.at
      : null;
  } catch (error) {
    console.warn(
      '[play-runner.daytona.runtime_completion_marker_unavailable]',
      {
        sandboxId: input.sandboxId,
        // Path is generated per attempt and contains no customer data.
        runtimeCompletedPath: input.runtimeCompletedPath,
        error: error instanceof Error ? error.message : String(error),
      },
    );
    return null;
  }
}

export type DaytonaSandboxDeleteOutcome =
  | {
      kind: 'deleted' | 'already_absent';
      organizationId: string | null;
    }
  | {
      kind: 'timed_out' | 'rate_limited' | 'failed';
      organizationId: string | null;
      code: string;
      detail: string;
    };

export async function deleteDaytonaSandboxByIdWithOutcome(input: {
  sandboxId: string;
  timeoutSeconds?: number;
  expectedOrganizationId?: string | null;
  allowUnscopedAlreadyAbsent?: boolean;
}): Promise<DaytonaSandboxDeleteOutcome> {
  const sandboxId = input.sandboxId?.trim();
  const expectedOrganizationId = input.expectedOrganizationId?.trim() || null;
  if (!sandboxId) {
    return {
      kind: 'failed',
      organizationId: expectedOrganizationId,
      code: 'invalid_sandbox_id',
      detail: 'Sandbox ID is required.',
    };
  }
  try {
    const { clientOptions } = loadDaytonaRequiredConfig();
    const daytona = daytonaSdkClientFactory.createFull(clientOptions);
    const sandbox = await daytona.get(sandboxId);
    const observedOrganizationId = sandbox.organizationId?.trim() || null;
    if (
      expectedOrganizationId &&
      observedOrganizationId !== expectedOrganizationId
    ) {
      return {
        kind: 'failed',
        organizationId: observedOrganizationId,
        code: 'wrong_routing_domain',
        detail: 'Sandbox belongs to a different Daytona organization.',
      };
    }
    await daytona.delete(sandbox, input.timeoutSeconds ?? 30);
    return {
      kind: 'deleted',
      organizationId: observedOrganizationId,
    };
  } catch (error) {
    const failure = describeDaytonaLookupFailure(error);
    // Cleanup is an idempotent "ensure absent" operation. Daytona returning
    // not-found means another cleanup owner already satisfied the obligation.
    // The durable expected organization came from the sandbox returned by the
    // same organization-scoped credential at creation time. When an explicit
    // worker organization is configured it must still match that evidence;
    // deployments which rely only on the provider-returned organization retain
    // that durable creation-domain proof. Legacy eager cleanup retains its
    // previous unscoped behavior through the explicit compatibility option.
    if (failure.httpStatus === 404) {
      const configuredOrganizationId =
        process.env.DAYTONA_ORGANIZATION_ID?.trim() || null;
      if (
        !input.allowUnscopedAlreadyAbsent &&
        (!expectedOrganizationId ||
          (configuredOrganizationId &&
            configuredOrganizationId !== expectedOrganizationId))
      ) {
        return {
          kind: 'failed',
          organizationId: configuredOrganizationId,
          code: expectedOrganizationId
            ? 'wrong_routing_domain'
            : 'missing_routing_domain',
          detail:
            'Daytona returned not-found without an exact creation-domain match.',
        };
      }
      console.info('[play-runner.daytona.reclaim_sandbox_already_absent]', {
        sandboxId,
      });
      return {
        kind: 'already_absent',
        organizationId:
          configuredOrganizationId ?? expectedOrganizationId ?? null,
      };
    }
    console.warn('[play-runner.daytona.reclaim_sandbox_delete_failed]', {
      sandboxId,
      failure,
    });
    const timedOut =
      failure.httpStatus === 408 ||
      /(?:timeout|timed out|ETIMEDOUT)/i.test(
        `${failure.errorCode ?? ''} ${failure.detail}`,
      );
    return {
      kind:
        failure.httpStatus === 429
          ? 'rate_limited'
          : timedOut
            ? 'timed_out'
            : 'failed',
      organizationId: expectedOrganizationId,
      code:
        failure.errorCode ??
        (failure.httpStatus ? `http_${failure.httpStatus}` : 'delete_failed'),
      detail: failure.detail,
    };
  }
}

export async function deleteDaytonaSandboxById(input: {
  sandboxId: string;
  timeoutSeconds?: number;
}): Promise<boolean> {
  const outcome = await deleteDaytonaSandboxByIdWithOutcome({
    ...input,
    allowUnscopedAlreadyAbsent: true,
  });
  return outcome.kind === 'deleted' || outcome.kind === 'already_absent';
}

function formatDaytonaError(error: unknown): string {
  if (!(error instanceof Error)) {
    return String(error);
  }
  const errorCode = (error as Error & { code?: unknown }).code;
  const code = typeof errorCode === 'string' ? ` code=${errorCode}` : '';
  const cause =
    error.cause && error.cause !== error
      ? ` cause=${formatDaytonaError(error.cause)}`
      : '';
  const details =
    error.name && error.name !== 'Error'
      ? [error.name, error.message].filter(Boolean).join(': ')
      : error.message || error.name || 'Error';
  return `${details || 'Error'}${code}${cause}`;
}

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

function formatDaytonaUploadAttemptTimings(
  timings: readonly DaytonaUploadAttemptTiming[],
): string {
  return `attempts=[${timings
    .map(
      (timing) =>
        `{attempt=${timing.attempt}, sandboxId=${timing.sandboxId}, acquireElapsedMs=${timing.acquireElapsedMs}, uploadElapsedMs=${timing.uploadElapsedMs}, attemptElapsedMs=${timing.attemptElapsedMs}, totalElapsedMs=${timing.totalElapsedMs}}`,
    )
    .join(', ')}]`;
}

function emitDaytonaStage(
  _callbacks: Parameters<PlayRunnerBackend['execute']>[1],
  context: DaytonaExecutionContext,
  stage: string,
  extra: Record<string, unknown> = {},
): void {
  const payload = {
    workflowId: context.workflowId ?? null,
    runId: context.runId ?? null,
    playName: context.playName ?? null,
    stage,
    ...extra,
  };
  console.info('[play-runner.daytona.stage]', JSON.stringify(payload));
}

async function downloadDaytonaTextFile(
  sandbox: DaytonaSandbox,
  remotePath: string,
): Promise<string> {
  const file = await sandbox.fs.downloadFile(remotePath, 15);
  return file.toString('utf-8');
}

async function recoverDaytonaCommandExecution(input: {
  sandbox: DaytonaSandbox;
  outputPath: string;
  exitCodePath: string;
  timeoutMs?: number;
}): Promise<{ execution: DaytonaExecution | null; lastError: unknown }> {
  const deadline =
    Date.now() + (input.timeoutMs ?? DAYTONA_COMMAND_RECOVERY_TIMEOUT_MS);
  let lastError: unknown = null;

  while (Date.now() < deadline) {
    try {
      const exitCodeText = await downloadDaytonaTextFile(
        input.sandbox,
        input.exitCodePath,
      );
      const output = await downloadDaytonaTextFile(
        input.sandbox,
        input.outputPath,
      );
      const exitCode = Number.parseInt(exitCodeText.trim(), 10);
      return {
        execution: {
          exitCode: Number.isFinite(exitCode) ? exitCode : null,
          result: output,
        },
        lastError: null,
      };
    } catch (error) {
      lastError = error;
      await sleep(DAYTONA_COMMAND_RECOVERY_POLL_MS);
    }
  }

  return { execution: null, lastError };
}

function createDaytonaCancelledResult(
  config: PlayRunnerExecutionConfig,
): PlayRunnerResult {
  return {
    status: 'failed',
    error: DAYTONA_CANCELLED_ERROR,
    logs: [],
    stats: {},
    steps: [],
    checkpoint: config.checkpoint ?? null,
    tableNamespace: null,
  };
}

function createDaytonaFailedResult(input: {
  config: PlayRunnerExecutionConfig;
  error: string;
  runtimeTiming?: PlayRunnerRuntimeTiming;
}): PlayRunnerResult {
  return {
    status: 'failed',
    error: input.error,
    logs: [],
    stats: {},
    steps: [],
    checkpoint: input.config.checkpoint ?? null,
    tableNamespace: null,
    ...(input.runtimeTiming ? { runtimeTiming: input.runtimeTiming } : {}),
  };
}

/**
 * Classify a failed runner result as a retryable runtime-initialization /
 * transport failure. Exported for the absurd worker's detached wake leg
 * (B2-final): the backend no longer sees the runner result (it arrives via the
 * gateway terminal push), so the fresh-sandbox retry decision moved to the
 * worker, which throws a retryable error into the engine retry ladder.
 */
export function retryableRuntimeInitializationFailureReason(
  result: PlayRunnerResult,
):
  | 'runtime_postgres_connect'
  | 'runtime_api_transport'
  | 'runtime_receipt_gateway_transport'
  | null {
  if (result.status !== 'failed') return null;
  if (RUNTIME_POSTGRES_CONNECT_RETRY_PATTERN.test(result.error)) {
    return 'runtime_postgres_connect';
  }
  if (RUNTIME_RECEIPT_GATEWAY_TRANSPORT_RETRY_PATTERN.test(result.error)) {
    return 'runtime_receipt_gateway_transport';
  }
  const rowsProcessed = Number(result.stats?.rowsProcessed ?? 0);
  if (
    rowsProcessed === 0 &&
    result.steps.length === 0 &&
    RUNTIME_API_TRANSPORT_RETRY_PATTERN.test(result.error)
  ) {
    return 'runtime_api_transport';
  }
  return null;
}

function isRetryableDaytonaInfrastructureFailure(error: unknown): boolean {
  if (error instanceof DaytonaRunnerInitializationError) return false;
  const message = error instanceof Error ? error.message : String(error);
  if (!message || /cancelled|aborted/i.test(message)) {
    return false;
  }
  return DAYTONA_INFRASTRUCTURE_RETRY_PATTERN.test(message);
}

function isRetryableUnstartedDaytonaRunner(
  error: unknown,
): error is DaytonaRunnerInitializationError {
  if (!(error instanceof DaytonaRunnerInitializationError)) return false;
  const diagnostic = error.startupDiagnostic;
  // The sandbox runner cannot execute customer code until the worker observes
  // its heartbeat, parks the scheduler attempt, and the gateway acknowledges
  // that waiting state. If no heartbeat was observed and Daytona has no exit
  // evidence, deleting this fenced sandbox and retrying once cannot duplicate
  // customer side effects.
  return (
    diagnostic.commandFound &&
    diagnostic.sessionFound &&
    diagnostic.exitCode === null &&
    diagnostic.exitCodeFile === null &&
    !diagnostic.schedulerReadinessObserved
  );
}

function prepareDaytonaExecution(
  input: PlayRunnerPrepareInput,
  callbacks: Parameters<PlayRunnerBackend['execute']>[1],
): DaytonaPreparedExecution {
  const { clientOptions: daytonaOptions } = loadDaytonaRequiredConfig();
  const daytona = daytonaSdkClientFactory.create(daytonaOptions);
  const bundlePromise = buildPlayRunnerBundle();
  const sandboxLifecycle = createOneShotDaytonaSandboxLifecycle({
    daytona,
    context: input.context,
    emitStage: (stage, extra) =>
      emitDaytonaStage(callbacks, input.context, stage, extra),
    observeCreateCall: callbacks?.onRuntimeLifecycleEvent,
    reserveSandboxCapacity: callbacks?.reserveSandboxCapacity
      ? () => callbacks.reserveSandboxCapacity!('daytona')
      : undefined,
    releaseSandboxCapacity: callbacks?.releaseSandboxCapacity,
  });
  return {
    kind: 'daytona',
    bundlePromise,
    sandboxLifecycle,
    dispose: sandboxLifecycle.dispose,
  };
}

function isDaytonaPreparedExecution(
  prepared: PlayRunnerPreparedExecution | undefined,
): prepared is DaytonaPreparedExecution {
  return prepared?.kind === 'daytona';
}

export const daytonaPlayRunnerBackend: PlayRunnerBackend = {
  async prepare(input, callbacks) {
    return prepareDaytonaExecution(input, callbacks);
  },

  async execute(config, callbacks, prepared) {
    try {
      validateDaytonaExecutionContext(config.context);
    } catch (error) {
      return {
        status: 'failed',
        error: error instanceof Error ? error.message : String(error),
        logs: [],
        stats: {},
        steps: [],
        checkpoint: config.checkpoint ?? null,
        tableNamespace: null,
      };
    }
    // Push execution is THE Daytona shape (B2-final): the worker parks after
    // the detached start and finalizes from the runner's pushed terminal.
    // There is deliberately no worker-babysat fallback — a launch without the
    // push config is a wiring bug, not a mode.
    const push = config.context.runnerPushExecution;
    if (!push) {
      return {
        status: 'failed',
        error:
          'Daytona runner backend requires push-execution config (context.runnerPushExecution). Worker-babysat Daytona execution was removed; submit through the absurd scheduler, which stamps it.',
        logs: [],
        stats: {},
        steps: [],
        checkpoint: config.checkpoint ?? null,
        tableNamespace: null,
      };
    }
    const readRunnerReadiness = callbacks?.readDetachedRunnerReadiness;
    if (!readRunnerReadiness) {
      if (isDaytonaPreparedExecution(prepared)) {
        await prepared.dispose?.();
      }
      return {
        status: 'failed',
        error:
          'Daytona runner backend requires the scheduler-owned detached-runner readiness port.',
        logs: [],
        stats: {},
        steps: [],
        checkpoint: config.checkpoint ?? null,
        tableNamespace: null,
      };
    }
    let preparedExecution: DaytonaPreparedExecution;
    try {
      preparedExecution = isDaytonaPreparedExecution(prepared)
        ? prepared
        : prepareDaytonaExecution({ context: config.context }, callbacks);
    } catch (error) {
      return {
        status: 'failed',
        error: error instanceof Error ? error.message : String(error),
        logs: [],
        stats: {},
        steps: [],
        checkpoint: config.checkpoint ?? null,
        tableNamespace: null,
      };
    }

    if (callbacks?.cancellationSignal?.aborted) {
      await preparedExecution.dispose?.();
      return createDaytonaCancelledResult(config);
    }

    const bundlePromise = preparedExecution.bundlePromise;
    const sandboxLifecycle = preparedExecution.sandboxLifecycle;
    const startedAt = sandboxLifecycle.startedAt;
    const executionTimeoutSeconds =
      config.context.sandboxRuntimeLimits?.timeoutSeconds ??
      STANDARD_PLAY_RUNTIME_LIMIT_SECONDS;
    const runtimeTiming: PlayRunnerRuntimeTiming = { backend: 'daytona' };
    const sandboxCleanup = createDaytonaSandboxCleanupManager();
    const withCleanup = (result: PlayRunnerResult): PlayRunnerResult => {
      return sandboxCleanup.withCleanup(
        {
          ...result,
          runtimeTiming: result.runtimeTiming ?? runtimeTiming,
        },
        {
          cancellationCleanupStarted,
        },
      );
    };
    let cancellationRequested = false;
    let cancellationCleanupStarted = false;
    let activeAcquiredResource: AcquiredDaytonaSandbox | null = null;
    const uploadDeadlineBreaches: DaytonaUploadAttemptTiming[] = [];
    const reportedSandboxIds = new Set<string>();
    const reportedSandboxEndTimes = new Map<string, number | null>();
    const runtimeResourceReportErrors = new Set<unknown>();
    const reportRuntimeResource = async (acquired: AcquiredDaytonaSandbox) => {
      const sandbox = acquired.sandbox;
      const billingEndedAt = acquired.billingEndedAt ?? null;
      if (
        reportedSandboxIds.has(sandbox.id) &&
        reportedSandboxEndTimes.get(sandbox.id) === billingEndedAt
      ) {
        return;
      }
      try {
        await callbacks?.onRuntimeResourceAcquired?.({
          kind: 'daytona_sandbox',
          sandboxId: sandbox.id,
          runtimeEnvironment:
            process.env.DEEPLINE_RUNTIME_ENVIRONMENT === 'preview'
              ? 'preview'
              : 'production',
          daytonaEnvironment:
            process.env.DEEPLINE_RUNTIME_ENVIRONMENT === 'preview'
              ? 'preview'
              : 'production',
          daytonaOrganizationId: acquired.daytonaOrganizationId,
          billingStartedAt: acquired.billingStartedAt,
          billingEndedAt,
          maxBillingDurationSeconds:
            (config.context.sandboxRuntimeLimits?.timeoutSeconds ??
              STANDARD_PLAY_RUNTIME_LIMIT_SECONDS) +
            PLAY_RUNNER_TERMINAL_GRACE_SECONDS,
          cpu: typeof sandbox.cpu === 'number' ? sandbox.cpu : null,
          memoryGiB: typeof sandbox.memory === 'number' ? sandbox.memory : null,
          diskGiB: typeof sandbox.disk === 'number' ? sandbox.disk : null,
          ...(acquired.sandboxCapacityLeaseId
            ? { sandboxCapacityLeaseId: acquired.sandboxCapacityLeaseId }
            : {}),
        });
      } catch (error) {
        runtimeResourceReportErrors.add(error);
        throw error;
      }
      reportedSandboxIds.add(sandbox.id);
      reportedSandboxEndTimes.set(sandbox.id, billingEndedAt);
    };
    const reportRetiringRuntimeResource = async (
      acquired: AcquiredDaytonaSandbox,
    ) => {
      try {
        await reportRuntimeResource(acquired);
      } catch (error) {
        if (error instanceof RuntimeResourceFenceLostError) throw error;
        console.warn('[play-runner.daytona.resource_report_deferred]', {
          sandboxId: acquired.sandbox.id,
          error: error instanceof Error ? error.message : String(error),
        });
      }
    };
    let cancelExecution!: (error: Error) => void;
    const cancellationPromise = new Promise<never>((_resolve, reject) => {
      cancelExecution = reject;
    });
    const onCancel = () => {
      cancellationRequested = true;
      if (activeAcquiredResource && !activeAcquiredResource.billingEndedAt) {
        activeAcquiredResource.billingEndedAt = Date.now();
      }
      if (cancellationCleanupStarted) {
        cancelExecution(new Error(DAYTONA_CANCELLED_ERROR));
        return;
      }
      if (sandboxCleanup.cleanupActiveSandboxForCancellation()) {
        cancellationCleanupStarted = true;
      } else {
        cancellationCleanupStarted = true;
        void preparedExecution.dispose?.();
      }
      cancelExecution(new Error(DAYTONA_CANCELLED_ERROR));
    };
    callbacks?.cancellationSignal?.addEventListener('abort', onCancel, {
      once: true,
    });
    const throwIfCancellationRequested = () => {
      if (cancellationRequested || callbacks?.cancellationSignal?.aborted) {
        onCancel();
        throw new Error(DAYTONA_CANCELLED_ERROR);
      }
    };
    let acquiredSandboxPromise = sandboxLifecycle.acquiredSandboxPromise;
    try {
      for (
        let executionAttempt = 1;
        executionAttempt <= DAYTONA_INFRASTRUCTURE_MAX_ATTEMPTS;
        executionAttempt += 1
      ) {
        let sandboxForAttempt: DaytonaSandbox | null = null;
        const attemptStartedAt = Date.now();
        try {
          const acquiredSandbox = await Promise.race([
            acquiredSandboxPromise,
            cancellationPromise,
          ]);
          activeAcquiredResource = acquiredSandbox;
          sandboxCleanup.activate(acquiredSandbox);
          runtimeTiming.daytonaCreateMs =
            acquiredSandbox.billingStartedAt - startedAt;
          const sandbox = acquiredSandbox.sandbox;
          sandboxForAttempt = sandbox;
          await reportRuntimeResource(acquiredSandbox);
          // Account for every acquired retry sandbox before customer code runs.
          // This keeps resource persistence failures on the pre-side-effect
          // side of the execution boundary and makes cancellation billing
          // complete without a post-terminal write race.
          await sandboxLifecycle.settlePendingCreates();
          for (const acquired of sandboxLifecycle.acquiredSandboxes()) {
            await reportRuntimeResource(acquired);
          }
          throwIfCancellationRequested();

          const { workdir: workDir } = loadDaytonaRunnerPathsConfig();
          emitDaytonaStage(callbacks, config.context, 'workdir:resolved', {
            sandboxId: sandbox.id,
            workDir,
          });

          // Session allocation is independent of payload staging. Start it
          // now so Daytona control-plane latency overlaps compression/upload.
          const sessionId = `deepline-play-${randomUUID()}`;
          const detachedRunner = prepareDetachedDaytonaRunner({
            sandbox,
            sessionId,
            readiness: readRunnerReadiness,
            cancellation: cancellationPromise,
          });

          const uploadStartedAt = Date.now();
          let uploadDeadlineTimer: ReturnType<typeof setTimeout> | null = null;
          const uploadDeadlinePromise = new Promise<never>(
            (_resolve, reject) => {
              uploadDeadlineTimer = setTimeout(() => {
                const now = Date.now();
                const timing: DaytonaUploadAttemptTiming = {
                  attempt: executionAttempt,
                  sandboxId: sandbox.id,
                  acquireElapsedMs: uploadStartedAt - attemptStartedAt,
                  uploadElapsedMs: now - uploadStartedAt,
                  attemptElapsedMs: now - attemptStartedAt,
                  totalElapsedMs: now - startedAt,
                };
                uploadDeadlineBreaches.push(timing);
                reject(
                  new DaytonaUploadDeadlineBreachError({
                    deadlineMs: DAYTONA_UPLOAD_ATTEMPT_DEADLINE_MS,
                    attemptTimings: uploadDeadlineBreaches,
                  }),
                );
              }, DAYTONA_UPLOAD_ATTEMPT_DEADLINE_MS);
            },
          );
          let stagedPayload: Awaited<
            ReturnType<typeof stageDaytonaRunnerPayload>
          >;
          try {
            stagedPayload = await Promise.race([
              stageDaytonaRunnerPayload({
                sandbox,
                bundlePromise,
                config,
                workDir,
                startedAt,
                emitStage: (stage, extra) =>
                  emitDaytonaStage(callbacks, config.context, stage, extra),
              }),
              uploadDeadlinePromise,
              cancellationPromise,
            ]);
          } catch (error) {
            if (error instanceof DaytonaUploadDeadlineBreachError) {
              const latestTiming =
                error.attemptTimings[error.attemptTimings.length - 1];
              emitDaytonaStage(callbacks, config.context, 'upload:timeout', {
                sandboxId: sandbox.id,
                attempt: executionAttempt,
                deadlineMs: DAYTONA_UPLOAD_ATTEMPT_DEADLINE_MS,
                uploadElapsedMs: latestTiming?.uploadElapsedMs ?? null,
                attemptElapsedMs: latestTiming?.attemptElapsedMs ?? null,
                elapsedMs: Date.now() - startedAt,
              });
              if (executionAttempt < DAYTONA_UPLOAD_MAX_ATTEMPTS) {
                acquiredSandbox.billingEndedAt = Date.now();
                await reportRetiringRuntimeResource(acquiredSandbox);
                sandboxCleanup.stashActiveSandboxForRetry();
                emitDaytonaStage(callbacks, config.context, 'recreate:start', {
                  previousSandboxId: sandbox.id,
                  attempt: executionAttempt + 1,
                  reason: 'upload_deadline',
                  elapsedMs: Date.now() - startedAt,
                });
                acquiredSandboxPromise = sandboxLifecycle.createFreshSandbox();
                continue;
              }
            }
            throw error;
          } finally {
            if (uploadDeadlineTimer) {
              clearTimeout(uploadDeadlineTimer);
            }
          }
          runtimeTiming.daytonaUploadMs = Date.now() - uploadStartedAt;
          throwIfCancellationRequested();

          emitDaytonaStage(callbacks, config.context, 'execute:start', {
            sandboxId: sandbox.id,
            timeoutSeconds: executionTimeoutSeconds,
            mode: 'session_detached',
          });
          // Push execution (B2-final, park-and-wake): start the runner command
          // DETACHED via a Daytona session and return a `detached_runner`
          // suspension only after the scheduler observes the runner's durable
          // gateway heartbeat. A Daytona cmdId alone does not prove liveness.
          // After that bounded startup check, the worker parks (releasing its
          // claim and slot) and wakes on terminal push or the ceiling timeout.
          let start: Awaited<ReturnType<typeof detachedRunner.start>>;
          try {
            start = await detachedRunner.start({
              command: stagedPayload.command,
              exitCodePath: stagedPayload.exitCodePath,
              startupDiagnosticPath: stagedPayload.startupDiagnosticPath,
            });
          } catch (error) {
            if (
              cancellationRequested ||
              callbacks?.cancellationSignal?.aborted
            ) {
              onCancel();
              throw new Error(DAYTONA_CANCELLED_ERROR);
            }
            // Start never launched customer code; let the infrastructure retry
            // classification below decide fresh-sandbox retry vs loud failure.
            throw error;
          }
          const runnerAttempt = Math.max(
            0,
            Math.floor(config.context.runAttempt ?? 0),
          );
          emitDaytonaStage(callbacks, config.context, 'execute:detached', {
            sandboxId: sandbox.id,
            sessionId: start.sessionId,
            cmdId: start.cmdId,
            runnerAttempt,
            ceilingSeconds: executionTimeoutSeconds,
            terminalGraceSeconds: PLAY_RUNNER_TERMINAL_GRACE_SECONDS,
            outputPath: stagedPayload.outputPath,
            exitCodePath: stagedPayload.exitCodePath,
            elapsedMs: Date.now() - startedAt,
          });
          return {
            status: 'suspended',
            suspension: {
              kind: 'detached_runner',
              boundaryId: `detached-runner:${push.runId}:${runnerAttempt}`,
              runnerAttempt,
              sandboxProvider: 'daytona',
              runtimeSandboxRef: {
                schemaVersion: 1,
                provider: 'daytona',
                resourceId: sandbox.id,
                routingDomain:
                  activeAcquiredResource?.daytonaOrganizationId ?? null,
              },
              sandboxId: sandbox.id,
              sessionId: start.sessionId,
              cmdId: start.cmdId,
              outputPath: stagedPayload.outputPath,
              exitCodePath: stagedPayload.exitCodePath,
              runtimeCompletedPath: stagedPayload.runtimeCompletedPath,
              terminationDiagnosticPath:
                stagedPayload.terminationDiagnosticPath,
              startedAtMs: Date.now(),
              heartbeatTimeoutMs: push.leaseSeconds * 1_000,
              runtimeLimitSeconds: executionTimeoutSeconds,
              ceilingMs:
                (executionTimeoutSeconds + PLAY_RUNNER_TERMINAL_GRACE_SECONDS) *
                1_000,
            },
            logs: [],
            stats: {},
            steps: [],
            checkpoint: config.checkpoint ?? {
              completedBatches: {},
              completedToolBatches: {},
              resolvedWaterfalls: {},
              resolvedBoundaries: {},
            },
            tableNamespace: null,
            runtimeTiming,
          };
        } catch (error) {
          if (cancellationRequested || callbacks?.cancellationSignal?.aborted) {
            onCancel();
            throw new Error(DAYTONA_CANCELLED_ERROR);
          }
          const retryReason = isRetryableUnstartedDaytonaRunner(error)
            ? 'runner_startup_not_live'
            : isRetryableDaytonaInfrastructureFailure(error)
              ? 'daytona_infrastructure'
              : null;
          if (
            executionAttempt < DAYTONA_INFRASTRUCTURE_MAX_ATTEMPTS &&
            retryReason
          ) {
            emitDaytonaStage(callbacks, config.context, 'execute:retry', {
              sandboxId: sandboxForAttempt?.id ?? null,
              attempt: executionAttempt + 1,
              reason: retryReason,
              error: error instanceof Error ? error.message : String(error),
              elapsedMs: Date.now() - startedAt,
            });
            if (sandboxForAttempt) {
              activeAcquiredResource!.billingEndedAt = Date.now();
              await reportRetiringRuntimeResource(activeAcquiredResource!);
              sandboxCleanup.stashActiveSandboxForRetry();
            }
            acquiredSandboxPromise = sandboxLifecycle.createFreshSandbox();
            continue;
          }
          throw error;
        }
      }

      return withCleanup(
        createDaytonaFailedResult({
          config,
          error: 'Daytona play runner exhausted retry attempts.',
          runtimeTiming,
        }),
      );
    } catch (error) {
      // Resource persistence belongs to the scheduler control plane. Preserve
      // its typed capacity/fence errors so Absurd can defer or fence the
      // attempt; converting them into a runner failure would terminally fail a
      // play that never began executing customer code. The sandbox exists
      // before that durable callback can succeed, so synchronously delete it
      // before returning the control-plane error. The old fire-and-forget
      // cleanup path was skipped by this rethrow and leaked the acquisition.
      if (runtimeResourceReportErrors.has(error)) {
        const sandbox = sandboxCleanup.currentSandbox();
        if (sandbox) {
          try {
            await sandbox.delete(30);
            console.info(
              '[play-runner.daytona.resource_report_failure_cleanup_done]',
              { sandboxId: sandbox.id },
            );
          } catch (cleanupError) {
            throw new AggregateError(
              [error, cleanupError],
              `Failed to persist or delete acquired Daytona sandbox ${sandbox.id}.`,
            );
          }
        }
        throw error;
      }
      if (isRuntimeSandboxCapacityLimitError(error)) {
        throw error;
      }
      if (error instanceof DaytonaSandboxAcquisitionUnavailableError) {
        throw error;
      }
      emitDaytonaStage(callbacks, config.context, 'execute:error', {
        sandboxId: sandboxCleanup.currentSandbox()?.id ?? null,
        runnerAttempt: Math.max(0, Math.floor(config.context.runAttempt ?? 0)),
        error: formatDaytonaError(error),
        operation: 'runner_backend',
        elapsedMs: Date.now() - startedAt,
      });
      return withCleanup(
        createDaytonaFailedResult({
          config,
          error: formatDaytonaError(error),
          runtimeTiming,
        }),
      );
    } finally {
      callbacks?.cancellationSignal?.removeEventListener('abort', onCancel);
      if (cancellationRequested) {
        await sandboxLifecycle.settlePendingCreates();
        for (const acquired of sandboxLifecycle.acquiredSandboxes()) {
          try {
            await reportRuntimeResource(acquired);
          } catch (error) {
            console.error(
              '[play-runner.daytona.cancel_resource_report_failed]',
              {
                sandboxId: acquired.sandbox.id,
                error: error instanceof Error ? error.message : String(error),
              },
            );
          }
        }
      }
    }
  },
};