import { statSync } from "node:fs";
import { join } from "node:path";

import { getConfig } from "../../../config/loader.js";
import {
  isMemoryEnabled,
  isMemoryV1Active,
  isMemoryV3Live,
  usesConceptPageMemory,
} from "../../../config/memory-v3-gate.js";
import type { AssistantConfig } from "../../../config/types.js";
import {
  checkDiskPressureBackgroundGate,
  diskPressureBackgroundSkipLogFields,
  shouldLogDiskPressureBackgroundSkip,
} from "../../../daemon/disk-pressure-background-gate.js";
import {
  getMemoryCheckpoint,
  setMemoryCheckpoint,
} from "../../../persistence/checkpoints.js";
import {
  type CleanupJobKind,
  getLastScheduledCleanupEnqueueMs,
  markScheduledCleanupEnqueued,
} from "../../../persistence/cleanup-schedule-state.js";
import {
  maybeRunDbMaintenance,
  maybeRunPassiveWalCheckpoint,
} from "../../../persistence/db-maintenance.js";
import {
  EmbeddingBillingBlockError,
  extractHttpStatus,
  recordBillingBlock,
} from "../../../persistence/embeddings/embedding-billing-breaker.js";
import { QdrantCircuitOpenError } from "../../../persistence/embeddings/qdrant-circuit-breaker.js";
import {
  BackendUnavailableError,
  classifyError,
  RETRY_MAX_ATTEMPTS,
  retryDelayForAttempt,
} from "../../../persistence/job-utils.js";
import {
  claimMemoryJobs,
  completeMemoryJob,
  deferMemoryJob,
  EMBED_JOB_TYPES,
  enqueueMemoryJob,
  enqueuePruneOldConversationsJob,
  enqueuePruneOldLlmRequestLogsJob,
  enqueuePruneOldToolInvocationsJob,
  failMemoryJob,
  failStalledJobs,
  hasActiveJobOfType,
  MEMORY_V2_CONSOLIDATION_JOB_TRIGGERS,
  type MemoryJob,
  type MemoryJobType,
  MESSAGE_LEXICAL_JOB_TYPES,
  rescheduleMemoryJob,
  resetRunningJobsToPending,
  SLOW_LLM_JOB_TYPES,
} from "../../../persistence/jobs-store.js";
import { isJobQueueResolution, type JobHandler } from "../../types.js";
import { sweepOrphanConversationMemoryTables } from "./conversation-memory-orphan-sweep.js";
import { getLogger } from "./logging.js";
import { sweepOrphanMemoryRetrospectiveConversations } from "./memory-retrospective-startup-cleanup.js";
import { getWorkspaceDir } from "./paths.js";
// SUBSTRATE (v2+v3) — feeds `enqueueSubstrateMaintenanceJobs`.
import {
  type ConsolidationFailureKind,
  countBufferLines,
  readConsolidationFailureState,
} from "./substrate/consolidation-job.js";
import { resolveSubstrateTuning } from "./substrate/tuning.js";
// V1 — delete with v1. Feeds `enqueueV1MaintenanceJobs`.
import { hasPkbBufferContent } from "./v1/pkb-schedule.js";
import { spawnMemoryWorkerProcess } from "./worker-control.js";

const log = getLogger("memory-jobs-worker");

const jobHandlers = new Map<string, JobHandler>();

/**
 * Register a handler for a job type. Later registrations overwrite earlier ones
 * for the same type.
 */
export function registerJobHandler(type: string, handler: JobHandler): void {
  jobHandlers.set(type, handler);
}

const AUTOMATIC_CONSOLIDATION_JOB_PAYLOAD = {
  trigger: MEMORY_V2_CONSOLIDATION_JOB_TRIGGERS.automatic,
} as const;

/**
 * Minimum buffer entries required for a scheduled consolidation run. The
 * time-based schedule noops when `memory/buffer.md` has fewer non-empty lines
 * than this threshold — the LLM cost of a full consolidation pass outweighs
 * the benefit when the buffer is nearly empty. Mirrors the heartbeat
 * max-consecutive-runs skip pattern. Manual "Run now" and the size-based
 * trigger are not affected, and a non-empty buffer left unwritten for a full
 * interval drains regardless (see the staleness override in
 * {@link maybeEnqueueGraphMaintenanceJobs}) so a small buffer can never sit
 * unconsolidated forever.
 */
export const MIN_BUFFER_LINES_FOR_CONSOLIDATION = 10;

/**
 * V1 job types that read or write the v1 Qdrant collection via
 * `getQdrantClient()`. While v1 is not the live tier, the v1 client is
 * intentionally left uninitialized in `startup.ts`, so these handlers would
 * throw `BackendUnavailableError` and accumulate as a deferred backlog. Stale
 * rows from indexer.ts and other unguarded enqueue sites must short-circuit
 * here for the same reason the v1 graph handlers do (`isStaleV1GraphJob` in
 * `job-handlers.ts`) — one shared condition, `isMemoryV1Active`.
 *
 * Completing these as a no-op in that state is safe: their live write paths
 * keep re-enqueuing them, so nothing is lost. The one-shot
 * `sweep_orphaned_graph_node_points` cleanup is deliberately NOT in this set —
 * it has no re-enqueue, so `processJob` holds it pending (see
 * {@link SweepPostponedOffV1Error}) instead of losing it to a no-op
 * completion.
 */
const V1_QDRANT_JOB_TYPES = new Set<MemoryJobType>([
  "embed_segment",
  "embed_summary",
  "embed_media",
  "embed_attachment",
  "embed_graph_node",
  "embed_pkb_file",
  "rebuild_index",
  "delete_qdrant_vectors",
]);

/**
 * The one-shot cacheless graph-node sweep (migration 341) can only run against
 * the v1 Qdrant collection. Thrown from {@link processJob} when the job is
 * claimed while v1 is not the live tier, so {@link handleJobError} reschedules
 * it — keeping it pending with no attempt or deferral spent — until v1 is
 * active again, rather than completing it as a no-op and losing the cleanup on
 * a later return to v1 (a rollback off the substrate, or Memory switched back
 * on).
 */
class SweepPostponedOffV1Error extends Error {
  constructor() {
    super("Cacheless graph-node sweep postponed while v1 is not the live tier");
    this.name = "SweepPostponedOffV1Error";
  }
}

/**
 * Reschedule window for the postponed cacheless graph-node sweep: long enough
 * that re-checking the gate costs only a trivial claim a few times a day,
 * short enough that a rollback to v1 runs the cleanup within the same day.
 */
const SWEEP_POSTPONE_OFF_V1_MS = 6 * 60 * 60 * 1000;

/**
 * Job types whose handlers have been removed. Existing rows may still sit in
 * the database — the worker completes them silently instead of throwing.
 */
