import { randomUUID } from 'node:crypto';
import { readFile } from 'node:fs/promises';
import { gzipSync } from 'node:zlib';
import {
  RUNNER_TERMINAL_CHECKPOINT_DISPOSITION,
  type PlayRunnerExecutionConfig,
} from '@shared_libs/play-runtime/protocol';
import {
  PLAY_RUNTIME_CONTRACT,
  PLAY_RUNTIME_CONTRACT_HEADER,
} from '@shared_libs/play-runtime/runtime-contract';
import {
  PLAY_RUNNER_TERMINAL_GRACE_SECONDS,
  PLAY_RUNNER_STARTUP_GRACE_SECONDS,
  STANDARD_PLAY_RUNTIME_LIMIT_SECONDS,
} from '@shared_libs/play-runtime/runtime-constants';
import {
  RUNTIME_RELIABILITY_ENV,
  RUNTIME_RELIABILITY_SEV_OVERRIDE_ENV,
} from '@shared_libs/play-runtime/runtime-reliability-policy';
import {
  STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS,
  validatePlaySandboxRuntimeLimits,
} from '@shared_libs/play-runtime/sandbox-runtime-limits';
import {
  RUNNER_POST_TERMINAL_DIAGNOSTIC_MAX_BYTES,
  RUNNER_TERMINAL_PUSH_MAX_BODY_BYTES,
} from '@shared_libs/play-runtime/output-size-limits';
import {
  RUNTIME_SANDBOX_KILLED_MESSAGE,
  RUNTIME_SANDBOX_OOM_MESSAGE,
} from '@shared_libs/play-runtime/run-failure';
import { compactPlayArtifactForRuntimeTransport } from '@shared_libs/plays/artifact-transport';
import type {
  DaytonaExecutionContext,
  DaytonaSandbox,
  DaytonaStageEmitter,
} from './daytona-lifecycle';
import { buildDaytonaRuntimeWatchdogSource } from './daytona-runtime-watchdog';

type DaytonaRemoteRuntimeContext = PlayRunnerExecutionConfig['context'];

type DaytonaPayloadEnvelope = {
  schemaVersion: 1;
  runnerCode: string;
  artifactBundledCode: string;
  artifactSourceMap: string;
  config: PlayRunnerExecutionConfig;
  /** Crash-containment epilogue script (see buildDaytonaCrashTerminalPusherSource). */
  crashPusherCode: string;
};

/**
 * The crash pusher must be able to recover one complete terminal event after
 * a runner exits before its primary gateway push. A stdout `result` event is
 * smaller than the accepted runner-terminal request envelope (the request also
 * carries action + runId), so this covers every terminal that the gateway can
 * have accepted, including a resume-required suspended checkpoint. Leave room
 * for the runner's primary-push retry diagnostics written after that event.
 */
export const DAYTONA_CRASH_PUSHER_POST_EVENT_DIAGNOSTIC_HEADROOM_BYTES =
  RUNNER_POST_TERMINAL_DIAGNOSTIC_MAX_BYTES;
export const DAYTONA_CRASH_PUSHER_OUTPUT_TAIL_MAX_BYTES =
  RUNNER_TERMINAL_PUSH_MAX_BODY_BYTES +
  DAYTONA_CRASH_PUSHER_POST_EVENT_DIAGNOSTIC_HEADROOM_BYTES;

export type StagedDaytonaPayload = {
  workDir: string;
  command: string;
  outputPath: string;
  exitCodePath: string;
  runtimeCompletedPath: string;
  terminationDiagnosticPath: string;
  progressEventPath: string;
  startupDiagnosticPath: string;
};

export type RemoteRunnerPayloadSandbox = {
  id: string;
  uploadFile(content: Buffer, path: string): Promise<void>;
};

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

/**
 * Sandboxes intentionally receive no general worker environment. The bounded,
 * non-secret reliability controls are the exception: copy only their declared
 * names so an incident override affects the runner's gateway/egress deadlines
 * too, without ever widening this into an environment pass-through.
 */
function runtimeReliabilityEnvironmentPrefix(): string {
  const names = [
    RUNTIME_RELIABILITY_SEV_OVERRIDE_ENV,
    ...Object.values(RUNTIME_RELIABILITY_ENV),
  ];
  const entries = names.flatMap((name) => {
    const value = process.env[name];
    return value === undefined || value.trim() === ''
      ? []
      : [`${name}=${shellQuote(value)}`];
  });
  return entries.length > 0 ? `${entries.join(' ')} ` : '';
}

