import { refreshBackgroundWakeIntent } from "../background-wake/publisher.js";
import { getConfig } from "../config/loader.js";
import {
  checkDiskPressureBackgroundGate,
  diskPressureBackgroundSkipLogFields,
  shouldLogDiskPressureBackgroundSkip,
} from "../daemon/disk-pressure-background-gate.js";
import { processMessage } from "../daemon/process-message.js";
import { INTERNAL_GUARDIAN_TRUST_CONTEXT } from "../daemon/trust-context.js";
import { emitNotificationSignal } from "../notifications/emit-signal.js";
import { getConversation } from "../persistence/conversation-crud.js";
import { isLifecycleQuiesced } from "../persistence/lifecycle-quiesce.js";
import { invalidateAssistantInferredItemsForConversation } from "../plugins/defaults/memory/task-memory-cleanup.js";
import { wakeAgentForOpportunity } from "../runtime/agent-wake.js";
import { broadcastMessage } from "../runtime/assistant-event-hub.js";
import { runBackgroundJob } from "../runtime/background-job-runner.js";
import { publishConversationListChanged } from "../runtime/sync/resource-sync-events.js";
import { runSequencesOnce } from "../sequence/engine.js";
import type { TurnFailure } from "../telemetry/turn-outcome.js";
import { recordWatchdogEvent } from "../telemetry/watchdog-events-store.js";
import { getLogger } from "../util/logger.js";
import { describeScheduleSource } from "../util/schedule-source-key.js";
import {
  createWorkerSupervisor,
  type WorkerSupervisor,
} from "../util/worker-process.js";
import { runWatchersOnce } from "../watcher/engine.js";
import { normalizeCapabilityManifest } from "../workflows/capabilities.js";
import { getWorkflowRunManager } from "../workflows/run-manager.js";
import { declarationExistsOnDisk } from "./plugin-schedule-declarations.js";
import { isPluginSchedulesEnabled } from "./plugin-schedules-gate.js";
import { hasSetConstructs } from "./recurrence-engine.js";
import { applyRetryDecision, decideRetry } from "./retry-policy.js";
import { runScript, type ScriptResult } from "./run-script.js";
import {
  claimDueSchedules,
  completeOneShot,
  completeScheduleRun,
  createScheduleRun,
  deferClaimedSchedule,
  failOneShotPermanently,
  getLastScheduleConversationId,
  recordAdministrativeSkipRun,
  resetRetryCount,
  resolveScheduleConversationGroupId,
  retryOneShot,
  type RoutingIntent,
  type ScheduleJob,
  scheduleRetry,
  setScheduleRunConversationId,
} from "./schedule-store.js";
import { buildWakeScheduleOptions } from "./wake-schedule-options.js";
import {
  isScheduleWorkerAdministrativelyStopped,
  probeScheduleWorker,
  spawnScheduleWorkerProcess,
  startScheduleWorker,
  stopScheduleWorker,
} from "./worker-control.js";

const log = getLogger("scheduler");

import type { ScheduleMessageOptions } from "./scheduler-types.js";

/**
 * Run a scheduled message through the daemon's agent pipeline, translating the
 * schedule's `trustClass` into the trust context `processMessage` expects.
 *
 * Returns the turn's failure outcome (if any) so the caller can record a run
 * whose LLM call failed as an error. Such a turn resolves normally rather than
 * throwing, so `turnFailure` is the only failure signal on the happy return.
 */
async function dispatchScheduleMessage(
  conversationId: string,
  message: string,
  options?: ScheduleMessageOptions,
): Promise<{ turnFailure?: TurnFailure }> {
  const { turnFailure } = await processMessage(
    conversationId,
    message,
    options
      ? {
          ...(options.trustClass
            ? {
                trustContext: {
                  sourceChannel: "vellum",
                  trustClass: options.trustClass,
                },
              }
            : {}),
          ...(options.taskRunId ? { taskRunId: options.taskRunId } : {}),
          ...(options.overrideProfile
            ? { overrideProfile: options.overrideProfile }
            : {}),
          ...(options.cronRunId ? { cronRunId: options.cronRunId } : {}),
        }
      : undefined,
  );
  return { ...(turnFailure ? { turnFailure } : {}) };
}

/** Build a schedule-run error message from a turn's failure outcome. */
function describeTurnFailure(turnFailure: TurnFailure): string {
  return turnFailure.failureCode
    ? `Agent turn failed during its LLM call (${turnFailure.failureCode})`
    : "Agent turn failed during its LLM call";
}