const LEGACY_JOB_TYPES = new Set([
  "embed_item",
  "extract_items",
  "batch_extract",
  "extract_entities",
  "cleanup_stale_superseded_items",
  "backfill_entity_relations",
  "refresh_weekly_summary",
  "refresh_monthly_summary",
  "journal_carry_forward",
  "generate_capability_cards",
  "generate_thread_starters",
  "memory_v2_rebuild_edges",
  // Retired memory-v3 job types — handlers were removed in the v3 rip. Kept
  // here so pre-upgrade rows enqueued by the old write path drop gracefully.
  "memory_v3_consolidate",
  "memory_v3_index_maintenance",
  "memory_v3_edge_learning",
  "memory_proc_distill",
  // Retired analyze-conversation job type — pre-upgrade pending rows drop
  // gracefully.
  "conversation_analyze",
]);

export const POLL_INTERVAL_MIN_MS = 1_500;
export const POLL_INTERVAL_MAX_MS = 30_000;

export interface MemoryJobsWorker {
  runOnce(): Promise<number>;
  stop(): void;
}

/**
 * Daemon-lifecycle entry point: spawn the memory jobs worker as a child of the
 * daemon (`detached: false`, so it appears in `assistant ps` and is torn down
 * on shutdown). The worker process is the sole drainer of the memory job queue.
 * Fire-and-forget — a worker failure must never block boot. A worker that comes
 * up late is still the desired sole drainer, so `terminateOnTimeout` is
 * deliberately not set.
 *
 * This must not be used as the standalone worker process's entry — that would
 * recurse and fork-bomb. `worker.ts` calls {@link startMemoryJobsWorkerLoop}
 * directly.
 */
export function startMemoryJobsWorker(): void {
  void spawnMemoryWorkerProcess({
    terminateOnTimeout: false,
    detached: false,
  })
    .then(({ pid, alreadyRunning }) =>
      log.info(
        { pid, alreadyRunning },
        alreadyRunning
          ? "Memory worker process already running — reusing it"
          : "Memory worker process started",
      ),
    )
    .catch((err) => log.warn({ err }, "Failed to start memory worker process"));
}

/**
 * Run the memory jobs worker loop on the caller's event loop: poll for
 * claimable jobs with adaptive backoff until {@link MemoryJobsWorker.stop} is
 * called. This is the worker loop itself, driven by the standalone worker
 * process (`worker.ts`).
 */
export function startMemoryJobsWorkerLoop(): MemoryJobsWorker {
  const recovered = resetRunningJobsToPending();
  if (recovered > 0) {
    log.info({ recovered }, "Recovered stale running memory jobs");
  }

  // Restore each cleanup job's cadence from its persisted checkpoint so a
  // restart resumes counting from the last enqueue instead of re-firing every
  // job on boot. Runs after resetRunningJobsToPending (which has already
  // touched the DB), so migrations are settled. Best-effort: on failure the
  // throttle stays at 0 and jobs fire on the first tick (the pre-persistence
  // behavior), which is a safe degradation.
  try {
    seedCleanupScheduleFromCheckpoints();
  } catch (err) {
    log.warn(
      { err },
      "Failed to seed cleanup schedule from checkpoints; jobs will fire on the first tick",
    );
  }

  // After running-job recovery (so legitimate in-flight retries aren't
  // swept), clean up orphan memory-retrospective background conversations
  // left behind by daemon crashes mid-job. Best-effort and detached — worker
  // startup never blocks on the sweep, and failures only log. Concurrent
  // ticking is safe: the sweep reads the active-job set and the orphan
  // candidates in one synchronous block after its awaited baseline loads, so
  // a retrospective forked by a mid-sweep tick is protected by its
  // pending/running job.
  void sweepOrphanMemoryRetrospectiveConversations().catch((err: unknown) => {
    log.warn(
      { err },
      "Memory-retrospective startup cleanup failed; continuing worker startup",
    );
  });

  // Also catch up on relocated conversation-keyed tables orphaned while the
  // plugin was disabled: with memory off, the conversation-deleted hook never
  // fired, so deletes during that window left rows behind (the pre-Wave-2 main
  // DB cascade caught these regardless of plugin state). Detached and
  // best-effort, same as the sweep above.
  void sweepOrphanConversationMemoryTables().catch((err: unknown) => {
    log.warn(
      { err },
      "Relocated-memory-table orphan sweep failed; continuing worker startup",
    );
  });

  let stopped = false;
  let tickRunning = false;
  let timer: ReturnType<typeof setTimeout>;
  let currentIntervalMs = POLL_INTERVAL_MIN_MS;

  const tick = async () => {
    if (stopped || tickRunning) {
      return;
    }
    tickRunning = true;
    try {
      const processed = await runMemoryJobsOnce({
        enableScheduledCleanup: true,
      });
      if (processed > 0) {
        // Per-tick claim budget equals the lane caps, so when a tick
        // processed work the next tick must run immediately to drain any
        // remaining backlog. Holding the 1.5s floor between ticks would cap
        // sustained throughput at lane-cap jobs per 1.5s and starve large
        // backlogs of short jobs.
        currentIntervalMs = 0;
      } else {
        currentIntervalMs = Math.min(
          Math.max(currentIntervalMs * 2, POLL_INTERVAL_MIN_MS),
          POLL_INTERVAL_MAX_MS,
        );
      }
    } catch (err) {
      log.error({ err }, "Memory worker tick failed");
      currentIntervalMs = Math.min(
        Math.max(currentIntervalMs * 2, POLL_INTERVAL_MIN_MS),
        POLL_INTERVAL_MAX_MS,
      );
    } finally {
      tickRunning = false;
    }
  };

  const scheduleTick = () => {
    if (stopped) {
      return;
    }
    timer = setTimeout(() => {
      void tick().then(() => {
        if (!stopped) {
          scheduleTick();
        }
      });
    }, currentIntervalMs);
    (timer as NodeJS.Timeout).unref?.();
  };

  void tick().then(() => {
    if (!stopped) {
      scheduleTick();
    }
  });

  return {
    async runOnce(): Promise<number> {
      return runMemoryJobsOnce({ enableScheduledCleanup: true });
    },
    stop(): void {
      stopped = true;
      clearTimeout(timer);
    },
  };
}

type ProcessGroup = (group: MemoryJob[]) => Promise<number>;

