import { scheduledTaskKnowledgeSource } from "@opengeni/contracts";
import {
  dbSearchPath,
  getSettings,
  resolveFirstPartyDelegationSecret,
  resolveNatsCalloutConfig,
  resolveNatsControlPlaneAuth,
  retryStartupDependency,
  startupRetryOptions,
  temporalConnectionOptions,
} from "@opengeni/config";
import type {
  ScheduledTask,
  ScheduledTaskOverlapPolicy,
  ScheduledTaskScheduleSpec,
} from "@opengeni/contracts";
import {
  assertRuntimeDatabasePosture,
  createDb,
  markSessionWorkflowWakeDelivered,
  runtimeDatabaseReadyCheck,
  type Database,
} from "@opengeni/db";
import { createNatsEventBus, type ResponderConnection } from "@opengeni/events";
import {
  createObservability,
  logStartupDependencyRetry,
  type Observability,
} from "@opengeni/observability";
import { createObjectStorage } from "@opengeni/storage";
import { createNativeRemoteMcpCredentialsPort } from "@opengeni/core/remote-mcp-credentials";
import { isArtifactRuntimeConfigured } from "@opengeni/artifact-tool/runtime/development";
import {
  resolveCatalogSettings,
  SESSION_WORKFLOW_WAKE_DISPATCHER_SCHEDULE_ID,
} from "@opengeni/core";
import {
  Connection,
  Client as TemporalClient,
  ScheduleNotFoundError,
  ScheduleOverlapPolicy,
  WorkflowExecutionAlreadyStartedError,
} from "@temporalio/client";
import type { ScheduleOptions, ScheduleSpec, ScheduleUpdateOptions } from "@temporalio/client";
import { createAppComposition, type DocumentIndexClient, type SessionWorkflowClient } from "./app";
import { observabilityEventLogger } from "./observability";
import { startAuthCalloutResponder } from "./sandbox/auth-callout";
import { startHelloIngestion, startMetricsIngestion } from "./sandbox/metrics-ingestion";
import { startSlackInteractionPump } from "./integrations/slack-interactions";
import { startMemorySlackPublicationPump } from "./memory-slack-delivery";
import { startTemporalScheduleCleanupPump } from "./temporal-schedule-cleanup";
import { cleanupScheduledTaskConnectorAuthorization } from "./scheduled-task-deletion";
import {
  EDITABLE_ARTIFACT_LIVE_WEBSOCKET_MAX_MESSAGE_BYTES,
  EditableArtifactWebSocketTransport,
} from "./editable-artifact-websocket";
import type { ApiWebSocketConnection } from "./api-websocket";
import { InteractionFrameProxyTransport } from "./interaction-frame-proxy";
import { apiRequestBindingsForTransportPeer } from "./http/request-source";
import {
  createStandaloneEditableArtifactApplication,
  type StandaloneEditableArtifactApplication,
} from "./editable-artifact-production";
import { installApiFatalProcessBoundary } from "./fatal-process-boundary";

/**
 * A REJECT_DUPLICATE start collides on the deterministic workflowId when the
 * same manual trigger token fires twice. Temporal surfaces that as
 * WorkflowExecutionAlreadyStartedError; the caller treats it as an idempotent
 * no-op rather than a failure.
 */
function isWorkflowAlreadyStarted(error: unknown): boolean {
  return error instanceof WorkflowExecutionAlreadyStartedError;
}

const TEMPORAL_MONTHS = [
  "JANUARY",
  "FEBRUARY",
  "MARCH",
  "APRIL",
  "MAY",
  "JUNE",
  "JULY",
  "AUGUST",
  "SEPTEMBER",
  "OCTOBER",
  "NOVEMBER",
  "DECEMBER",
] as const;