/** Emit the attention signal for a `notify`-mode schedule firing. */
async function emitScheduleNotifySignal(payload: {
  id: string;
  label: string;
  message: string;
  routingIntent: RoutingIntent;
  routingHints: Record<string, unknown>;
  groupId: string;
  deepLinkConversationId?: string;
}): Promise<void> {
  await emitNotificationSignal({
    sourceEventName: "schedule.notify",
    sourceChannel: "scheduler",
    sourceContextId: payload.id,
    attentionHints: {
      requiresAction: true,
      urgency: "high",
      isAsyncBackground: false,
      visibleInSourceNow: false,
    },
    contextPayload: {
      scheduleId: payload.id,
      label: payload.label,
      message: payload.message,
      ...(payload.deepLinkConversationId
        ? { deepLinkConversationId: payload.deepLinkConversationId }
        : {}),
    },
    routingIntent: payload.routingIntent,
    routingHints: payload.routingHints,
    conversationMetadata: {
      groupId: payload.groupId,
      scheduleJobId: payload.id,
      source: "schedule",
    },
    dedupeKey: `schedule:notify:${payload.id}:${Date.now()}`,
    throwOnError: true,
  });
}

/** Emit the attention signal for a watcher notification. */
function emitWatcherNotifySignal(notification: {
  title: string;
  body: string;
}): void {
  void emitNotificationSignal({
    sourceEventName: "watcher.notification",
    sourceChannel: "watcher",
    sourceContextId: `watcher-${Date.now()}`,
    attentionHints: {
      requiresAction: false,
      urgency: "low",
      isAsyncBackground: true,
      visibleInSourceNow: false,
    },
    contextPayload: {
      title: notification.title,
      body: notification.body,
    },
    dedupeKey: `watcher:notification:${crypto.randomUUID()}`,
  });
}

/** Broadcast + refresh the conversation list when a schedule creates one. */
function broadcastScheduleConversationCreated(info: {
  conversationId: string;
  scheduleJobId: string;
  title: string;
}): void {
  broadcastMessage({
    type: "schedule_conversation_created",
    conversationId: info.conversationId,
    scheduleJobId: info.scheduleJobId,
    title: info.title,
  });
  publishConversationListChanged("created");
}

export interface SchedulerHandle {
  runOnce(): Promise<number>;
  runDueWorkOnce(
    options?: SchedulerRunDueWorkOptions,
  ): Promise<SchedulerDueWorkResult>;
  stop(): void;
}

export interface SchedulerRunDueWorkOptions {
  deadlineAt?: number;
  minStartBudgetMs?: number;
}

export interface SchedulerDueWorkResult {
  claimed: number;
  completed: number;
  failed: number;
  skipped: number;
}

const TICK_INTERVAL_MS = 15_000;

/**
 * Maximum number of times a wake can be retried after a timeout before
 * being permanently failed. At 15-second scheduler intervals, 20 retries
 * ≈ 5 minutes of total retry window.
 */
const WAKE_MAX_RETRIES = 20;

/**
 * Apply retry policy on schedule-execution failure. Retries are scheduled by
 * `applyRetryDecision`; once retries are exhausted, the `emitAlert` callback
 * fires an `activity.failed` notification so the user sees that a job
 * permanently failed rather than just silently disappearing.
 */
async function handleExecutionFailure(params: {
  job: ScheduleJob;
  errorMsg: string;
  isOneShot: boolean;
}): Promise<void> {
  const decision = decideRetry(params.job);
  await applyRetryDecision({
    job: params.job,
    isOneShot: params.isOneShot,
    errorMsg: params.errorMsg,
    decision,
    scheduleRetry,
    failOneShotPermanently,
    resetRetryCount,
    emitAlert: (_title, _summary, dedupKey) =>
      emitScheduleActivityFailed({
        jobId: params.job.id,
        jobName: params.job.name,
        errorMessage: params.errorMsg,
        dedupKey,
      }),
    log,
  });
}

/** The running scheduler, retained so shutdown can stop it. */
let instance: SchedulerHandle | null = null;

/** The schedule worker liveness watchdog, disposed by {@link stopScheduler}. */
let scheduleWorkerSupervisor: WorkerSupervisor | null = null;

/**
 * Notify the user once per outage when the worker cannot be respawned after
 * repeated attempts, so schedules being paused is not silent. A stable
 * hourly dedupe key keeps it to one notification per outage window rather than
 * one per tick.
 */