export async function runMemoryJobsOnce(
  options: { enableScheduledCleanup?: boolean } = {},
): Promise<number> {
  const config = getConfig();
  // While memory is disabled the queue still drains the MESSAGE-LEXICAL job
  // types — host-owned message-search indexing that shares this queue but is
  // not a memory feature. Every memory lane and every maintenance enqueue
  // stays idle in that state; only the lexical types are claimable.
  const memoryEnabled = isMemoryEnabled(config);
  const enableScheduledCleanup =
    options.enableScheduledCleanup === true && memoryEnabled;

  const diskPressureGate = checkDiskPressureBackgroundGate("background-work");
  if (diskPressureGate.action === "skip") {
    if (shouldLogDiskPressureBackgroundSkip("memory-jobs-worker")) {
      log.warn(
        {
          source: "memory",
          ...diskPressureBackgroundSkipLogFields(diskPressureGate),
        },
        "Memory jobs worker skipped during disk pressure cleanup mode",
      );
    }
    return 0;
  }

  // Fail jobs that have been running longer than the configured timeout
  const timedOut = failStalledJobs(config.memory.jobs.stalledJobTimeoutMs);
  if (timedOut > 0) {
    log.warn({ timedOut }, "Timed out stalled memory jobs");
  }

  const cfgSlow = Math.max(1, config.memory.jobs.slowLlmConcurrency);
  const cfgFast = Math.max(1, config.memory.jobs.fastConcurrency);
  const cfgEmbed = Math.max(1, config.memory.jobs.embedConcurrency);

  // Claim per-lane budgets so a backlog of slow LLM jobs cannot starve fast
  // jobs (and vice versa). The Qdrant circuit breaker still gates only the
  // embed lane inside `claimMemoryJobs`. With memory disabled, the slow and
  // embed lanes get no budget and the fast lane is restricted to the
  // message-lexical types (they ride the fast lane).
  const claimed = claimMemoryJobs(
    {
      slowLlm: memoryEnabled ? cfgSlow : 0,
      fast: cfgFast,
      embed: memoryEnabled ? cfgEmbed : 0,
    },
    memoryEnabled ? undefined : MESSAGE_LEXICAL_JOB_TYPES,
  );

  if (claimed.length === 0) {
    if (enableScheduledCleanup) {
      maybeEnqueueScheduledCleanupJobs(config);
    }
    maybeEnqueueGraphMaintenanceJobs(config);
    maybeEnqueueRetrospectiveSweepJob(config);
    if (memoryEnabled) {
      await maybeRunDbMaintenance();
      await maybeRunPassiveWalCheckpoint();
    }
    return 0;
  }

  const slowSet = new Set<MemoryJobType>(SLOW_LLM_JOB_TYPES);
  const embedSet = new Set<MemoryJobType>(EMBED_JOB_TYPES);
  const slowJobs: MemoryJob[] = [];
  const fastJobs: MemoryJob[] = [];
  const embedJobs: MemoryJob[] = [];
  for (const job of claimed) {
    if (slowSet.has(job.type)) {
      slowJobs.push(job);
    } else if (embedSet.has(job.type)) {
      embedJobs.push(job);
    } else {
      fastJobs.push(job);
    }
  }

  const processGroup: ProcessGroup = async (group) => {
    let groupProcessed = 0;
    for (const job of group) {
      try {
        const resolution = await processJob(job, config);
        applyQueueResolution(job, resolution);
        groupProcessed += 1;
      } catch (err) {
        try {
          handleJobError(job, err);
        } catch (handlerErr) {
          log.error(
            { err: handlerErr, jobId: job.id, type: job.type },
            "handleJobError itself threw, job left in running status",
          );
        }
        // A billing block (402) is deterministic — every subsequent embed
        // call will fail identically. Defer the remaining embed jobs in
        // this batch instead of burning a network round-trip on each one.
        if (
          err instanceof EmbeddingBillingBlockError ||
          (embedSet.has(job.type) && extractHttpStatus(err) === 402)
        ) {
          for (const remaining of group.slice(group.indexOf(job) + 1)) {
            deferMemoryJob(remaining.id);
          }
          break;
        }
      }
    }
    return groupProcessed;
  };

  // Run all three lanes in parallel. Each lane runs its own bounded task pool
  // so a slow `graph_consolidate` cannot block embed or fast jobs from making
  // progress, and per-`(type, conversationId)` grouping inside each lane keeps
  // same-conversation jobs serialized.
  const [slowProcessed, fastProcessed, embedProcessed] = await Promise.all([
    runLanePool(slowJobs, cfgSlow, processGroup),
    runLanePool(fastJobs, cfgFast, processGroup),
    runLanePool(embedJobs, cfgEmbed, processGroup),
  ]);

  if (enableScheduledCleanup) {
    maybeEnqueueScheduledCleanupJobs(config);
  }
  maybeEnqueueGraphMaintenanceJobs(config);
  maybeEnqueueRetrospectiveSweepJob(config);
  await maybeRunDbMaintenance();
  await maybeRunPassiveWalCheckpoint();
  return slowProcessed + fastProcessed + embedProcessed;
}

/**
 * Run a single lane's jobs through a bounded task pool of size `concurrency`.
 *
 * Jobs targeting different conversations (via payload.conversationId) are
 * placed in separate groups and run in parallel up to the lane's concurrency
 * cap. Jobs targeting the same conversation — or global jobs without a
 * conversationId — share a group and run sequentially to avoid checkpoint
 * races.
 */
async function runLanePool(
  jobs: MemoryJob[],
  concurrency: number,
  processGroup: ProcessGroup,
): Promise<number> {
  if (jobs.length === 0) {
    return 0;
  }

  const groups = new Map<string, MemoryJob[]>();
  for (const job of jobs) {
    const convId =
      typeof job.payload.conversationId === "string"
        ? job.payload.conversationId
        : null;
    const groupKey = convId ? `${job.type}:${convId}` : job.type;
    let group = groups.get(groupKey);
    if (!group) {
      group = [];
      groups.set(groupKey, group);
    }
    group.push(job);
  }

  let processed = 0;
  const typeGroups = [...groups.values()];

  if (typeGroups.length <= concurrency) {
    const results = await Promise.allSettled(typeGroups.map(processGroup));
    for (const result of results) {
      if (result.status === "fulfilled") {
        processed += result.value;
      } else {
        log.error(
          { err: result.reason },
          "Memory job group rejected unexpectedly — jobs in this batch may have been dropped",
        );
      }
    }
    return processed;
  }

  // Task pool: keep `concurrency` groups in flight at all times so a new group
  // starts the instant any slot frees up.
  let nextIdx = 0;
  const startNext = (): Promise<void> | undefined => {
    if (nextIdx >= typeGroups.length) {
      return undefined;
    }
    const group = typeGroups[nextIdx++]!;
    return processGroup(group)
      .then(
        (count) => {
          processed += count;
        },
        (err) => {
          log.error(
            { err },
            "Memory job group rejected unexpectedly — jobs in this batch may have been dropped",
          );
        },
      )
      .then(() => startNext());
  };

  const workers = Array.from(
    { length: Math.min(concurrency, typeGroups.length) },
    () => startNext()!,
  );
  await Promise.all(workers);
  return processed;
}

