import { z } from "zod";
import { MachinePresetName } from "@trigger.dev/core/v3";
import { BoolEnv } from "./utils/boolEnv";
import { isValidDatabaseUrl } from "./utils/db";
import { isValidRegex } from "./utils/regex";
import { isValidDuration } from "./services/realtime/duration.server";

// `z.string()` constrained to a `parseDuration`-parseable string (e.g.
// `7d`, `1h`). Validated at boot so a typo'd duration fails fast.
function durationString() {
  return z.string().refine(isValidDuration, "must be a duration like 7d, 30d, 365d, 1h, 1y");
}

// Parses a CSV of machine preset names (e.g. "small-1x,small-2x") into a
// non-empty array of MachinePresetName. Used by COMPUTE_TEMPLATE_MACHINE_PRESETS
// and its _REQUIRED variant. Adds zod issues for empty input or unknown names.
const parseMachinePresetCsv = (raw: string, ctx: z.RefinementCtx): MachinePresetName[] => {
  const names = raw
    .split(",")
    .map((s) => s.trim())
    .filter(Boolean);
  if (names.length === 0) {
    ctx.addIssue({
      code: z.ZodIssueCode.custom,
      message: "must list at least one machine preset",
    });
    return z.NEVER;
  }
  const out: MachinePresetName[] = [];
  for (const name of names) {
    const parsed = MachinePresetName.safeParse(name);
    if (!parsed.success) {
      ctx.addIssue({
        code: z.ZodIssueCode.custom,
        message: `unknown machine preset: "${name}"`,
      });
      return z.NEVER;
    }
    out.push(parsed.data);
  }
  return out;
};

const GithubAppEnvSchema = z.preprocess(
  (val) => {
    const obj = val as any;
    if (!obj || !obj.GITHUB_APP_ENABLED) {
      return { ...obj, GITHUB_APP_ENABLED: "0" };
    }
    return obj;
  },
  z.discriminatedUnion("GITHUB_APP_ENABLED", [
    z.object({
      GITHUB_APP_ENABLED: z.literal("1"),
      GITHUB_APP_ID: z.string(),
      GITHUB_APP_PRIVATE_KEY: z.string(),
      GITHUB_APP_WEBHOOK_SECRET: z.string(),
      GITHUB_APP_SLUG: z.string(),
    }),
    z.object({
      GITHUB_APP_ENABLED: z.literal("0"),
    }),
  ])
);

// eventually we can make all S2 env vars required once the S2 OSS version is out
const S2EnvSchema = z.preprocess(
  (val) => {
    const obj = val as any;
    if (!obj || !obj.S2_ENABLED) {
      return { ...obj, S2_ENABLED: "0" };
    }
    return obj;
  },
  z.discriminatedUnion("S2_ENABLED", [
    z.object({
      S2_ENABLED: z.literal("1"),
      S2_ACCESS_TOKEN: z.string(),
      S2_DEPLOYMENT_LOGS_BASIN_NAME: z.string(),
      S2_DEPLOYMENT_STREAMS_LOCAL: z.string().default("0"),
    }),
    z.object({
      S2_ENABLED: z.literal("0"),
    }),
  ])
);

// Previously published secret values must never be accepted, including when
// an existing deployment or external secret manager still supplies one.
const INSECURE_SECRET_VALUES = [
  "managed-secret",
  "2818143646516f6fffd707b36f334bbb",
  "44da78b7bbb0dfe709cf38931d25dcdd",
  "f686147ab967943ebbe9ed3b496e465a",
  "447c29678f9eaf289e9c4b70d3dd8a7f",
];

// Escape hatch for deployments that can't rotate a published default yet (e.g.
// ENCRYPTION_KEY protects existing data). Read raw: a refine can't see the
// sibling parsed flag.
const allowInsecureDefaultSecrets = ["true", "1"].includes(
  (process.env.ALLOW_INSECURE_DEFAULT_SECRETS ?? "").toLowerCase().trim()
);

const isNotInsecureSecret = (value: string) =>
  allowInsecureDefaultSecrets || !INSECURE_SECRET_VALUES.includes(value);

const INSECURE_SECRET_MESSAGE =
  "must not be a known-insecure published default; set a strong, unique value. If you cannot rotate it yet (e.g. it protects existing encrypted data or active sessions), set ALLOW_INSECURE_DEFAULT_SECRETS=1 to boot while you migrate.";

/** Optional int env var; blank/whitespace normalises to undefined (z.coerce turns "" into 0). */
const OptionalIntEnv = z.preprocess(
  (v) => (typeof v === "string" && v.trim() === "" ? undefined : v),
  z.coerce.number().int().optional()
);

/** Optional boolean env var; blank/whitespace/unset normalises to undefined (so it falls back). */
const OptionalBoolEnv = z.preprocess((v) => {
  if (typeof v !== "string" || v.trim() === "") return undefined;
  return ["true", "1"].includes(v.toLowerCase().trim());
}, z.boolean().optional());

/** Boolean env var with a default where blank/whitespace falls back to the default instead of parsing as false. */
const BoolEnvWithDefault = (defaultValue: boolean) =>
  z.preprocess((v) => {
    if (typeof v !== "string" || v.trim() === "") return undefined;
    return ["true", "1"].includes(v.toLowerCase().trim());
  }, z.boolean().default(defaultValue));

/** Int env var with a default where a blank/whitespace value falls back to the default instead of coercing to 0. */
const IntEnvWithDefault = (defaultValue: number) =>
  z.preprocess(
    (v) => (typeof v === "string" && v.trim() === "" ? undefined : v),
    z.coerce.number().int().default(defaultValue)
  );

/**
 * Optional int env var for a limit that can be switched off. Blank, whitespace and `0` all mean
 * "no limit" and normalise to undefined; anything else that is set must be greater than zero.
 */
const OptionalLimitEnv = z.preprocess((v) => {
  if (typeof v === "string" && (v.trim() === "" || Number(v.trim()) === 0)) {
    return undefined;
  }

  return v === 0 ? undefined : v;
}, z.coerce.number().int().positive().optional());