function gzipUtf8(value: string): Buffer {
  return gzipSync(Buffer.from(value, 'utf-8'));
}

function gzipJson(value: DaytonaPayloadEnvelope): Buffer {
  return gzipUtf8(JSON.stringify(value));
}

function nodeMaterializePayloadCommand(input: {
  envelopePath: string;
  runnerPath: string;
  configPath: string;
  artifactCodePath: string;
  artifactSourceMapPath: string;
  crashPusherPath: string;
}): string {
  const script =
    "const fs=require('node:fs');const zlib=require('node:zlib');const p=JSON.parse(zlib.gunzipSync(fs.readFileSync(process.argv[1])));if(p.schemaVersion!==1)throw new Error('Unsupported Daytona payload envelope');fs.writeFileSync(process.argv[2],p.runnerCode);fs.writeFileSync(process.argv[3],JSON.stringify(p.config));fs.writeFileSync(process.argv[4],p.artifactBundledCode);fs.writeFileSync(process.argv[5],p.artifactSourceMap);fs.writeFileSync(process.argv[6],p.crashPusherCode);";
  return [
    'node',
    '-e',
    shellQuote(script),
    shellQuote(input.envelopePath),
    shellQuote(input.runnerPath),
    shellQuote(input.configPath),
    shellQuote(input.artifactCodePath),
    shellQuote(input.artifactSourceMapPath),
    shellQuote(input.crashPusherPath),
  ].join(' ');
}

/**
 * Crash-containment epilogue (push execution). The parked worker no longer
 * holds anything that returns when the runner process dies, so a runner that
 * exits WITHOUT pushing its terminal (play code calling `process.exit`, OOM
 * SIGKILL of the node process, a crash before the in-process push) would only
 * be detected by the ceiling timeout. The staged shell command therefore runs
 * this tiny script AFTER the runner exits, unconditionally:
 *
 *  1. parse the captured output for the structured `result` event (the runner
 *     may have printed it even when its own gateway push failed), and
 *  2. push it — or a synthesized crash failure naming the exit code — to the
 *     gateway's `runner_terminal` action.
 *
 * The gateway records terminals FIRST-WRITE-WINS and absurd wake events are
 * first-emit-wins, so a duplicate push after a successful in-runner push is a
 * no-op; this needs no coordination with the runner.
 *
 * Exported (with the builder) for the unit test that executes the script
 * against a real local HTTP server.
 */