// ── Job error handling ─────────────────────────────────────────────

function handleJobError(job: MemoryJob, err: unknown): void {
  if (err instanceof SweepPostponedOffV1Error) {
    rescheduleMemoryJob(job.id, SWEEP_POSTPONE_OFF_V1_MS);
    log.debug(
      { jobId: job.id, type: job.type },
      "Cacheless graph-node sweep held pending while v1 is not the live tier",
    );
    return;
  }

  if (err instanceof EmbeddingBillingBlockError) {
    const result = deferMemoryJob(job.id);
    if (result === "failed") {
      log.error(
        { jobId: job.id, type: job.type },
        "Billing breaker open, job exceeded max deferrals",
      );
    } else {
      log.debug(
        { jobId: job.id, type: job.type },
        "Billing breaker open, deferring job",
      );
    }
    return;
  }

  // Detect 402 billing exhaustion from any embedding backend and trip the
  // billing breaker so subsequent embed jobs short-circuit at claim time.
  if (EMBED_JOB_TYPES.includes(job.type) && extractHttpStatus(err) === 402) {
    recordBillingBlock();
    const result = deferMemoryJob(job.id);
    if (result === "failed") {
      log.error(
        { jobId: job.id, type: job.type },
        "Embedding billing block (402), job exceeded max deferrals",
      );
    } else {
      log.warn(
        { jobId: job.id, type: job.type },
        "Embedding billing block (402), deferring job",
      );
    }
    return;
  }

  if (err instanceof BackendUnavailableError) {
    const result = deferMemoryJob(job.id);
    if (result === "failed") {
      log.error(
        { jobId: job.id, type: job.type },
        "Embedding backend unavailable, job exceeded max deferrals",
      );
    } else {
      log.debug(
        { jobId: job.id, type: job.type },
        "Embedding backend unavailable, deferring job",
      );
    }
  } else if (err instanceof QdrantCircuitOpenError) {
    const result = deferMemoryJob(job.id);
    if (result === "failed") {
      log.error(
        { jobId: job.id, type: job.type },
        "Qdrant circuit breaker open, job exceeded max deferrals",
      );
    } else {
      log.debug(
        { jobId: job.id, type: job.type },
        "Qdrant circuit breaker open, deferring job",
      );
    }
  } else {
    const message = err instanceof Error ? err.message : String(err);
    const category = classifyError(err);
    if (category === "retryable") {
      const delay = retryDelayForAttempt(job.attempts + 1);
      failMemoryJob(job.id, message, {
        retryDelayMs: delay,
        maxAttempts: RETRY_MAX_ATTEMPTS,
      });
      log.warn(
        { err, jobId: job.id, type: job.type, delay, category },
        "Memory job failed (retryable)",
      );
    } else {
      failMemoryJob(job.id, message, { maxAttempts: 1 });
      log.warn(
        { err, jobId: job.id, type: job.type, category },
        "Memory job failed (fatal)",
      );
    }
  }
}

// ── Job dispatch ───────────────────────────────────────────────────

/**
 * Apply a handler's returned {@link JobQueueResolution} to its claimed row.
 * Handlers that return anything else resolve as `completed` (the historical
 * contract: failure is signaled by throwing). This is the worker's half of
 * the outcome-truthfulness boundary: the persisted `memory_jobs.status` must
 * reflect the handler's actual outcome, so a handler that reports failure
 * through a returned value (e.g. the retrospective's `wake_failed`, the
 * consolidation's `run_failed`) dead-letters or retries instead of silently
 * completing.
 */
function applyQueueResolution(job: MemoryJob, resolution: unknown): void {
  if (!isJobQueueResolution(resolution)) {
    completeMemoryJob(job.id);
    return;
  }
  switch (resolution.queueResolution) {
    case "completed": {
      completeMemoryJob(job.id);
      return;
    }
    case "failed": {
      failMemoryJob(job.id, resolution.errorMessage ?? "handler failed", {
        maxAttempts: 1,
      });
      return;
    }
    case "retryable": {
      failMemoryJob(job.id, resolution.errorMessage ?? "handler failed", {
        retryDelayMs:
          resolution.retryDelayMs ?? retryDelayForAttempt(job.attempts + 1),
        maxAttempts: RETRY_MAX_ATTEMPTS,
      });
      return;
    }
    case "deferred": {
      deferMemoryJob(job.id, {
        ...(resolution.deferralExhaustedMessage
          ? { exhaustedMessage: resolution.deferralExhaustedMessage }
          : {}),
      });
      return;
    }
  }
}

async function processJob(
  job: MemoryJob,
  config: AssistantConfig,
): Promise<unknown> {
  // Dispatch-level half of the v1-staleness guard, on the same condition the
  // handler-level half uses (`isStaleV1GraphJob` in `job-handlers.ts`): v1 work
  // runs only while v1 is the live tier, which memory being off is not (see
  // `isMemoryV1Active`). Memory-off jobs of these types are unclaimable
  // upstream — `runMemoryJobsOnce` restricts the claim to the message-lexical
  // types — so this arm is defense in depth for a hot config flip mid-batch.
  if (!isMemoryV1Active(config)) {
    if (V1_QDRANT_JOB_TYPES.has(job.type)) {
      return;
    }
    if (job.type === "sweep_orphaned_graph_node_points") {
      throw new SweepPostponedOffV1Error();
    }
  }
  const handler = jobHandlers.get(job.type);
  if (handler) {
    return await handler(job, config);
  }

  const rawType = (job as { type: string }).type;
  if (LEGACY_JOB_TYPES.has(rawType)) {
    log.debug({ jobId: job.id, type: rawType }, "Dropping legacy job");
    return;
  }
  throw new Error(`Unknown memory job type: ${rawType}`);
}

const MS_PER_DAY = 24 * 60 * 60 * 1000;

const CLEANUP_JOB_KINDS: readonly CleanupJobKind[] = [
  "conversations",
  "llm_request_logs",
  "tool_invocations",
];

const CLEANUP_ENQUEUE_CHECKPOINT_KEYS: Record<CleanupJobKind, string> = {
  conversations: "cleanup:last_enqueue:conversations",
  llm_request_logs: "cleanup:last_enqueue:llm_request_logs",
  tool_invocations: "cleanup:last_enqueue:tool_invocations",
};