export async function createTemporalWorkflowClient(
  settings: ReturnType<typeof getSettings>,
  db: Database,
): Promise<{
  client: SessionWorkflowClient;
  documentIndexer: DocumentIndexClient;
  close: () => Promise<void>;
}> {
  const connection = await Connection.connect(temporalConnectionOptions(settings));
  const temporal = new TemporalClient({
    connection,
    namespace: settings.temporalNamespace,
  });
  const client: SessionWorkflowClient = {
    triggerAutomationRun: async ({ accountId, workspaceId, runId }) => {
      try {
        await temporal.workflow.start("automationRunWorkflow", {
          taskQueue: settings.temporalTaskQueue,
          workflowId: `automation-run:${runId}`,
          workflowIdReusePolicy: "REJECT_DUPLICATE",
          args: [{ accountId, workspaceId, runId }],
        });
      } catch (error) {
        if (isWorkflowAlreadyStarted(error)) return;
        throw error;
      }
    },
    signalUserMessage: async ({ eventId, workflowId }) => {
      await temporal.workflow.getHandle(workflowId).signal("userMessage", eventId);
    },
    wakeSessionWorkflow: async ({
      accountId,
      workspaceId,
      sessionId,
      workflowId,
      wakeRevision,
      interruptionRequested,
      onSignalAccepted,
    }) => {
      await temporal.workflow.signalWithStart("sessionWorkflow", {
        taskQueue: settings.temporalTaskQueue,
        workflowId,
        workflowIdReusePolicy: "ALLOW_DUPLICATE",
        args: [{ accountId, workspaceId, sessionId }],
        signal: interruptionRequested ? "sessionControl" : "queueChanged",
      });
      onSignalAccepted?.();
      return await markSessionWorkflowWakeDelivered(db, {
        accountId,
        workspaceId,
        sessionId,
        temporalWorkflowId: workflowId,
        wakeRevision,
      });
    },
    requestSessionWorkflowWakeDispatch: async () => {
      await temporal.schedule
        .getHandle(SESSION_WORKFLOW_WAKE_DISPATCHER_SCHEDULE_ID)
        .trigger(ScheduleOverlapPolicy.BUFFER_ONE);
    },
    signalCodexCapacity: async ({
      accountId,
      workspaceId,
      sessionId,
      workflowId,
      wakeRevision,
      workflowWakeRevision,
    }) => {
      await temporal.workflow.signalWithStart("sessionWorkflow", {
        taskQueue: settings.temporalTaskQueue,
        workflowId,
        workflowIdReusePolicy: "ALLOW_DUPLICATE",
        args: [{ accountId, workspaceId, sessionId }],
        signal: "codexCapacityChanged",
        signalArgs: [wakeRevision],
      });
      await markSessionWorkflowWakeDelivered(db, {
        accountId,
        workspaceId,
        sessionId,
        temporalWorkflowId: workflowId,
        wakeRevision: workflowWakeRevision,
      });
    },
    signalApprovalDecision: async ({
      accountId,
      workspaceId,
      sessionId,
      eventId,
      workflowId,
      workflowWakeRevision,
    }) => {
      await temporal.workflow.signalWithStart("sessionWorkflow", {
        taskQueue: settings.temporalTaskQueue,
        workflowId,
        workflowIdReusePolicy: "ALLOW_DUPLICATE",
        args: [{ accountId, workspaceId, sessionId }],
        signal: "approvalDecision",
        signalArgs: [eventId],
      });
      await markSessionWorkflowWakeDelivered(db, {
        accountId,
        workspaceId,
        sessionId,
        temporalWorkflowId: workflowId,
        wakeRevision: workflowWakeRevision,
      });
    },
    syncScheduledTask: async ({ task }) => {
      const schedule = temporal.schedule.getHandle(task.temporalScheduleId);
      if (task.schedule.type === "manual") {
        try {
          await schedule.delete();
        } catch (error) {
          if (!(error instanceof ScheduleNotFoundError)) throw error;
        }
        return;
      }
      const options = temporalScheduleOptions(task, settings.temporalTaskQueue);
      try {
        await schedule.update(() => temporalScheduleUpdateOptions(options));
      } catch (error) {
        if (!shouldCreateScheduleAfterUpdateError(error)) {
          throw error;
        }
        await temporal.schedule.create(options);
      }
    },
    deleteScheduledTaskSchedule: async ({ temporalScheduleId }) => {
      try {
        await temporal.withDeadline(Date.now() + 5_000, async () => {
          await temporal.schedule.getHandle(temporalScheduleId).delete();
        });
      } catch (error) {
        if (error instanceof ScheduleNotFoundError) return;
        throw error;
      }
    },
    triggerScheduledTask: async ({
      task,
      agentRunUsageIdempotencyKey,
      triggerWorkflowId,
      initiator,
      triggerType = "manual",
    }) => {
      // Deterministic workflowId (derived from the trigger token by the
      // caller) + REJECT_DUPLICATE makes a retried manual trigger idempotent:
      // the second start collides on the id and is rejected instead of
      // spawning a second run. The shared idempotency key dedupes the charge.
      const workflowId = triggerWorkflowId;
      try {
        await temporal.workflow.start("scheduledTaskFireWorkflow", {
          taskQueue: settings.temporalTaskQueue,
          workflowId,
          workflowIdReusePolicy: "REJECT_DUPLICATE",
          args: [
            {
              accountId: task.accountId,
              workspaceId: task.workspaceId,
              taskId: task.id,
              triggerType,
              agentRunUsageIdempotencyKey,
              initiator,
            },
          ],
        });
      } catch (error) {
        // A duplicate trigger token started this run already; treat the retry
        // as a no-op so the (idempotent) usage charge stays the only effect.
        if (isWorkflowAlreadyStarted(error)) {
          return;
        }
        throw error;
      }
    },
    startRigVerification: async ({ workspaceId, changeId, versionId, workflowId }) => {
      const targetId = changeId ?? versionId;
      if (!targetId) {
        throw new Error("rig verification requires changeId or versionId");
      }
      try {
        await temporal.workflow.start("rigVerificationWorkflow", {
          taskQueue: settings.temporalTaskQueue,
          workflowId: workflowId ?? `rig-verification-${targetId}-${crypto.randomUUID()}`,
          workflowIdReusePolicy: "REJECT_DUPLICATE",
          args: [
            {
              workspaceId,
              ...(changeId ? { changeId } : {}),
              ...(versionId ? { versionId } : {}),
            },
          ],
        });
      } catch (error) {
        // A lost start acknowledgement is indistinguishable from a retry. The
        // caller-owned workflow id makes both cases the same successful start.
        if (isWorkflowAlreadyStarted(error)) return;
        throw error;
      }
    },
    check: async () => {
      await connection.workflowService.getSystemInfo({});
    },
  };
  const documentIndexer: DocumentIndexClient = {
    indexDocument: async (input) => {
      const { documentId } = input;
      const workflowId = `document-index-${documentId}-${crypto.randomUUID()}`;
      await temporal.workflow.start("documentIndexWorkflow", {
        taskQueue: settings.temporalTaskQueue,
        workflowId,
        args: [input],
      });
    },
  };
  return {
    client,
    documentIndexer,
    close: async () => {
      await connection.close();
    },
  };
}