function emitScheduleWorkerDownNotification(consecutiveFailures: number): void {
  void emitNotificationSignal({
    sourceEventName: "activity.failed",
    sourceChannel: "scheduler",
    sourceContextId: "schedule-worker",
    dedupeKey: `schedule-worker-down:${Math.floor(Date.now() / 3_600_000)}`,
    contextPayload: {
      jobName: "Scheduled tasks",
      errorMessage:
        "The assistant could not restart its schedule runner, so scheduled tasks are paused. They resume automatically once it recovers.",
      errorKind: "exception",
      consecutiveFailures,
    },
    attentionHints: {
      requiresAction: false,
      urgency: "medium",
      isAsyncBackground: true,
      visibleInSourceNow: false,
    },
  }).catch((emitErr) => {
    log.warn(
      { err: emitErr },
      "Failed to emit schedule-worker-down notification",
    );
  });
}

export function startScheduler(): SchedulerHandle {
  // Schedule execution is owned by the schedule worker process; spawn it now as
  // a child of the daemon so it is running immediately. Fire-and-forget — a
  // worker failure must never block boot. The daemon's own tick below runs only
  // watchers and sequences.
  startScheduleWorker();

  // Liveness watchdog: the worker calls process.exit() on any uncaught error
  // and nothing respawned it, so schedules could stop for days silently. Probe
  // and respawn off the existing tick — recovery latency drops from days to
  // ~one tick. Detects process death, not a wedged-but-alive worker.
  scheduleWorkerSupervisor = createWorkerSupervisor({
    label: "Schedule worker",
    probe: probeScheduleWorker,
    respawn: () => spawnScheduleWorkerProcess({ detached: false }),
    isSuppressed: isScheduleWorkerAdministrativelyStopped,
    killChild: (pid) => {
      try {
        process.kill(pid, "SIGTERM");
      } catch {
        // Worker already gone — nothing to kill.
      }
    },
    onRespawn: (pid) => {
      log.warn(
        { pid },
        "Schedule worker was not running — respawned by watchdog",
      );
      recordWatchdogEvent({
        checkName: "schedule_worker_respawn",
        detail: { pid },
      });
    },
    onPersistentFailure: (consecutiveFailures, err) => {
      log.error(
        { err, consecutiveFailures },
        "Schedule worker repeatedly failed to respawn — schedules are paused",
      );
      recordWatchdogEvent({
        checkName: "schedule_worker_down",
        value: consecutiveFailures,
        detail: { consecutiveFailures },
      });
      emitScheduleWorkerDownNotification(consecutiveFailures);
    },
  });

  let stopped = false;
  let tickRunning = false;

  const tick = async () => {
    if (stopped || tickRunning) {
      return;
    }
    tickRunning = true;
    try {
      // Respawn the schedule worker if it has died (fire-and-forget; never
      // throws). Idempotent — a live PID file short-circuits to alreadyRunning.
      void scheduleWorkerSupervisor?.ensureAlive();
      await runScheduleOnce();
    } catch (err) {
      log.error({ err }, "Schedule tick failed");
    } finally {
      tickRunning = false;
    }
  };

  const timer = setInterval(() => {
    void tick();
  }, TICK_INTERVAL_MS);
  timer.unref();
  void tick();

  instance = {
    async runOnce(): Promise<number> {
      return runScheduleOnce();
    },
    async runDueWorkOnce(
      options?: SchedulerRunDueWorkOptions,
    ): Promise<SchedulerDueWorkResult> {
      return runScheduleDueWorkOnce(options);
    },
    stop(): void {
      stopped = true;
      clearInterval(timer);
    },
  };

  // Publish the initial background wake intent now that the scheduler is live
  // and its schedules are visible to `computeNextBackgroundWakeIntent`.
  refreshBackgroundWakeIntent("daemon-startup");

  return instance;
}

/**
 * Stop the running scheduler if one was started, and SIGTERM the schedule
 * worker process if one is running.
 */
export function stopScheduler(): void {
  // Stop the tick FIRST so no tick can respawn the worker we are about to kill.
  if (instance) {
    instance.stop();
    instance = null;
  }
  // No further respawns; a respawn already in flight kills its child on resolve.
  scheduleWorkerSupervisor?.dispose();
  scheduleWorkerSupervisor = null;
  // Then SIGTERM the running worker via its PID file.
  stopScheduleWorker();
}

/** The running scheduler, or null if one was never started. */
export function getScheduler(): SchedulerHandle | null {
  return instance;
}