/**
 * Seed the in-memory cleanup throttle from persisted checkpoints so each job's
 * cadence survives a daemon restart. Without this the throttle would start at 0
 * on every boot, re-firing every cleanup job immediately regardless of when it
 * last ran — which for a long retention window (e.g. 30-day conversation
 * pruning) turns a frequent restart cycle into a prune on every boot.
 *
 * A job with no checkpoint (never enqueued on this instance) keeps its default
 * 0, so it fires once on the first tick and then persists its timestamp. On a
 * fresh instance that first prune is a harmless no-op (nothing is old enough to
 * delete yet); on an upgrade it clears whatever has already aged out.
 *
 * Must run after DB migrations settle — the worker startup path already
 * satisfies this (it touches the DB before calling here).
 */
export function seedCleanupScheduleFromCheckpoints(): void {
  for (const kind of CLEANUP_JOB_KINDS) {
    const raw = getMemoryCheckpoint(CLEANUP_ENQUEUE_CHECKPOINT_KEYS[kind]);
    if (raw === null) {
      continue;
    }
    const parsed = Number.parseInt(raw, 10);
    if (Number.isFinite(parsed) && parsed > 0) {
      markScheduledCleanupEnqueued(kind, parsed);
    }
  }
}

/**
 * Record that a cleanup job for `kind` was just enqueued: advance the in-memory
 * throttle and persist the timestamp so the cadence survives a restart. A
 * config-driven throttle reset (ConfigWatcher) only clears the in-memory value;
 * the next enqueue re-persists here, so the checkpoint self-heals on the tick
 * after a retention change.
 */
function recordCleanupEnqueued(kind: CleanupJobKind, nowMs: number): void {
  markScheduledCleanupEnqueued(kind, nowMs);
  setMemoryCheckpoint(CLEANUP_ENQUEUE_CHECKPOINT_KEYS[kind], String(nowMs));
}

/**
 * Enqueue periodic cleanup jobs, each on a cadence equal to its own retention
 * window. A job that keeps data for N is re-enqueued at most once per N:
 * pruning that retains LLM logs for 1h runs hourly, pruning that retains
 * conversations for 30d runs every 30d. A non-positive window disables its job
 * (`conversationRetentionDays`/`auditLog.retentionDays` of 0, or
 * `llmRequestLogRetentionMs` of `null`/0) — a 0 window would otherwise make the
 * cadence 0 and busy-loop the enqueue. Each job's throttle is tracked
 * independently in cleanup-schedule-state (and persisted via checkpoints so the
 * cadence survives restarts), and enqueue is deduped in jobs-store, so repeated
 * calls remain safe.
 *
 * Exported for tests; the worker calls it on every idle/drain tick.
 *
 * Returns true if at least one job was enqueued this call.
 */
export function maybeEnqueueScheduledCleanupJobs(
  config: AssistantConfig,
  nowMs = Date.now(),
): boolean {
  const cleanup = config.memory.cleanup;
  if (!cleanup.enabled) {
    return false;
  }

  // A job is due when at least its full retention window has elapsed since the
  // last enqueue for that job. The throttle is seeded from a persisted
  // checkpoint at startup and reset to 0 when ConfigWatcher observes a
  // retention change, so a due job also fires promptly after a config change
  // while an unchanged one resumes its cadence across restarts.
  const isDue = (kind: CleanupJobKind, intervalMs: number): boolean =>
    nowMs - getLastScheduledCleanupEnqueueMs(kind) >= intervalMs;

  let enqueuedAny = false;

  if (
    cleanup.conversationRetentionDays > 0 &&
    isDue("conversations", cleanup.conversationRetentionDays * MS_PER_DAY)
  ) {
    const jobId = enqueuePruneOldConversationsJob(
      cleanup.conversationRetentionDays,
    );
    recordCleanupEnqueued("conversations", nowMs);
    enqueuedAny = true;
    log.debug(
      { jobId, retentionDays: cleanup.conversationRetentionDays },
      "Enqueued scheduled prune_old_conversations",
    );
  }

  // A retention of `null` or `0` disables LLM-request-log pruning (keep
  // forever). `0` must be excluded here, not just `null`: the cadence interval
  // equals the retention window, so `isDue("llm_request_logs", 0)` reduces to
  // `nowMs - lastEnqueue >= 0` — always true — which would re-enqueue the prune
  // on every idle/drain tick and spin a sqlite3 subprocess many times a second.
  if (
    cleanup.llmRequestLogRetentionMs !== null &&
    cleanup.llmRequestLogRetentionMs > 0 &&
    isDue("llm_request_logs", cleanup.llmRequestLogRetentionMs)
  ) {
    const jobId = enqueuePruneOldLlmRequestLogsJob(
      cleanup.llmRequestLogRetentionMs,
    );
    recordCleanupEnqueued("llm_request_logs", nowMs);
    enqueuedAny = true;
    log.debug(
      { jobId, retentionMs: cleanup.llmRequestLogRetentionMs },
      "Enqueued scheduled prune_old_llm_request_logs",
    );
  }

  // Audit-log (tool_invocations) retention is configured separately under
  // `auditLog.retentionDays`; its prune cadence follows that window.
  if (
    config.auditLog.retentionDays > 0 &&
    isDue("tool_invocations", config.auditLog.retentionDays * MS_PER_DAY)
  ) {
    const jobId = enqueuePruneOldToolInvocationsJob(
      config.auditLog.retentionDays,
    );
    recordCleanupEnqueued("tool_invocations", nowMs);
    enqueuedAny = true;
    log.debug(
      { jobId, retentionDays: config.auditLog.retentionDays },
      "Enqueued scheduled prune_old_tool_invocations",
    );
  }

  return enqueuedAny;
}

// ── Graph maintenance scheduling ──────────────────────────────────

const GRAPH_DECAY_INTERVAL_MS = 60 * 60 * 1000; // 1 hour
const GRAPH_CONSOLIDATE_INTERVAL_MS = 4 * 60 * 60 * 1000; // 4 hours
const GRAPH_PATTERN_SCAN_INTERVAL_MS = 24 * 60 * 60 * 1000; // 1 day
const GRAPH_NARRATIVE_INTERVAL_MS = 7 * 24 * 60 * 60 * 1000; // 1 week
// Backstop cadence for v3 self-maintenance. The primary trigger is the
// post-consolidation follow-up (see `consolidation-job.ts`); this interval only
// covers the case where that follow-up is missed (enqueue failure). A
// conservative cadence is fine since
// the maintenance pass is idempotent and cheap when there's nothing to do.
const GRAPH_V3_MAINTAIN_INTERVAL_MS = 6 * 60 * 60 * 1000; // 6 hours