export async function startApi(
  options: {
    settings?: ReturnType<typeof getSettings>;
    observability?: Observability;
  } = {},
) {
  const settings = options.settings ?? getSettings();
  const observability =
    options.observability ?? createObservability(settings, { component: "api" });
  // Step I: standalone → dbSchema unset → searchPath undefined → today's plain
  // handle (public). Embedded → scoped to the dedicated schema + the host's RLS
  // strategy.
  const searchPath = dbSearchPath(settings);
  const dbClient = createDb(settings.databaseUrl, {
    max: 32,
    ...(searchPath ? { searchPath } : {}),
    rlsStrategy: settings.rlsStrategy,
  });
  let bus: Awaited<ReturnType<typeof createNatsEventBus>> | undefined;
  let workflowClient: Awaited<ReturnType<typeof createTemporalWorkflowClient>> | undefined;
  const retryOptions = startupRetryOptions(settings);
  const onRetry = (event: Parameters<typeof logStartupDependencyRetry>[1]) =>
    logStartupDependencyRetry(observability, event);
  const databasePosture = {
    rlsStrategy: settings.rlsStrategy,
    expectedRole: settings.runtimeDatabaseRole,
    targetSchema: settings.dbSchema.trim() || "public",
    organizationTenancyCanonicalActivationEnabled:
      settings.organizationTenancyCanonicalActivationEnabled,
  } as const;
  // The PRIVILEGED control-plane NATS login (M-AUTH): when the server runs with
  // auth_callout, api/worker authenticate as a static account user permitted to
  // request exact generation-fenced agent RPC subjects. Null in local dev
  // (anonymous connect — the bus default).
  const controlPlaneAuth = resolveNatsControlPlaneAuth(settings);
  try {
    // Browser, Computer, Terminal, and Files can cold-create the home sandbox
    // directly from this API process. Resolve the same private-registry Modal
    // image before the API becomes reachable so the first human interaction
    // does not pay the provider import/build latency. This mirrors the turn
    // worker's startup boundary and is a no-op for provider-native image IDs,
    // public images, and non-Modal backends.
    const { ensureModalRegistryImage } = await import("@opengeni/runtime/sandbox");
    await retryStartupDependency(
      "Modal private-registry image",
      () => ensureModalRegistryImage(settings),
      { ...retryOptions, onRetry },
    );
    await retryStartupDependency(
      "PostgreSQL runtime posture",
      () => assertRuntimeDatabasePosture(dbClient.db, databasePosture),
      { ...retryOptions, onRetry },
    );
    const resolvedCatalog = await retryStartupDependency(
      "model catalog",
      () => resolveCatalogSettings(dbClient.db, settings),
      { ...retryOptions, onRetry },
    );
    observability.info("OpenGeni model catalog resolved", {
      catalogSource: resolvedCatalog.source,
      catalogVersion: resolvedCatalog.version,
    });
    bus = await retryStartupDependency(
      "NATS",
      () =>
        createNatsEventBus(
          settings.natsUrl,
          controlPlaneAuth
            ? { user: controlPlaneAuth.user, pass: controlPlaneAuth.password }
            : undefined,
          { logger: observabilityEventLogger(observability) },
        ),
      {
        ...retryOptions,
        onRetry,
      },
    );
    workflowClient = await retryStartupDependency(
      "Temporal",
      () => createTemporalWorkflowClient(settings, dbClient.db),
      {
        ...retryOptions,
        onRetry,
      },
    );
  } catch (error) {
    await Promise.allSettled([bus?.close(), workflowClient?.close(), dbClient.close()]);
    throw error;
  }
  if (!bus || !workflowClient) {
    await dbClient.close();
    throw new Error("OpenGeni API startup dependencies were not initialized");
  }
  const objectStorage = createObjectStorage(settings);
  let editableArtifactComposition: StandaloneEditableArtifactApplication | undefined;
  if (objectStorage && isArtifactRuntimeConfigured()) {
    try {
      editableArtifactComposition = await createStandaloneEditableArtifactApplication({
        db: dbClient.db,
        bus,
        objectStorage,
      });
    } catch (error) {
      await Promise.allSettled([bus.close(), workflowClient.close(), dbClient.close()]);
      throw error;
    }
  }
  const { app, routeDeps } = createAppComposition({
    settings,
    connectionCredentials: createNativeRemoteMcpCredentialsPort(settings, dbClient.db),
    db: dbClient.db,
    bus,
    workflowClient: workflowClient.client,
    documentIndexer: workflowClient.documentIndexer,
    objectStorage,
    ...(editableArtifactComposition
      ? {
          editableArtifacts: editableArtifactComposition.application,
          editableArtifactExports: editableArtifactComposition.durableExports,
          editableArtifactAgent: editableArtifactComposition.agent,
          editableArtifactOfficeImports: editableArtifactComposition.officeImports,
        }
      : {}),
    observability,
    readinessChecks: {
      db: runtimeDatabaseReadyCheck(dbClient.db, databasePosture),
    },
  });
  if (!routeDeps.editableArtifacts) {
    observability.warn("editable artifact engine is not composed", {
      subsystem: "editable_artifacts",
      behavior: "http_and_websocket_fail_closed",
    });
  }
  const artifactWebSockets = new EditableArtifactWebSocketTransport(routeDeps.editableArtifacts);
  const interactionFrameProxies = new InteractionFrameProxyTransport(
    resolveFirstPartyDelegationSecret(settings),
  );
  const server = Bun.serve<ApiWebSocketConnection>({
    hostname: settings.apiHost,
    port: settings.apiPort,
    idleTimeout: 255,
    fetch: (request, bunServer) => {
      if (interactionFrameProxies.handles(request)) {
        return interactionFrameProxies.upgrade(request, bunServer);
      }
      if (artifactWebSockets.handles(request)) {
        return artifactWebSockets.upgrade(request, bunServer);
      }
      return app.fetch(
        request,
        apiRequestBindingsForTransportPeer(bunServer.requestIP(request)?.address),
      );
    },
    websocket: {
      maxPayloadLength: EDITABLE_ARTIFACT_LIVE_WEBSOCKET_MAX_MESSAGE_BYTES,
      backpressureLimit: 16 * 1024 * 1024,
      closeOnBackpressureLimit: true,
      open: (socket) => socket.data.attach(socket),
      message: (socket, message) => socket.data.receive(message),
      close: (socket) => socket.data.transportClosed(),
    },
  });
  const stopSlackInteractionPump = settings.slackSigningSecret
    ? startSlackInteractionPump(routeDeps)
    : undefined;
  const stopMemorySlackPublicationPump = startMemorySlackPublicationPump(routeDeps);
  const stopTemporalScheduleCleanupPump = startTemporalScheduleCleanupPump({
    db: dbClient.db,
    cleanupConnectorAuthorization: async (claim) =>
      await cleanupScheduledTaskConnectorAuthorization(routeDeps, claim),
    deleteSchedule: async (temporalScheduleId) => {
      await workflowClient.client.deleteScheduledTaskSchedule({ temporalScheduleId });
    },
    observability,
  });
  // M10 — start the metrics-ingestion consumer (agent heartbeats → DB last-sample
  // + downsampled series), gated on the selfhosted flag. A no-op when disabled.
  let stopMetricsIngestion: (() => void) | undefined;
  // Reconcile enrollments.has_display to the LIVE capability the agent reports in
  // its connect Hello (has_display was frozen at the enroll-time snapshot). Gated
  // on the same selfhosted flag.
  let stopHelloIngestion: (() => void) | undefined;
  // M-AUTH — start the NATS auth-callout responder (the tenancy boundary): it
  // validates an agent's enrollment bearer presented at NATS connect and mints a
  // workspace-scoped user JWT. Gated on the selfhosted flag + a resolvable callout
  // config; without the callout plane it never starts (selfhosted agents simply
  // cannot connect — graceful). It runs on its OWN connection (the callout auth
  // user), separate from the privileged control-plane bus.
  let authCalloutResponder: ResponderConnection | undefined;
  if (settings.sandboxSelfhostedEnabled) {
    stopMetricsIngestion = startMetricsIngestion({
      db: dbClient.db,
      bus,
      observability,
    });
    stopHelloIngestion = startHelloIngestion({
      db: dbClient.db,
      bus,
      observability,
    });
    observability.info("OpenGeni machine-metrics + hello ingestion consumers started", {});

    const callout = resolveNatsCalloutConfig(settings);
    if (callout) {
      try {
        authCalloutResponder = await startAuthCalloutResponder(
          { db: dbClient.db, settings, callout, observability },
          settings.natsUrl,
        );
      } catch {
        // A responder start failure must not crash the API (other planes work); log
        // loudly — selfhosted agents will fail to connect until it is up.
        observability.error("OpenGeni NATS auth-callout responder failed to start", {
          errorClass: "NatsAuthCalloutOperationError",
          errorCode: "nats_auth_callout_start_failed",
          origin: "api",
        });
      }
    } else {
      observability.warn(
        "OpenGeni selfhosted enabled but the NATS auth-callout plane is not configured; selfhosted agents cannot connect",
        {},
      );
    }
  }
  observability.info("OpenGeni API listening", {
    host: settings.apiHost,
    port: settings.apiPort,
  });
  return {
    server,
    close: async () => {
      server.stop(true);
      stopSlackInteractionPump?.();
      await stopMemorySlackPublicationPump();
      stopMetricsIngestion?.();
      stopHelloIngestion?.();
      await stopTemporalScheduleCleanupPump();
      await Promise.allSettled([
        Promise.resolve(editableArtifactComposition?.close()),
        authCalloutResponder?.close(),
        bus.close(),
        workflowClient.close(),
        dbClient.close(),
      ]);
    },
  };
}