export async function runScheduleOnce(): Promise<number> {
  const result = await runScheduleDueWorkOnce();
  return result.completed + result.failed + result.skipped;
}

/**
 * Run the daemon's due background work: watchers and sequences. Schedule
 * execution is owned by the schedule worker process (see `runDueSchedulesOnce`,
 * driven by `worker.ts`), so this daemon-side path never claims schedules and
 * reports none of them as its own pending work.
 */
export async function runScheduleDueWorkOnce(
  options: SchedulerRunDueWorkOptions = {},
): Promise<SchedulerDueWorkResult> {
  const minStartBudgetMs = options.minStartBudgetMs ?? 0;
  const result: SchedulerDueWorkResult = {
    claimed: 0,
    completed: 0,
    failed: 0,
    skipped: 0,
  };

  if (
    options.deadlineAt != null &&
    options.deadlineAt - Date.now() < minStartBudgetMs
  ) {
    return result;
  }

  const diskPressureGate = checkDiskPressureBackgroundGate("background-work");
  if (diskPressureGate.action === "skip") {
    if (shouldLogDiskPressureBackgroundSkip("scheduler")) {
      log.warn(
        {
          source: "schedule",
          ...diskPressureBackgroundSkipLogFields(diskPressureGate),
        },
        "Schedule tick skipped during disk pressure cleanup mode",
      );
    }
    return result;
  }

  // The drain quiesce gates live inside claimDueWatchers and
  // claimDueEnrollments, immediately before their claim writes.

  // ── Watchers (event-driven polling) ────────────────────────────────
  try {
    const watcherProcessed = await runWatchersOnce(emitWatcherNotifySignal);
    result.completed += watcherProcessed;
  } catch (err) {
    log.error({ err }, "Watcher tick failed");
  }

  // ── Sequences (multi-step outreach) ──────────────────────────────
  try {
    const sequenceProcessed = await runSequencesOnce();
    result.completed += sequenceProcessed;
  } catch (err) {
    log.error({ err }, "Sequence engine tick failed");
  }

  const processed = result.completed + result.failed + result.skipped;
  if (processed > 0) {
    log.info({ processed }, "Schedule tick complete");
  }
  return result;
}

/**
 * Claim and execute every due schedule (all modes: notify, script, wake,
 * workflow, execute). Driven by the schedule worker process's tick
 * (`worker.ts`). Claims are atomic in the schedule store, so overlapping ticks
 * cannot double-run a job another claim already took.
 */
/** How far a claimed-but-quiesced schedule is pushed back into the queue. */
const QUIESCE_DEFER_MS = 30_000;