export const GRAPH_MAINTENANCE_CHECKPOINTS = {
  decay: "graph_maintenance:decay:last_run",
  consolidate: "graph_maintenance:consolidate:last_run",
  patternScan: "graph_maintenance:pattern_scan:last_run",
  narrative: "graph_maintenance:narrative:last_run",
  // FROZEN: persisted checkpoint key — never rename the value.
  memoryV2Consolidate: "memory_v2_consolidate_last_run",
  memoryV3Maintain: "memory_v3_maintain_last_run",
  pkbFiling: "pkb_filing_last_run",
  pkbCompaction: "pkb_compaction_last_run",
} as const;

/**
 * Durable checkpoint (epoch-ms) tracking the last scheduled retrospective
 * sweep so the cadence survives daemon restarts.
 */
export const RETROSPECTIVE_SWEEP_CHECKPOINT = "retro_sweep:last_run";

/**
 * Enqueue the scheduled `memory_retrospective_sweep` job once its interval has
 * elapsed — the timer-driven backstop for retrospective triggers that an
 * abnormal turn end (crash / IPC drop) skipped. See
 * `memory-retrospective-sweep.ts`.
 *
 * First-run seeding: a missing checkpoint is seeded to `nowMs` WITHOUT
 * enqueuing, so the first sweep fires one full interval after startup instead
 * of treating the sweep as immediately overdue — a `?? "0"` fallback would make
 * `nowMs - 0 >= sweepIntervalMs` true on the first tick and enqueue
 * retrospective LLM work the moment the worker starts, violating the
 * no-startup-LLM-work invariant. Mirrors the PKB schedule's null-checkpoint
 * seed.
 *
 * Deduped against an in-flight sweep so a slow scan can't stack copies; the
 * checkpoint still advances in that case to hold the cadence steady.
 *
 * `memory.retrospective.enabled` is checked here as well as inside the
 * enqueue funnel, so a disabled retrospective costs no scan at all rather
 * than a full conversation scan whose every enqueue is then declined. The
 * checkpoint is left untouched while disabled, so re-enabling fires the
 * first sweep on the next tick, which is the desired catch-up.
 *
 * Exported for tests; the worker calls it on every idle/drain tick. Returns
 * true if a sweep job was enqueued this call.
 */
export function maybeEnqueueRetrospectiveSweepJob(
  config: AssistantConfig,
  nowMs = Date.now(),
): boolean {
  if (!isMemoryEnabled(config)) {
    return false;
  }

  if (!config.memory.retrospective.enabled) {
    return false;
  }

  const checkpoint = getMemoryCheckpoint(RETROSPECTIVE_SWEEP_CHECKPOINT);
  if (checkpoint === null) {
    setMemoryCheckpoint(RETROSPECTIVE_SWEEP_CHECKPOINT, String(nowMs));
    return false;
  }

  const lastRun = parseInt(checkpoint, 10);
  const sweepIntervalMs = config.memory.retrospective.sweepIntervalMs;
  if (nowMs - lastRun < sweepIntervalMs) {
    return false;
  }

  if (hasActiveJobOfType("memory_retrospective_sweep")) {
    setMemoryCheckpoint(RETROSPECTIVE_SWEEP_CHECKPOINT, String(nowMs));
    return false;
  }

  enqueueMemoryJob("memory_retrospective_sweep", {});
  setMemoryCheckpoint(RETROSPECTIVE_SWEEP_CHECKPOINT, String(nowMs));
  return true;
}

/**
 * Whether `hour` falls inside the PKB jobs' configured active window. A `null`
 * bound on either side means no restriction. Windows may wrap midnight
 * (start > end, e.g. 22–6).
 */
function isWithinPkbActiveHours(
  hour: number,
  start: number | null,
  end: number | null,
): boolean {
  if (start == null || end == null) {
    return true;
  }
  if (start <= end) {
    return hour >= start && hour < end;
  }
  return hour >= start || hour < end;
}

/** Line count of the memory buffer, the scheduler's consolidation gate. */
function memoryBufferLineCount(): number {
  return countBufferLines(join(getWorkspaceDir(), "memory", "buffer.md"));
}

/**
 * Milliseconds since `memory/buffer.md` was last written; `0` (fresh) when
 * the file is missing or unreadable. Drives the min-lines staleness override:
 * a fresh mtime means entries are still arriving and waiting for more is
 * reasonable; a stale mtime means the buffer has settled below the minimum
 * and would otherwise never drain.
 */
function memoryBufferIdleMs(nowMs: number): number {
  try {
    const stat = statSync(join(getWorkspaceDir(), "memory", "buffer.md"));
    return Math.max(0, nowMs - stat.mtimeMs);
  } catch {
    return 0;
  }
}

// Failure backoff for automatic consolidation enqueues. A failed run never
// trims the buffer, so without backoff the size trigger re-enqueues a
// fast-failing run on every worker poll (~1.5s), each iteration persisting a
// full background conversation. Two curves, selected by the most recent
// failure's kind (see `ConsolidationFailureState`):
//   - transient (network blip, model hiccup, timeout): 5min doubling,
//     capped at 30min — transient failures never meaningfully delay
//     consolidation;
//   - billing (non-retryable PROVIDER_BILLING): 1h doubling, capped at
//     max(6h, the configured interval) — retrying can't succeed until the
//     account is funded.
// Manual "run now" enqueues go through the routes layer, not this schedule,
// and are never gated.
const CONSOLIDATION_TRANSIENT_BACKOFF_BASE_MS = 5 * 60 * 1000;
const CONSOLIDATION_TRANSIENT_BACKOFF_CAP_MS = 30 * 60 * 1000;
const CONSOLIDATION_BILLING_BACKOFF_BASE_MS = 60 * 60 * 1000;
const CONSOLIDATION_BILLING_BACKOFF_MIN_CAP_MS = 6 * 60 * 60 * 1000;

/**
 * Backoff window after `consecutiveFailures` failed consolidation runs.
 * Billing: `min(1h * 2^(n-1), max(6h, intervalMs))`. Transient:
 * `min(5min * 2^(n-1), 30min)`.
 */
export function consolidationFailureBackoffMs(
  kind: ConsolidationFailureKind,
  consecutiveFailures: number,
  intervalMs: number,
): number {
  if (kind === "billing") {
    const capMs = Math.max(
      CONSOLIDATION_BILLING_BACKOFF_MIN_CAP_MS,
      intervalMs,
    );
    return Math.min(
      CONSOLIDATION_BILLING_BACKOFF_BASE_MS * 2 ** (consecutiveFailures - 1),
      capMs,
    );
  }
  return Math.min(
    CONSOLIDATION_TRANSIENT_BACKOFF_BASE_MS * 2 ** (consecutiveFailures - 1),
    CONSOLIDATION_TRANSIENT_BACKOFF_CAP_MS,
  );
}