if (import.meta.main) {
  const fatalBoundary = installApiFatalProcessBoundary();
  try {
    const settings = getSettings();
    const observability = createObservability(settings, { component: "api" });
    fatalBoundary.attachObservability(observability);
    await startApi({ settings, observability });
    fatalBoundary.markRunning();
  } catch (error) {
    await fatalBoundary.reportStartupFailure(error);
  }
}

export function temporalOverlapPolicy(policy: ScheduledTaskOverlapPolicy): ScheduleOverlapPolicy {
  if (policy === "skip") {
    return ScheduleOverlapPolicy.SKIP;
  }
  if (policy === "buffer_one") {
    return ScheduleOverlapPolicy.BUFFER_ONE;
  }
  return ScheduleOverlapPolicy.ALLOW_ALL;
}

export function shouldCreateScheduleAfterUpdateError(error: unknown): boolean {
  return error instanceof ScheduleNotFoundError;
}

export function temporalScheduleSpec(schedule: ScheduledTaskScheduleSpec): ScheduleSpec {
  if (schedule.type === "manual") {
    throw new Error("manual scheduled tasks do not have a Temporal Schedule spec");
  }
  if (schedule.type === "interval") {
    return {
      intervals: [temporalIntervalSpec(schedule)],
      ...(schedule.startAt ? { startAt: new Date(schedule.startAt) } : {}),
      ...(schedule.endAt ? { endAt: new Date(schedule.endAt) } : {}),
    };
  }
  if (schedule.type === "calendar") {
    return {
      calendars: [
        {
          hour: schedule.hour,
          minute: schedule.minute,
          second: 0,
          ...(schedule.daysOfWeek ? { dayOfWeek: schedule.daysOfWeek } : {}),
        },
      ],
      timezone: schedule.timeZone,
    };
  }
  const runAt = new Date(schedule.runAt);
  return {
    calendars: [
      {
        year: runAt.getUTCFullYear(),
        month: temporalMonth(runAt.getUTCMonth()),
        dayOfMonth: runAt.getUTCDate(),
        hour: runAt.getUTCHours(),
        minute: runAt.getUTCMinutes(),
        second: runAt.getUTCSeconds(),
      },
    ],
    timezone: "UTC",
  };
}