export function buildDaytonaCrashTerminalPusherSource(
  runtimeLimitSeconds = STANDARD_PLAY_RUNTIME_LIMIT_SECONDS,
): string {
  const runtimeLimitMessage = `The play reached its configured ${runtimeLimitSeconds} second runtime limit and was stopped.`;
  const runtimeLimitDetail = `${runtimeLimitMessage} Completed row state was preserved; run a smaller batch or continue from the persisted rows.`;
  return `
const fs = require('node:fs');
const configPath = process.argv[2];
const exitCodeRaw = Number.parseInt(process.argv[3] ?? '', 10);
const outputPath = process.argv[4];
const runtimeLimitMarkerPath = process.argv[5] || '';
const oomKillBaselinePath = process.argv[6] || '';
const memoryEventsPath = process.argv[7] || '/sys/fs/cgroup/memory.events';
const terminationDiagnosticPath = process.argv[8] || '';
const exitCode = Number.isFinite(exitCodeRaw) ? exitCodeRaw : null;
const MAX_OUTPUT_BYTES = ${DAYTONA_CRASH_PUSHER_OUTPUT_TAIL_MAX_BYTES};
const MAX_POST_TERMINAL_DIAGNOSTIC_BYTES = ${RUNNER_POST_TERMINAL_DIAGNOSTIC_MAX_BYTES};
let postTerminalDiagnosticBytesRemaining = MAX_POST_TERMINAL_DIAGNOSTIC_BYTES;

function truncateUtf8(value, maxBytes) {
  if (Buffer.byteLength(value, 'utf8') <= maxBytes) return value;
  const encoded = Buffer.from(value, 'utf8');
  for (let end = Math.min(encoded.length, maxBytes); end > 0; end -= 1) {
    const candidate = encoded.subarray(0, end).toString('utf8');
    if (Buffer.byteLength(candidate, 'utf8') <= maxBytes) return candidate;
  }
  return '';
}

function writeDiagnostic(line) {
  if (postTerminalDiagnosticBytesRemaining <= 1) return;
  const bounded = truncateUtf8(
    String(line),
    postTerminalDiagnosticBytesRemaining - 1,
  );
  if (!bounded) return;
  console.log(bounded);
  postTerminalDiagnosticBytesRemaining -= Buffer.byteLength(bounded, 'utf8') + 1;
}

function readResultFromOutput() {
  try {
    const stat = fs.statSync(outputPath);
    const start = Math.max(0, stat.size - MAX_OUTPUT_BYTES);
    const fd = fs.openSync(outputPath, 'r');
    const buffer = Buffer.alloc(stat.size - start);
    fs.readSync(fd, buffer, 0, buffer.length, start);
    fs.closeSync(fd);
    const lines = buffer.toString('utf-8').split(/\\r?\\n/);
    for (let index = lines.length - 1; index >= 0; index -= 1) {
      const line = lines[index].trim();
      if (!line.startsWith('{')) continue;
      try {
        const event = JSON.parse(line);
        if (
          event &&
          event.type === 'result' &&
          event.result &&
          ['completed', 'failed', 'suspended'].includes(event.result.status)
        ) {
          return event.result;
        }
      } catch {}
    }
  } catch {}
  return null;
}

function outputHasExplicitOomSignature() {
  try {
    const stat = fs.statSync(outputPath);
    const start = Math.max(0, stat.size - MAX_OUTPUT_BYTES);
    const fd = fs.openSync(outputPath, 'r');
    const buffer = Buffer.alloc(stat.size - start);
    fs.readSync(fd, buffer, 0, buffer.length, start);
    fs.closeSync(fd);
    return /javascript heap out of memory|fatal error:.*(?:heap|allocation).*memory/i.test(buffer.toString('utf-8'));
  } catch {}
  return false;
}

function cgroupOomKillObserved() {
  try {
    if (!oomKillBaselinePath) return false;
    const baseline = Number.parseInt(fs.readFileSync(oomKillBaselinePath, 'utf8').trim(), 10);
    const events = fs.readFileSync(memoryEventsPath, 'utf8');
    const match = events.match(/^oom_kill\\s+(\\d+)$/m);
    const current = match ? Number.parseInt(match[1], 10) : Number.NaN;
    return Number.isFinite(baseline) && Number.isFinite(current) && current > baseline;
  } catch {}
  return false;
}

function watchdogObservedChildSigkill() {
  try {
    if (!terminationDiagnosticPath) return false;
    const diagnostic = JSON.parse(
      fs.readFileSync(terminationDiagnosticPath, 'utf8'),
    );
    return Boolean(
      diagnostic &&
        diagnostic.schemaVersion === 1 &&
        diagnostic.reason === 'child_exit' &&
        diagnostic.childExitCode === null &&
        diagnostic.childSignal === 'SIGKILL',
    );
  } catch {}
  return false;
}

function terminalResultForTransport(result) {
  if (
    result &&
    ${JSON.stringify(Object.values(RUNNER_TERMINAL_CHECKPOINT_DISPOSITION))}.includes(
      result.checkpointDisposition,
    )
  ) {
    return result;
  }
  if (result && result.status === 'suspended') {
    return {
      ...result,
      checkpointDisposition: '${RUNNER_TERMINAL_CHECKPOINT_DISPOSITION.INCLUDED_RESUME_REQUIRED}',
      checkpointBytes: Buffer.byteLength(JSON.stringify(result.checkpoint)),
    };
  }
  const checkpoint = result && result.checkpoint;
  const terminal = { ...(result || {}) };
  delete terminal.checkpoint;
  if (checkpoint === null || checkpoint === undefined) {
    return {
      ...terminal,
      checkpointDisposition: '${RUNNER_TERMINAL_CHECKPOINT_DISPOSITION.ABSENT}',
      checkpointBytes: 0,
    };
  }
  return {
    ...terminal,
    checkpointDisposition: '${RUNNER_TERMINAL_CHECKPOINT_DISPOSITION.OMITTED_TERMINAL_REPLAY}',
    checkpointBytes: Buffer.byteLength(JSON.stringify(checkpoint)),
  };
}

async function main() {
  const config = JSON.parse(fs.readFileSync(configPath, 'utf-8'));
  const context = (config && config.context) || {};
  const push = context.runnerPushExecution;
  const gateway = String(context.receiptGatewayBaseUrl || '').replace(/\\/$/, '');
  const token = context.executorToken;
  if (!push || !push.runId || !gateway || !token) {
    writeDiagnostic('[crash-terminal] no push config; skipping');
    return;
  }
  const parsedResult = readResultFromOutput();
  // Exit codes are authored-program output too: a play may intentionally call
  // process.exit(124). Only the separate parent watchdog writes this marker.
  const runtimeLimitExceeded = Boolean(
    runtimeLimitMarkerPath && fs.existsSync(runtimeLimitMarkerPath),
  );
  // 137 is SIGKILL, not proof of an OOM. A watchdog, platform eviction, or
  // operator action can produce the same status. Only label this an OOM when
  // the captured runtime output carries an explicit V8 allocation signature.
  // The shell sees the watchdog's exit code, not necessarily the nested
  // runner child's signal. Read the watchdog's strict marker so a child
  // SIGKILL does not collapse into the wrapper's generic exit code 1.
  const sandboxKilled = exitCode === 137 || watchdogObservedChildSigkill();
  const sandboxOom =
    sandboxKilled && (outputHasExplicitOomSignature() || cgroupOomKillObserved());
  const synthesizedError = runtimeLimitExceeded
    ? 'RUNTIME_LIMIT_EXCEEDED: ' + ${JSON.stringify(runtimeLimitMessage)}
    : sandboxOom
      ? 'RUNTIME_SANDBOX_OOM: ' + ${JSON.stringify(RUNTIME_SANDBOX_OOM_MESSAGE)}
      : sandboxKilled
        ? 'RUNTIME_SANDBOX_KILLED: ' + ${JSON.stringify(RUNTIME_SANDBOX_KILLED_MESSAGE)}
    :
      'Daytona play runner exited with code ' +
      (exitCode === null ? 'unknown' : exitCode) +
      ' without pushing a terminal (crash containment epilogue).';
  const result = terminalResultForTransport(parsedResult || {
    status: 'failed',
    error: synthesizedError,
    errors: runtimeLimitExceeded
      ? [{
          code: 'RUNTIME_LIMIT_EXCEEDED',
          phase: 'runtime',
          message: ${JSON.stringify(runtimeLimitDetail)},
          retryable: false,
          cause: synthesizedError,
        }]
      : undefined,
    ...(sandboxOom
      ? { errors: [{
          code: 'RUNTIME_SANDBOX_OOM',
          phase: 'infrastructure',
          message: ${JSON.stringify(RUNTIME_SANDBOX_OOM_MESSAGE)},
          retryable: false,
          cause: synthesizedError,
        }] }
      : {}),
    ...(sandboxKilled && !sandboxOom
      ? { errors: [{
          code: 'RUNTIME_SANDBOX_KILLED',
          phase: 'infrastructure',
          message: ${JSON.stringify(RUNTIME_SANDBOX_KILLED_MESSAGE)},
          retryable: true,
          cause: synthesizedError,
        }] }
      : {}),
    logs: [],
    stats: {},
    steps: [],
    checkpointDisposition: '${RUNNER_TERMINAL_CHECKPOINT_DISPOSITION.ABSENT}',
    checkpointBytes: 0,
    tableNamespace: null,
  });
  for (let attempt = 1; attempt <= 3; attempt += 1) {
    try {
      const controller = new AbortController();
      const timer = setTimeout(() => controller.abort(), 10000);
      const response = await fetch(gateway + '/api/v2/plays/runtime', {
        method: 'POST',
        headers: {
          'content-type': 'application/json',
          authorization: 'Bearer ' + token,
          '${PLAY_RUNTIME_CONTRACT_HEADER}': '${String(PLAY_RUNTIME_CONTRACT)}',
        },
        body: JSON.stringify({
          action: 'runner_terminal',
          runId: push.runId,
          result,
        }),
        signal: controller.signal,
        redirect: 'manual',
      });
      clearTimeout(timer);
      if (response.ok) {
        writeDiagnostic('[crash-terminal] pushed status=' + result.status);
        return;
      }
      writeDiagnostic('[crash-terminal] rejected http=' + response.status);
      // Deterministic scope/fence rejections cannot succeed on retry.
      if (response.status >= 400 && response.status < 500) return;
    } catch (error) {
      writeDiagnostic(
        '[crash-terminal] transport ' + (error && error.message ? error.message : String(error)),
      );
    }
    await new Promise((resolveDelay) => setTimeout(resolveDelay, attempt * 1000));
  }
  writeDiagnostic('[crash-terminal] exhausted retries');
}

main().catch((error) => {
  writeDiagnostic('[crash-terminal] fatal ' + (error && error.message ? error.message : String(error)));
});
`;
}