/**
 * Milliseconds until the consolidation failure backoff expires; `0` when no
 * failure state is recorded or the window has already elapsed. Exported so
 * the create-memory route's consolidation nudge honors the same backoff as
 * the scheduler.
 */
export function consolidationBackoffRemainingMs(
  intervalMs: number,
  nowMs: number,
): number {
  const state = readConsolidationFailureState();
  if (state === null) {
    return 0;
  }
  const backoffMs = consolidationFailureBackoffMs(
    state.kind,
    state.consecutiveFailures,
    intervalMs,
  );
  return Math.max(0, state.lastFailureAt + backoffMs - nowMs);
}

/**
 * Enqueue periodic graph maintenance jobs.
 *
 * Mutually exclusive between v1 and concept-page memory:
 *   - concept-page memory active ({@link usesConceptPageMemory}) → only one
 *     buffer-drainer is scheduled ({@link enqueueSubstrateMaintenanceJobs}).
 *   - inactive → the four v1 entries (decay, consolidate, pattern_scan,
 *     narrative) are scheduled instead ({@link enqueueV1MaintenanceJobs}).
 *
 * The `memory/buffer.md` is shared, so exactly one consolidator owns the drain
 * at a time. When concept-page memory is active, the concept-page consolidator
 * (`memory_v2_consolidate`) is the sole buffer-drainer.
 *
 * Read/write paths route to concept pages when the gate is on, so v1 graph
 * data goes unread; running v1 maintenance alongside it is wasted compute and
 * LLM spend. The v1 code path remains live so disabling concept-page memory
 * fully re-engages v1.
 *
 * Uses durable checkpoints so intervals survive daemon restarts — jobs only
 * fire when the actual elapsed time since last run exceeds the interval.
 * Sweep is intentionally not on this schedule: it is debounced from the
 * live `graph_extract` trigger path (see `indexMessageNow` in `indexer.ts`)
 * so it runs on the same idle/message-count cadence.
 *
 * Independently of the v1/concept-page split, a flag-gated
 * `memory_v3_maintain` backstop is scheduled when a v3 path is active so the
 * topic tree self-heals even if the primary post-consolidation follow-up
 * enqueue is missed ({@link enqueueV3BackstopJobs}).
 */
export function maybeEnqueueGraphMaintenanceJobs(
  config: AssistantConfig,
  nowMs = Date.now(),
): void {
  if (!isMemoryEnabled(config)) {
    return;
  }

  if (usesConceptPageMemory(config.memory)) {
    enqueueSubstrateMaintenanceJobs(config, nowMs);
  } else {
    // V1 — delete with v1. Dropping only the banner-marked function body below
    // would leave this call dangling: collapse the branch to the substrate arm.
    enqueueV1MaintenanceJobs(config, nowMs);
  }

  // v3 self-maintenance backstop. Orthogonal to the mutual exclusion above:
  // it owns its own checkpoint and operates on the v3 topic tree. Gated on
  // the same config that gates the v3 plugin so it stays inert when v3 is
  // off. The job handler itself no-ops when v3 is off, so this guard is
  // belt-and-suspenders that also avoids a wasted enqueue.
  if (isMemoryV3Live(config)) {
    enqueueV3BackstopJobs(nowMs);
  }
}

// ── SUBSTRATE (v2+v3) maintenance ─────────────────────────────────

/**
 * Substrate maintenance entries; scheduled only while concept-page memory is
 * active. The concept-page consolidator (`memory_v2_consolidate`) is the sole
 * buffer-drainer, enqueued on an interval cadence plus a size-based trigger,
 * both gated by the consolidation failure backoff.
 */
function enqueueSubstrateMaintenanceJobs(
  config: AssistantConfig,
  nowMs: number,
): void {
  const tuning = resolveSubstrateTuning(config.memory);

  // The single buffer-drainer entry, shared by the interval cadence and the
  // size-based trigger below.
  const consolidateEntry = {
    key: GRAPH_MAINTENANCE_CHECKPOINTS.memoryV2Consolidate,
    intervalMs: tuning.consolidation_interval_hours * 60 * 60 * 1000,
    jobType: "memory_v2_consolidate" as MemoryJobType,
  };

  const lastRun = parseInt(
    getMemoryCheckpoint(consolidateEntry.key) ?? "0",
    10,
  );
  let enqueuedConsolidate = false;
  if (nowMs - lastRun >= consolidateEntry.intervalMs) {
    enqueuedConsolidate = maybeEnqueueScheduledConsolidation(
      consolidateEntry,
      nowMs,
    );
  }

  // Size-based trigger: when the shared buffer crosses the configured line
  // count, drain it now rather than waiting out the interval. Retargets to the
  // same consolidator the interval branch above selected.
  //
  // The size branch is checkpoint-blind by design (it must fire before the
  // interval elapses), so it dedupes against an already-active consolidate job
  // instead — otherwise it would re-enqueue on every worker tick while the
  // buffer stays over threshold, flooding the queue with redundant LLM work.
  // The failure backoff gates it too: a failed run never trims the buffer, so
  // the size trigger alone would re-fire a failing run on every tick. A
  // backoff skip leaves the checkpoint alone.
  const maxLines = tuning.consolidation_max_buffer_lines;
  if (
    !enqueuedConsolidate &&
    maxLines !== null &&
    !hasActiveJobOfType(consolidateEntry.jobType)
  ) {
    if (memoryBufferLineCount() >= maxLines) {
      const backoffRemainingMs = consolidationBackoffRemainingMs(
        consolidateEntry.intervalMs,
        nowMs,
      );
      if (backoffRemainingMs > 0) {
        log.debug(
          { backoffRemainingMs },
          "Size-triggered consolidation skipped: failure backoff active",
        );
      } else {
        enqueueMemoryJob(
          consolidateEntry.jobType,
          AUTOMATIC_CONSOLIDATION_JOB_PAYLOAD,
        );
        setMemoryCheckpoint(consolidateEntry.key, String(nowMs));
      }
    }
  }
}

/**
 * Interval-cadence arm of the substrate consolidation schedule: enqueue the
 * consolidate job unless the failure backoff or the minimum-buffer-lines gate
 * (with its staleness override) skips it. Returns true when the job was
 * enqueued.
 */