function temporalIntervalSpec(
  schedule: Extract<ScheduledTaskScheduleSpec, { type: "interval" }>,
): NonNullable<ScheduleSpec["intervals"]>[number] {
  const every = `${schedule.everySeconds}s` as `${number}s`;
  if (!schedule.startAt) {
    return { every };
  }

  // Temporal interval schedules match Epoch + (n * every) + offset. Its
  // top-level startAt only filters matching times before that boundary, so it
  // does not itself anchor the cadence. Derive the phase from startAt to make
  // the stored OpenGeni timestamp the first interval boundary rather than the
  // next epoch-aligned match.
  const everyMilliseconds = BigInt(schedule.everySeconds) * 1_000n;
  const startMilliseconds = BigInt(new Date(schedule.startAt).getTime());
  const offsetMilliseconds =
    ((startMilliseconds % everyMilliseconds) + everyMilliseconds) % everyMilliseconds;
  return {
    every,
    ...(offsetMilliseconds === 0n ? {} : { offset: `${offsetMilliseconds}ms` as `${number}ms` }),
  };
}

function temporalMonth(monthIndex: number) {
  return TEMPORAL_MONTHS[monthIndex]!;
}

function temporalScheduleOptions(task: ScheduledTask, taskQueue: string): ScheduleOptions {
  return {
    scheduleId: task.temporalScheduleId,
    spec: temporalScheduleSpec(task.schedule),
    action: {
      type: "startWorkflow",
      workflowType: "scheduledTaskFireWorkflow",
      taskQueue,
      args: [
        {
          accountId: task.accountId,
          workspaceId: task.workspaceId,
          taskId: task.id,
          triggerType: "scheduled",
        },
      ],
    },
    policies: {
      overlap: temporalOverlapPolicy(task.overlapPolicy),
      catchupWindow: "24h",
      pauseOnFailure: false,
    },
    state: {
      paused: scheduledTaskEffectivelyPaused(task),
      ...(task.schedule.type === "once" ? { remainingActions: 1 } : {}),
    },
    memo: {
      accountId: task.accountId,
      workspaceId: task.workspaceId,
      scheduledTaskId: task.id,
      name: task.name,
    },
  };
}

function scheduledTaskEffectivelyPaused(task: ScheduledTask): boolean {
  if (task.status === "paused") return true;
  if (!scheduledTaskKnowledgeSource(task)) return false;
  const control = task.metadata.knowledgeSourceSync;
  if (!control || typeof control !== "object") return false;
  const record = control as Record<string, unknown>;
  return record.sourceEnabled === false || record.connectionPaused === true;
}

function temporalScheduleUpdateOptions(options: ScheduleOptions): ScheduleUpdateOptions {
  return {
    spec: options.spec,
    action: options.action,
    ...(options.policies ? { policies: options.policies } : {}),
    state: options.state ?? {},
    ...(options.searchAttributes ? { searchAttributes: options.searchAttributes } : {}),
    ...(options.typedSearchAttributes
      ? { typedSearchAttributes: options.typedSearchAttributes }
      : {}),
  };
}