function remoteRuntimeContextForDaytona(
  context: DaytonaExecutionContext,
): DaytonaRemoteRuntimeContext {
  const gatewayBaseUrl = context.receiptGatewayBaseUrl?.trim();
  if (!gatewayBaseUrl) {
    throw new Error(
      'Daytona execution requires DEEPLINE_RUNTIME_RECEIPT_GATEWAY_URL. Refusing to put app credentials or database sessions in the customer-code sandbox.',
    );
  }
  const executionGatewayBaseUrl =
    context.executionGatewayBaseUrl?.trim() || gatewayBaseUrl;
  return {
    ...context,
    // The generic runtime transport owns receipts, runner terminal publication,
    // heartbeats, sheets, and runtime control, so it must stay on the receipt
    // gateway. Only ctx.tools.execute selects executionGatewayBaseUrl below.
    // Otherwise a receipt gateway rollout silently moves all durable control
    // traffic to the relay and can leave a run parked after its sandbox starts.
    baseUrl: gatewayBaseUrl,
    executionGatewayBaseUrl,
    receiptGatewayBaseUrl: gatewayBaseUrl,
    // The execution relay owns the Postgres-backed invocation fence even though
    // receipt control traffic has a different origin. This must stay explicit:
    // origin equality was only a legacy shortcut for the combined gateway.
    requestDurableInvocationFence: true,
    durableInvocationFence: true,
    vercelProtectionBypassToken: null,
    dbSessionStrategy: 'gateway_only',
    preloadedDbSessions: undefined,
    postgresSessionUnwrapKey: undefined,
  };
}