function maybeEnqueueScheduledConsolidation(
  entry: { key: string; intervalMs: number; jobType: MemoryJobType },
  nowMs: number,
): boolean {
  // Failure backoff: skip WITHOUT advancing the checkpoint so the enqueue
  // fires on the first tick after the window elapses instead of a full
  // interval later.
  const backoffRemainingMs = consolidationBackoffRemainingMs(
    entry.intervalMs,
    nowMs,
  );
  if (backoffRemainingMs > 0) {
    log.debug(
      { backoffRemainingMs },
      "Scheduled consolidation skipped: failure backoff active",
    );
    return false;
  }
  // Noop scheduled consolidation when the buffer has too few entries to
  // justify an LLM run — mirrors the heartbeat max-consecutive-runs skip.
  // The checkpoint advances so the next check fires after the regular
  // interval. Manual "Run now" is unaffected (routes layer, not schedule).
  const bufferLines = memoryBufferLineCount();
  if (bufferLines < MIN_BUFFER_LINES_FOR_CONSOLIDATION) {
    // Staleness override: the minimum only defers while entries are still
    // arriving. Once a non-empty buffer has sat unwritten for a full
    // interval, drain it anyway — otherwise a buffer that never reaches the
    // minimum re-skips every interval forever and its facts never become
    // concept pages.
    const stale =
      bufferLines > 0 && memoryBufferIdleMs(nowMs) >= entry.intervalMs;
    if (!stale) {
      log.debug(
        "Scheduled consolidation skipped: buffer under minimum line threshold",
      );
      setMemoryCheckpoint(entry.key, String(nowMs));
      return false;
    }
  }
  enqueueMemoryJob(entry.jobType, AUTOMATIC_CONSOLIDATION_JOB_PAYLOAD);
  setMemoryCheckpoint(entry.key, String(nowMs));
  return true;
}

// ── V1 (legacy engine) maintenance — delete with v1 ───────────────

/**
 * v1-only maintenance entries; scheduled only when the legacy graph engine is
 * the live memory tier. Covers the four v1 graph lifecycle jobs (decay,
 * consolidate, pattern_scan, narrative) and the PKB filing/compaction
 * schedule.
 */
function enqueueV1MaintenanceJobs(
  config: AssistantConfig,
  nowMs: number,
): void {
  const schedule: Array<{
    key: string;
    intervalMs: number;
    jobType: MemoryJobType;
  }> = [
    {
      key: GRAPH_MAINTENANCE_CHECKPOINTS.decay,
      intervalMs: GRAPH_DECAY_INTERVAL_MS,
      jobType: "graph_decay",
    },
    {
      key: GRAPH_MAINTENANCE_CHECKPOINTS.consolidate,
      intervalMs: GRAPH_CONSOLIDATE_INTERVAL_MS,
      jobType: "graph_consolidate",
    },
    {
      key: GRAPH_MAINTENANCE_CHECKPOINTS.patternScan,
      intervalMs: GRAPH_PATTERN_SCAN_INTERVAL_MS,
      jobType: "graph_pattern_scan",
    },
    {
      key: GRAPH_MAINTENANCE_CHECKPOINTS.narrative,
      intervalMs: GRAPH_NARRATIVE_INTERVAL_MS,
      jobType: "graph_narrative_refine",
    },
  ];
  for (const { key, intervalMs, jobType } of schedule) {
    const lastRun = parseInt(getMemoryCheckpoint(key) ?? "0", 10);
    if (nowMs - lastRun >= intervalMs) {
      enqueueMemoryJob(jobType, {});
      setMemoryCheckpoint(key, String(nowMs));
    }
  }

  // PKB filing/compaction — v1-only, like the v1 graph entries above (under
  // concept-page memory the consolidation job owns periodic background memory
  // processing). Same durable-checkpoint pattern, with four PKB-specific gates:
  //  - no checkpoint yet (fresh workspace, or the first tick after an
  //    upgrade): seed it to now WITHOUT enqueuing, so the first run lands a
  //    full interval later instead of an LLM job firing at boot;
  //  - outside the configured active-hours window: skip AND advance the
  //    checkpoint, so the next attempt lands a full interval later (the
  //    interval cadence, not a busy-retry against a closed window);
  //  - filing with an empty buffer: skip and advance — no work, no LLM run
  //    (mirrors the consolidation minimum-line skip above);
  //  - either PKB job already pending/running: skip WITHOUT advancing, so the
  //    next worker tick retries. Filing and compaction both rewrite the PKB
  //    tree, so at most one of the two is ever in the queue.
  const filingConfig = config.filing;
  const withinActiveHours = isWithinPkbActiveHours(
    new Date(nowMs).getHours(),
    filingConfig.activeHoursStart ?? null,
    filingConfig.activeHoursEnd ?? null,
  );
  const pkbSchedule: Array<{
    key: string;
    intervalMs: number;
    jobType: MemoryJobType;
    enabled: boolean;
    hasWork: () => boolean;
  }> = [
    {
      key: GRAPH_MAINTENANCE_CHECKPOINTS.pkbFiling,
      intervalMs: filingConfig.intervalMs,
      jobType: "pkb_filing",
      enabled: filingConfig.enabled,
      hasWork: () => hasPkbBufferContent(),
    },
    {
      key: GRAPH_MAINTENANCE_CHECKPOINTS.pkbCompaction,
      intervalMs: filingConfig.compactionIntervalMs,
      jobType: "pkb_compaction",
      enabled: filingConfig.compactionEnabled,
      hasWork: () => true,
    },
  ];
  for (const { key, intervalMs, jobType, enabled, hasWork } of pkbSchedule) {
    if (!enabled) {
      continue;
    }
    const checkpoint = getMemoryCheckpoint(key);
    if (checkpoint === null) {
      setMemoryCheckpoint(key, String(nowMs));
      continue;
    }
    const lastRun = parseInt(checkpoint, 10);
    if (nowMs - lastRun < intervalMs) {
      continue;
    }
    if (!withinActiveHours || !hasWork()) {
      setMemoryCheckpoint(key, String(nowMs));
      continue;
    }
    if (
      hasActiveJobOfType("pkb_filing") ||
      hasActiveJobOfType("pkb_compaction")
    ) {
      continue;
    }
    enqueueMemoryJob(jobType, {});
    setMemoryCheckpoint(key, String(nowMs));
  }
}

// ── V3 backstop ───────────────────────────────────────────────────

/**
 * v3 self-maintenance backstop on its own durable checkpoint. The
 * post-consolidation follow-up in `consolidation-job.ts` is the primary
 * trigger; this interval only self-heals when that follow-up is missed
 * (failed enqueue).
 */
function enqueueV3BackstopJobs(nowMs: number): void {
  const lastRun = parseInt(
    getMemoryCheckpoint(GRAPH_MAINTENANCE_CHECKPOINTS.memoryV3Maintain) ?? "0",
    10,
  );
  if (nowMs - lastRun >= GRAPH_V3_MAINTAIN_INTERVAL_MS) {
    enqueueMemoryJob("memory_v3_maintain", {});
    setMemoryCheckpoint(
      GRAPH_MAINTENANCE_CHECKPOINTS.memoryV3Maintain,
      String(nowMs),
    );
  }
}