const EnvironmentSchema = z
  .object({
    NODE_ENV: z.union([z.literal("development"), z.literal("production"), z.literal("test")]),
    DATABASE_URL: z
      .string()
      .refine(
        isValidDatabaseUrl,
        "DATABASE_URL is invalid, for details please check the additional output above this message."
      ),
    DATABASE_CONNECTION_LIMIT: z.coerce.number().int().default(10),
    DATABASE_POOL_TIMEOUT: z.coerce.number().int().default(60),
    DATABASE_CONNECTION_TIMEOUT: z.coerce.number().int().default(20),
    DATABASE_WRITER_POOL_TIMEOUT: OptionalIntEnv,
    DATABASE_WRITER_CONNECTION_TIMEOUT: OptionalIntEnv,
    DATABASE_READ_REPLICA_POOL_TIMEOUT: OptionalIntEnv,
    DATABASE_READ_REPLICA_CONNECTION_TIMEOUT: OptionalIntEnv,
    DATABASE_TRANSACTION_MAX_WAIT_MS: IntEnvWithDefault(10000),
    DATABASE_TRANSACTION_START_RETRY_ENABLED: BoolEnvWithDefault(true),
    DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS: IntEnvWithDefault(3),
    DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS: IntEnvWithDefault(50),
    DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS: IntEnvWithDefault(250),
    DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC: IntEnvWithDefault(50),
    DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST: IntEnvWithDefault(100),
    RUN_OPS_DATABASE_TRANSACTION_MAX_WAIT_MS: OptionalIntEnv,
    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_ENABLED: OptionalBoolEnv,
    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS: OptionalIntEnv,
    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS: OptionalIntEnv,
    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS: OptionalIntEnv,
    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC: OptionalIntEnv,
    RUN_OPS_DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_TRANSACTION_MAX_WAIT_MS: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_ENABLED: OptionalBoolEnv,
    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_MAX_ATTEMPTS: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MIN_MS: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BACKOFF_MAX_MS: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BUDGET_PER_SEC: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_TRANSACTION_START_RETRY_BUDGET_BURST: OptionalIntEnv,
    // Dashboard-agent conversation store. Cloud points this at a dedicated
    // database; when unset it falls back to DATABASE_URL (OSS), where
    // the tables live in the isolated `trigger_dashboard_agent` schema.
    DASHBOARD_AGENT_DATABASE_URL: z.string().optional(),
    // The secret key (tr_*) for the runtime environment the dashboard-agent task
    // is deployed to. The chat session is created in that environment via the
    // standard chat.agent SDK flow. When unset, the live agent is disabled — the
    // conversation store / History still work, no chat can start.
    DASHBOARD_AGENT_SECRET_KEY: z.string().optional(),
    // Pins agent sessions to a specific deployed version (paired with
    // --skip-promotion deploys); unset => the project env's current version.
    DASHBOARD_AGENT_VERSION: z.string().optional(),
    // Global default for the `hasDashboardAgentAccess` flag. "0" (off) ships the
    // agent dark; flip to "1" to enable it for everyone at GA. Per-org overrides
    // (org featureFlags) win regardless.
    DASHBOARD_AGENT_ENABLED: z.string().default("0"),
    // Gates the create-org management API endpoint (default off).
    ORG_CREATION_API_ENABLED: z.string().default("0"),
    // "1" gives admins/impersonators an everywhere-preview (default off),
    // separate from the per-org rollout flag above.
    DASHBOARD_AGENT_ADMIN_PREVIEW: z.string().default("0"),
    // Anthropic key for the dashboard agent's Head Start route only (the warm
    // first-turn step-1 LLM call runs in this process). The agent run itself
    // uses its own key on the Trigger side. When unset, Head Start is disabled
    // and the first turn falls back to the normal cold-start path.
    ANTHROPIC_API_KEY: z.string().optional(),
    DIRECT_URL: z
      .string()
      .refine(
        isValidDatabaseUrl,
        "DIRECT_URL is invalid, for details please check the additional output above this message."
      ),
    DATABASE_READ_REPLICA_URL: z.string().optional(),
    // --- Run-ops DB split — Cloud-only scaling concern; OFF by default. ---
    // Explicit positive opt-in. Split behavior is unreachable unless this is true
    // AND the distinct-DB sentinel confirms the two URLs are physically distinct DBs.
    RUN_OPS_SPLIT_ENABLED: BoolEnv.default(false),
    RUN_OPS_WORKER_VERSION_FRESH_READ_ENABLED: BoolEnv.default(true),
    // Canonical connection URL for the dedicated NEW run-ops DB — drives the runtime pool, the split
    // decision, replication, and migrations. Optional so single-DB installs never set it.
    RUN_OPS_DATABASE_URL: z
      .string()
      .refine(isValidDatabaseUrl, "RUN_OPS_DATABASE_URL is invalid")
      .optional(),
    // The LEGACY run-ops DB. Now a CONNECTED DSN (Track 2): when split is on and this is set it builds
    // an INDEPENDENT legacy Prisma client, no longer an alias of the control-plane client (nor merely
    // the sentinel's probe target). Unset -> legacy reuses the control-plane client / DATABASE_URL, so
    // single-DB and self-host installs boot byte-identical.
    RUN_OPS_LEGACY_DATABASE_URL: z
      .string()
      .refine(isValidDatabaseUrl, "RUN_OPS_LEGACY_DATABASE_URL is invalid")
      .optional(),
    // The NEW dedicated run-ops DB read replica. Optional; self-host never sets it.
    // Refined (unlike the unrefined control-plane DATABASE_READ_REPLICA_URL) so a malformed run-ops
    // replica URL fails boot loudly rather than silently degrading — do not align it down to the CP shape.
    RUN_OPS_DATABASE_READ_REPLICA_URL: z
      .string()
      .refine(isValidDatabaseUrl, "RUN_OPS_DATABASE_READ_REPLICA_URL is invalid")
      .optional(),
    // The LEGACY run-ops DB read replica (Track 2). Unset -> the legacy replica handle falls back to the
    // legacy WRITER (as $replica does with no CP replica). Set in production so legacy reads hit the reader.
    RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL: z
      .string()
      .refine(isValidDatabaseUrl, "RUN_OPS_LEGACY_DATABASE_READ_REPLICA_URL is invalid")
      .optional(),
    // Optional cap for the unpooled new run-ops read replica. Unset falls back to DATABASE_CONNECTION_LIMIT.
    RUN_OPS_DATABASE_READ_REPLICA_CONNECTION_LIMIT: OptionalIntEnv,
    RUN_OPS_DATABASE_WRITER_POOL_TIMEOUT: OptionalIntEnv,
    RUN_OPS_DATABASE_WRITER_CONNECTION_TIMEOUT: OptionalIntEnv,
    RUN_OPS_DATABASE_READ_REPLICA_POOL_TIMEOUT: OptionalIntEnv,
    RUN_OPS_DATABASE_READ_REPLICA_CONNECTION_TIMEOUT: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_WRITER_POOL_TIMEOUT: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_WRITER_CONNECTION_TIMEOUT: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_READ_REPLICA_POOL_TIMEOUT: OptionalIntEnv,
    RUN_OPS_LEGACY_DATABASE_READ_REPLICA_CONNECTION_TIMEOUT: OptionalIntEnv,
    // Direct DSN for applying the full @trigger.dev/database migrations to the LEGACY run-ops DB, keeping
    // its schema current after the control plane moves off it. Direct, not pooled — migrations never run
    // over a pooler. Optional; unset -> the entrypoint's legacy migrate step is skipped.
    RUN_OPS_LEGACY_DIRECT_URL: z
      .string()
      .refine(isValidDatabaseUrl, "RUN_OPS_LEGACY_DIRECT_URL is invalid")
      .optional(),
    // Advisory control-plane co-residency sentinel enforcement (Track 2, T2.3). Default OFF; the advisory
    // arm always emits its metric, this only turns a still-co-resident pair into a hard boot failure.
    RUN_OPS_EXPECT_CONTROL_PLANE_SPLIT: BoolEnv.default(false),
    // --- Control-plane datasource repoint. Additive-only. ---
    // Optional control-plane DB. Unset (self-host/single-DB) -> getClient()/getReplicaClient() fall back to
    // DATABASE_URL/DATABASE_READ_REPLICA_URL, so boot is byte-identical. When set, these point at the
    // dedicated control-plane DSN; moving off the shared DB is an ops config change, not a code edit.
    CONTROL_PLANE_DATABASE_URL: z
      .string()
      .refine(
        (v) => v === undefined || isValidDatabaseUrl(v),
        "CONTROL_PLANE_DATABASE_URL is invalid"
      )
      .optional(),
    CONTROL_PLANE_DATABASE_READ_REPLICA_URL: z.string().optional(),
    CONTROL_PLANE_DATABASE_WRITER_DRIVER_ADAPTER: z.string().default("0"),
    CONTROL_PLANE_DATABASE_REPLICA_DRIVER_ADAPTER: z.string().default("0"),
    CONTROL_PLANE_DEQUEUE_READS_FROM_REPLICA: z.string().default("0"),
    RUN_OPS_DATABASE_WRITER_DRIVER_ADAPTER: z.string().default("0"),
    RUN_OPS_DATABASE_REPLICA_DRIVER_ADAPTER: z.string().default("0"),
    RUN_OPS_LEGACY_DATABASE_WRITER_DRIVER_ADAPTER: z.string().default("0"),
    RUN_OPS_LEGACY_DATABASE_REPLICA_DRIVER_ADAPTER: z.string().default("0"),
    // Control-plane cache relax knobs. Unset -> defaults (DEFAULT_CP_CACHE_TTL_MS / _MAX_ENTRIES).
    CONTROL_PLANE_CACHE_TTL_MS: z.coerce.number().int().optional(),
    CONTROL_PLANE_CACHE_MAX_ENTRIES: z.coerce.number().int().optional(),
    // Webhook feature data-plane DB (WebhookEndpoint + WebhookDelivery). Unset -> the webhook
    // clients reuse the main prisma / $replica, so this is connection-neutral until you split.
    WEBHOOK_DATABASE_URL: z.string().optional(),
    WEBHOOK_DATABASE_READ_REPLICA_URL: z.string().optional(),
    WEBHOOK_DATABASE_CONNECTION_LIMIT: z.coerce.number().int().optional(),
    SESSION_SECRET: z.string().min(1).refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE),
    MAGIC_LINK_SECRET: z.string().min(1).refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE),
    ENCRYPTION_KEY: z
      .string()
      .refine(
        (val) => Buffer.from(val, "utf8").length === 32,
        "ENCRYPTION_KEY must be exactly 32 bytes"
      )
      .refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE),
    WHITELISTED_EMAILS: z
      .string()
      .refine(isValidRegex, "WHITELISTED_EMAILS must be a valid regex.")
      .optional(),
    ADMIN_EMAILS: z.string().refine(isValidRegex, "ADMIN_EMAILS must be a valid regex.").optional(),
    REMIX_APP_PORT: z.string().optional(),
    // Opt-in, dev-only: stream this process's logs over a local telnet/TCP socket on this port.
    // Read directly from process.env in server.ts (before this schema loads); declared here for discoverability.
    WEBAPP_TELNET_LOGS_PORT: z.coerce.number().optional(),
    LOGIN_ORIGIN: z.string().default("http://localhost:3030"),
    LOGIN_RATE_LIMITS_ENABLED: BoolEnv.default(true),
    APP_ORIGIN: z.string().default("http://localhost:3030"),
    // Extra exact origins (comma separated) added to the document `img-src` CSP,
    // e.g. an SSO host serving profile images. Wildcards are refused.
    CSP_IMG_SRC_ALLOWLIST: z.string().optional(),
    API_ORIGIN: z.string().optional(),
    // Alternative API origin for deployed runs whose org has the
    // internalApiOriginEnabled feature flag on. Unset = flag is a no-op.
    INTERNAL_API_ORIGIN: z.string().optional(),
    // Global default for internalApiOriginEnabled when an org hasn't set it.
    INTERNAL_API_ORIGIN_ENABLED: z.string().default("0"),
    STREAM_ORIGIN: z.string().optional(),
    ELECTRIC_ORIGIN: z.string().default("http://localhost:3060"),
    // A comma separated list of electric origins to shard into different electric instances by environmentId
    // example: "http://localhost:3060,http://localhost:3061,http://localhost:3062"
    ELECTRIC_ORIGIN_SHARDS: z.string().optional(),
    APP_ENV: z.string().default(process.env.NODE_ENV),
    SERVICE_NAME: z.string().default("trigger.dev webapp"),
    SENTRY_DSN: z.string().optional(),
    POSTHOG_PROJECT_KEY: z.string().default("phc_LFH7kJiGhdIlnO22hTAKgHpaKhpM8gkzWAFvHmf5vfS"),
    // Upstream hosts the /ph reverse proxy forwards to (defaults: PostHog Cloud
    // EU). The client points api_host at the same-origin /ph path; the proxy
    // fans out to the ingest vs assets host by path.
    POSTHOG_INGEST_HOST: z.string().default("eu.i.posthog.com"),
    POSTHOG_ASSETS_HOST: z.string().default("eu-assets.i.posthog.com"),
    // PostHog app host, used for the browser toolbar (ui_host) and the server
    // client. Set to https://us.posthog.com for a US project (also switch the
    // ingest/assets hosts to their us.i / us-assets equivalents).
    POSTHOG_HOST: z.string().default("https://eu.posthog.com"),
    TRIGGER_TELEMETRY_DISABLED: z.string().optional(),
    AUTH_GITHUB_CLIENT_ID: z.string().optional(),
    AUTH_GITHUB_CLIENT_SECRET: z.string().optional(),
    AUTH_GOOGLE_CLIENT_ID: z.string().optional(),
    AUTH_GOOGLE_CLIENT_SECRET: z.string().optional(),
    EMAIL_TRANSPORT: z.enum(["resend", "smtp", "aws-ses"]).optional(),
    FROM_EMAIL: z.string().optional(),
    REPLY_TO_EMAIL: z.string().optional(),
    RESEND_API_KEY: z.string().optional(),
    SMTP_HOST: z.string().optional(),
    SMTP_PORT: z.coerce.number().optional(),
    SMTP_SECURE: BoolEnv.optional(),
    SMTP_USER: z.string().optional(),
    SMTP_PASSWORD: z.string().optional(),

    PLAIN_API_KEY: z.string().optional(),
    PLAIN_CUSTOMER_CARDS_SECRET: z.string().optional(),
    PLAIN_CUSTOMER_CARDS_KEY: z.string().optional(),
    PLAIN_CUSTOMER_CARDS_HEADERS: z.string().optional(),
    // How often each replica reloads the global flags snapshot from the DB.
    // Sets kill/ramp propagation latency.
    GLOBAL_FLAGS_RELOAD_INTERVAL_MS: z.coerce.number().int().min(1000).default(5000),
    WORKER_ENABLED: z.string().default("true"),
    GRACEFUL_SHUTDOWN_TIMEOUT: z.coerce.number().int().default(60000),
    DISABLE_SSE: z.string().optional(),
    OPENAI_API_KEY: z.string().optional(),

    // Redis options
    REDIS_HOST: z.string().optional(),
    REDIS_READER_HOST: z.string().optional(),
    REDIS_READER_PORT: z.coerce.number().optional(),
    REDIS_PORT: z.coerce.number().optional(),
    REDIS_USERNAME: z.string().optional(),
    REDIS_PASSWORD: z.string().optional(),
    REDIS_TLS_DISABLED: z.string().optional(),

    RATE_LIMIT_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    RATE_LIMIT_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    RATE_LIMIT_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    RATE_LIMIT_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    RATE_LIMIT_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    RATE_LIMIT_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    RATE_LIMIT_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
    RATE_LIMIT_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

    CACHE_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    CACHE_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    CACHE_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    CACHE_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    CACHE_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    CACHE_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    CACHE_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
    CACHE_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

    TASK_META_CACHE_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    TASK_META_CACHE_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    TASK_META_CACHE_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    TASK_META_CACHE_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    TASK_META_CACHE_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),
    TASK_META_CACHE_CURRENT_ENV_TTL_SECONDS: z.coerce.number().default(86400),

    // Runs-list empty-state check: how far back the ClickHouse "does this env have any run"
    // probe looks. Bounds the prove-absence partition scan. 0 = unbounded ("any run ever").
    RUN_LIST_HAS_RUNS_LOOKBACK_DAYS: z.coerce.number().default(30),
    // SWR TTLs for the empty-state has-runs cache (memory + Redis).
    RUN_LIST_HAS_RUNS_CACHE_FRESH_MS: z.coerce.number().default(86_400_000),
    RUN_LIST_HAS_RUNS_CACHE_STALE_MS: z.coerce.number().default(604_800_000),
    TASK_META_CACHE_BY_WORKER_TTL_SECONDS: z.coerce.number().default(2592000),

    REALTIME_STREAMS_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    REALTIME_STREAMS_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    REALTIME_STREAMS_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    REALTIME_STREAMS_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    REALTIME_STREAMS_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    REALTIME_STREAMS_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    REALTIME_STREAMS_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),
    REALTIME_STREAMS_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),
    REALTIME_STREAMS_INACTIVITY_TIMEOUT_MS: z.coerce.number().int().default(60000), // 1 minute

    REALTIME_MAXIMUM_CREATED_AT_FILTER_AGE_IN_MS: z.coerce
      .number()
      .int()
      .default(24 * 60 * 60 * 1000), // 1 day in milliseconds

    // Master switch for the native realtime backend; off = Electric serves everything, publishes no-op.
    REALTIME_BACKEND_NATIVE_ENABLED: z.string().default("0"),
    // Default backend when an org has no `realtimeBackend` override and no global flag row is set.
    REALTIME_BACKEND_DEFAULT: z.enum(["electric", "native", "shadow"]).default("electric"),
    // Live long-poll backstop hold (ms); matches Electric's ~20s cadence.
    REALTIME_BACKEND_NATIVE_LIVE_POLL_TIMEOUT_MS: z.coerce.number().int().default(20_000),
    // Jitter ratio on the live-poll hold (0.15 = ±15%) to avoid synchronized refetch herds.
    REALTIME_BACKEND_NATIVE_LIVE_POLL_JITTER_RATIO: z.coerce.number().default(0.15),
    // Hard cap on the tag-list snapshot size.
    REALTIME_BACKEND_NATIVE_MAX_LIST_RESULTS: z.coerce.number().int().default(1_000),
    // TTL/size of the coalescing cache for the multi-run resolve+hydrate (same-filter feeds share one query).
    REALTIME_BACKEND_NATIVE_RUNSET_CACHE_TTL_MS: z.coerce.number().int().default(1_000),
    REALTIME_BACKEND_NATIVE_RUNSET_CACHE_MAX_ENTRIES: z.coerce.number().int().default(5_000),
    // Size/TTL of the per-handle working-set cache used to diff multi-run live polls.
    REALTIME_BACKEND_NATIVE_WORKING_SET_MAX_ENTRIES: z.coerce.number().int().default(10_000),
    REALTIME_BACKEND_NATIVE_WORKING_SET_TTL_MS: z.coerce.number().int().default(300_000),
    // Bucket (ms) the tag-list createdAt floor is quantized to so same-tag feeds share a cache entry; 0 disables.
    REALTIME_BACKEND_NATIVE_RUNSET_CREATED_AT_BUCKET_MS: z.coerce.number().int().default(60_000),
    // Leading-edge throttle (ms) on per-env wake delivery; 0 wakes on every change.
    REALTIME_BACKEND_NATIVE_ENV_WAKE_COALESCE_WINDOW_MS: z.coerce.number().int().default(250),
    // "1" shares per-connection replay cursors fleet-wide via Redis, so a load-balancer hop reads the connection's true inter-poll gap instead of cold-resolving.
    REALTIME_BACKEND_NATIVE_SHARED_REPLAY_CURSORS: z.string().default("1"),
    // "1" holds a multi-run live poll open on a non-matching wake instead of replying up-to-date.
    REALTIME_BACKEND_NATIVE_HOLD_ON_EMPTY: z.string().default("1"),
    // Max concurrent fresh ClickHouse resolves per instance (reconnect-stampede gate); 0 disables.
    REALTIME_BACKEND_NATIVE_RESOLVE_ADMISSION_LIMIT: z.coerce.number().int().default(16),
    // Replay window (ms) for buffered change records delivered to newly-armed feeds; 0 disables.
    REALTIME_BACKEND_NATIVE_REPLAY_WINDOW_MS: z.coerce.number().int().default(2_000),
    // Cap on buffered recent records per env (latest record per run).
    REALTIME_BACKEND_NATIVE_REPLAY_MAX_RUNS: z.coerce.number().int().default(512),
    // Keep an env subscribed + buffering this long (ms) after its last feed closes; 0 disables.
    REALTIME_BACKEND_NATIVE_UNSUBSCRIBE_LINGER_MS: z.coerce.number().int().default(5_000),
    // Fallback per-env concurrent-connection limit when the org has none configured.
    REALTIME_BACKEND_NATIVE_DEFAULT_CONCURRENCY_LIMIT: z.coerce.number().int().default(100_000),
    // TTL/size of the single-run read-through cache that collapses duplicate refetch bursts.
    REALTIME_BACKEND_NATIVE_RUN_CACHE_TTL_MS: z.coerce.number().int().default(250),
    REALTIME_BACKEND_NATIVE_RUN_CACHE_MAX_ENTRIES: z.coerce.number().int().default(5_000),
    // TTL/size of the per-org realtimeBackend flag cache used to pick the serving backend.
    REALTIME_BACKEND_FLAG_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
    REALTIME_BACKEND_FLAG_CACHE_MAX_ENTRIES: z.coerce.number().int().default(50_000),
    REALTIME_BACKEND_NATIVE_RUN_READS_FROM_PRIMARY: z.string().default("0"),
    // "1" enables the read-your-writes gate: wake hydrates wait out the measured replica lag
    // (anchored to the change record's updatedAtMs) and stale reads are retried.
    REALTIME_BACKEND_NATIVE_REPLICA_LAG_GATE_ENABLED: z.string().default("1"),
    // Reader-side lag probe cadence while the router is active; probing pauses when idle.
    REALTIME_BACKEND_NATIVE_REPLICA_LAG_SAMPLE_INTERVAL_MS: z.coerce.number().int().default(250),
    REALTIME_BACKEND_NATIVE_REPLICA_LAG_IDLE_AFTER_MS: z.coerce.number().int().default(30_000),
    // The lag estimate is the max sample inside this window (spikes widen it immediately).
    REALTIME_BACKEND_NATIVE_REPLICA_LAG_WINDOW_MS: z.coerce.number().int().default(5_000),
    // Estimate before the first sample lands (and the floor when probing is unavailable).
    REALTIME_BACKEND_NATIVE_REPLICA_LAG_DEFAULT_MS: z.coerce.number().int().default(30),
    // Safety margin (clock skew + scheduling) added on top of the lag estimate.
    REALTIME_BACKEND_NATIVE_REPLICA_LAG_MARGIN_MS: z.coerce.number().int().default(10),
    // Hard cap on any single gate delay — a sick replica degrades freshness, never liveness.
    REALTIME_BACKEND_NATIVE_REPLICA_LAG_MAX_DELAY_MS: z.coerce.number().int().default(1_000),
    // Re-hydrate attempts for rows the tripwire still finds stale after the delay.
    REALTIME_BACKEND_NATIVE_STALE_HYDRATE_RETRIES: z.coerce.number().int().default(3),
    // How long a tripwire-observed staleness floors the lag estimate (vanilla-PG replicas
    // can't measure mid-apply lag, so observations carry the estimate between races).
    REALTIME_BACKEND_NATIVE_REPLICA_LAG_OBSERVED_FLOOR_TTL_MS: z.coerce
      .number()
      .int()
      .default(60_000),

    PUBSUB_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    PUBSUB_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    PUBSUB_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    PUBSUB_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    PUBSUB_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    PUBSUB_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    PUBSUB_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
    PUBSUB_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

    // Dedicated pub/sub Redis for the native realtime backend; falls back to PUBSUB_REDIS_* then REDIS_*.
    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.PUBSUB_REDIS_HOST ?? process.env.REDIS_HOST),
    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform((v) => {
        if (v !== undefined) return v;
        const raw = process.env.PUBSUB_REDIS_PORT ?? process.env.REDIS_PORT;
        return raw ? parseInt(raw) : undefined;
      }),
    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.PUBSUB_REDIS_USERNAME ?? process.env.REDIS_USERNAME),
    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.PUBSUB_REDIS_PASSWORD ?? process.env.REDIS_PASSWORD),
    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.PUBSUB_REDIS_TLS_DISABLED ?? process.env.REDIS_TLS_DISABLED ?? "false"),
    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_CLUSTER_MODE_ENABLED: z
      .string()
      .default(process.env.PUBSUB_REDIS_CLUSTER_MODE_ENABLED ?? "0"),
    // Use sharded pub/sub (SSUBSCRIBE/SPUBLISH) in cluster mode; "0" forces classic pub/sub.
    REALTIME_BACKEND_NATIVE_PUBSUB_REDIS_SHARDED_ENABLED: z.string().default("1"),

    DEFAULT_ENV_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(100),
    DEFAULT_ENV_EXECUTION_CONCURRENCY_BURST_FACTOR: z.coerce.number().default(1.0),
    DEFAULT_ORG_EXECUTION_CONCURRENCY_LIMIT: z.coerce.number().int().default(300),
    DEFAULT_DEV_ENV_EXECUTION_ATTEMPTS: z.coerce.number().int().positive().default(1),

    //API Rate limiting
    /**
     * @example "60s"
     * @example "1m"
     * @example "1h"
     * @example "1d"
     * @example "1000ms"
     * @example "1000s"
     */
    API_RATE_LIMIT_REFILL_INTERVAL: z.string().default("10s"), // refill 250 tokens every 10 seconds
    API_RATE_LIMIT_MAX: z.coerce.number().int().default(750), // allow bursts of 750 requests
    API_RATE_LIMIT_REFILL_RATE: z.coerce.number().int().default(250), // refix 250 tokens every 10 seconds
    API_RATE_LIMIT_REQUEST_LOGS_ENABLED: z.string().default("0"),
    API_RATE_LIMIT_REJECTION_LOGS_ENABLED: z.string().default("1"),
    API_RATE_LIMIT_LIMITER_LOGS_ENABLED: z.string().default("0"),

    API_RATE_LIMIT_JWT_WINDOW: z.string().default("1m"),
    API_RATE_LIMIT_JWT_TOKENS: z.coerce.number().int().default(60),

    // Separate budget for deploy-flow endpoints, see deploymentRateLimit.server.ts
    DEPLOYMENT_RATE_LIMIT_REFILL_INTERVAL: z.string().default("10s"),
    DEPLOYMENT_RATE_LIMIT_MAX: z.coerce.number().int().default(1500),
    DEPLOYMENT_RATE_LIMIT_REFILL_RATE: z.coerce.number().int().default(500),
    DEPLOYMENT_RATE_LIMIT_REQUEST_LOGS_ENABLED: z.string().default("0"),
    DEPLOYMENT_RATE_LIMIT_REJECTION_LOGS_ENABLED: z.string().default("1"),
    DEPLOYMENT_RATE_LIMIT_LIMITER_LOGS_ENABLED: z.string().default("0"),

    // Per-IP rate limit for the unauthenticated OTLP ingestion endpoints
    // (/otel/*). Bounds unauthenticated request rates. Opt-in
    // (disabled by default): because it keys on the source IP, it is only
    // safe to enable when each client presents a distinct IP through a proxy
    // that appends the real client IP to X-Forwarded-For. Enabling it where
    // many clients share one egress IP (e.g. behind NAT or a shared proxy)
    // would collapse that traffic into a single bucket and could throttle
    // legitimate telemetry. Set OTLP_RATE_LIMIT_ENABLED=1 to enable, then tune
    // OTLP_RATE_LIMIT_MAX / OTLP_RATE_LIMIT_WINDOW for expected volume.
    OTLP_RATE_LIMIT_ENABLED: z.string().default("0"),
    OTLP_RATE_LIMIT_WINDOW: z
      .string()
      .regex(/^\d+ ?(?:ms|s|m|h|d)$/)
      .default("1m"),
    OTLP_RATE_LIMIT_MAX: z.coerce.number().int().positive().default(3000),

    DEPOT_TOKEN: z.string().optional(),
    DEPOT_ORG_ID: z.string().optional(),
    DEPOT_REGION: z.string().default("us-east-1"),

    // Deployment registry (v3)
    DEPLOY_REGISTRY_HOST: z.string().min(1),
    DEPLOY_REGISTRY_USERNAME: z.string().optional(),
    DEPLOY_REGISTRY_PASSWORD: z.string().optional(),
    DEPLOY_REGISTRY_NAMESPACE: z.string().min(1).default("trigger"),
    DEPLOY_REGISTRY_ECR_TAGS: z.string().optional(), // csv, for example: "key1=value1,key2=value2"
    DEPLOY_REGISTRY_ECR_ASSUME_ROLE_ARN: z.string().optional(),
    DEPLOY_REGISTRY_ECR_ASSUME_ROLE_EXTERNAL_ID: z.string().optional(),
    DEPLOY_REGISTRY_ECR_DEFAULT_REPOSITORY_POLICY: z.string().optional(), // raw IAM policy JSON applied to every repo created by the webapp

    // Deployment registry (v4) - falls back to v3 registry if not specified
    V4_DEPLOY_REGISTRY_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.DEPLOY_REGISTRY_HOST)
      .pipe(z.string().min(1)), // Ensure final type is required string
    V4_DEPLOY_REGISTRY_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.DEPLOY_REGISTRY_USERNAME),
    V4_DEPLOY_REGISTRY_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.DEPLOY_REGISTRY_PASSWORD),
    V4_DEPLOY_REGISTRY_NAMESPACE: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.DEPLOY_REGISTRY_NAMESPACE)
      .pipe(z.string().min(1).default("trigger")), // Ensure final type is required string
    V4_DEPLOY_REGISTRY_ECR_TAGS: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.DEPLOY_REGISTRY_ECR_TAGS),
    V4_DEPLOY_REGISTRY_ECR_ASSUME_ROLE_ARN: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.DEPLOY_REGISTRY_ECR_ASSUME_ROLE_ARN),
    V4_DEPLOY_REGISTRY_ECR_ASSUME_ROLE_EXTERNAL_ID: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.DEPLOY_REGISTRY_ECR_ASSUME_ROLE_EXTERNAL_ID),
    V4_DEPLOY_REGISTRY_ECR_DEFAULT_REPOSITORY_POLICY: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.DEPLOY_REGISTRY_ECR_DEFAULT_REPOSITORY_POLICY),

    // Compute gateway (template creation during deploy finalize)
    COMPUTE_GATEWAY_URL: z.string().optional(),
    COMPUTE_GATEWAY_AUTH_TOKEN: z.string().optional(),
    COMPUTE_TEMPLATE_SHADOW_ROLLOUT_PCT: z.string().optional(),
    // Comma-separated machine preset names to build boot snapshots for on
    // deploy (e.g. "small-1x,small-2x,medium-1x"). Default: "small-1x".
    COMPUTE_TEMPLATE_MACHINE_PRESETS: z
      .string()
      .default("small-1x")
      .transform(parseMachinePresetCsv),
    // Subset of COMPUTE_TEMPLATE_MACHINE_PRESETS that must succeed for a
    // required-mode deploy to be considered successful. Failures of presets
    // outside this list are logged but don't fail the deploy. Defaults to the
    // full COMPUTE_TEMPLATE_MACHINE_PRESETS list when unset (everything required).
    COMPUTE_TEMPLATE_MACHINE_PRESETS_REQUIRED: z
      .string()
      .optional()
      .transform((v, ctx) =>
        parseMachinePresetCsv(v ?? process.env.COMPUTE_TEMPLATE_MACHINE_PRESETS ?? "small-1x", ctx)
      ),

    DEPLOY_IMAGE_PLATFORM: z.string().default("linux/amd64"),
    DEPLOY_TIMEOUT_MS: z.coerce
      .number()
      .int()
      .default(60 * 1000 * 8), // 8 minutes
    DEPLOY_QUEUE_TIMEOUT_MS: z.coerce
      .number()
      .int()
      .default(60 * 1000 * 15), // 15 minutes

    // When enabled, reject deploys made by v3 CLI versions (i.e. payloads that
    // omit the `type` field). v4 CLI versions always send `type` ("MANAGED" or "V1"),
    // so they are unaffected. Defaults to off so detection can run in
    // log-only mode before enforcement.
    DEPRECATE_V3_CLI_DEPLOYS_ENABLED: z.string().default("0"),

    // Verify the deploy image exists before promoting. Disable for out-of-band/air-gapped push. ECR only.
    DEPLOY_IMAGE_VERIFICATION_ENABLED: BoolEnv.default(true),

    OBJECT_STORE_BASE_URL: z.string().optional(),
    OBJECT_STORE_BUCKET: z.string().optional(),
    OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),
    OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
    OBJECT_STORE_REGION: z.string().optional(),
    OBJECT_STORE_SERVICE: z.string().default("s3"),

    // Protocol to use for new uploads (e.g., "s3", "r2"). Data without protocol uses default provider above.
    // If specified, you must configure the corresponding provider using OBJECT_STORE_{PROTOCOL}_* env vars.
    // Example: OBJECT_STORE_DEFAULT_PROTOCOL=s3 requires OBJECT_STORE_S3_BASE_URL, OBJECT_STORE_S3_ACCESS_KEY_ID, etc.
    // Enables zero-downtime migration between providers (old data keeps working, new data uses new provider).
    OBJECT_STORE_DEFAULT_PROTOCOL: z
      .string()
      .regex(/^[a-z0-9]+$/)
      .optional(),

    ARTIFACTS_OBJECT_STORE_BUCKET: z.string().optional(),
    ARTIFACTS_OBJECT_STORE_BASE_URL: z.string().optional(),
    ARTIFACTS_OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),
    ARTIFACTS_OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
    ARTIFACTS_OBJECT_STORE_REGION: z.string().optional(),
    EVENTS_BATCH_SIZE: z.coerce.number().int().default(100),
    EVENTS_BATCH_INTERVAL: z.coerce.number().int().default(1000),
    EVENTS_DEFAULT_LOG_RETENTION: z.coerce.number().int().default(7),
    EVENTS_MIN_CONCURRENCY: z.coerce.number().int().default(1),
    EVENTS_MAX_CONCURRENCY: z.coerce.number().int().default(10),
    EVENTS_MAX_BATCH_SIZE: z.coerce.number().int().default(500),
    EVENTS_MEMORY_PRESSURE_THRESHOLD: z.coerce.number().int().default(5000),
    EVENTS_LOAD_SHEDDING_THRESHOLD: z.coerce.number().int().default(100000),
    EVENTS_LOAD_SHEDDING_ENABLED: z.string().default("1"),

    MANAGED_WORKER_SECRET: z.string().min(1).refine(isNotInsecureSecret, INSECURE_SECRET_MESSAGE),

    // Allow booting with a known-insecure published default secret. Temporary
    // bridge for deployments that can't rotate yet; rotate as soon as possible.
    ALLOW_INSECURE_DEFAULT_SECRETS: BoolEnv.default(false),

    // Tenant scoping on worker actions is header-driven (folded into the engine snapshot read) and
    // needs no flag. This is only the no-header fallback: when "1", a worker action on a run created
    // after WORKLOAD_TOKEN_CUTOFF without a verified env header is rejected; runs on or before the
    // cutoff pass (grandfathered). Default off = no run-row read, byte-for-byte today's behavior.
    WORKLOAD_CREATED_AT_GATE_ENABLED: z.string().default("0"),
    WORKLOAD_TOKEN_CUTOFF: z.string().datetime().optional(),

    // Development OTEL environment variables
    DEV_OTEL_EXPORTER_OTLP_ENDPOINT: z.string().optional(),
    DEV_OTEL_METRICS_ENDPOINT: z.string().optional(),
    // If this is set to 1, then the below variables are used to configure the batch processor for spans and logs
    DEV_OTEL_BATCH_PROCESSING_ENABLED: z.string().default("0"),
    DEV_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
    DEV_OTEL_SPAN_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
    DEV_OTEL_SPAN_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
    DEV_OTEL_SPAN_MAX_QUEUE_SIZE: z.string().default("512"),
    DEV_OTEL_LOG_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
    DEV_OTEL_LOG_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
    DEV_OTEL_LOG_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
    DEV_OTEL_LOG_MAX_QUEUE_SIZE: z.string().default("512"),
    DEV_OTEL_METRICS_EXPORT_INTERVAL_MILLIS: z.string().optional(),
    DEV_OTEL_METRICS_EXPORT_TIMEOUT_MILLIS: z.string().optional(),
    DEV_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS: z.string().optional(),

    PROD_OTEL_BATCH_PROCESSING_ENABLED: z.string().default("0"),
    PROD_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
    PROD_OTEL_SPAN_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
    PROD_OTEL_SPAN_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
    PROD_OTEL_SPAN_MAX_QUEUE_SIZE: z.string().default("512"),
    PROD_OTEL_LOG_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
    PROD_OTEL_LOG_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
    PROD_OTEL_LOG_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
    PROD_OTEL_LOG_MAX_QUEUE_SIZE: z.string().default("512"),
    PROD_OTEL_METRICS_EXPORT_INTERVAL_MILLIS: z.string().optional(),
    PROD_OTEL_METRICS_EXPORT_TIMEOUT_MILLIS: z.string().optional(),
    PROD_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS: z.string().optional(),

    TRIGGER_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT: z.string().default("1024"),
    TRIGGER_OTEL_LOG_ATTRIBUTE_COUNT_LIMIT: z.string().default("1024"),
    TRIGGER_OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT: z.string().default("131072"),
    TRIGGER_OTEL_LOG_ATTRIBUTE_VALUE_LENGTH_LIMIT: z.string().default("131072"),
    TRIGGER_OTEL_SPAN_EVENT_COUNT_LIMIT: z.string().default("10"),
    TRIGGER_OTEL_LINK_COUNT_LIMIT: z.string().default("2"),
    TRIGGER_OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT: z.string().default("10"),
    TRIGGER_OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT: z.string().default("10"),

    CHECKPOINT_THRESHOLD_IN_MS: z.coerce.number().int().default(30000),

    // Internal OTEL environment variables
    INTERNAL_OTEL_TRACE_EXPORTER_URL: z.string().optional(),
    INTERNAL_OTEL_TRACE_EXPORTER_AUTH_HEADERS: z.string().optional(),
    INTERNAL_OTEL_TRACE_LOGGING_ENABLED: z.string().default("1"),
    // this means 1/20 traces or 5% of traces will be sampled (sampled = recorded)
    INTERNAL_OTEL_TRACE_SAMPLING_RATE: z.string().default("20"),
    INTERNAL_OTEL_TRACE_INSTRUMENT_PRISMA_ENABLED: z.string().default("0"),
    INTERNAL_OTEL_TRACE_DISABLED: z.string().default("0"),
    DISABLE_HTTP_INSTRUMENTATION: BoolEnv.default(false),

    INTERNAL_OTEL_LOG_EXPORTER_URL: z.string().optional(),
    INTERNAL_OTEL_METRIC_EXPORTER_URL: z.string().optional(),
    INTERNAL_OTEL_METRIC_EXPORTER_AUTH_HEADERS: z.string().optional(),
    INTERNAL_OTEL_METRIC_EXPORTER_ENABLED: z.string().default("0"),
    INTERNAL_OTEL_METRIC_EXPORTER_INTERVAL_MS: z.coerce.number().int().default(30_000),
    INTERNAL_OTEL_HOST_METRICS_ENABLED: BoolEnv.default(true),
    INTERNAL_OTEL_NODEJS_METRICS_ENABLED: BoolEnv.default(true),
    INTERNAL_OTEL_ADDITIONAL_DETECTORS_ENABLED: BoolEnv.default(true),

    ORG_SLACK_INTEGRATION_CLIENT_ID: z.string().optional(),
    ORG_SLACK_INTEGRATION_CLIENT_SECRET: z.string().optional(),

    /** Vercel integration OAuth credentials */
    VERCEL_INTEGRATION_CLIENT_ID: z.string().optional(),
    VERCEL_INTEGRATION_CLIENT_SECRET: z.string().optional(),
    VERCEL_INTEGRATION_APP_SLUG: z.string().optional(),

    /** These enable the alerts feature in v3 */
    ALERT_EMAIL_TRANSPORT: z.enum(["resend", "smtp", "aws-ses"]).optional(),
    ALERT_FROM_EMAIL: z.string().optional(),
    ALERT_REPLY_TO_EMAIL: z.string().optional(),
    ALERT_RESEND_API_KEY: z.string().optional(),
    ALERT_SMTP_HOST: z.string().optional(),
    ALERT_SMTP_PORT: z.coerce.number().optional(),
    ALERT_SMTP_SECURE: BoolEnv.optional(),
    ALERT_SMTP_USER: z.string().optional(),
    ALERT_SMTP_PASSWORD: z.string().optional(),
    ALERT_RATE_LIMITER_EMISSION_INTERVAL: z.coerce.number().int().default(2_500),
    ALERT_RATE_LIMITER_BURST_TOLERANCE: z.coerce.number().int().default(10_000),
    ALERT_RATE_LIMITER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    ALERT_RATE_LIMITER_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    ALERT_RATE_LIMITER_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    ALERT_RATE_LIMITER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    ALERT_RATE_LIMITER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    ALERT_RATE_LIMITER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    ALERT_RATE_LIMITER_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),
    ALERT_RATE_LIMITER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

    LOOPS_API_KEY: z.string().optional(),
    ATTIO_API_KEY: z.string().optional(),

    PROD_TASK_HEARTBEAT_INTERVAL_MS: z.coerce.number().int().optional(),

    /* Usage settings */
    USAGE_EVENT_URL: z.string().optional(),
    PROD_USAGE_HEARTBEAT_INTERVAL_MS: z.coerce.number().int().optional(),

    CENTS_PER_RUN: z.coerce.number().default(0),

    EVENT_LOOP_MONITOR_ENABLED: z.string().default("1"),
    MAXIMUM_LIVE_RELOADING_EVENTS: z.coerce.number().int().default(1000),
    MAXIMUM_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000),
    MAXIMUM_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(10_000),
    // Emergency circuit breaker: when set, clamps the trace summary and detailed
    // summary span limits on both event store paths to this value. Unset = disabled.
    TRACE_VIEW_EMERGENCY_SPAN_CAP: z.coerce.number().int().positive().optional(),
    TASK_PAYLOAD_OFFLOAD_THRESHOLD: z.coerce.number().int().default(524_288), // 512KB
    BATCH_PAYLOAD_OFFLOAD_THRESHOLD: z.coerce.number().int().optional(), // Defaults to TASK_PAYLOAD_OFFLOAD_THRESHOLD if not set
    TASK_PAYLOAD_MAXIMUM_SIZE: z.coerce.number().int().default(3_145_728), // 3MB
    BATCH_TASK_PAYLOAD_MAXIMUM_SIZE: z.coerce.number().int().default(1_000_000), // 1MB
    TASK_RUN_METADATA_MAXIMUM_SIZE: z.coerce.number().int().default(262_144), // 256KB

    MAXIMUM_DEV_QUEUE_SIZE: z.coerce.number().int().optional(),
    MAXIMUM_DEPLOYED_QUEUE_SIZE: z.coerce.number().int().optional(),
    QUEUE_SIZE_CACHE_TTL_MS: z.coerce.number().int().optional().default(1_000), // 1 second
    QUEUE_SIZE_CACHE_MAX_SIZE: z.coerce.number().int().optional().default(5_000),
    QUEUE_SIZE_CACHE_ENABLED: z.coerce.number().int().optional().default(1),
    MAX_BATCH_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),
    MAX_BATCH_AND_WAIT_V2_TRIGGER_ITEMS: z.coerce.number().int().default(500),

    // 2-phase batch API settings
    STREAMING_BATCH_MAX_ITEMS: z.coerce.number().int().default(1_000), // Max items in streaming batch
    STREAMING_BATCH_ITEM_MAXIMUM_SIZE: z.coerce.number().int().default(3_145_728),
    // Number of streamed batch items ingested concurrently in Phase 2. Peak
    // in-flight memory per request ≈ this × STREAMING_BATCH_ITEM_MAXIMUM_SIZE,
    // so raise with care. Set to 1 for fully sequential ingestion.
    STREAMING_BATCH_INGEST_CONCURRENCY: z.coerce.number().int().positive().default(10),
    BATCH_RATE_LIMIT_REFILL_RATE: z.coerce.number().int().default(100),
    BATCH_RATE_LIMIT_MAX: z.coerce.number().int().default(1200),
    BATCH_RATE_LIMIT_REFILL_INTERVAL: z.string().default("10s"),
    BATCH_CONCURRENCY_LIMIT_DEFAULT: z.coerce.number().int().default(5),
    /**
     * How long a created batch may remain unsealed before the seal-timeout reaper
     * aborts it and resumes any blocked parent with an error. Must exceed the SDK's
     * worst-case stream-retry budget (maxAttempts x server request timeout).
     * Doubles as the TTL of the phase 2 streaming grant, so the grant and the reaper
     * always agree on how long a batch is allowed to be sealing.
     */
    BATCH_SEAL_TIMEOUT_MS: z.coerce.number().int().positive().default(1_800_000),
    /**
     * Number of phase 2 (`POST /api/v3/batches/:id/items`) requests a created batch is
     * granted, exempt from the general API rate limit. Sized above the SDK's stream
     * maxAttempts so a batch admitted by the batch limiter can always finish streaming.
     */
    BATCH_STREAM_GRANT_ATTEMPTS: z.coerce.number().int().positive().default(10),

    REALTIME_STREAM_VERSION: z.enum(["v1", "v2"]).default("v1"),
    REALTIME_STREAM_MAX_LENGTH: z.coerce.number().int().default(1000),
    REALTIME_STREAM_TTL: z.coerce
      .number()
      .int()
      .default(60 * 60 * 24), // 1 day in seconds
    BATCH_METADATA_OPERATIONS_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
    BATCH_METADATA_OPERATIONS_FLUSH_ENABLED: z.string().default("1"),
    BATCH_METADATA_OPERATIONS_FLUSH_LOGGING_ENABLED: z.string().default("1"),

    // Run Engine 2.0
    RUN_ENGINE_WORKER_COUNT: z.coerce.number().int().default(4),
    RUN_ENGINE_TASKS_PER_WORKER: z.coerce.number().int().default(10),
    RUN_ENGINE_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(10),
    RUN_ENGINE_WORKER_POLL_INTERVAL: z.coerce.number().int().default(100),
    RUN_ENGINE_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(100),
    RUN_ENGINE_TIMEOUT_PENDING_EXECUTING: z.coerce.number().int().default(60_000),
    RUN_ENGINE_TIMEOUT_PENDING_CANCEL: z.coerce.number().int().default(60_000),
    RUN_ENGINE_TIMEOUT_EXECUTING: z.coerce.number().int().default(300_000), // 5 minutes
    RUN_ENGINE_TIMEOUT_EXECUTING_WITH_WAITPOINTS: z.coerce.number().int().default(300_000), // 5 minutes
    RUN_ENGINE_TIMEOUT_SUSPENDED: z.coerce
      .number()
      .int()
      .default(60_000 * 10),
    RUN_ENGINE_DEBUG_WORKER_NOTIFICATIONS: BoolEnv.default(false),
    RUN_ENGINE_PARENT_QUEUE_LIMIT: z.coerce.number().int().default(1000),
    RUN_ENGINE_CONCURRENCY_LIMIT_BIAS: z.coerce.number().default(0.75),
    RUN_ENGINE_AVAILABLE_CAPACITY_BIAS: z.coerce.number().default(0.3),
    RUN_ENGINE_QUEUE_AGE_RANDOMIZATION_BIAS: z.coerce.number().default(0.25),
    RUN_ENGINE_REUSE_SNAPSHOT_COUNT: z.coerce.number().int().default(0),
    RUN_ENGINE_MAXIMUM_ENV_COUNT: z.coerce.number().int().optional(),
    RUN_ENGINE_RUN_QUEUE_SHARD_COUNT: z.coerce.number().int().default(4),
    // Queue metrics ingestion (Redis Stream -> ClickHouse). The runtime on/off is the
    // `queue_metrics:enabled` Redis key; these gate emitter construction + consumer boot.
    QUEUE_METRICS_EMIT_ENABLED: z.string().default("0"),
    QUEUE_METRICS_CONSUMER_ENABLED: z.string().default("0"),
    QUEUE_METRICS_STREAM_SHARD_COUNT: z.coerce.number().int().default(4),
    QUEUE_METRICS_CONSUMER_BATCH_SIZE: z.coerce.number().int().default(1000),
    // Counter stream (exact counts, loss-intolerant). Unset host => the run-queue Redis;
    // set it to a dedicated instance so counter backlog never competes with the run queue.
    QUEUE_METRICS_REDIS_HOST: z.string().optional(),
    QUEUE_METRICS_REDIS_PORT: z.coerce.number().optional(),
    QUEUE_METRICS_REDIS_USERNAME: z.string().optional(),
    QUEUE_METRICS_REDIS_PASSWORD: z.string().optional(),
    QUEUE_METRICS_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
    // Default depends on where the stream lives: see metricsDefinition() in
    // queueMetrics.server.ts (2M on the shared run-queue Redis, 8M on a dedicated one).
    QUEUE_METRICS_COUNTER_STREAM_MAXLEN: z.coerce.number().int().optional(),
    // TTL (seconds) on the per-(queue,op) cumulative odometer key, refreshed on every write.
    // Idle-past-TTL queues purge and self-heal (restart from 1) on return; default 7 days.
    QUEUE_METRICS_COUNTER_ODOMETER_TTL_SECONDS: z.coerce.number().int().default(604_800),
    // Per-env distinct queue_name cap (0 = unlimited); overflow maps to "__overflow__".
    QUEUE_METRICS_MAX_QUEUE_NAMES_PER_ENV: z.coerce.number().int().default(1000),
    QUEUE_METRICS_MAX_CONCURRENCY_KEYS_PER_QUEUE: z.coerce.number().int().default(10_000),
    // Fraction (0..1) of ops that emit a gauge; counters are never sampled. Dial below 1
    // only if EngineCPU is too high in slow-path-heavy regions (hurts low-traffic queues).
    QUEUE_METRICS_GAUGE_SAMPLE_RATE: z.coerce.number().min(0).max(1).default(1),
    /**
     * Lists the queue-metrics tables in the Query page, its schema docs, the schema API and the
     * AI query context. Off by default so the tables are not advertised before ingestion is
     * enabled. Listing only: a query naming one of these tables still runs either way.
     */
    QUEUE_METRICS_QUERY_TABLES_VISIBLE: z.string().default("0"),
    RUN_ENGINE_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
    RUN_ENGINE_RETRY_WARM_START_THRESHOLD_MS: z.coerce.number().int().default(30_000),
    RUN_ENGINE_PROCESS_WORKER_QUEUE_DEBOUNCE_MS: z.coerce.number().int().default(200),
    RUN_ENGINE_DEQUEUE_BLOCKING_TIMEOUT_SECONDS: z.coerce.number().int().default(10),
    RUN_ENGINE_DEQUEUE_DISABLED_WORKER_QUEUES: z.string().optional(),
    RUN_ENGINE_MASTER_QUEUE_CONSUMERS_INTERVAL_MS: z.coerce.number().int().default(1000),
    // Off by default. Enable on a single service (e.g. the engine worker) so only one
    // instance reports worker queue length, rather than every replica.
    RUN_ENGINE_WORKER_QUEUE_OBSERVER_ENABLED: z.string().default("0"),
    RUN_ENGINE_WORKER_QUEUE_OBSERVER_INTERVAL_MS: z.coerce.number().int().default(30_000),
    // Comma-separated cloud providers to exclude from worker queue length observation.
    RUN_ENGINE_WORKER_QUEUE_OBSERVER_EXCLUDED_CLOUD_PROVIDERS: z.string().default("digitalocean"),
    RUN_ENGINE_MASTER_QUEUE_COOLOFF_PERIOD_MS: z.coerce.number().int().default(10_000),
    RUN_ENGINE_MASTER_QUEUE_COOLOFF_COUNT_THRESHOLD: z.coerce.number().int().default(10),
    RUN_ENGINE_MASTER_QUEUE_CONSUMER_DEQUEUE_COUNT: z.coerce.number().int().default(10),
    RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_SCHEDULE: z.string().optional(),
    RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_SCHEDULE: z.string().optional(),
    RUN_ENGINE_CONCURRENCY_SWEEPER_SCAN_JITTER_IN_MS: z.coerce.number().int().optional(),
    RUN_ENGINE_CONCURRENCY_SWEEPER_PROCESS_MARKED_JITTER_IN_MS: z.coerce.number().int().optional(),

    // TTL System settings for automatic run expiration
    RUN_ENGINE_TTL_SYSTEM_DISABLED: BoolEnv.default(false),
    RUN_ENGINE_TTL_SYSTEM_SHARD_COUNT: z.coerce.number().int().optional(),
    RUN_ENGINE_TTL_SYSTEM_POLL_INTERVAL_MS: z.coerce.number().int().default(1_000),
    RUN_ENGINE_TTL_SYSTEM_BATCH_SIZE: z.coerce.number().int().default(100),
    RUN_ENGINE_TTL_WORKER_CONCURRENCY: z.coerce.number().int().default(1),
    RUN_ENGINE_TTL_WORKER_BATCH_MAX_SIZE: z.coerce.number().int().default(50),
    RUN_ENGINE_TTL_CONSUMERS_DISABLED: BoolEnv.default(false),
    RUN_ENGINE_TTL_WORKER_BATCH_MAX_WAIT_MS: z.coerce.number().int().default(5_000),

    /** Optional maximum TTL for all runs (e.g. "14d"). If set, runs without an explicit TTL
     *  will use this as their TTL, and runs with a TTL larger than this will be clamped. */
    RUN_ENGINE_DEFAULT_MAX_TTL: z.string().optional(),

    RUN_ENGINE_RUN_LOCK_DURATION: z.coerce.number().int().default(5000),
    RUN_ENGINE_RUN_LOCK_AUTOMATIC_EXTENSION_THRESHOLD: z.coerce.number().int().default(1000),
    RUN_ENGINE_RUN_LOCK_MAX_RETRIES: z.coerce.number().int().default(10),
    RUN_ENGINE_RUN_LOCK_BASE_DELAY: z.coerce.number().int().default(100),
    RUN_ENGINE_RUN_LOCK_MAX_DELAY: z.coerce.number().int().default(3000),
    RUN_ENGINE_RUN_LOCK_BACKOFF_MULTIPLIER: z.coerce.number().default(1.8),
    RUN_ENGINE_RUN_LOCK_JITTER_FACTOR: z.coerce.number().default(0.15),
    RUN_ENGINE_RUN_LOCK_MAX_TOTAL_WAIT_TIME: z.coerce.number().int().default(15000),

    RUN_ENGINE_SUSPENDED_HEARTBEAT_RETRIES_MAX_COUNT: z.coerce.number().int().default(12),
    RUN_ENGINE_SUSPENDED_HEARTBEAT_RETRIES_MAX_DELAY_MS: z.coerce
      .number()
      .int()
      .default(60_000 * 60 * 6),
    RUN_ENGINE_SUSPENDED_HEARTBEAT_RETRIES_INITIAL_DELAY_MS: z.coerce
      .number()
      .int()
      .default(60_000),
    RUN_ENGINE_SUSPENDED_HEARTBEAT_RETRIES_FACTOR: z.coerce.number().default(2),

    /**
     * Optional ceiling on how long a debounced run can be pushed back, measured from the first
     * trigger. Unset by default: a continuously triggered debounce key is pushed back for as
     * long as the triggers keep coming, and `debounce.maxDelay` on the trigger is the only
     * bound. Setting this applies a ceiling to every debounced run that does not carry its own
     * `maxDelay`, and any `delay` at or above it is rejected at trigger time. It is a default
     * rather than an enforced limit: a trigger that sets `maxDelay` uses that value even when it
     * is longer than this. `0` and blank both mean no ceiling.
     */
    RUN_ENGINE_MAXIMUM_DEBOUNCE_DURATION_MS: OptionalLimitEnv,

    /**
     * Bucket size in milliseconds used to quantize the newly computed `delayUntil`
     * in the debounce system. Quantization collapses concurrent triggers on the
     * same hot debounce key onto the same target time so the unlocked fast-path
     * skip is effective. Set to 0 to disable. Default: 1000ms (1s).
     */
    RUN_ENGINE_DEBOUNCE_QUANTIZE_NEW_DELAY_UNTIL_MS: z.coerce.number().int().min(0).default(1000),

    /**
     * Whether the unlocked fast-path skip is enabled in the debounce system.
     * Acts as a kill switch in case the fast-path needs to be disabled in
     * production without a redeploy. Default: "1" (enabled).
     */
    RUN_ENGINE_DEBOUNCE_FAST_PATH_SKIP_ENABLED: z.string().default("1"),

    RUN_ENGINE_WORKER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    RUN_ENGINE_WORKER_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    RUN_ENGINE_WORKER_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    RUN_ENGINE_WORKER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    RUN_ENGINE_WORKER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    RUN_ENGINE_WORKER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    RUN_ENGINE_WORKER_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),

    RUN_ENGINE_RUN_QUEUE_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    RUN_ENGINE_RUN_QUEUE_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    RUN_ENGINE_RUN_QUEUE_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    RUN_ENGINE_RUN_QUEUE_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    RUN_ENGINE_RUN_QUEUE_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    RUN_ENGINE_RUN_QUEUE_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    RUN_ENGINE_RUN_QUEUE_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),

    RUN_ENGINE_RUN_LOCK_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    RUN_ENGINE_RUN_LOCK_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    RUN_ENGINE_RUN_LOCK_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    RUN_ENGINE_RUN_LOCK_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    RUN_ENGINE_RUN_LOCK_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    RUN_ENGINE_RUN_LOCK_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    RUN_ENGINE_RUN_LOCK_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),

    RUN_ENGINE_DEV_PRESENCE_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    RUN_ENGINE_DEV_PRESENCE_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    RUN_ENGINE_DEV_PRESENCE_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    RUN_ENGINE_DEV_PRESENCE_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    RUN_ENGINE_DEV_PRESENCE_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    RUN_ENGINE_DEV_PRESENCE_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    RUN_ENGINE_DEV_PRESENCE_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),

    //API Rate limiting
    /**
     * @example "60s"
     * @example "1m"
     * @example "1h"
     * @example "1d"
     * @example "1000ms"
     * @example "1000s"
     */
    RUN_ENGINE_RATE_LIMIT_REFILL_INTERVAL: z.string().default("10s"), // refill 250 tokens every 10 seconds
    RUN_ENGINE_RATE_LIMIT_MAX: z.coerce.number().int().default(1200), // allow bursts of 750 requests
    RUN_ENGINE_RATE_LIMIT_REFILL_RATE: z.coerce.number().int().default(400), // refix 250 tokens every 10 seconds
    RUN_ENGINE_RATE_LIMIT_REQUEST_LOGS_ENABLED: z.string().default("0"),
    RUN_ENGINE_RATE_LIMIT_REJECTION_LOGS_ENABLED: z.string().default("1"),
    RUN_ENGINE_RATE_LIMIT_LIMITER_LOGS_ENABLED: z.string().default("0"),

    RUN_ENGINE_RELEASE_CONCURRENCY_ENABLED: z.string().default("0"),
    RUN_ENGINE_RELEASE_CONCURRENCY_DISABLE_CONSUMERS: z.string().default("0"),
    RUN_ENGINE_RELEASE_CONCURRENCY_MAX_TOKENS_RATIO: z.coerce.number().default(1),
    RUN_ENGINE_RELEASE_CONCURRENCY_RELEASINGS_MAX_AGE: z.coerce
      .number()
      .int()
      .default(60_000 * 30),
    RUN_ENGINE_RELEASE_CONCURRENCY_RELEASINGS_POLL_INTERVAL: z.coerce
      .number()
      .int()
      .default(60_000),
    RUN_ENGINE_RELEASE_CONCURRENCY_MAX_RETRIES: z.coerce.number().int().default(3),
    RUN_ENGINE_RELEASE_CONCURRENCY_CONSUMERS_COUNT: z.coerce.number().int().default(1),
    RUN_ENGINE_RELEASE_CONCURRENCY_POLL_INTERVAL: z.coerce.number().int().default(500),
    RUN_ENGINE_RELEASE_CONCURRENCY_BATCH_SIZE: z.coerce.number().int().default(10),

    RUN_ENGINE_WORKER_ENABLED: z.string().default("1"),
    RUN_ENGINE_WORKER_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
    RUN_ENGINE_RUN_QUEUE_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    RUN_ENGINE_TREAT_PRODUCTION_EXECUTION_STALLS_AS_OOM: z.string().default("0"),
    RUN_ENGINE_READ_REPLICA_SNAPSHOTS_SINCE_ENABLED: z.string().default("0"),
    RUN_ENGINE_SNAPSHOTS_SINCE_REPLICA_RETRY_MIN_MS: z.coerce.number().int().default(50),
    RUN_ENGINE_SNAPSHOTS_SINCE_REPLICA_RETRY_MAX_MS: z.coerce.number().int().default(200),
    RUN_ENGINE_DEBOUNCE_USE_REPLICA_FOR_FAST_PATH_READ: z.string().default("0"),

    /** How long should the presence ttl last */
    DEV_PRESENCE_SSE_TIMEOUT: z.coerce.number().int().default(30_000),
    DEV_PRESENCE_TTL_MS: z.coerce.number().int().default(5_000),
    DEV_PRESENCE_POLL_MS: z.coerce.number().int().default(1_000),
    /** How many ms to wait until dequeuing again, if there was a run last time */
    DEV_DEQUEUE_INTERVAL_WITH_RUN: z.coerce.number().int().default(250),
    /** How many ms to wait until dequeuing again, if there was no run last time */
    DEV_DEQUEUE_INTERVAL_WITHOUT_RUN: z.coerce.number().int().default(1_000),
    /** The max number of runs per API call that we'll dequeue in DEV */
    DEV_DEQUEUE_MAX_RUNS_PER_PULL: z.coerce.number().int().default(10),

    /** The maximum concurrent local run processes executing at once in dev. This is a hard limit */
    DEV_MAX_CONCURRENT_RUNS: z.coerce.number().int().optional(),

    /** The CLI should connect to this for dev runs */
    DEV_ENGINE_URL: z.string().default(process.env.APP_ORIGIN ?? "http://localhost:3030"),

    COMMON_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
    COMMON_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
    COMMON_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
    COMMON_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
    COMMON_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
    COMMON_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(50),
    COMMON_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
    COMMON_WORKER_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),

    COMMON_WORKER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    COMMON_WORKER_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    COMMON_WORKER_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    COMMON_WORKER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    COMMON_WORKER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    COMMON_WORKER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    COMMON_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
    COMMON_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

    // Global default for the scheduled worker-queue split. When "1", runs in a
    // scheduled lineage (rootTriggerSource === "schedule") are routed to a
    // dedicated `<region>:scheduled` worker queue so a separate consumer fleet
    // can dequeue them independently of standard/agent runs. The per-org
    // `workerQueueScheduledSplitEnabled` feature flag overrides this default in
    // BOTH directions (an org set to false stays on the single queue even when
    // this is "1"; an org set to true splits even when this is "0"). Never
    // applies to DEVELOPMENT environments.
    TRIGGER_WORKER_QUEUE_SCHEDULED_SPLIT_ENABLED: z.string().default("0"),

    TRIGGER_MOLLIFIER_ENABLED: z.string().default("0"),
    // Separate switch for the drainer (consumer side) so it can be split
    // off onto a dedicated worker service. Unset → inherits
    // TRIGGER_MOLLIFIER_ENABLED, so single-container self-hosters don't have to
    // flip two switches. Multi-replica drainers are correct — `popAndMarkDraining`
    // is an atomic RPOP + status flip in one Lua call, so only one replica
    // can win any given entry — but inefficient: polling load (SMEMBERS +
    // per-env scans) multiplies by N, and `TRIGGER_MOLLIFIER_DRAIN_CONCURRENCY`
    // is per-process so engine load also multiplies. Splitting the drainer
    // onto a dedicated worker keeps that traffic off the request-serving
    // replicas. `TRIGGER_MOLLIFIER_ENABLED` is still the master kill switch;
    // setting this to "1" while `TRIGGER_MOLLIFIER_ENABLED` is "0" is a
    // no-op because the gate-side singleton refuses to construct a buffer
    // when the system is off.
    TRIGGER_MOLLIFIER_DRAINER_ENABLED: z
      .string()
      .default(process.env.TRIGGER_MOLLIFIER_ENABLED ?? "0"),
    TRIGGER_MOLLIFIER_SHADOW_MODE: z.string().default("0"),
    TRIGGER_MOLLIFIER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    TRIGGER_MOLLIFIER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    TRIGGER_MOLLIFIER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    TRIGGER_MOLLIFIER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    TRIGGER_MOLLIFIER_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),
    TRIGGER_MOLLIFIER_TRIP_WINDOW_MS: z.coerce.number().int().positive().default(200),
    TRIGGER_MOLLIFIER_TRIP_THRESHOLD: z.coerce.number().int().positive().default(100),
    TRIGGER_MOLLIFIER_HOLD_MS: z.coerce.number().int().positive().default(500),
    TRIGGER_MOLLIFIER_DRAIN_CONCURRENCY: z.coerce.number().int().positive().default(50),
    TRIGGER_MOLLIFIER_DRAIN_MAX_ATTEMPTS: z.coerce.number().int().positive().default(3),
    TRIGGER_MOLLIFIER_DRAIN_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().positive().default(30_000),
    TRIGGER_MOLLIFIER_DRAIN_MAX_ORGS_PER_TICK: z.coerce.number().int().positive().default(500),
    // Per-env per-tick pop cap. The drainer rotates one env per org per
    // tick; this bounds how many entries it pops from that env before
    // dispatching them through the shared `DRAIN_CONCURRENCY`-bounded
    // limiter. Default matches `DRAIN_CONCURRENCY` so a single-env burst
    // uses the full handler-parallelism budget — for 20k buffered on one
    // env this is the difference between ~17m (one-pop-per-tick × ~50ms)
    // and ~20s (400 ticks × concurrent engine.trigger). Org/env fairness
    // is preserved because the per-tick env selection is unchanged; only
    // the in-env pop count grows.
    TRIGGER_MOLLIFIER_DRAIN_BATCH_SIZE: z.coerce.number().int().positive().default(50),
    // Periodic sweep that scans buffer queue LISTs for entries whose
    // dwell exceeds the stale threshold. Independent of the drainer —
    // its job is exactly to make a stuck/offline drainer visible to
    // ops. Defaults: explicitly opt-in (a separate kill switch from
    // the mollifier itself), run every 5 minutes, alert on anything
    // that's been dwelling for 5+ minutes (matches the sweep interval
    // — "anything still here when we check" is the simplest threshold
    // that converges).
    //
    // The sweep was previously defaulting to inherit
    // `TRIGGER_MOLLIFIER_ENABLED`, which meant any deployment already
    // running with the mollifier on would auto-start the sweep worker
    // on upgrade — turning on new background load with no explicit
    // rollout step. Hard-defaulting to "0" preserves the intent of
    // exposing the sweep as a separate switch.
    TRIGGER_MOLLIFIER_STALE_SWEEP_ENABLED: z.string().default("0"),
    TRIGGER_MOLLIFIER_STALE_SWEEP_INTERVAL_MS: z.coerce
      .number()
      .int()
      .positive()
      .default(5 * 60_000),
    TRIGGER_MOLLIFIER_STALE_SWEEP_THRESHOLD_MS: z.coerce
      .number()
      .int()
      .positive()
      .default(5 * 60_000),
    // Bounds for one stale-sweep pass (see mollifierStaleSweep.server.ts).
    // Max entries scanned per env and max orgs visited per tick — together
    // they cap the Redis traffic / wall-time of a single sweep pass.
    TRIGGER_MOLLIFIER_STALE_SWEEP_MAX_ENTRIES_PER_ENV: z.coerce
      .number()
      .int()
      .positive()
      .default(1000),
    TRIGGER_MOLLIFIER_STALE_SWEEP_MAX_ORGS_PER_PASS: z.coerce
      .number()
      .int()
      .positive()
      .default(100),

    // --- Mollifier buffer internals (wired into MollifierBuffer in
    // mollifierBuffer.server.ts). ---
    // Grace TTL applied to the entry hash on drainer ack so direct reads
    // (retrieve, trace) have a safety net while PG replica lag settles.
    TRIGGER_MOLLIFIER_ACK_GRACE_TTL_SECONDS: z.coerce.number().int().positive().default(30),
    // ioredis per-request retry limit on the buffer's Redis client.
    TRIGGER_MOLLIFIER_REDIS_MAX_RETRIES_PER_REQUEST: z.coerce.number().int().positive().default(20),
    // ioredis reconnect backoff envelope for the buffer client: the base
    // grows by `STEP_MS` per attempt, capped at `MAX_MS`, then equal-jittered.
    TRIGGER_MOLLIFIER_REDIS_RECONNECT_STEP_MS: z.coerce.number().int().positive().default(50),
    TRIGGER_MOLLIFIER_REDIS_RECONNECT_MAX_MS: z.coerce.number().int().positive().default(1000),

    // --- Mollifier drainer loop internals (wired into MollifierDrainer in
    // mollifierDrainer.server.ts). ---
    // Tick gap when a tick drained nothing; under backlog ticks run back-to-back.
    TRIGGER_MOLLIFIER_DRAIN_POLL_INTERVAL_MS: z.coerce.number().int().positive().default(100),
    // Cap on the drainer's exponential backoff after consecutive runOnce errors.
    TRIGGER_MOLLIFIER_DRAIN_MAX_BACKOFF_MS: z.coerce.number().int().positive().default(5_000),
    // Floor for the drainer's backoff base (so a tiny poll interval doesn't
    // collapse the backoff to near-zero during a sustained outage).
    TRIGGER_MOLLIFIER_DRAIN_BACKOFF_FLOOR_MS: z.coerce.number().int().positive().default(100),
    // Required margin between the drainer's shutdown deadline and
    // GRACEFUL_SHUTDOWN_TIMEOUT; boot fails loud if the timeout leaves less.
    TRIGGER_MOLLIFIER_DRAIN_SHUTDOWN_MARGIN_MS: z.coerce.number().int().positive().default(1_000),

    // How often the draining-tracker ZSET cardinality is polled into the gauge.
    TRIGGER_MOLLIFIER_DRAINING_GAUGE_INTERVAL_MS: z.coerce
      .number()
      .int()
      .positive()
      .default(15_000),

    // --- Pre-gate idempotency claim (idempotencyClaim.server.ts). ---
    // TTL on the claim key (and the upper clamp on the customer-derived
    // claim TTL), how long a waiter blocks before timing out, and the
    // waiter poll interval.
    TRIGGER_MOLLIFIER_CLAIM_TTL_SECONDS: z.coerce.number().int().positive().default(30),
    // Pipeline floor: the claim never shrinks below this even for a short customer key TTL, so it
    // can't expire mid-pipeline and let a loser re-claim (cross-DB duplicate under the split).
    TRIGGER_MOLLIFIER_CLAIM_MIN_TTL_SECONDS: z.coerce.number().int().positive().default(5),
    TRIGGER_MOLLIFIER_CLAIM_WAIT_MS: z.coerce.number().int().positive().default(5_000),
    TRIGGER_MOLLIFIER_CLAIM_POLL_MS: z.coerce.number().int().positive().default(25),

    // --- Buffered-run mutate-with-fallback wait loop (mutateWithFallback.server.ts). ---
    // Ceiling on the wait-for-materialisation loop, initial poll gap, the
    // backoff ceiling, and the exponential growth factor (a float).
    TRIGGER_MOLLIFIER_MUTATE_SAFETY_NET_MS: z.coerce.number().int().positive().default(2_000),
    TRIGGER_MOLLIFIER_MUTATE_POLL_STEP_MS: z.coerce.number().int().positive().default(20),
    TRIGGER_MOLLIFIER_MUTATE_MAX_POLL_STEP_MS: z.coerce.number().int().positive().default(250),
    TRIGGER_MOLLIFIER_MUTATE_BACKOFF_FACTOR: z.coerce.number().gt(1).default(1.7),

    // --- Buffered-run metadata CAS retry loop (applyMetadataMutation.server.ts). ---
    // Retry budget for concurrent metadata writers, and the jittered
    // conflict-backoff envelope: random in [0, base + attempt * step) ms.
    TRIGGER_MOLLIFIER_METADATA_MAX_RETRIES: z.coerce.number().int().positive().default(12),
    TRIGGER_MOLLIFIER_METADATA_BACKOFF_BASE_MS: z.coerce.number().int().positive().default(5),
    TRIGGER_MOLLIFIER_METADATA_BACKOFF_STEP_MS: z.coerce.number().int().positive().default(5),

    BATCH_TRIGGER_PROCESS_JOB_VISIBILITY_TIMEOUT_MS: z.coerce
      .number()
      .int()
      .default(60_000 * 5), // 5 minutes

    BATCH_TRIGGER_CACHED_RUNS_CHECK_ENABLED: BoolEnv.default(false),

    BATCH_TRIGGER_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
    BATCH_TRIGGER_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
    BATCH_TRIGGER_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
    BATCH_TRIGGER_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
    BATCH_TRIGGER_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
    BATCH_TRIGGER_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(20),
    BATCH_TRIGGER_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
    BATCH_TRIGGER_WORKER_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),

    BATCH_TRIGGER_WORKER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    BATCH_TRIGGER_WORKER_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    BATCH_TRIGGER_WORKER_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    BATCH_TRIGGER_WORKER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    BATCH_TRIGGER_WORKER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    BATCH_TRIGGER_WORKER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    BATCH_TRIGGER_WORKER_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),
    BATCH_TRIGGER_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

    // BatchQueue DRR settings (Run Engine v2)
    BATCH_QUEUE_DRR_QUANTUM: z.coerce.number().int().default(25),
    BATCH_QUEUE_MAX_DEFICIT: z.coerce.number().int().default(100),
    BATCH_QUEUE_CONSUMER_COUNT: z.coerce.number().int().default(3),
    BATCH_QUEUE_CONSUMER_INTERVAL_MS: z.coerce.number().int().default(50),
    BATCH_QUEUE_WORKER_ENABLED: BoolEnv.default(true),
    // Number of master queue shards for horizontal scaling
    BATCH_QUEUE_SHARD_COUNT: z.coerce.number().int().default(1),
    // Maximum queues to fetch from master queue per iteration
    BATCH_QUEUE_MASTER_QUEUE_LIMIT: z.coerce.number().int().default(1000),
    // Enable worker queue for two-stage processing (claim messages, push to worker queue, process from worker queue)
    BATCH_QUEUE_WORKER_QUEUE_ENABLED: BoolEnv.default(true),
    // Worker queue blocking timeout in seconds (for two-stage processing, only used when BATCH_QUEUE_WORKER_QUEUE_ENABLED is true)
    BATCH_QUEUE_WORKER_QUEUE_TIMEOUT_SECONDS: z.coerce.number().int().default(10),
    // Global rate limit: max items processed per second across all consumers
    // If not set, no global rate limiting is applied
    BATCH_QUEUE_GLOBAL_RATE_LIMIT: z.coerce.number().int().positive().optional(),
    // Max items in the worker queue before claiming pauses (protects visibility timeouts)
    // If not set, no depth limit is applied
    BATCH_QUEUE_WORKER_QUEUE_MAX_DEPTH: z.coerce.number().int().positive().optional(),

    ADMIN_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
    ADMIN_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
    ADMIN_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
    ADMIN_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
    ADMIN_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
    ADMIN_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(20),
    ADMIN_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
    ADMIN_WORKER_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),

    ADMIN_WORKER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    ADMIN_WORKER_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    ADMIN_WORKER_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    ADMIN_WORKER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    ADMIN_WORKER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    ADMIN_WORKER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    ADMIN_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
    ADMIN_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

    ALERTS_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
    ALERTS_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
    ALERTS_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
    ALERTS_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
    ALERTS_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(100),
    ALERTS_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(50),
    ALERTS_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
    ALERTS_WORKER_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),

    ALERTS_WORKER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    ALERTS_WORKER_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    ALERTS_WORKER_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    ALERTS_WORKER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    ALERTS_WORKER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    ALERTS_WORKER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    ALERTS_WORKER_REDIS_TLS_DISABLED: z.string().default(process.env.REDIS_TLS_DISABLED ?? "false"),
    ALERTS_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

    BILLING_LIMIT_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
    BILLING_LIMIT_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
    BILLING_LIMIT_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
    BILLING_LIMIT_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
    BILLING_LIMIT_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
    BILLING_LIMIT_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(20),
    BILLING_LIMIT_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(60_000),
    BILLING_LIMIT_WORKER_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    BILLING_LIMIT_RECONCILE_INTERVAL_MS: z.coerce.number().int().default(90_000),
    BILLING_LIMIT_WORKER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    BILLING_LIMIT_WORKER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    BILLING_LIMIT_WORKER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    BILLING_LIMIT_WORKER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    BILLING_LIMIT_WORKER_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),

    SCHEDULE_ENGINE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
    SCHEDULE_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
    SCHEDULE_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
    SCHEDULE_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
    SCHEDULE_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
    SCHEDULE_WORKER_IMMEDIATE_POLL_INTERVAL: z.coerce.number().int().default(50),
    SCHEDULE_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(50),
    SCHEDULE_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(30_000),
    SCHEDULE_WORKER_DISTRIBUTION_WINDOW_SECONDS: z.coerce.number().int().default(30),
    SCHEDULE_WORKER_CRON_SPREAD_FRACTION: z.coerce
      .number()
      .catch(0)
      .default(0)
      .transform((value) => (Number.isFinite(value) ? Math.min(1, Math.max(0, value)) : 0)),

    SCHEDULE_WORKER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    SCHEDULE_WORKER_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    SCHEDULE_WORKER_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    SCHEDULE_WORKER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    SCHEDULE_WORKER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    SCHEDULE_WORKER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    SCHEDULE_WORKER_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),
    SCHEDULE_WORKER_REDIS_CLUSTER_MODE_ENABLED: z.string().default("0"),

    WEBHOOK_ENGINE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
    WEBHOOK_WORKER_ENABLED: z.string().default(process.env.WORKER_ENABLED ?? "true"),
    WEBHOOK_WORKER_CONCURRENCY_LIMIT: z.coerce.number().int().default(50),
    WEBHOOK_WORKER_CONCURRENCY_WORKERS: z.coerce.number().int().default(2),
    WEBHOOK_WORKER_CONCURRENCY_TASKS_PER_WORKER: z.coerce.number().int().default(10),
    WEBHOOK_WORKER_POLL_INTERVAL: z.coerce.number().int().default(1000),
    WEBHOOK_WORKER_SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().default(30_000),
    WEBHOOK_ENABLED: z.string().default("0"),

    WEBHOOK_WORKER_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    WEBHOOK_WORKER_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    WEBHOOK_WORKER_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    WEBHOOK_WORKER_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    WEBHOOK_WORKER_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),

    WEBHOOK_PARTITION_ENSURE_SCHEDULE: z.string().optional(),
    WEBHOOK_PARTITION_ENSURE_JITTER_MS: z.coerce.number().int().optional(),
    WEBHOOK_PARTITION_LOOKAHEAD_DAYS: z.coerce.number().int().default(10),
    WEBHOOK_PARTITION_RETENTION_DAYS: z.coerce.number().int().default(60),

    // Ingest hot-path cache for the endpoint + resolved signing secret (keyed by opaqueId). 0 disables.
    WEBHOOK_ENDPOINT_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
    WEBHOOK_ENDPOINT_CACHE_MAX_SIZE: z.coerce.number().int().default(10_000),

    WEBHOOK_INGRESS_ENABLED: z.string().default("1"),

    // Public origin for the webhook ingress URL shown to users / returned by the API. Defaults to the
    // API origin; set to a dedicated host (e.g. https://webhook.trigger.dev) when one is fronted.
    WEBHOOK_INGRESS_ORIGIN: z.string().optional(),
    WEBHOOK_INGRESS_BODY_SIZE_LIMIT_MB: z.coerce.number().int().default(1),
    WEBHOOK_INGRESS_RATE_LIMIT_WINDOW: z.string().default("10s"),
    WEBHOOK_INGRESS_RATE_LIMIT_TOKENS: z.coerce.number().int().default(100),
    WEBHOOK_INGRESS_IP_RATE_LIMIT_WINDOW: z.string().default("10s"),
    WEBHOOK_INGRESS_IP_RATE_LIMIT_TOKENS: z.coerce.number().int().default(300),
    WEBHOOK_FRONT_GATE_DEFAULT_TTL_SECONDS: z.coerce
      .number()
      .int()
      .default(6 * 60 * 60),
    WEBHOOK_FRONT_GATE_MAX_TTL_SECONDS: z.coerce
      .number()
      .int()
      .default(6 * 60 * 60),

    TASK_EVENT_PARTITIONING_ENABLED: z.string().default("0"),
    TASK_EVENT_PARTITIONED_WINDOW_IN_SECONDS: z.coerce.number().int().default(60), // 1 minute

    DEPLOYMENTS_AUTORELOAD_POLL_INTERVAL_MS: z.coerce.number().int().default(5_000),
    BULK_ACTION_AUTORELOAD_POLL_INTERVAL_MS: z.coerce.number().int().default(1_000),
    QUEUES_AUTORELOAD_POLL_INTERVAL_MS: z.coerce.number().int().default(5_000),

    SLACK_BOT_TOKEN: z.string().optional(),
    SLACK_SIGNUP_REASON_CHANNEL_ID: z.string().optional(),

    // kapa.ai — read by the root loader; unset turns Ask AI off, and ⌘I then opens the agent
    // for users who have agent access, or does nothing for everyone else.
    KAPA_AI_WEBSITE_ID: z.string().optional(),

    // BetterStack
    BETTERSTACK_API_KEY: z.string().optional(),
    BETTERSTACK_STATUS_PAGE_ID: z.string().optional(),

    RUN_REPLICATION_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    RUN_REPLICATION_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    RUN_REPLICATION_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    RUN_REPLICATION_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    RUN_REPLICATION_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    RUN_REPLICATION_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    RUN_REPLICATION_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),

    RUN_REPLICATION_CLICKHOUSE_URL: z.string().optional(),
    RUN_REPLICATION_ENABLED: z.string().default("0"),
    RUN_REPLICATION_SLOT_NAME: z.string().default("task_runs_to_clickhouse_v1"),
    RUN_REPLICATION_PUBLICATION_NAME: z.string().default("task_runs_to_clickhouse_v1_publication"),
    RUN_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(2),
    RUN_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
    RUN_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
    RUN_REPLICATION_MAX_POISON_STRIPS_PER_BATCH: z.coerce.number().int().default(1),
    RUN_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
    RUN_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
    RUN_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),
    RUN_REPLICATION_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
    RUN_REPLICATION_CLICKHOUSE_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    RUN_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS: z.coerce.number().int().default(10_000),
    RUN_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS: z.coerce.number().int().default(500),
    RUN_REPLICATION_WAIT_FOR_ASYNC_INSERT: z.string().default("0"),
    RUN_REPLICATION_KEEP_ALIVE_ENABLED: z.string().default("0"),
    RUN_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
    RUN_REPLICATION_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
    // Retry configuration for insert operations
    RUN_REPLICATION_INSERT_MAX_RETRIES: z.coerce.number().int().default(3),
    RUN_REPLICATION_INSERT_BASE_DELAY_MS: z.coerce.number().int().default(100),
    RUN_REPLICATION_INSERT_MAX_DELAY_MS: z.coerce.number().int().default(2000),
    RUN_REPLICATION_INSERT_STRATEGY: z.enum(["insert", "insert_async"]).default("insert"),
    RUN_REPLICATION_DISABLE_PAYLOAD_INSERT: z.string().default("0"),
    RUN_REPLICATION_DISABLE_ERROR_FINGERPRINTING: z.string().default("0"),

    // Connection URL for the LEGACY runs-replication source (the runs-CDC slot on the legacy runs DB, plus
    // the admin recovery route). Direct, not pooled: replication can't run over a pooler. Optional; unset ->
    // falls back to DATABASE_URL, so nothing changes today.
    RUN_REPLICATION_LEGACY_DATABASE_URL: z
      .string()
      .refine(isValidDatabaseUrl, "RUN_REPLICATION_LEGACY_DATABASE_URL is invalid")
      .optional(),

    // --- Run-ops DB split — second replication source (the NEW dedicated run-ops DB). ---
    // Cloud-only; only consulted when isSplitEnabled() is true. Self-host never sets these.
    // Connection URL for the run-ops DB used by the runs-replication source. Required when the split is
    // enabled (unset → boot fails via SplitReplicationMisconfiguredError, no silent fallback). Kept
    // separate from the app's RUN_OPS_DATABASE_URL: replication can't run over a pooler, so this is direct.
    RUN_REPLICATION_RUN_OPS_DATABASE_URL: z
      .string()
      .refine(isValidDatabaseUrl, "RUN_REPLICATION_RUN_OPS_DATABASE_URL is invalid")
      .optional(),
    RUN_REPLICATION_NEW_SLOT_NAME: z.string().default("task_runs_to_clickhouse_v2"),
    RUN_REPLICATION_NEW_PUBLICATION_NAME: z
      .string()
      .default("task_runs_to_clickhouse_v2_publication"),
    RUN_REPLICATION_NEW_ENABLED: z.string().default("0"),
    // Origin generations packed into _version via composeTaskRunVersion.
    // Legacy DB = 0, new dedicated run-ops DB = 1. Exposed as env so the mapping is auditable
    // per-deploy, but DEFAULTS encode the canonical legacy=0 / new=1 contract.
    RUN_REPLICATION_LEGACY_ORIGIN_GENERATION: z.coerce.number().int().default(0),
    RUN_REPLICATION_NEW_ORIGIN_GENERATION: z.coerce.number().int().default(1),

    // Run-ops id mint cutover — per-env, canary-first, OFF by default.
    // Even when on, an env mints run-ops ids only if its per-org runOpsMintKind flag is
    // "runOpsId" AND isSplitEnabled() is true. Cache mirrors REALTIME_BACKEND_FLAG_CACHE_*.
    RUN_OPS_MINT_ENABLED: BoolEnv.default(false),
    RUN_OPS_MINT_FLAG_CACHE_TTL_MS: z.coerce.number().int().default(30_000),
    RUN_OPS_MINT_FLAG_CACHE_MAX_ENTRIES: z.coerce.number().int().default(10_000),
    // Deterministic wall-clock cutover after a runOpsMintKind flip. Must exceed the sum
    // of RUN_OPS_MINT_FLAG_CACHE_TTL_MS and the control-plane cache TTL so every process
    // (stale or fresh) resolves to the same kind for the whole window. See mintFlipGrace.ts.
    RUN_OPS_MINT_FLIP_GRACE_MS: z.coerce.number().int().default(90_000),

    // Session replication (Postgres → ClickHouse sessions_v1). Shares Redis
    // with the runs replicator for leader locking but has its own slot and
    // publication so the two consume independently.
    SESSION_REPLICATION_CLICKHOUSE_URL: z.string().optional(),
    SESSION_REPLICATION_ENABLED: z.string().default("0"),
    SESSION_REPLICATION_SLOT_NAME: z.string().default("sessions_to_clickhouse_v1"),
    SESSION_REPLICATION_PUBLICATION_NAME: z
      .string()
      .default("sessions_to_clickhouse_v1_publication"),
    // Connection URL for the sessions-replication slot. Direct, not pooled: replication can't run over a
    // pooler. Optional; unset -> falls back to DATABASE_URL, so nothing changes today.
    SESSION_REPLICATION_DATABASE_URL: z
      .string()
      .refine(isValidDatabaseUrl, "SESSION_REPLICATION_DATABASE_URL is invalid")
      .optional(),
    SESSION_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(1),
    SESSION_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
    SESSION_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
    SESSION_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
    SESSION_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce.number().int().default(10_000),
    SESSION_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS: z.coerce.number().int().default(10_000),
    SESSION_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS: z.coerce.number().int().default(500),
    SESSION_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),
    SESSION_REPLICATION_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    SESSION_REPLICATION_CLICKHOUSE_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    SESSION_REPLICATION_WAIT_FOR_ASYNC_INSERT: z.string().default("0"),
    SESSION_REPLICATION_KEEP_ALIVE_ENABLED: z.string().default("0"),
    SESSION_REPLICATION_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
    SESSION_REPLICATION_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
    SESSION_REPLICATION_INSERT_STRATEGY: z.enum(["insert", "insert_async"]).default("insert"),
    SESSION_REPLICATION_INSERT_MAX_RETRIES: z.coerce.number().int().default(3),
    SESSION_REPLICATION_INSERT_BASE_DELAY_MS: z.coerce.number().int().default(100),
    SESSION_REPLICATION_INSERT_MAX_DELAY_MS: z.coerce.number().int().default(2000),

    // Webhook deliveries replication (Postgres → ClickHouse webhook_deliveries_v1).
    // Shares Redis with the runs replicator for leader locking but has its own
    // slot and publication so the two consume independently. The source table is
    // a partitioned parent, so the publication is created with
    // publish_via_partition_root.
    WEBHOOK_DELIVERIES_REPLICATION_CLICKHOUSE_URL: z.string().optional(),
    WEBHOOK_DELIVERIES_REPLICATION_ENABLED: z.string().default("0"),
    WEBHOOK_DELIVERIES_REPLICATION_SLOT_NAME: z
      .string()
      .default("webhook_deliveries_to_clickhouse_v1"),
    WEBHOOK_DELIVERIES_REPLICATION_PUBLICATION_NAME: z
      .string()
      .default("webhook_deliveries_to_clickhouse_v1_publication"),
    WEBHOOK_DELIVERIES_REPLICATION_MAX_FLUSH_CONCURRENCY: z.coerce.number().int().default(1),
    WEBHOOK_DELIVERIES_REPLICATION_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
    WEBHOOK_DELIVERIES_REPLICATION_FLUSH_BATCH_SIZE: z.coerce.number().int().default(100),
    WEBHOOK_DELIVERIES_REPLICATION_LEADER_LOCK_TIMEOUT_MS: z.coerce.number().int().default(30_000),
    WEBHOOK_DELIVERIES_REPLICATION_LEADER_LOCK_EXTEND_INTERVAL_MS: z.coerce
      .number()
      .int()
      .default(10_000),
    WEBHOOK_DELIVERIES_REPLICATION_LEADER_LOCK_ADDITIONAL_TIME_MS: z.coerce
      .number()
      .int()
      .default(10_000),
    WEBHOOK_DELIVERIES_REPLICATION_LEADER_LOCK_RETRY_INTERVAL_MS: z.coerce
      .number()
      .int()
      .default(500),
    WEBHOOK_DELIVERIES_REPLICATION_ACK_INTERVAL_SECONDS: z.coerce.number().int().default(10),
    WEBHOOK_DELIVERIES_REPLICATION_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    WEBHOOK_DELIVERIES_REPLICATION_WAIT_FOR_ASYNC_INSERT: z.string().default("0"),
    WEBHOOK_DELIVERIES_REPLICATION_INSERT_STRATEGY: z
      .enum(["insert", "insert_async"])
      .default("insert"),
    WEBHOOK_DELIVERIES_REPLICATION_INSERT_MAX_RETRIES: z.coerce.number().int().default(3),
    WEBHOOK_DELIVERIES_REPLICATION_INSERT_BASE_DELAY_MS: z.coerce.number().int().default(100),
    WEBHOOK_DELIVERIES_REPLICATION_INSERT_MAX_DELAY_MS: z.coerce.number().int().default(2000),

    // Clickhouse
    CLICKHOUSE_URL: z.string(),
    // Optional read replica endpoint. Read-only clients (logs, query, admin, runsList,
    // engine, realtime) default to this when their own URL is unset; writes always stay on
    // CLICKHOUSE_URL. Events reads opt in separately via EVENTS_READER_CLICKHOUSE_URL (no
    // fallback here). Must share storage with the CLICKHOUSE_URL warehouse.
    CLICKHOUSE_READER_URL: z.string().optional(),
    CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
    CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
    CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
    CLICKHOUSE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
    CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),

    // Logs Query Settings
    CLICKHOUSE_LOGS_LIST_MAX_MEMORY_USAGE: z.coerce.number().int().default(1_000_000_000),
    CLICKHOUSE_LOGS_LIST_MAX_BYTES_BEFORE_EXTERNAL_SORT: z.coerce
      .number()
      .int()
      .default(256_000_000),
    CLICKHOUSE_LOGS_LIST_MAX_THREADS: z.coerce.number().int().default(2),
    CLICKHOUSE_LOGS_LIST_MAX_ROWS_TO_READ: z.coerce.number().int().default(10_000_000),
    CLICKHOUSE_LOGS_LIST_MAX_EXECUTION_TIME: z.coerce.number().int().default(120),
    // Bound read-in-order memory on object-storage reads: each part opens a per-column read
    // stream, and the default ~1 MiB+ S3 buffers dominate peak memory. These two byte sizes
    // cap the per-stream buffers and exist on every supported ClickHouse, so they are always on.
    CLICKHOUSE_LOGS_LIST_PREFETCH_BUFFER_SIZE: z.coerce
      .number()
      .int()
      .nonnegative()
      .default(262_144),
    CLICKHOUSE_LOGS_LIST_MAX_READ_BUFFER_SIZE: z.coerce
      .number()
      .int()
      .nonnegative()
      .default(262_144),
    // The decisive lever on Cloud SharedMergeTree, but it only exists on newer ClickHouse and
    // is a no-op on local-disk MergeTree, so it is opt-in: unset means it is never sent (safe on
    // any self-hosted version). Set to 0 on object-storage deployments to get the memory win.
    CLICKHOUSE_LOGS_LIST_FILESYSTEM_CACHE_PREFER_BIGGER_BUFFER_SIZE: z.coerce
      .number()
      .int()
      .nonnegative()
      .optional(),

    // Logs list pagination tuning (page sizing + recent-first probe windows).
    LOGS_LIST_DEFAULT_PAGE_SIZE: z.coerce.number().int().positive().default(50),
    LOGS_LIST_MAX_PAGE_SIZE: z.coerce.number().int().positive().default(100),
    // Days back from the page ceiling to probe before widening to the full requested window,
    // comma-separated. Empty disables narrowing (a single full-window query).
    LOGS_LIST_RECENT_FIRST_PROBE_DAYS: z
      .string()
      .default("1,7")
      .transform((s) =>
        s
          .split(",")
          .map((v) => Number(v.trim()))
          .filter((n) => Number.isFinite(n) && n > 0)
      ),

    // Query feature flag
    QUERY_FEATURE_ENABLED: z.string().default("1"),

    // AI features (Prompts, Models, AI Metrics sidebar section)
    AI_FEATURES_ENABLED: z.string().default("0"),

    // Logs page ClickHouse URL (for logs queries)
    LOGS_CLICKHOUSE_URL: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.CLICKHOUSE_READER_URL ?? process.env.CLICKHOUSE_URL),

    // Query page ClickHouse limits (for TSQL queries)
    QUERY_CLICKHOUSE_URL: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.CLICKHOUSE_READER_URL ?? process.env.CLICKHOUSE_URL),
    QUERY_CLICKHOUSE_MAX_EXECUTION_TIME: z.coerce.number().int().default(10),
    QUERY_CLICKHOUSE_MAX_MEMORY_USAGE: z.coerce.number().int().default(1_073_741_824), // 1GB in bytes
    QUERY_CLICKHOUSE_MAX_AST_ELEMENTS: z.coerce.number().int().default(4_000_000),
    QUERY_CLICKHOUSE_MAX_EXPANDED_AST_ELEMENTS: z.coerce.number().int().default(4_000_000),
    QUERY_CLICKHOUSE_MAX_BYTES_BEFORE_EXTERNAL_GROUP_BY: z.coerce.number().int().default(0),
    QUERY_CLICKHOUSE_MAX_RETURNED_ROWS: z.coerce.number().int().default(10_000),

    // Query page concurrency limits
    QUERY_DEFAULT_ORG_CONCURRENCY_LIMIT: z.coerce.number().int().default(3),
    QUERY_GLOBAL_CONCURRENCY_LIMIT: z.coerce.number().int().default(100),

    // Metric widget concurrency limits
    METRIC_WIDGET_DEFAULT_ORG_CONCURRENCY_LIMIT: z.coerce.number().int().default(30),

    // Admin ClickHouse URL (for admin dashboard queries like missing models)
    ADMIN_CLICKHOUSE_URL: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.CLICKHOUSE_READER_URL ?? process.env.CLICKHOUSE_URL),

    EVENTS_CLICKHOUSE_URL: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.CLICKHOUSE_URL),
    // Events read replica (traces/spans/logs). No CLICKHOUSE_READER_URL fallback by design: this write-capable client opts in explicitly.
    EVENTS_READER_CLICKHOUSE_URL: z.string().optional(),
    EVENTS_CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
    EVENTS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
    EVENTS_CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
    EVENTS_CLICKHOUSE_LOG_LEVEL: z.enum(["log", "error", "warn", "info", "debug"]).default("info"),
    EVENTS_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
    // ClickHouse client used by @internal/run-engine's PendingVersionSystem.
    // Kept on its own URL + pool so this low-QPS path can't contend with
    // the main analytics client (CLICKHOUSE_URL). Falls back to the main
    // URL when unset so unconfigured environments still work.
    RUN_ENGINE_CLICKHOUSE_URL: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.CLICKHOUSE_READER_URL ?? process.env.CLICKHOUSE_URL),
    RUN_ENGINE_CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
    RUN_ENGINE_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
    RUN_ENGINE_CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(5),
    RUN_ENGINE_CLICKHOUSE_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    RUN_ENGINE_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
    // Dedicated ClickHouse pool for the native backend's tag/batch id resolution; falls back to CLICKHOUSE_URL.
    REALTIME_BACKEND_NATIVE_CLICKHOUSE_URL: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.CLICKHOUSE_READER_URL ?? process.env.CLICKHOUSE_URL),
    REALTIME_BACKEND_NATIVE_CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
    REALTIME_BACKEND_NATIVE_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce
      .number()
      .int()
      .optional(),
    REALTIME_BACKEND_NATIVE_CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
    REALTIME_BACKEND_NATIVE_CLICKHOUSE_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    REALTIME_BACKEND_NATIVE_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
    // Dedicated ClickHouse pool for the runs list (dashboard + API). Lets us point
    // the highest-traffic read path at a read replica without moving ingest/replication
    // writes off CLICKHOUSE_URL. Falls back to CLICKHOUSE_URL when unset.
    RUNS_LIST_CLICKHOUSE_URL: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.CLICKHOUSE_READER_URL ?? process.env.CLICKHOUSE_URL),
    RUNS_LIST_CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
    RUNS_LIST_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
    RUNS_LIST_CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
    RUNS_LIST_CLICKHOUSE_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    RUNS_LIST_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
    /**
     * Dedicated ClickHouse service for queue metrics: the ingestion consumer's inserts and every
     * queue-metrics read (dashboards, queue pages, run inspector, health report) go through it, so
     * metrics traffic never competes with runs-list or trace reads. Unset keeps the previous
     * wiring: inserts on CLICKHOUSE_URL, reads on the query pool.
     */
    QUEUE_METRICS_CLICKHOUSE_URL: z.string().optional(),
    /** Reader split so the consumer's inserts can never land on a read endpoint. Defaults to QUEUE_METRICS_CLICKHOUSE_URL. */
    QUEUE_METRICS_CLICKHOUSE_READER_URL: z.string().optional(),
    QUEUE_METRICS_CLICKHOUSE_KEEP_ALIVE_ENABLED: z.string().default("1"),
    QUEUE_METRICS_CLICKHOUSE_KEEP_ALIVE_IDLE_SOCKET_TTL_MS: z.coerce.number().int().optional(),
    QUEUE_METRICS_CLICKHOUSE_MAX_OPEN_CONNECTIONS: z.coerce.number().int().default(10),
    QUEUE_METRICS_CLICKHOUSE_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    QUEUE_METRICS_CLICKHOUSE_COMPRESSION_REQUEST: z.string().default("1"),
    EVENTS_CLICKHOUSE_BATCH_SIZE: z.coerce.number().int().default(1000),
    EVENTS_CLICKHOUSE_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
    METRICS_CLICKHOUSE_BATCH_SIZE: z.coerce.number().int().default(10000),
    METRICS_CLICKHOUSE_FLUSH_INTERVAL_MS: z.coerce.number().int().default(1000),
    METRICS_CLICKHOUSE_MAX_CONCURRENCY: z.coerce.number().int().default(3),
    EVENTS_CLICKHOUSE_INSERT_STRATEGY: z.enum(["insert", "insert_async"]).default("insert"),
    EVENTS_CLICKHOUSE_WAIT_FOR_ASYNC_INSERT: z.string().default("1"),
    EVENTS_CLICKHOUSE_ASYNC_INSERT_MAX_DATA_SIZE: z.coerce.number().int().default(10485760),
    EVENTS_CLICKHOUSE_ASYNC_INSERT_BUSY_TIMEOUT_MS: z.coerce.number().int().default(5000),
    EVENTS_CLICKHOUSE_START_TIME_MAX_AGE_MS: z.coerce
      .number()
      .int()
      .default(60_000 * 5), // 5 minutes
    EVENT_REPOSITORY_DEFAULT_STORE: z
      .enum(["postgres", "clickhouse", "clickhouse_v2"])
      .default("postgres"),
    EVENT_REPOSITORY_DEBUG_LOGS_DISABLED: BoolEnv.default(false),
    EVENT_REPOSITORY_POSTGRES_WRITES_DISABLED: BoolEnv.default(false),
    EVENTS_CLICKHOUSE_MAX_TRACE_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(25_000),
    EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(5_000),
    EVENTS_CLICKHOUSE_MAX_LIVE_RELOADING_SETTING: z.coerce.number().int().default(2000),

    // OTLP ingest transform worker pool (opt-in). When enabled, decode/convert/enrich run in a
    // worker_threads pool instead of the request event loop; the single consolidated insert path
    // is unchanged.
    OTEL_TRANSFORM_WORKER_POOL_ENABLED: BoolEnv.default(false),
    OTEL_TRANSFORM_WORKER_POOL_SIZE: z.coerce.number().int().optional(),
    OTEL_TRANSFORM_WORKER_PATH: z.string().optional(),

    // Organization data stores registry
    ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce
      .number()
      .int()
      .default(60 * 1000), // 1 minute

    // LLM cost tracking
    LLM_COST_TRACKING_ENABLED: BoolEnv.default(true),
    LLM_PRICING_RELOAD_INTERVAL_MS: z.coerce
      .number()
      .int()
      .default(5 * 60 * 1000), // 5 minutes
    LLM_PRICING_RELOAD_CHANNEL: z.string().default("llm-registry:reload"),
    LLM_PRICING_RELOAD_DEBOUNCE_MS: z.coerce.number().int().default(1000),
    // Whether to subscribe this process to the LLM_PRICING_RELOAD_CHANNEL.
    // Default off — only OTel-ingesting services need real-time pricing
    // freshness; dashboard/worker processes are fine on the existing
    // 5-minute periodic reload. In multi-service deployments, set this to
    // true on the span-ingesting services.
    LLM_PRICING_RELOAD_PUBSUB_ENABLED: BoolEnv.default(false),
    LLM_PRICING_SEED_ON_STARTUP: BoolEnv.default(false),
    LLM_PRICING_READY_TIMEOUT_MS: z.coerce.number().int().default(500),
    LLM_METRICS_BATCH_SIZE: z.coerce.number().int().default(5000),
    LLM_METRICS_FLUSH_INTERVAL_MS: z.coerce.number().int().default(2000),
    LLM_METRICS_MAX_BATCH_SIZE: z.coerce.number().int().default(10000),
    LLM_METRICS_MAX_CONCURRENCY: z.coerce.number().int().default(2),

    // Bootstrap
    TRIGGER_BOOTSTRAP_ENABLED: z.string().default("0"),
    TRIGGER_BOOTSTRAP_WORKER_GROUP_NAME: z.string().optional(),
    TRIGGER_BOOTSTRAP_WORKER_TOKEN_PATH: z.string().optional(),

    // Machine presets
    MACHINE_PRESETS_OVERRIDE_PATH: z.string().optional(),

    // CLI package tag (e.g. "latest", "v4-beta", "4.0.0") - used for setup commands
    TRIGGER_CLI_TAG: z.string().default("latest"),

    HEALTHCHECK_DATABASE_DISABLED: z.string().default("0"),

    REQUEST_IDEMPOTENCY_REDIS_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_HOST),
    REQUEST_IDEMPOTENCY_REDIS_READER_HOST: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_READER_HOST),
    REQUEST_IDEMPOTENCY_REDIS_READER_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) =>
          v ?? (process.env.REDIS_READER_PORT ? parseInt(process.env.REDIS_READER_PORT) : undefined)
      ),
    REQUEST_IDEMPOTENCY_REDIS_PORT: z.coerce
      .number()
      .optional()
      .transform(
        (v) => v ?? (process.env.REDIS_PORT ? parseInt(process.env.REDIS_PORT) : undefined)
      ),
    REQUEST_IDEMPOTENCY_REDIS_USERNAME: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_USERNAME),
    REQUEST_IDEMPOTENCY_REDIS_PASSWORD: z
      .string()
      .optional()
      .transform((v) => v ?? process.env.REDIS_PASSWORD),
    REQUEST_IDEMPOTENCY_REDIS_TLS_DISABLED: z
      .string()
      .default(process.env.REDIS_TLS_DISABLED ?? "false"),

    REQUEST_IDEMPOTENCY_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),

    REQUEST_IDEMPOTENCY_TTL_IN_MS: z.coerce
      .number()
      .int()
      .default(60_000 * 60 * 24),

    // Bulk action
    BULK_ACTION_BATCH_SIZE: z.coerce.number().int().default(100),
    BULK_ACTION_BATCH_DELAY_MS: z.coerce.number().int().default(200),
    BULK_ACTION_SUBBATCH_CONCURRENCY: z.coerce.number().int().default(5),
    /// Max number of concurrent in-flight (PENDING) bulk replays per environment.
    BULK_ACTION_MAX_CONCURRENT_REPLAYS: z.coerce.number().int().default(3),
    /// Max number of explicit run IDs accepted in a single bulk action create request.
    BULK_ACTION_MAX_RUN_IDS: z.coerce.number().int().default(500),

    // AI Run Filter
    AI_RUN_FILTER_MODEL: z.string().optional(),

    EVENT_LOOP_MONITOR_THRESHOLD_MS: z.coerce.number().int().default(100),
    EVENT_LOOP_MONITOR_UTILIZATION_INTERVAL_MS: z.coerce.number().int().default(1000),
    EVENT_LOOP_MONITOR_UTILIZATION_SAMPLE_RATE: z.coerce.number().default(0.05),
    EVENT_LOOP_MONITOR_NOTIFY_ENABLED: z.string().default("0"),

    VERY_SLOW_QUERY_THRESHOLD_MS: z.coerce.number().int().optional(),

    REALTIME_STREAMS_S2_BASIN: z.string().optional(),
    REALTIME_STREAMS_S2_ACCESS_TOKEN: z.string().optional(),
    REALTIME_STREAMS_S2_ENDPOINT: z.string().optional(),
    REALTIME_STREAMS_S2_ACCOUNT_URL: z.string().default("https://a.s2.dev/v1"),
    REALTIME_STREAMS_S2_BASIN_URL: z.string().default("https://{basin}.b.s2.dev/v1"),
    REALTIME_STREAMS_S2_SKIP_ACCESS_TOKENS: z.enum(["true", "false"]).default("false"),
    REALTIME_STREAMS_S2_ACCESS_TOKEN_EXPIRATION_IN_MS: z.coerce
      .number()
      .int()
      .default(60_000 * 60 * 24), // 1 day
    REALTIME_STREAMS_S2_LOG_LEVEL: z
      .enum(["log", "error", "warn", "info", "debug"])
      .default("info"),
    REALTIME_STREAMS_S2_FLUSH_INTERVAL_MS: z.coerce.number().int().default(100),
    REALTIME_STREAMS_S2_MAX_RETRIES: z.coerce.number().int().default(10),
    REALTIME_STREAMS_S2_WAIT_SECONDS: z.coerce.number().int().default(60),
    // When "true", provision a dedicated S2 basin per org and stamp
    // `streamBasinName` on new rows. Off keeps everything on the single
    // basin defined by `REALTIME_STREAMS_S2_BASIN`.
    REALTIME_STREAMS_PER_ORG_BASINS_ENABLED: z.enum(["true", "false"]).default("false"),
    // Per-org basin name = `{prefix}-{env}-org-{orgId}`.
    REALTIME_STREAMS_BASIN_NAME_PREFIX: z.string().default("triggerdotdev"),
    REALTIME_STREAMS_BASIN_NAME_ENV: z.string().default("dev"),
    REALTIME_STREAMS_BASIN_DEFAULT_RETENTION: durationString().default("30d"),
    REALTIME_STREAMS_BASIN_STORAGE_CLASS: z.enum(["express", "standard"]).default("express"),
    REALTIME_STREAMS_BASIN_DELETE_ON_EMPTY_MIN_AGE: durationString().default("1h"),
    REALTIME_STREAMS_DEFAULT_VERSION: z.enum(["v1", "v2"]).default("v1"),
    WAIT_UNTIL_TIMEOUT_MS: z.coerce.number().int().default(600_000),

    // Private connections
    PRIVATE_CONNECTIONS_ENABLED: z.string().optional(),
    PRIVATE_CONNECTIONS_AWS_ACCOUNT_IDS: z.string().optional(),

    // Force RBAC to not use the plugin
    RBAC_FORCE_FALLBACK: BoolEnv.default(false),

    // Per-process pool sizes for an RBAC plugin that owns its own database
    // client (the fallback queries through Prisma and ignores these). Writes
    // are rare role mutations; reads run on the per-request auth hot path.
    RBAC_DATABASE_WRITER_CONNECTION_LIMIT: z.coerce.number().int().default(2),
    RBAC_DATABASE_READER_CONNECTION_LIMIT: z.coerce.number().int().default(5),

    // Force SSO to not use the plugin (contributors without the cloud
    // plugin installed can opt in to a clean OSS-only experience).
    SSO_FORCE_FALLBACK: BoolEnv.default(false),

    // Per-process pool sizes for an SSO plugin that owns its own database
    // client (the fallback queries through Prisma and ignores these). Writes
    // are rare config mutations and webhook processing; reads run on the
    // login path.
    SSO_DATABASE_WRITER_CONNECTION_LIMIT: z.coerce.number().int().default(2),
    SSO_DATABASE_READER_CONNECTION_LIMIT: z.coerce.number().int().default(5),
    // Emit a console.log when the SSO fallback is selected because no
    // plugin is installed. Default off so OSS deployments stay quiet.
    SSO_LOG_FALLBACK: BoolEnv.default(false),
    // Master deploy gate for the whole SSO feature. Default OFF so the
    // image can ship dark and be flipped on only once the SSO plugin's
    // backing services are available. When false, the SSO controller is
    // forced to the OSS fallback — login link hidden, SSO login disabled,
    // settings inert, and session re-validation skipped.
    SSO_ENABLED: BoolEnv.default(false),
    // How often (seconds) a live SSO session is re-validated against the
    // identity provider. The check is single-flight per user, so this is
    // the minimum interval between plugin round-trips, not a per-request
    // cost. Defaults to 5 minutes: every active SSO user drives one
    // billing→IdP round-trip per window, so a seconds-scale default
    // exhausts vendor rate limits at trivial user counts (masked by
    // fail-open, so it degrades silently).
    SSO_SESSION_REVALIDATION_INTERVAL_SECONDS: z.coerce.number().int().positive().default(300),
    // Hard timeout (ms) on the re-validation round-trip. If the SSO plugin
    // doesn't answer within this window the check fails OPEN (session kept)
    // and emits a `sso.revalidation.timeout` warn log — alert on an
    // elevated rate of those to catch a slow/unhealthy SSO dependency.
    SSO_SESSION_REVALIDATION_TIMEOUT_MS: z.coerce.number().int().positive().default(2000),
  })
  .and(GithubAppEnvSchema)
  .and(S2EnvSchema)
  .superRefine((env, ctx) => {
    const presets = new Set(env.COMPUTE_TEMPLATE_MACHINE_PRESETS);
    for (const required of env.COMPUTE_TEMPLATE_MACHINE_PRESETS_REQUIRED) {
      if (!presets.has(required)) {
        ctx.addIssue({
          code: z.ZodIssueCode.custom,
          path: ["COMPUTE_TEMPLATE_MACHINE_PRESETS_REQUIRED"],
          message: `"${required}" is not in COMPUTE_TEMPLATE_MACHINE_PRESETS`,
        });
      }
    }
  });

export type Environment = z.infer<typeof EnvironmentSchema>;
export const env = EnvironmentSchema.parse(process.env);

if (env.ALLOW_INSECURE_DEFAULT_SECRETS) {
  const insecure = (
    [
      ["SESSION_SECRET", env.SESSION_SECRET],
      ["MAGIC_LINK_SECRET", env.MAGIC_LINK_SECRET],
      ["ENCRYPTION_KEY", env.ENCRYPTION_KEY],
      ["MANAGED_WORKER_SECRET", env.MANAGED_WORKER_SECRET],
    ] as const
  )
    .filter(([, value]) => INSECURE_SECRET_VALUES.includes(value))
    .map(([name]) => name);

  if (insecure.length > 0) {
    console.warn(
      `⚠️  ALLOW_INSECURE_DEFAULT_SECRETS is enabled and these secrets still use a known-insecure published default: ${insecure.join(
        ", "
      )}. This is insecure - rotate them as soon as you can.`
    );
  }
}