export async function runDueSchedulesOnce(
  now: number = Date.now(),
): Promise<SchedulerDueWorkResult> {
  const result: SchedulerDueWorkResult = {
    claimed: 0,
    completed: 0,
    failed: 0,
    skipped: 0,
  };
  const mark = (status: "completed" | "failed" | "skipped") => {
    result[status] += 1;
  };

  const diskPressureGate = checkDiskPressureBackgroundGate("background-work");
  if (diskPressureGate.action === "skip") {
    if (shouldLogDiskPressureBackgroundSkip("scheduler-schedules")) {
      log.warn(
        {
          source: "schedule",
          ...diskPressureBackgroundSkipLogFields(diskPressureGate),
        },
        "Due-schedule run skipped during disk pressure cleanup mode",
      );
    }
    return result;
  }

  // The drain quiesce gate lives inside claimDueSchedules, immediately
  // before the claim writes.
  const jobs = await claimDueSchedules(now);
  result.claimed = jobs.length;
  for (const job of jobs) {
    // Lease re-check per claimed job: a lease armed between the batch claim
    // and this job's start returns the job to the queue untouched instead of
    // starting work the drain snapshot cannot see — notify mode especially,
    // which emits before any run row exists.
    if (isLifecycleQuiesced()) {
      try {
        await deferClaimedSchedule(job.id, Date.now() + QUIESCE_DEFER_MS);
      } catch (err) {
        log.warn(
          { err, jobId: job.id },
          "Failed to defer claimed schedule under quiesce",
        );
      }
      result.skipped += 1;
      continue;
    }

    // Fire-time gate for plugin-sourced rows, covering every way the source
    // can go away under an armed row. `declarationExistsOnDisk` is the probe
    // the enable path uses, and it answers for all of them: a `.disabled`
    // sentinel, a plugin directory a local uninstall removed, a manifest that
    // no longer parses, and a declaration that is simply gone. Turning the
    // feature flag off retires the whole surface.
    // The probe is deliberately disk-only. Schedule execution runs in the
    // schedule worker process, which activates no plugins, so the daemon's
    // in-memory activation ledger is empty here and reading it would skip
    // every plugin schedule. Activation is gated where the daemon owns it: the
    // reconciler decides what arms, and the run-now route refuses by hand.
    // The reconciler is what disarms the rows any of these own, and it runs
    // on its own schedule, so re-reading here is what makes the change take
    // effect immediately: a row still armed (or already claimed) at that
    // moment cannot run the plugin's code. The probe costs a stat and a
    // manifest read, paid only when a sourced row fires.
    const sourceKey = job.sourceKey;
    if (
      sourceKey !== null &&
      (!isPluginSchedulesEnabled() ||
        !(await declarationExistsOnDisk(sourceKey)))
    ) {
      const sourcePlugin = describeScheduleSource(sourceKey) ?? sourceKey;
      log.info(
        { jobId: job.id, name: job.name, plugin: sourcePlugin },
        "Schedule not run: its plugin schedule source is unavailable",
      );
      // cron_runs has no skip status, so the skip is recorded as an error run
      // to stay visible in the schedule's history. It is an administrative
      // skip, not an attempt: the retry budget is left alone and no retry is
      // scheduled, so the row stays on its normal cadence.
      await recordAdministrativeSkipRun(
        job.id,
        `Schedule not run: plugin "${sourcePlugin}" is disabled, uninstalled, or no longer declares this schedule.`,
      );
      mark("skipped");
      continue;
    }

    const isOneShot = job.expression == null;

    // ── Notify mode (one-shot or recurring) ─────────────────────────
    if (job.mode === "notify") {
      let failed = false;
      try {
        log.info(
          { jobId: job.id, name: job.name, isOneShot },
          "Firing schedule notification",
        );
        await emitScheduleNotifySignal({
          id: job.id,
          label: job.name,
          message: job.message,
          routingIntent: job.routingIntent,
          routingHints: job.routingHints,
          groupId: resolveScheduleConversationGroupId(job),
          ...(job.createdFromConversationId
            ? { deepLinkConversationId: job.createdFromConversationId }
            : {}),
        });
        if (isOneShot) {
          const successRunId = await createScheduleRun(
            job.id,
            `notify-ok:${job.id}`,
          );
          await completeScheduleRun(successRunId, { status: "ok" });
          await completeOneShot(job.id);
        } else {
          // Track recurring notify-mode success so lastStatus resets to ok
          // and retryCount clears after a transient failure.
          const runId = await createScheduleRun(job.id, `notify-ok:${job.id}`);
          await completeScheduleRun(runId, { status: "ok" });
        }
      } catch (err) {
        log.warn(
          { err, jobId: job.id, name: job.name, isOneShot },
          "Schedule notification failed",
        );
        const errorMsg = err instanceof Error ? err.message : String(err);
        const errorRunId = await createScheduleRun(
          job.id,
          `notify-error:${job.id}`,
        );
        await completeScheduleRun(errorRunId, {
          status: "error",
          error: errorMsg,
        });
        await handleExecutionFailure({ job, errorMsg, isOneShot });
        failed = true;
      }
      mark(failed ? "failed" : "completed");
      continue;
    }

    // ── Script mode (shell command, no LLM) ────────────────────────
    if (job.mode === "script") {
      if (!job.script) {
        log.warn(
          { jobId: job.id, name: job.name },
          "Script schedule has no script command — skipping",
        );
        mark("skipped");
        continue;
      }
      const runId = await createScheduleRun(job.id, `script:${job.id}`);
      let failed = false;
      try {
        log.info(
          { jobId: job.id, name: job.name, isOneShot },
          "Executing script schedule",
        );
        const result: ScriptResult = await runScript(job.script, {
          timeoutMs: job.timeoutMs ?? undefined,
          scheduleRunId: runId,
          scheduleId: job.id,
        });
        await completeScheduleRun(runId, {
          status: result.exitCode === 0 ? "ok" : "error",
          output: result.stdout || undefined,
          error: result.stderr || undefined,
        });
        if (result.exitCode === 0) {
          if (isOneShot) {
            await completeOneShot(job.id);
          }
        } else {
          const errorMsg =
            result.stderr || "Script exited with non-zero status";
          await handleExecutionFailure({ job, errorMsg, isOneShot });
          failed = true;
        }
      } catch (err) {
        const errorMsg = err instanceof Error ? err.message : String(err);
        log.warn(
          { err, jobId: job.id, name: job.name, isOneShot },
          "Script schedule execution failed",
        );
        await completeScheduleRun(runId, { status: "error", error: errorMsg });
        await handleExecutionFailure({ job, errorMsg, isOneShot });
        failed = true;
      }
      mark(failed ? "failed" : "completed");
      continue;
    }

    // ── Wake mode (resume an existing conversation) ─────────────────
    if (job.mode === "wake") {
      const { wakeConversationId } = job;
      if (!wakeConversationId) {
        log.warn(
          { jobId: job.id, name: job.name },
          "Wake schedule missing wakeConversationId — completing as no-op",
        );
        if (isOneShot) {
          await completeOneShot(job.id);
        }
        mark("skipped");
        continue;
      }

      let failed = false;
      try {
        log.info(
          { jobId: job.id, name: job.name, wakeConversationId, isOneShot },
          "Executing wake schedule",
        );
        const result = await wakeAgentForOpportunity(
          buildWakeScheduleOptions(job, wakeConversationId),
        );

        if (result.reason === "timeout" && isOneShot) {
          // The conversation is busy processing a tool call. Retry on
          // the next scheduler tick unless we've exceeded the retry cap.
          if (job.retryCount >= WAKE_MAX_RETRIES) {
            log.warn(
              {
                jobId: job.id,
                name: job.name,
                wakeConversationId,
                retryCount: job.retryCount,
              },
              "Wake timed out and exceeded max retries — permanently failing",
            );
            await failOneShotPermanently(job.id);
          } else {
            log.warn(
              {
                jobId: job.id,
                name: job.name,
                wakeConversationId,
                retryCount: job.retryCount,
              },
              "Wake timed out waiting for idle conversation — will retry on next tick",
            );
            await retryOneShot(job.id);
          }
          mark("skipped");
          continue;
        }

        // Guard: if the wake was not invoked for any reason (timeout on
        // a recurring schedule, not_found, archived, no_resolver), skip
        // the success feed event — the wake did not actually fire.
        if (!result.invoked) {
          log.warn(
            {
              jobId: job.id,
              name: job.name,
              wakeConversationId,
              reason: result.reason,
            },
            "Wake not invoked; skipping feed event",
          );
          if (isOneShot) {
            await completeOneShot(job.id);
          }
          mark("skipped");
          continue;
        }

        if (isOneShot) {
          const successRunId = await createScheduleRun(
            job.id,
            `wake-ok:${job.id}`,
          );
          await completeScheduleRun(successRunId, { status: "ok" });
          await completeOneShot(job.id);
        }
      } catch (err) {
        log.warn(
          { err, jobId: job.id, name: job.name, wakeConversationId, isOneShot },
          "Wake schedule execution failed",
        );
        const errorMsg = err instanceof Error ? err.message : String(err);
        const wakeErrorRunId = await createScheduleRun(
          job.id,
          `wake-error:${job.id}`,
        );
        await completeScheduleRun(wakeErrorRunId, {
          status: "error",
          error: errorMsg,
        });
        await handleExecutionFailure({ job, errorMsg, isOneShot });
        failed = true;
      }
      mark(failed ? "failed" : "completed");
      continue;
    }

    // ── Workflow mode (trigger a saved workflow by name) ────────────
    if (job.mode === "workflow") {
      if (!job.workflowName) {
        log.warn(
          { jobId: job.id, name: job.name },
          "Workflow schedule has no workflowName — skipping",
        );
        mark("skipped");
        continue;
      }
      const runId = await createScheduleRun(job.id, `workflow:${job.id}`);
      let failed = false;
      try {
        log.info(
          {
            jobId: job.id,
            name: job.name,
            workflowName: job.workflowName,
            isOneShot,
          },
          "Triggering workflow schedule",
        );
        const { runId: workflowRunId } = getWorkflowRunManager().start({
          name: job.workflowName,
          args: job.workflowArgs ?? {},
          // Where the completion summary is delivered (an agent wake). Prefer an
          // explicit wake target, then fall back to the conversation that
          // created the schedule — workflow schedules made via `schedule_create`
          // store that as `createdFromConversationId` and leave
          // `wakeConversationId` unset, so without this fallback their result
          // would surface only to live SSE listeners / the DB, never delivered.
          conversationId:
            job.wakeConversationId ??
            job.createdFromConversationId ??
            undefined,
          // The schedule's persisted capability manifest scopes the run; a
          // legacy schedule with null `capabilities` normalizes to the read-only
          // baseline.
          manifest: normalizeCapabilityManifest(job.capabilities),
          trustContext: INTERNAL_GUARDIAN_TRUST_CONTEXT,
        });
        // `start` launches the run fire-and-forget and returns synchronously;
        // a successful trigger is recorded as ok. Workflow completion/failure
        // is surfaced out-of-band via workflow events and the completion wake.
        await completeScheduleRun(runId, {
          status: "ok",
          output: `workflow run ${workflowRunId} started`,
        });
        if (isOneShot) {
          await completeOneShot(job.id);
        }
      } catch (err) {
        const errorMsg = err instanceof Error ? err.message : String(err);
        log.warn(
          { err, jobId: job.id, name: job.name, isOneShot },
          "Workflow schedule trigger failed",
        );
        await completeScheduleRun(runId, { status: "error", error: errorMsg });
        await handleExecutionFailure({ job, errorMsg, isOneShot });
        failed = true;
      }
      mark(failed ? "failed" : "completed");
      continue;
    }

    // ── Execute mode ────────────────────────────────────────────────

    // Legacy task-template schedules stored a `run_task:<id>` message that an
    // older scheduler special-cased to run a saved task template. That
    // capability has been removed. Record a failed run so the dead schedule is
    // visible, and never forward the raw `run_task:<id>` string to the agent.
    if (/^run_task:\S+$/.test(job.message)) {
      const runId = await createScheduleRun(job.id, null);
      await completeScheduleRun(runId, {
        status: "error",
        error: "Scheduled task templates are no longer supported.",
      });
      log.warn(
        { jobId: job.id, name: job.name },
        "Skipped unsupported task-template schedule (run_task)",
      );
      mark("failed");
      continue;
    }

    // Reuse the conversation from the last successful run when the flag is set
    // and a prior conversation still exists; otherwise route through the
    // shared `runBackgroundJob` runner (which bootstraps fresh, applies the
    // standard timeout, and emits `activity.failed` on any failure).
    const isRruleSetMsg =
      job.syntax === "rrule" &&
      job.expression != null &&
      hasSetConstructs(job.expression);

    let reusedConversationId: string | null = null;
    if (job.reuseConversation && !isOneShot) {
      const lastId = getLastScheduleConversationId(job.id);
      if (lastId && getConversation(lastId)) {
        reusedConversationId = lastId;
      }
    }

    log.info(
      {
        jobId: job.id,
        name: job.name,
        syntax: job.syntax,
        expression: job.expression,
        isRruleSet: isRruleSetMsg,
        isOneShot,
        ...(reusedConversationId
          ? { conversationId: reusedConversationId }
          : {}),
      },
      isOneShot ? "Executing one-shot schedule" : "Executing schedule",
    );

    let conversationId: string;
    let ok: boolean;
    let errorMsg: string | undefined;
    const conversationReused = reusedConversationId != null;
    let runConversationId = reusedConversationId;
    const runId = await createScheduleRun(job.id, reusedConversationId);

    if (reusedConversationId) {
      // Reuse path: dispatch the message into the existing conversation so it
      // is continued in place. `runBackgroundJob` unconditionally bootstraps a
      // new conversation and is therefore not a drop-in replacement for the
      // reuse semantics.
      conversationId = reusedConversationId;
      broadcastScheduleConversationCreated({
        conversationId,
        scheduleJobId: job.id,
        title: job.name,
      });
      try {
        const { turnFailure } = await dispatchScheduleMessage(
          conversationId,
          job.message,
          {
            trustClass: "guardian",
            cronRunId: runId,
            ...(job.inferenceProfile
              ? { overrideProfile: job.inferenceProfile }
              : {}),
          },
        );
        // A failed LLM call (e.g. an invalid provider) ends the turn without
        // throwing, so treat a reported turn failure as a run error rather
        // than recording "ok".
        if (turnFailure) {
          ok = false;
          errorMsg = describeTurnFailure(turnFailure);
        } else {
          ok = true;
        }
      } catch (err) {
        ok = false;
        errorMsg = err instanceof Error ? err.message : String(err);
      }
    } else {
      // Fresh-bootstrap path: route through the shared runner so failures
      // surface via `activity.failed` and we get the standard timeout +
      // error-classification policy applied to every background producer.
      // The runner fires `onConversationCreated` synchronously after bootstrap
      // (before `processMessage` starts) so the macOS sidebar gets the new
      // conversation immediately rather than after the up-to-30-min job ends.
      const result = await runBackgroundJob({
        jobName: `schedule:${job.id}`,
        source: "schedule",
        prompt: job.message,
        systemHint: `Schedule: ${job.name}`,
        trustContext: { sourceChannel: "vellum", trustClass: "guardian" },
        callSite: "mainAgent",
        cronRunId: runId,
        ...(job.inferenceProfile
          ? { overrideProfile: job.inferenceProfile }
          : {}),
        // Hard timeout for talk-mode scheduled turns: bounds a wedged turn so
        // it cannot block the next scheduler tick. Configurable via
        // timeouts.scheduleTurnTimeoutSec (default 1800s).
        timeoutMs: getConfig().timeouts.scheduleTurnTimeoutSec * 1000,
        origin: "schedule",
        groupId: resolveScheduleConversationGroupId(job),
        conversationType: "scheduled",
        scheduleJobId: job.id,
        suppressFailureNotifications: job.quiet === true,
        onConversationCreated: async (newConversationId) => {
          runConversationId = newConversationId;
          await setScheduleRunConversationId(runId, newConversationId);
          broadcastScheduleConversationCreated({
            conversationId: newConversationId,
            scheduleJobId: job.id,
            title: job.name,
          });
        },
      });
      // Bootstrap-failure path returns `{ ok: false, conversationId: "" }`.
      // Substitute a sentinel only for failures so the schedule-run DB row
      // carries a recognizable marker. Successful skips (e.g.
      // `pre_first_user_message`) also return `conversationId: ""` but with
      // `ok: true` — keep the empty ID to preserve their skip contract.
      conversationId =
        !result.ok && result.conversationId === ""
          ? `bootstrap-error:${job.id}`
          : result.conversationId;
      if (runConversationId !== conversationId) {
        runConversationId = conversationId;
        await setScheduleRunConversationId(runId, conversationId);
      }
      ok = result.ok;
      errorMsg = result.error?.message;
    }

    if (ok) {
      await completeScheduleRun(runId, { status: "ok" });
      if (isOneShot) {
        await completeOneShot(job.id);
      }
      mark("completed");
    } else {
      log.warn(
        {
          err: errorMsg,
          jobId: job.id,
          name: job.name,
          syntax: job.syntax,
          expression: job.expression,
          isRruleSet: isRruleSetMsg,
          isOneShot,
        },
        isOneShot
          ? "One-shot schedule execution failed"
          : "Schedule execution failed",
      );
      await completeScheduleRun(runId, { status: "error", error: errorMsg });
      await handleExecutionFailure({
        job,
        errorMsg: errorMsg ?? "Schedule run failed",
        isOneShot,
      });

      // Only skip invalidation when the conversation was *actually* reused,
      // i.e. it contains prior successful context worth preserving. When
      // reuseConversation is true but no prior conversation existed (first run
      // or deleted), the conversation is brand-new and should be invalidated
      // like any other failed conversation.
      if (!conversationReused) {
        try {
          invalidateAssistantInferredItemsForConversation(conversationId);
        } catch (cleanupErr) {
          log.warn(
            { err: cleanupErr, conversationId },
            "Failed to invalidate assistant-inferred memory items",
          );
        }
      }
      mark("failed");
    }
  }

  return result;
}

/**
 * Emit an `activity.failed` notification for a schedule whose retries have
 * been exhausted. Fires once when the retry policy has given up, so the
 * dedupeKey caller is the per-attempt key passed in by `applyRetryDecision`
 * (already includes the job id and a timestamp).
 */
function emitScheduleActivityFailed(args: {
  jobId: string;
  jobName: string;
  errorMessage: string;
  dedupKey: string;
}): void {
  emitNotificationSignal({
    sourceChannel: "scheduler",
    sourceContextId: args.jobId,
    sourceEventName: "activity.failed",
    dedupeKey: args.dedupKey,
    contextPayload: {
      jobName: `schedule:${args.jobName}`,
      errorMessage: args.errorMessage,
      errorKind: "exception",
    },
    attentionHints: {
      requiresAction: false,
      urgency: "medium",
      isAsyncBackground: true,
      visibleInSourceNow: false,
    },
  }).catch((emitErr) => {
    log.warn(
      {
        err: emitErr instanceof Error ? emitErr.message : String(emitErr),
        jobId: args.jobId,
      },
      "Failed to emit activity.failed notification for exhausted schedule",
    );
  });
}