async function timedRunnerPayloadUpload(input: {
  sandbox: RemoteRunnerPayloadSandbox;
  emitStage: DaytonaStageEmitter;
  label: string;
  path: string;
  content: Buffer;
}) {
  const startedAt = Date.now();
  await input.sandbox.uploadFile(input.content, input.path);
  input.emitStage(`upload:${input.label}:done`, {
    bytes: input.content.byteLength,
    elapsedMs: Date.now() - startedAt,
  });
}

export async function stageRunnerPayload(input: {
  sandbox: RemoteRunnerPayloadSandbox;
  bundlePromise: Promise<string>;
  config: PlayRunnerExecutionConfig;
  workDir: string;
  startedAt: number;
  emitStage: DaytonaStageEmitter;
}): Promise<StagedDaytonaPayload> {
  const workDir = input.workDir.replace(/\/$/, '');
  const runnerPath = `${workDir}/deepline-play-runner-${randomUUID()}.cjs`;
  const envelopePath = `${workDir}/deepline-play-payload-${randomUUID()}.json.gz`;
  const configPath = `${workDir}/deepline-play-config-${randomUUID()}.json`;
  const artifactCodePath = `${workDir}/deepline-play-artifact-${randomUUID()}.cjs`;
  const artifactSourceMapPath = `${artifactCodePath}.map`;
  const crashPusherPath = `${workDir}/deepline-play-crash-terminal-${randomUUID()}.cjs`;

  const remoteMaterializedFiles: Record<string, string> = {};
  const hasInlineCsv = Boolean(input.config.csvSourceContentBase64);
  input.emitStage('upload:start', {
    sandboxId: input.sandbox.id,
    materializedFileCount: Object.keys(input.config.materializedFiles).length,
    hasCsv: Boolean(
      input.config.csvSourceContentBase64 ||
      input.config.csvSourcePath ||
      input.config.csvSourceUrl,
    ),
  });

  const compactedArtifact = compactPlayArtifactForRuntimeTransport(
    input.config.artifact,
  );
  const uniqueLocalPaths = new Map<string, string>();
  for (const localPath of Object.values(input.config.materializedFiles)) {
    if (hasInlineCsv && localPath === input.config.csvSourcePath) {
      continue;
    }
    if (!uniqueLocalPaths.has(localPath)) {
      uniqueLocalPaths.set(localPath, `${workDir}/files/${randomUUID()}`);
    }
  }
  if (
    input.config.csvSourcePath &&
    !hasInlineCsv &&
    !uniqueLocalPaths.has(input.config.csvSourcePath)
  ) {
    uniqueLocalPaths.set(
      input.config.csvSourcePath,
      `${workDir}/files/${randomUUID()}`,
    );
  }

  await Promise.all(
    [...uniqueLocalPaths.entries()].map(async ([localPath, remotePath]) => {
      await timedRunnerPayloadUpload({
        sandbox: input.sandbox,
        emitStage: input.emitStage,
        label:
          localPath === input.config.csvSourcePath
            ? 'csv'
            : 'materialized_file',
        path: remotePath,
        content: await readFile(localPath),
      });
    }),
  );

  for (const [logicalPath, localPath] of Object.entries(
    input.config.materializedFiles,
  )) {
    const remotePath = uniqueLocalPaths.get(localPath);
    if (remotePath) {
      remoteMaterializedFiles[logicalPath] = remotePath;
    }
  }

  const remoteConfig: PlayRunnerExecutionConfig = {
    ...input.config,
    artifact: {
      ...compactedArtifact,
      bundledCode: '',
    },
    artifactTransport: {
      bundledCodePath: artifactCodePath,
      bundledCodeEncoding: 'utf8',
      sourceMapPath: artifactSourceMapPath,
    },
    workspaceRoot: input.workDir,
    csvSourcePath: hasInlineCsv
      ? null
      : input.config.csvSourcePath
        ? (uniqueLocalPaths.get(input.config.csvSourcePath) ??
          input.config.csvSourcePath)
        : null,
    csvSourceUrl: input.config.csvSourceUrl ?? null,
    csvSourceContentBase64: input.config.csvSourceContentBase64 ?? null,
    materializedFiles: remoteMaterializedFiles,
    context: remoteRuntimeContextForDaytona(input.config.context),
  };
  const runtimeLimitSeconds = validatePlaySandboxRuntimeLimits(
    input.config.context.sandboxRuntimeLimits ?? {
      ...STANDARD_PLAY_SANDBOX_RUNTIME_LIMITS,
    },
  ).timeoutSeconds;

  const envelopeUpload = input.bundlePromise.then((bundle) =>
    timedRunnerPayloadUpload({
      sandbox: input.sandbox,
      emitStage: input.emitStage,
      label: 'payload',
      path: envelopePath,
      content: gzipJson({
        schemaVersion: 1,
        runnerCode: bundle,
        artifactBundledCode: compactedArtifact.bundledCode,
        artifactSourceMap: input.config.artifact.sourceMap,
        config: remoteConfig,
        crashPusherCode:
          buildDaytonaCrashTerminalPusherSource(runtimeLimitSeconds),
      }),
    }),
  );
  await envelopeUpload;
  input.emitStage('upload:done', {
    sandboxId: input.sandbox.id,
    uploadedFileCount: uniqueLocalPaths.size + 1,
    elapsedMs: Date.now() - input.startedAt,
  });

  const outputPath = `${workDir}/deepline-play-output-${randomUUID()}.log`;
  const crashPusherLogPath = `${workDir}/deepline-play-crash-terminal-${randomUUID()}.log`;
  const exitCodePath = `${workDir}/deepline-play-exit-${randomUUID()}.txt`;
  const oomKillBaselinePath = `${workDir}/deepline-play-oom-kill-${randomUUID()}.txt`;
  const runtimeStartedPath = `${workDir}/deepline-play-runtime-started-${randomUUID()}.json`;
  const runtimeCompletedPath = `${workDir}/deepline-play-runtime-completed-${randomUUID()}.json`;
  const runtimeLimitMarkerPath = `${workDir}/deepline-play-runtime-limit-${randomUUID()}.txt`;
  const terminationDiagnosticPath = `${workDir}/deepline-play-termination-${randomUUID()}.json`;
  const progressEventPath = `${workDir}/deepline-play-progress-${randomUUID()}.jsonl`;
  const startupDiagnosticPath = `${workDir}/deepline-play-startup-${randomUUID()}.json`;
  const runnerTraceEnv =
    process.env.DEEPLINE_RUNTIME_RECEIPT_TRACE === '1'
      ? 'DEEPLINE_RUNTIME_RECEIPT_TRACE=1 '
      : '';
  const runtimeReliabilityEnv = runtimeReliabilityEnvironmentPrefix();
  const watchedRunnerCommand = [
    'node',
    '-e',
    shellQuote(buildDaytonaRuntimeWatchdogSource()),
    shellQuote(runnerPath),
    shellQuote(configPath),
    shellQuote(runtimeStartedPath),
    shellQuote(runtimeCompletedPath),
    String(runtimeLimitSeconds * 1_000),
    String(PLAY_RUNNER_STARTUP_GRACE_SECONDS * 1_000),
    String(PLAY_RUNNER_TERMINAL_GRACE_SECONDS * 1_000),
    shellQuote(runtimeLimitMarkerPath),
    shellQuote(terminationDiagnosticPath),
  ].join(' ');
  const runnerCommand = `${nodeMaterializePayloadCommand({
    envelopePath,
    runnerPath,
    configPath,
    artifactCodePath,
    artifactSourceMapPath,
    crashPusherPath,
  })} && DEEPLINE_PLAY_RUNNER_RUNTIME_STARTED_PATH=${shellQuote(runtimeStartedPath)} DEEPLINE_PLAY_RUNNER_RUNTIME_COMPLETED_PATH=${shellQuote(runtimeCompletedPath)} DEEPLINE_PLAY_RUNNER_PROGRESS_EVENT_PATH=${shellQuote(progressEventPath)} DEEPLINE_PLAY_RUNNER_STARTUP_DIAGNOSTIC_PATH=${shellQuote(startupDiagnosticPath)} DEEPLINE_PLAY_RUNNER_STDOUT_EVENT_MODE=compact ${runtimeReliabilityEnv}${runnerTraceEnv}${watchedRunnerCommand}`;
  // Crash-containment epilogue: runs UNCONDITIONALLY after the runner exits and
  // pushes the parsed (or synthesized) terminal to the gateway so the parked
  // worker wakes within seconds of ANY runner death — process.exit abuse, OOM
  // SIGKILL, or a crash before the in-process push. Idempotent against a
  // successful in-runner push (terminals are first-write-wins).
  // Its own retry diagnostics are kept in a separate log: appending them to
  // `outputPath` after a near-gateway-sized result could hide that result from
  // a later bounded-tail recovery read.
  const command = `rm -f ${shellQuote(outputPath)} ${shellQuote(crashPusherLogPath)} ${shellQuote(exitCodePath)} ${shellQuote(oomKillBaselinePath)} ${shellQuote(runtimeStartedPath)} ${shellQuote(runtimeCompletedPath)} ${shellQuote(runtimeLimitMarkerPath)} ${shellQuote(terminationDiagnosticPath)} ${shellQuote(progressEventPath)} ${shellQuote(`${progressEventPath}.*`)} ${shellQuote(startupDiagnosticPath)}; ( awk '$1 == "oom_kill" { print $2 }' /sys/fs/cgroup/memory.events 2>/dev/null || true ) > ${shellQuote(oomKillBaselinePath)}; ( ${runnerCommand} ) > ${shellQuote(outputPath)} 2>&1; code=$?; printf '%s' "$code" > ${shellQuote(exitCodePath)}; ${runtimeReliabilityEnv}node ${shellQuote(crashPusherPath)} ${shellQuote(configPath)} "$code" ${shellQuote(outputPath)} ${shellQuote(runtimeLimitMarkerPath)} ${shellQuote(oomKillBaselinePath)} /sys/fs/cgroup/memory.events ${shellQuote(terminationDiagnosticPath)} > ${shellQuote(crashPusherLogPath)} 2>&1 || true; printf 'deepline runner output captured: %s\\n' ${shellQuote(outputPath)}; exit "$code"`;

  return {
    workDir: input.workDir,
    command,
    outputPath,
    exitCodePath,
    runtimeCompletedPath,
    terminationDiagnosticPath,
    progressEventPath,
    startupDiagnosticPath,
  };
}

export async function stageDaytonaRunnerPayload(
  input: Omit<Parameters<typeof stageRunnerPayload>[0], 'sandbox'> & {
    sandbox: DaytonaSandbox;
  },
): Promise<StagedDaytonaPayload> {
  return await stageRunnerPayload({
    ...input,
    sandbox: {
      id: input.sandbox.id,
      uploadFile: (content, path) => input.sandbox.fs.uploadFile(content, path),
    },
  });
}