// 重构前 index.ts 只读参考：commit 6a04144d3f5103014f75b637d7e6bc2f45bf007f，blob 56e590c1a6b5ca7ad128bf2c992f60e46c355a58；可用 `git show 6a04144d3f5103014f75b637d7e6bc2f45bf007f:src/components/microservices/code-queue/src/index.ts` 查看。
import { spawnSync } from "node:child_process";
import { copyFileSync, existsSync, mkdirSync, readFileSync, readdirSync, statSync } from "node:fs";
import { dirname, resolve } from "node:path";
import { Database } from "bun:sqlite";
import postgres from "postgres";
import { createHourlyJsonlWriter, logRetentionBytesForService } from "../../../shared/src/rotating-jsonl";
import type {
  AttemptSummary,
  CgroupMemoryUsage,
  CodexEventSummary,
  CodexRunResult,
  JudgeFailureDetails,
  JudgeProbeCase,
  JudgeResult,
  JsonValue,
  LiveOutput,
  OutputChannel,
  PersistedState,
  PromptHistoryItem,
  QueueRecord,
  QueuedStatusReason,
  QueueTask,
  QueueTaskRequest,
  RunMode,
  RuntimeConfig,
  CodeQueueServiceRole,
  SchedulerActiveRunHeartbeat,
  TaskStatus,
  WorkdirRecord,
} from "./types";
import {
  codeAgentPortForModel,
  defaultCodeModels,
  extractRecord,
  extractString,
  codeModelProviderSourceContract,
  normalizeCodeExecutionMode,
  normalizeCodeModel,
  normalizeRequestedCodeExecutionMode,
  terminalStatus,
} from "./code-agent/common";
import type { ActiveRun, ActiveRunSlotWaiter } from "./code-agent/common";
import { configureCodexPort, runCodexTurn } from "./code-agent/codex";
import { configureOpenCodePort, runOpenCodeTurn } from "./code-agent/opencode";
import {
  compactRetryTaskContext,
  compactContinuationPromptTargetChars,
  configureJudge,
  explicitUserInterrupt,
  judgeFailContinuationPrompt,
  judgeReasonForPrompt,
  judgeTaskInputDiagnostics,
  judgeTask,
  queueRecoveryRetryPrompt,
  retryPrompt,
} from "./judge";
import { defaultJudgeProbeCases } from "./judge-probes";
import { injectCodeQueueEnvironmentHint, promptWithCodeQueueEnvironmentHint, userPromptForDisplay } from "./prompts";
import { configureDevContainers, devContainerStatus, ensureTaskExecutionContainer, startDevContainer } from "./dev-containers";
import { appendOutput, appendOutputArchive, configureTaskOutput, taskFullOutput } from "./task-output";
import {
  buildDevContainerPlan,
  configureProviderRuntime,
  containerTunnelStartScript,
  defaultWorkdirForProvider,
  devContainerPingScript,
  executionModeOptions,
  executionProviderOptions,
  masterKeyReadScript,
  masterKeySetupScript,
  masterProxyEvidenceScript,
  masterProxyFinishScript,
  masterProxyPrepareScript,
  normalizeProviderId,
  normalizeTaskProviderId,
  providerIsMain,
  remoteAppServerCommand,
  remoteCodexConfigInstallScript,
  remoteCodexRuntimePrepareScript,
  remoteContainerStartScript,
  remoteHostWorkdirForTask,
  remoteKeyInstallScript,
  resolveTaskCwd,
  runCodeQueueSsh,
  shellQuote,
  throwIfCommandFailed,
  windowsNativeAppServerCommand,
} from "./provider-runtime";
import {
  armIdleNotification,
  backfillClaudeQqTaskNotifications,
  claudeQqNotificationItems,
  claudeQqNotificationOutboxItemCount,
  claudeQqNotificationOutboxStats,
  configureNotifications,
  drainClaudeQqNotificationOutbox,
  loadClaudeQqNotificationOutboxFromDatabase,
  maybeNotifyQueueIdle,
  notificationTargetConfigured,
  notificationTargetLabel,
  notifyTaskTerminal,
  scheduleClaudeQqNotificationDrain,
} from "./notifications";
import {
  configureQueueApi,
  loadAllTasksForRead,
  outputChunkResponse,
  perQueueSummaries,
  queueSummary,
  queueSummaryForHealth,
  queueSummaryForResponse,
  taskMatchesSearch,
  taskPageRows,
  taskSearchTerms,
  tasksOverviewResponse,
  taskForListResponse,
  transcriptChunkResponse,
} from "./queue-api";
import { buildExecutionDiagnostics, buildSchedulerHeartbeat, normalizeSchedulerHeartbeatStaleMs, schedulerHeartbeatStaleMs, schedulerHeartbeatStaleMsMax, staleRecoveryCandidate } from "./execution-diagnostics";
import { ReferenceTaskLookupError, configureReferences, injectReferencedTaskContext, taskReferenceIds } from "./references";
import { classifyTransientDatabaseError, databaseStorageHealth, isTransientDatabaseError, restoreDirtyFlushBatch, takeDirtyFlushBatch } from "./database-resilience";
import {
  applyOaTraceStatsToTaskJson,
  configureOaEvents,
  oaEventPublisherStatus,
  publishCodeQueueJudgeEvent,
  publishCodeQueueQueueUpdated,
  publishCodeQueueTaskUpdated,
  publishCodeQueueTraceStatsSnapshot,
  publishCodeQueueTraceStep,
  outputTraceKind,
  readOaTraceStatsForTask,
  readOaTraceStatsForTaskAttempts,
  readOaTraceStatsForTasks,
  readOaTraceStepsForTask,
} from "./oa-events";
import { collectRuntimePreflight, runtimePreflightJson } from "./runtime-preflight";
import { collectSkillAvailability, collectSkillSyncPreflight, skillAvailabilityJson, skillSyncPreflightJson } from "./skill-availability";
import { createSteerId, findSteerTraceConfirmation, normalizeSteerId, normalizeSteerPromptText, steerDuplicateDecision, steerPromptHash, steerTraceConfirmationJson, steerTraceText } from "./steer-confirmation";
import { createResumeId, findResumeTraceConfirmation, normalizeResumeId, resumeDuplicateDecision, resumePromptHash, resumeTraceConfirmationJson, resumeTraceText } from "./resume-confirmation";
import { configureSelfTests, runJudgeInfraSelfTest, runQueueClaimMoveSelfTest, runQueueOrderingSelfTest, runReferenceInjectionSelfTest, runStaleActiveRecoverySelfTest, runTracePortSelfTest, runTraceSummaryContractSelfTest } from "./self-tests";
import {
  codexToolLifecycleStartedBeforeIn,
  configureTaskView,
  formatCommandOutput,
  isCodexToolLifecycleOutput,
  lastAssistantMessage,
  promptLineCount,
  recordNumberField,
  recordStringField,
  safePreview,
  setAttemptFeedbackPrompt,
  setAttemptInputPrompt,
  statsDaysFromUrl,
  taskForMetaResponse,
  taskForResponse,
  taskListStepCount,
  taskOutputMaxSeq as taskViewOutputMaxSeq,
  taskStatisticsSummary,
  taskSummaryResponse,
  timestampMs,
  nonNegativeElapsed,
  taskTraceStepDetailResponse,
  taskTraceStepsResponse,
  taskTraceSummaryResponse,
  taskPromptDetailResponse,
} from "./task-view";

type SqlClient = postgres.Sql;
type SqlExecutor = postgres.Sql | postgres.TransactionSql;

const recentLogs: JsonValue[] = [];
const serviceStartedAt = new Date().toISOString();
const defaultQueueId = "default";
const firstPaintOverviewWarmUrl = "http://code-queue.local/api/tasks/overview?limit=12&transcriptLimit=1&compact=1&selected=0&includeActive=0&stats=0&skipTrace=1";
let firstPaintOverviewWarmInFlight: Promise<void> | null = null;
const judgeFailRetryLimit = 3;
const fallbackJudgeRetryLimit = 3;
const judgeProbeBehaviorVersion = "code-queue-judge-probe:v1";
const maxTaskAttempts = 99;
const referenceInjectionMaxRounds: number | null = null;
const retryBackoffBaseMs = 1000;
const retryBackoffMaxMs = 10 * 60 * 1000;
const orphanedActiveTaskRecoveryStaleMs = normalizeSchedulerHeartbeatStaleMs(5 * 60 * 1000);
const queueIdPattern = /^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$/u;
const queueNameMaxLength = 80;
const workdirMaxLength = 512;
const config = readConfig();

const logger = createLogger("code-queue", config.logFile);
configureOaEvents({ baseUrl: config.oaEventFlowBaseUrl, logger, nowIso });
const providerGatewayEgressProxy = configureProviderGatewayEgressProxy();
if (providerGatewayEgressProxy.enabled) {
  const proxyEnv = providerGatewayEgressProxy.proxyUrl;
  process.env.HTTP_PROXY = proxyEnv;
  process.env.HTTPS_PROXY = proxyEnv;
  process.env.http_proxy = proxyEnv;
  process.env.https_proxy = proxyEnv;
  process.env.ALL_PROXY = proxyEnv;
  process.env.all_proxy = proxyEnv;
  process.env.NO_PROXY = providerGatewayEgressProxy.noProxy;
  process.env.no_proxy = providerGatewayEgressProxy.noProxy;
}
const state = emptyState();
let processing = false;
const processingQueues = new Set<string>();
const mergingQueues = new Set<string>();
const activeRuns = new Map<string, ActiveRun>();
const activeRunSlotReservations = new Set<string>();
const activeRunSlotWaiters: ActiveRunSlotWaiter[] = [];
const devContainerEnsurePromises = new Map<string, Promise<void>>();
let nextActiveRunSlotWaiterId = 1;
let devReadyCache: { checkedAtMs: number; value: JsonValue } | null = null;
let persistTimer: ReturnType<typeof setTimeout> | null = null;
let persistDirty = false;
let shutdownRequested = false;
let serviceReady = false;
let sql: SqlClient = createSqlClient();
let sqlRotationInFlight: Promise<void> | null = null;
const schedulerReconcileAudit = {
  lastRunAt: null as string | null,
  lastReason: null as string | null,
  lastLoaded: 0,
  lastChanged: 0,
  lastRecovered: 0,
  lastError: null as string | null,
};

function createSqlClient(): SqlClient {
  return postgres(config.databaseUrl, {
    max: config.databasePoolMax,
    idle_timeout: 20,
    connect_timeout: 10,
    connection: { application_name: "unidesk-code-queue" },
  });
}

let databaseReady = false;
let databaseLastError: string | null = null;
let databaseLastErrorDetail: unknown = null;
let databaseFlushTimer: ReturnType<typeof setTimeout> | null = null;
let databaseFlushInFlight = false;
let databaseFlushFailureCount = 0;
let databaseLastFlushFailureAt: string | null = null;
let databaseNextFlushRetryAt: string | null = null;
let databaseClientRotationCount = 0;
let databaseLastClientRotationAt: string | null = null;
let databaseLastClientRotationReason: string | null = null;
let databaseTransientUncaughtSuppressedCount = 0;
let databaseLastTransientUncaughtSuppressedAt: string | null = null;
const dirtyDatabaseTaskIds = new Set<string>();
const dirtyDatabaseQueueIds = new Set<string>();
const workdirRecords = new Map<string, WorkdirRecord>();

function envString(name: string, fallback: string): string {
  const value = process.env[name];
  return value === undefined || value.length === 0 ? fallback : value;
}

function envNullableString(name: string): string | null {
  const value = process.env[name];
  return value === undefined || value.length === 0 ? null : value;
}

function envRequiredString(name: string): string {
  const value = process.env[name];
  if (value === undefined || value.trim().length === 0) throw new Error(`${name} is required`);
  return value;
}

function envNumber(name: string, fallback: number): number {
  const raw = process.env[name];
  if (raw === undefined || raw.length === 0) return fallback;
  const value = Number(raw);
  return Number.isFinite(value) && value > 0 ? Math.floor(value) : fallback;
}

function envNonNegativeNumber(name: string, fallback: number): number {
  const raw = process.env[name];
  if (raw === undefined || raw.length === 0) return fallback;
  const value = Number(raw);
  return Number.isFinite(value) && value >= 0 ? Math.floor(value) : fallback;
}

function clampTaskAttempts(value: number): number {
  const parsed = Number(value);
  if (!Number.isFinite(parsed)) return maxTaskAttempts;
  return Math.max(1, Math.min(maxTaskAttempts, Math.floor(parsed)));
}

function reserveNextAttemptBudget(task: QueueTask): boolean {
  const nextAttempt = task.attempts.length + 1;
  if (nextAttempt > maxTaskAttempts) return false;
  task.maxAttempts = clampTaskAttempts(Math.max(task.maxAttempts, nextAttempt));
  return true;
}

function envBool(name: string, fallback: boolean): boolean {
  const raw = process.env[name];
  if (raw === undefined || raw.trim().length === 0) return fallback;
  const value = raw.trim().toLowerCase();
  if (value === "1" || value === "true" || value === "yes" || value === "on") return true;
  if (value === "0" || value === "false" || value === "no" || value === "off") return false;
  return fallback;
}

function serviceRoleValue(raw: string): CodeQueueServiceRole {
  const normalized = raw.trim().toLowerCase();
  if (normalized === "read" || normalized === "write" || normalized === "scheduler" || normalized === "combined") return normalized;
  return "combined";
}

function serviceRoleAllowsWrite(role: CodeQueueServiceRole): boolean {
  return role === "combined" || role === "write";
}

function serviceRoleAllowsScheduler(role: CodeQueueServiceRole): boolean {
  return role === "combined" || role === "scheduler";
}

function serviceRoleReadOnly(role: CodeQueueServiceRole): boolean {
  return role === "read";
}

function envList(name: string, fallback: string[]): string[] {
  const raw = process.env[name];
  const source = raw === undefined || raw.length === 0 ? fallback.join(",") : raw;
  return Array.from(new Set(source.split(",").map((item) => item.trim()).filter(Boolean)));
}

function envModelReasoningEfforts(name: string, fallback: Record<string, string>): Record<string, string> {
  const raw = process.env[name];
  const map: Record<string, string> = { ...fallback };
  if (raw === undefined || raw.trim().length === 0) return map;
  for (const item of raw.split(/[,;]+/u)) {
    const [model, effort] = item.split("=", 2).map((part) => part.trim());
    if (model && effort) map[model.toLowerCase()] = effort;
  }
  return map;
}

function withRequiredModel(models: string[], model: string): string[] {
  return models.includes(model) ? models : [model, ...models];
}

function uniqueModels(models: string[]): string[] {
  return Array.from(new Set(models.map((item) => item.trim()).filter(Boolean)));
}






function sandboxValue(raw: string): RuntimeConfig["sandbox"] {
  if (raw === "read-only" || raw === "workspace-write" || raw === "danger-full-access") return raw;
  return "danger-full-access";
}

function approvalValue(raw: string): RuntimeConfig["approvalPolicy"] {
  if (raw === "untrusted" || raw === "on-failure" || raw === "on-request" || raw === "never") return raw;
  return "never";
}

function readConfig(): RuntimeConfig {
  const defaultModel = normalizeCodeModel(envString("CODE_QUEUE_DEFAULT_MODEL", "gpt-5.5"));
  const codeModels = uniqueModels(withRequiredModel(envList("CODE_QUEUE_MODELS", defaultCodeModels).map(normalizeCodeModel), defaultModel));
  const dataDir = envString("CODE_QUEUE_DATA_DIR", "/var/lib/unidesk/code-queue");
  const notifyTargetTypeRaw = envString("CODE_QUEUE_NOTIFY_CLAUDEQQ_TARGET_TYPE", "private").toLowerCase();
  const mainProviderId = envString("CODE_QUEUE_MAIN_PROVIDER_ID", "D601");
  const devContainerDefaultProviderId = envString("CODE_QUEUE_DEV_CONTAINER_DEFAULT_PROVIDER_ID", "D601");
  const remoteDefaultWorkdir = envString("CODE_QUEUE_REMOTE_WORKDIR", "/home/ubuntu");
  const defaultWorkdir = envString("CODE_QUEUE_WORKDIR", "/workspace");
  const devContainerMasterHost = envString("CODE_QUEUE_DEV_CONTAINER_MASTER_HOST", "74.48.78.17");
  const defaultProviderGatewayProxyHost = `unidesk-provider-gateway-${mainProviderId}`;
  const serviceRole = serviceRoleValue(envString("CODE_QUEUE_SERVICE_ROLE", "combined"));
  const executionProviderIds = Array.from(new Set([
    mainProviderId,
    ...envList("CODE_QUEUE_EXECUTION_PROVIDER_IDS", [devContainerDefaultProviderId]),
  ].map((item) => item.trim()).filter(Boolean)));
  return {
    host: envString("HOST", "0.0.0.0"),
    port: envNumber("PORT", 4222),
    dataDir,
    instanceId: envString("CODE_QUEUE_INSTANCE_ID", mainProviderId),
    environment: envString("UNIDESK_ENV", ""),
    namespace: envString("UNIDESK_NAMESPACE", ""),
    databaseName: envString("UNIDESK_DATABASE_NAME", envString("UNIDESK_DEV_DATABASE_NAME", "")),
    deployServiceId: envString("UNIDESK_DEPLOY_SERVICE_ID", "code-queue"),
    deployRef: envString("UNIDESK_DEPLOY_REF", ""),
    deployCommit: envString("CODE_QUEUE_DEPLOY_COMMIT", "unknown"),
    deployRequestedCommit: envString("CODE_QUEUE_DEPLOY_REQUESTED_COMMIT", ""),
    serviceRole,
    schedulerEnabled: envBool("CODE_QUEUE_SCHEDULER_ENABLED", serviceRoleAllowsScheduler(serviceRole)),
    schedulerPollIntervalMs: Math.max(500, Math.min(30_000, envNumber("CODE_QUEUE_SCHEDULER_POLL_INTERVAL_MS", 2000))),
    startupOaBackfillEnabled: envBool("CODE_QUEUE_STARTUP_OA_BACKFILL_ENABLED", false),
    outputArchiveDir: envString("CODE_QUEUE_OUTPUT_ARCHIVE_DIR", resolve(dataDir, "output-archive")),
    logFile: envString("LOG_FILE", "/var/log/unidesk/code-queue.jsonl"),
    defaultWorkdir,
    mainProviderId,
    remoteDefaultWorkdir,
    executionProviderIds,
    remoteCodexEnvKeys: envList("CODE_QUEUE_REMOTE_CODEX_ENV_KEYS", ["OPENAI_API_KEY", "CRS_OAI_KEY", "OPENAI_BASE_URL", "OPENAI_API_BASE", "DEEPSEEK_API_KEY", "DEEPSEEK_API_BASE", "DEEPSEEK_MODEL", "MINIMAX_API_KEY", "MINIMAX_API_BASE", "MINIMAX_MODEL", "MINIMAX_M3_MODEL", "GH_TOKEN", "GITHUB_TOKEN", "GH_HOST", "GITHUB_API_URL", "GH_REPO"]),
    skillsPath: envString("UNIDESK_SKILLS_PATH", "/root/.agents/skills"),
    runnerSkillsSourcePath: envString("CODE_QUEUE_RUNNER_SKILLS_SOURCE_PATH", "/home/ubuntu/.agents/skills"),
    resolvedRunnerSkillsPath,
    codexHome: envString("CODE_QUEUE_CODEX_HOME", "/var/lib/unidesk/code-queue/codex-home"),
    opencodeXdgDir: envString("CODE_QUEUE_OPENCODE_XDG_DIR", resolve(dataDir, "opencode-xdg")),
    sourceCodexConfig: envString("CODE_QUEUE_SOURCE_CODEX_CONFIG", "/root/.codex/config.toml"),
    codexSqliteLogExportEnabled: envBool("CODE_QUEUE_CODEX_SQLITE_LOG_EXPORT_ENABLED", true),
    codexSqliteLogExportIntervalMs: Math.max(10_000, Math.min(10 * 60_000, envNumber("CODE_QUEUE_CODEX_SQLITE_LOG_EXPORT_INTERVAL_MS", 60_000))),
    codexSqliteLogExportBatchSize: Math.max(100, Math.min(20_000, envNumber("CODE_QUEUE_CODEX_SQLITE_LOG_EXPORT_BATCH_SIZE", 500))),
    codexSqliteLogMaxBytes: logRetentionBytesForService("codex-app-server", ["CODE_QUEUE_CODEX_SQLITE_LOG_MAX_BYTES"]),
    memoryWatchdogThresholdBytes: Math.max(0, envNumber("CODE_QUEUE_MEMORY_WATCHDOG_THRESHOLD_BYTES", 0)),
    memoryWatchdogIntervalMs: Math.max(1000, Math.min(60_000, envNumber("CODE_QUEUE_MEMORY_WATCHDOG_INTERVAL_MS", 2000))),
    defaultModel,
    codexModels: codeModels.filter((model) => codeAgentPortForModel(model) === "codex"),
    defaultReasoningEffort: envNullableString("CODE_QUEUE_REASONING_EFFORT"),
    modelReasoningEfforts: envModelReasoningEfforts("CODE_QUEUE_MODEL_REASONING_EFFORTS", { "gpt-5.5": "xhigh" }),
    sandbox: sandboxValue(envString("CODE_QUEUE_SANDBOX", "danger-full-access")),
    approvalPolicy: approvalValue(envString("CODE_QUEUE_APPROVAL_POLICY", "never")),
    defaultMaxAttempts: clampTaskAttempts(envNumber("CODE_QUEUE_MAX_ATTEMPTS", maxTaskAttempts)),
    codeModels,
    deepseekApiKey: envString("DEEPSEEK_API_KEY", ""),
    deepseekApiBase: envString("DEEPSEEK_API_BASE", "https://api.deepseek.com").replace(/\/+$/u, ""),
    deepseekModel: envString("DEEPSEEK_MODEL", "deepseek-chat"),
    minimaxApiKey: envString("MINIMAX_API_KEY", ""),
    minimaxApiBase: envString("MINIMAX_API_BASE", "https://api.minimaxi.com/v1").replace(/\/+$/u, ""),
    minimaxModel: envString("MINIMAX_MODEL", "MiniMax-M2.7"),
    minimaxM3Model: envString("MINIMAX_M3_MODEL", "MiniMax-M3"),
    judgeTimeoutMs: envNumber("MINIMAX_JUDGE_TIMEOUT_MS", 90_000),
    judgeRepairAttempts: Math.max(0, Math.min(5, envNumber("MINIMAX_JUDGE_REPAIR_ATTEMPTS", 2))),
    judgeMaxTokens: Math.max(800, Math.min(4000, envNumber("MINIMAX_JUDGE_MAX_TOKENS", 1800))),
    turnNoActivityTimeoutMs: Math.max(60_000, Math.min(30 * 60_000, envNumber("CODEX_TURN_NO_ACTIVITY_TIMEOUT_MS", 6 * 60_000))),
    databaseUrl: envRequiredString("DATABASE_URL"),
    databasePoolMax: Math.max(1, Math.min(8, envNumber("CODE_QUEUE_DATABASE_POOL_MAX", envNumber("DATABASE_POOL_MAX", 2)))),
    databaseFlushIntervalMs: Math.max(100, Math.min(10_000, envNumber("CODE_QUEUE_DATABASE_FLUSH_INTERVAL_MS", 1000))),
    databaseClaimTimeoutMs: Math.max(1000, Math.min(60_000, envNumber("CODE_QUEUE_DATABASE_CLAIM_TIMEOUT_MS", 15_000))),
    oaEventFlowBaseUrl: envString("OA_EVENT_FLOW_BASE_URL", "http://oa-event-flow:4255").replace(/\/+$/u, ""),
    notifyClaudeQqEnabled: envBool("CODE_QUEUE_NOTIFY_CLAUDEQQ_ENABLED", false),
    notifyClaudeQqBaseUrl: envString("CODE_QUEUE_NOTIFY_CLAUDEQQ_BASE_URL", "http://claudeqq.unidesk.svc.cluster.local:3290").replace(/\/+$/u, ""),
    notifyClaudeQqTargetType: notifyTargetTypeRaw === "group" ? "group" : "private",
    notifyClaudeQqUserId: envString("CODE_QUEUE_NOTIFY_CLAUDEQQ_USER_ID", "645275593").trim(),
    notifyClaudeQqGroupId: envString("CODE_QUEUE_NOTIFY_CLAUDEQQ_GROUP_ID", "").trim(),
    notifyClaudeQqMaxResponseChars: Math.max(500, Math.min(50_000, envNumber("CODE_QUEUE_NOTIFY_CLAUDEQQ_MAX_RESPONSE_CHARS", 12_000))),
    notifyClaudeQqTimeoutMs: Math.max(1000, Math.min(60_000, envNumber("CODE_QUEUE_NOTIFY_CLAUDEQQ_TIMEOUT_MS", 15_000))),
    notifyClaudeQqSendAttempts: Math.max(1, Math.min(10, envNumber("CODE_QUEUE_NOTIFY_CLAUDEQQ_SEND_ATTEMPTS", 3))),
    notifyClaudeQqRetryIntervalMs: Math.max(5_000, Math.min(10 * 60_000, envNumber("CODE_QUEUE_NOTIFY_CLAUDEQQ_RETRY_INTERVAL_MS", 60_000))),
    notifyClaudeQqMaxOutboxItems: Math.max(100, Math.min(10_000, envNumber("CODE_QUEUE_NOTIFY_CLAUDEQQ_MAX_OUTBOX_ITEMS", 2000))),
    maxInMemoryOutputRecords: envNonNegativeNumber("CODE_QUEUE_IN_MEMORY_OUTPUT_RECORDS", 10),
    maxInMemoryEventRecords: envNonNegativeNumber("CODE_QUEUE_IN_MEMORY_EVENT_RECORDS", 10),
    maxActiveQueues: Math.max(0, Math.min(32, envNumber("CODE_QUEUE_MAX_ACTIVE_QUEUES", 0))),
    egressProxyEnabled: envBool("CODE_QUEUE_EGRESS_PROXY_ENABLED", true),
    egressProxyUrl: envString("CODE_QUEUE_EGRESS_PROXY_URL", `http://${defaultProviderGatewayProxyHost}:18789`).replace(/\/+$/u, ""),
    egressProxyNoProxy: envString("CODE_QUEUE_EGRESS_PROXY_NO_PROXY", [
      "localhost",
      "127.0.0.1",
      "::1",
      "host.docker.internal",
      defaultProviderGatewayProxyHost,
      devContainerMasterHost,
      "74.48.78.17",
      "backend-core",
      "oa-event-flow",
      "database",
      "hyueapi.com",
      ".hyueapi.com",
      "api.minimaxi.com",
      ".minimaxi.com",
    ].join(",")),
    devContainerMasterHost,
    devContainerDefaultProviderId,
    devContainerImage: envString("CODE_QUEUE_DEV_CONTAINER_IMAGE", ""),
    devContainerWorkdir: envString("CODE_QUEUE_DEV_CONTAINER_WORKDIR", remoteDefaultWorkdir),
    windowsNativeCodexBridgeDir: envString("CODE_QUEUE_WINDOWS_NATIVE_CODEX_BRIDGE_DIR", "/home/ubuntu/.unidesk/code-queue/windows-native-codex"),
    windowsNativeCodexCommand: envString("CODE_QUEUE_WINDOWS_NATIVE_CODEX_COMMAND", "codex app-server --listen stdio://"),
    windowsNativeCodexConnectHost: envString("CODE_QUEUE_WINDOWS_NATIVE_CODEX_CONNECT_HOST", "host.docker.internal"),
    windowsNativeCodexDefaultWorkdir: envString("CODE_QUEUE_WINDOWS_NATIVE_CODEX_DEFAULT_WORKDIR", "/mnt/f/Work/ConStart"),
    windowsNativeCodexIdleTimeoutMs: Math.max(30_000, Math.min(24 * 60 * 60_000, envNumber("CODE_QUEUE_WINDOWS_NATIVE_CODEX_IDLE_TIMEOUT_MS", 10 * 60_000))),
  };
}

function configureProviderGatewayEgressProxy(): { enabled: boolean; proxyUrl: string; noProxy: string; channel: string } {
  if (!config.egressProxyEnabled) {
    return { enabled: false, proxyUrl: "", noProxy: config.egressProxyNoProxy, channel: "provider-gateway" };
  }
  return {
    enabled: true,
    proxyUrl: config.egressProxyUrl,
    noProxy: config.egressProxyNoProxy,
    channel: "provider-gateway",
  };
}

async function providerGatewayEgressProxyStatus(): Promise<JsonValue> {
  if (!providerGatewayEgressProxy.enabled) return { enabled: false, channel: "provider-gateway" };
  const base = {
    enabled: true,
    channel: providerGatewayEgressProxy.channel,
    proxyUrl: providerGatewayEgressProxy.proxyUrl,
    noProxy: providerGatewayEgressProxy.noProxy,
  };
  const controller = new AbortController();
  const timer = setTimeout(() => controller.abort(), 600);
  try {
    const url = new URL(providerGatewayEgressProxy.proxyUrl);
    url.pathname = "/__unidesk/egress-proxy/health";
    url.search = "";
    const response = await fetch(url, { signal: controller.signal });
    const bodyText = await response.text();
    let upstream: JsonValue = bodyText.slice(0, 1000);
    try {
      upstream = JSON.parse(bodyText) as JsonValue;
    } catch {
      // Keep the bounded body text as evidence if the proxy returned a non-JSON failure page.
    }
    return { ...base, connected: response.ok, status: response.status, upstream };
  } catch (error) {
    return { ...base, connected: false, error: error instanceof Error ? error.message : String(error) };
  } finally {
    clearTimeout(timer);
  }
}

function runtimeIdentity(): Record<string, JsonValue> {
  return {
    environment: config.environment || null,
    namespace: config.namespace || null,
    databaseName: config.databaseName || null,
    serviceId: config.deployServiceId || "code-queue",
    deployRef: config.deployRef || null,
  };
}

function createLogger(service: string, logFile: string) {
  const writer = createHourlyJsonlWriter({
    baseLogFile: logFile,
    service,
    maxBytes: logRetentionBytesForService(service, ["CODE_QUEUE_SERVICE_LOG_MAX_BYTES"]),
  });
  writer.prune();
  return (level: "debug" | "info" | "warn" | "error", message: string, data?: JsonValue): void => {
    const entry = data === undefined
      ? { ts: new Date().toISOString(), service, level, message }
      : { ts: new Date().toISOString(), service, level, message, data };
    recentLogs.push(entry as JsonValue);
    while (recentLogs.length > 500) recentLogs.shift();
    const line = `${JSON.stringify(entry)}\n`;
    try {
      writer.appendLine(line, new Date(entry.ts));
    } catch (error) {
      console.error(JSON.stringify({ ts: new Date().toISOString(), service, level: "error", message: "log_write_failed", error: String(error) }));
    }
    const consoleMethod = level === "error" ? console.error : level === "warn" ? console.warn : console.log;
    consoleMethod(line.trimEnd());
  };
}

interface CodexSqliteLogRow {
  id: number;
  ts: number;
  ts_nanos: number;
  level: string;
  target: string;
  feedback_log_body: string | null;
  module_path: string | null;
  file: string | null;
  line: number | null;
  thread_id: string | null;
  process_uuid: string | null;
  estimated_bytes: number | null;
}

const codexSqliteLogExporter = {
  running: false,
  lastRunAt: null as string | null,
  lastExportedRows: 0,
  totalExportedRows: 0,
  lastDeletedRows: 0,
  lastVacuumAt: null as string | null,
  lastError: null as string | null,
};

function codexLogDate(row: CodexSqliteLogRow): Date {
  const seconds = Number(row.ts);
  const nanos = Number(row.ts_nanos);
  const millis = seconds * 1000 + Math.floor((Number.isFinite(nanos) ? nanos : 0) / 1_000_000);
  return new Date(Number.isFinite(millis) ? millis : Date.now());
}

function codexLogJson(row: CodexSqliteLogRow, sqlitePath: string): Record<string, JsonValue> {
  return {
    ts: codexLogDate(row).toISOString(),
    service: "codex-app-server",
    source: "codex-log-db",
    sqlitePath,
    id: row.id,
    level: row.level,
    target: row.target,
    message: row.feedback_log_body ?? "",
    modulePath: row.module_path,
    file: row.file,
    line: row.line,
    threadId: row.thread_id,
    processUuid: row.process_uuid,
    estimatedBytes: row.estimated_bytes ?? 0,
  };
}

function codexLogSqliteFiles(): string[] {
  if (!existsSync(config.codexHome)) return [];
  return readdirSync(config.codexHome, { withFileTypes: true })
    .filter((entry) => entry.isFile() && /^logs_\d+\.sqlite$/u.test(entry.name))
    .map((entry) => resolve(config.codexHome, entry.name))
    .sort();
}

function runSqliteMaintenance(db: Database, sqlitePath: string, forceVacuum: boolean): void {
  try {
    db.exec("PRAGMA wal_checkpoint(TRUNCATE)");
  } catch {
    // Active Codex app-server writers may hold a lock; the next pass will retry.
  }
  if (!forceVacuum) return;
  try {
    db.exec("VACUUM");
    codexSqliteLogExporter.lastVacuumAt = new Date().toISOString();
  } catch (error) {
    logger("warn", "codex_sqlite_log_vacuum_failed", { sqlitePath, error: errorToJson(error) });
  }
}

function exportCodexSqliteLogFile(sqlitePath: string, writer: ReturnType<typeof createHourlyJsonlWriter>): number {
  const db = new Database(sqlitePath);
  let exported = 0;
  let deleted = 0;
  try {
    db.exec("PRAGMA busy_timeout = 2000");
    const rows = db.query("SELECT id, ts, ts_nanos, level, target, feedback_log_body, module_path, file, line, thread_id, process_uuid, estimated_bytes FROM logs ORDER BY id ASC LIMIT ?")
      .all(config.codexSqliteLogExportBatchSize) as unknown as CodexSqliteLogRow[];
    for (const row of rows) {
      writer.appendJson(codexLogJson(row, sqlitePath), codexLogDate(row));
      exported += 1;
    }
    const maxId = rows.at(-1)?.id;
    if (typeof maxId === "number" && Number.isFinite(maxId)) {
      const result = db.query("DELETE FROM logs WHERE id <= ?").run(maxId);
      deleted = Number(result.changes ?? 0);
      codexSqliteLogExporter.lastDeletedRows = deleted;
    }
    const size = statSync(sqlitePath).size;
    runSqliteMaintenance(db, sqlitePath, size > config.codexSqliteLogMaxBytes);
    return exported;
  } finally {
    db.close();
  }
}

function exportCodexSqliteLogsOnce(): void {
  if (!config.codexSqliteLogExportEnabled || codexSqliteLogExporter.running) return;
  codexSqliteLogExporter.running = true;
  codexSqliteLogExporter.lastRunAt = new Date().toISOString();
  codexSqliteLogExporter.lastExportedRows = 0;
  try {
    const writer = createHourlyJsonlWriter({
      baseLogFile: config.logFile,
      service: "codex-app-server",
      maxBytes: config.codexSqliteLogMaxBytes,
    });
    let exported = 0;
    for (const sqlitePath of codexLogSqliteFiles()) {
      for (let batch = 0; batch < 12; batch += 1) {
        const batchRows = exportCodexSqliteLogFile(sqlitePath, writer);
        exported += batchRows;
        if (batchRows < config.codexSqliteLogExportBatchSize) break;
      }
    }
    writer.prune();
    codexSqliteLogExporter.lastExportedRows = exported;
    codexSqliteLogExporter.totalExportedRows += exported;
    codexSqliteLogExporter.lastError = null;
    if (exported > 0) logger("info", "codex_sqlite_logs_exported", { rows: exported, totalRows: codexSqliteLogExporter.totalExportedRows });
  } catch (error) {
    codexSqliteLogExporter.lastError = error instanceof Error ? error.message : String(error);
    logger("warn", "codex_sqlite_log_export_failed", { error: errorToJson(error) });
  } finally {
    codexSqliteLogExporter.running = false;
  }
}

function startCodexSqliteLogExporter(): void {
  if (!config.codexSqliteLogExportEnabled) return;
  setTimeout(exportCodexSqliteLogsOnce, 1000);
  setInterval(exportCodexSqliteLogsOnce, config.codexSqliteLogExportIntervalMs);
}

function readCgroupMemoryValue(path: string): number | null {
  try {
    const raw = readFileSync(path, "utf8").trim();
    if (raw === "max" || raw.length === 0) return null;
    const value = Number(raw);
    return Number.isFinite(value) && value > 0 ? value : null;
  } catch {
    return null;
  }
}

function readCgroupMemoryNonNegativeValue(path: string): number | null {
  try {
    const raw = readFileSync(path, "utf8").trim();
    if (raw === "max" || raw.length === 0) return null;
    const value = Number(raw);
    return Number.isFinite(value) && value >= 0 ? value : null;
  } catch {
    return null;
  }
}

function readCgroupMemoryStatValue(name: string): number {
  try {
    const raw = readFileSync("/sys/fs/cgroup/memory.stat", "utf8");
    for (const line of raw.split(/\n/u)) {
      const [key, value] = line.trim().split(/\s+/u);
      if (key !== name) continue;
      const parsed = Number(value);
      return Number.isFinite(parsed) && parsed > 0 ? parsed : 0;
    }
  } catch {
    // memory.stat is optional outside cgroup v2; fall back to raw memory.current.
  }
  return 0;
}

function readCgroupMemoryUsage(): CgroupMemoryUsage | null {
  const current = readCgroupMemoryValue("/sys/fs/cgroup/memory.current");
  if (current === null) return null;
  const inactiveFile = readCgroupMemoryStatValue("inactive_file");
  const swapMax = readCgroupMemoryNonNegativeValue("/sys/fs/cgroup/memory.swap.max");
  return {
    currentBytes: current,
    inactiveFileBytes: inactiveFile,
    workingSetBytes: Math.max(0, current - inactiveFile),
    swapCurrentBytes: readCgroupMemoryNonNegativeValue("/sys/fs/cgroup/memory.swap.current") ?? 0,
    swapMaxBytes: swapMax,
  };
}

function memoryWatchdogThreshold(): number | null {
  if (config.memoryWatchdogThresholdBytes > 0) return config.memoryWatchdogThresholdBytes;
  const cgroupMax = readCgroupMemoryValue("/sys/fs/cgroup/memory.max");
  return cgroupMax === null ? null : Math.floor(cgroupMax * 0.98);
}

function activeRunMemoryPressure(): (CgroupMemoryUsage & { thresholdBytes: number }) | null {
  const threshold = memoryWatchdogThreshold();
  if (threshold === null || threshold <= 0) return null;
  let usage = readCgroupMemoryUsage();
  if (usage === null || usage.workingSetBytes < threshold) return null;
  try {
    (Bun as unknown as { gc?: (force?: boolean) => void }).gc?.(true);
    usage = readCgroupMemoryUsage();
  } catch (error) {
    logger("debug", "memory_watchdog_gc_failed", { error: errorToJson(error) });
  }
  return usage !== null && usage.workingSetBytes >= threshold ? { ...usage, thresholdBytes: threshold } : null;
}

function startMemoryWatchdog(): void {
  const threshold = memoryWatchdogThreshold();
  if (threshold === null || threshold <= 0) return;
  const interval = setInterval(() => {
    if (shutdownRequested || activeRuns.size === 0) return;
    const usage = activeRunMemoryPressure();
    if (usage === null) return;
    const runs = Array.from(activeRuns.values());
    logger("warn", "memory_watchdog_stopping_active_runs", {
      currentBytes: usage.currentBytes,
      inactiveFileBytes: usage.inactiveFileBytes,
      workingSetBytes: usage.workingSetBytes,
      swapCurrentBytes: usage.swapCurrentBytes,
      swapMaxBytes: usage.swapMaxBytes,
      thresholdBytes: threshold,
      activeRunCount: runs.length,
      taskIds: runs.map((run) => run.taskId),
    });
    for (const run of runs) {
      const task = findTask(run.taskId);
      if (task !== null) {
        appendOutput(task, "error", `Code Queue memory watchdog stopped the active ${run.port} run at ${Math.round(usage.workingSetBytes / 1024 / 1024)}MiB working set (${Math.round(usage.currentBytes / 1024 / 1024)}MiB cgroup current) to keep the memory-capped container alive.\n`, "memory/watchdog");
      }
      run.app.stop();
    }
  }, config.memoryWatchdogIntervalMs);
  interval.unref?.();
}

function nowIso(): string {
  return new Date().toISOString();
}

function taskOutputMaxSeq(task: QueueTask): number {
  return taskViewOutputMaxSeq(task);
}

function resolveReasoningEffort(model: string, explicit?: string | null): string | null {
  const requested = explicit?.trim();
  if (requested) return requested;
  const modelEffort = config.modelReasoningEfforts[model.toLowerCase()];
  return modelEffort ?? config.defaultReasoningEffort;
}

function normalizeQueueId(value: unknown, fallback = defaultQueueId): string {
  const text = typeof value === "string" ? value.trim() : "";
  if (text.length === 0) return fallback;
  if (!queueIdPattern.test(text)) throw new Error("queueId must match /^[A-Za-z0-9][A-Za-z0-9_.-]{0,63}$/");
  return text;
}

function safeQueueId(value: unknown): string {
  try {
    return normalizeQueueId(value);
  } catch {
    return defaultQueueId;
  }
}

function normalizeQueueName(value: unknown, queueId: string): string {
  const fallback = queueId.trim().length > 0 ? queueId : defaultQueueId;
  const text = typeof value === "string" ? value.replace(/[\u0000-\u001F\u007F]+/gu, " ").replace(/\s+/gu, " ").trim() : "";
  if (text.length === 0) return fallback;
  if (text.length > queueNameMaxLength) throw new Error(`queue name must be ${queueNameMaxLength} characters or fewer`);
  return text;
}

function safeQueueName(value: unknown, queueId: string): string {
  try {
    return normalizeQueueName(value, queueId);
  } catch {
    return queueId;
  }
}

function normalizeWorkdirPath(value: unknown, providerId: string): string {
  const raw = typeof value === "string" ? value.trim() : "";
  if (raw.length === 0) throw new Error("workdir path is required");
  if (raw.length > workdirMaxLength) throw new Error(`workdir path must be ${workdirMaxLength} characters or fewer`);
  if (raw.includes("\u0000")) throw new Error("workdir path contains an invalid character");
  return resolveTaskCwd(providerId, raw).replace(/\/+$/u, "") || "/";
}

function workdirRecordKey(providerId: string, executionMode: ReturnType<typeof normalizeCodeExecutionMode>, path: string): string {
  return `${providerId}\u0000${executionMode}\u0000${path}`;
}

function sortedWorkdirRecords(): WorkdirRecord[] {
  return Array.from(workdirRecords.values()).sort((left, right) => {
    const providerDelta = left.providerId.localeCompare(right.providerId);
    if (providerDelta !== 0) return providerDelta;
    const modeDelta = left.executionMode.localeCompare(right.executionMode);
    if (modeDelta !== 0) return modeDelta;
    return left.path.localeCompare(right.path);
  });
}

function rememberWorkdir(providerIdValue: unknown, executionModeValue: unknown, pathValue: unknown, timestamp = nowIso()): WorkdirRecord {
  const providerId = normalizeTaskProviderId(providerIdValue);
  const executionMode = normalizeCodeExecutionMode(executionModeValue);
  const path = normalizeWorkdirPath(pathValue, providerId);
  const key = workdirRecordKey(providerId, executionMode, path);
  const existing = workdirRecords.get(key);
  if (existing !== undefined) {
    existing.updatedAt = timestamp;
    return existing;
  }
  const record: WorkdirRecord = { providerId, executionMode, path, createdAt: timestamp, updatedAt: timestamp };
  workdirRecords.set(key, record);
  return record;
}

function ensureDefaultWorkdirRecords(): void {
  const timestamp = nowIso();
  for (const provider of executionProviderOptions() as Array<Record<string, JsonValue>>) {
    const providerId = normalizeProviderId(provider.id);
    if (providerId === null) continue;
    rememberWorkdir(providerId, "default", String(provider.defaultWorkdir || defaultWorkdirForProvider(providerId)), timestamp);
    if (provider.supportsWindowsNativeCodex === true && typeof provider.windowsNativeDefaultWorkdir === "string" && provider.windowsNativeDefaultWorkdir.length > 0) {
      rememberWorkdir(providerId, "windows-native", provider.windowsNativeDefaultWorkdir, timestamp);
    }
  }
}

function ensureLocalWorkdir(path: string): { ok: boolean; created: boolean; error: string | null } {
  const existed = existsSync(path);
  mkdirSync(path, { recursive: true });
  return { ok: true, created: !existed, error: null };
}

function emptyState(): PersistedState {
  const at = nowIso();
  return { version: 1, updatedAt: at, nextSeq: 1, queues: [{ id: defaultQueueId, name: defaultQueueId, createdAt: at, updatedAt: at }], tasks: [] };
}

function ensureQueue(queueId: string, queueName?: unknown): QueueRecord {
  const id = normalizeQueueId(queueId);
  const existing = state.queues.find((queue) => queue.id === id);
  if (existing !== undefined) {
    existing.name = safeQueueName(existing.name, id);
    if (queueName !== undefined) existing.name = normalizeQueueName(queueName, id);
    return existing;
  }
  const at = nowIso();
  const queue = { id, name: normalizeQueueName(queueName, id), createdAt: at, updatedAt: at };
  state.queues.push(queue);
  state.queues.sort((left, right) => left.id.localeCompare(right.id));
  markQueueDirty(id);
  return queue;
}

function queueIdOf(task: QueueTask): string {
  return safeQueueId(task.queueId);
}

function taskQueueEnteredAt(task: QueueTask): string {
  const raw = (task as QueueTask & { queueEnteredAt?: unknown }).queueEnteredAt;
  const existing = taskTimestamp(typeof raw === "string" ? raw : null);
  if (existing !== null) return existing;
  const output = Array.isArray(task.output) ? task.output : [];
  const entryEvents = output
    .filter((item) => item.method === "queue/move" || item.method === "manual-retry")
    .map((item) => taskTimestamp(item.at))
    .filter((at): at is string => at !== null)
    .sort((left, right) => (timestampMs(right) ?? 0) - (timestampMs(left) ?? 0));
  return entryEvents[0] ?? taskTimestamp(task.createdAt) ?? taskTimestamp(task.updatedAt) ?? nowIso();
}

function outputPromptHistory(task: QueueTask): PromptHistoryItem[] {
  return task.output
    .filter((item) => item.channel === "user" && item.method === "turn/steer")
    .map((item) => ({
      seq: item.seq,
      at: item.at,
      method: "turn/steer" as const,
      text: normalizeSteerPromptText(item.text),
      ...(normalizeSteerId(item.itemId) === null ? {} : { steerId: normalizeSteerId(item.itemId) as string }),
    }))
    .filter((item) => item.text.trim().length > 0);
}

function mergePromptHistory(items: PromptHistoryItem[]): PromptHistoryItem[] {
  const byKey = new Map<string, PromptHistoryItem>();
  for (const item of items) {
    const seq = Number(item.seq);
    const text = normalizeSteerPromptText(String(item.text || ""));
    if (!Number.isFinite(seq) || text.trim().length === 0) continue;
    const steerId = normalizeSteerId((item as PromptHistoryItem & { steerId?: unknown }).steerId);
    byKey.set(`${seq}:${item.method}`, { seq, at: item.at || nowIso(), method: "turn/steer", text, ...(steerId === null ? {} : { steerId }) });
  }
  return Array.from(byKey.values()).sort((left, right) => left.seq - right.seq);
}

function judgeFailCountFromOutput(task: QueueTask): number {
  return (task.output ?? []).filter((item) => item.method === "judge" && /\bjudge=fail\b/u.test(item.text)).length;
}

function fallbackJudgeRetryCount(task: QueueTask): number {
  const attemptCount = (task.attempts ?? []).filter((attempt) => attempt.judge?.source === "fallback" && attempt.judge.decision === "retry").length;
  const outputCount = (task.output ?? []).filter((item) => item.method === "judge" && /\bjudge=retry\b/u.test(item.text) && /\bsource=fallback\b/u.test(item.text)).length;
  return Math.max(attemptCount, outputCount);
}

function fallbackJudgeIsProviderTransient(judge: JudgeResult | null | undefined): boolean {
  if (judge?.source !== "fallback" || judge.decision !== "retry") return false;
  const details = judge.failureDetails;
  if (details === null || details === undefined) return false;
  return details.provider === "minimax"
    && (details.stage === "request" || details.stage === "unknown" || details.timedOut);
}

function pruneTaskHotState(task: QueueTask): void {
  if (config.maxInMemoryOutputRecords > 0 && task.output.length > config.maxInMemoryOutputRecords) {
    task.output.splice(0, task.output.length - config.maxInMemoryOutputRecords);
  }
  if (config.maxInMemoryEventRecords > 0 && task.events.length > config.maxInMemoryEventRecords) {
    task.events.splice(0, task.events.length - config.maxInMemoryEventRecords);
  }
}

function taskRetainedTraceStepCount(task: QueueTask): number {
  return task.output.reduce((count, output) => count + (outputStartsTraceStep(task, output) ? 1 : 0), 0);
}

function normalizeTaskMetric(value: unknown): number | null {
  const parsed = Number(value);
  return Number.isFinite(parsed) && parsed >= 0 ? Math.floor(parsed) : null;
}

function taskCanHaveActiveTurn(status: TaskStatus): boolean {
  return status === "running" || status === "judging";
}

function normalizeSchedulerHeartbeat(task: QueueTask): SchedulerActiveRunHeartbeat | null {
  const heartbeat = task.schedulerHeartbeat;
  if (heartbeat === null || heartbeat === undefined || typeof heartbeat !== "object") return null;
  if (heartbeat.taskId !== task.id) return null;
  if (typeof heartbeat.lastLocalHeartbeatAt !== "string" || timestampMs(heartbeat.lastLocalHeartbeatAt) === null) return null;
  return {
    taskId: task.id,
    queueId: safeQueueId(heartbeat.queueId),
    attempt: Number.isInteger(heartbeat.attempt) && heartbeat.attempt >= 0 ? heartbeat.attempt : Math.max(0, Number(task.currentAttempt || 0)),
    activeTurnId: typeof heartbeat.activeTurnId === "string" && heartbeat.activeTurnId.length > 0 ? heartbeat.activeTurnId : null,
    codexThreadId: typeof heartbeat.codexThreadId === "string" && heartbeat.codexThreadId.length > 0 ? heartbeat.codexThreadId : null,
    owner: typeof heartbeat.owner === "string" && heartbeat.owner.length > 0 ? heartbeat.owner : config.instanceId,
    schedulerInstance: typeof heartbeat.schedulerInstance === "string" && heartbeat.schedulerInstance.length > 0 ? heartbeat.schedulerInstance : config.instanceId,
    executionPlane: "scheduler-execution-plane",
    agentPort: typeof heartbeat.agentPort === "string" && heartbeat.agentPort.length > 0 ? heartbeat.agentPort : codeAgentPortForModel(task.model),
    status: taskCanHaveActiveTurn(heartbeat.status) ? heartbeat.status : task.status,
    lastLocalHeartbeatAt: heartbeat.lastLocalHeartbeatAt,
    lastObservedAgentEventAt: typeof heartbeat.lastObservedAgentEventAt === "string" && timestampMs(heartbeat.lastObservedAgentEventAt) !== null ? heartbeat.lastObservedAgentEventAt : null,
    lastPersistedTraceAt: typeof heartbeat.lastPersistedTraceAt === "string" && timestampMs(heartbeat.lastPersistedTraceAt) !== null ? heartbeat.lastPersistedTraceAt : null,
    outputMaxSeq: Number.isInteger(heartbeat.outputMaxSeq) && heartbeat.outputMaxSeq >= 0 ? heartbeat.outputMaxSeq : taskViewOutputMaxSeq(task),
    source: "scheduler",
  };
}

function normalizeTask(task: QueueTask): QueueTask {
  task.queueId = safeQueueId(task.queueId);
  task.queueEnteredAt = taskQueueEnteredAt(task);
  task.output ??= [];
  task.events ??= [];
  task.attempts ??= [];
  task.readAt = typeof task.readAt === "string" && task.readAt.length > 0 ? task.readAt : null;
  if (task.status !== "succeeded" && task.status !== "failed" && task.status !== "canceled") {
    task.finishedAt = null;
    task.readAt = null;
  }
  task.activeTurnId ??= null;
  if (!taskCanHaveActiveTurn(task.status)) task.activeTurnId = null;
  task.providerId = normalizeTaskProviderId(task.providerId);
  task.model ||= config.defaultModel;
  task.executionMode = normalizeCodeExecutionMode(task.executionMode);
  task.requestedExecutionMode = normalizeRequestedCodeExecutionMode(task.requestedExecutionMode);
  task.cwd = resolveTaskCwd(task.providerId, task.cwd);
  task.reasoningEffort = resolveReasoningEffort(task.model, task.reasoningEffort);
  task.maxAttempts = clampTaskAttempts(task.maxAttempts || config.defaultMaxAttempts);
  task.basePrompt ||= userPromptForDisplay(task.prompt);
  task.referenceTaskIds ??= referenceTaskIdsFromPrompt(task.prompt);
  task.referenceInjection ??= null;
  task.judgeFailCount = Number.isInteger(task.judgeFailCount) && task.judgeFailCount >= 0 ? task.judgeFailCount : judgeFailCountFromOutput(task);
  const persistedPromptHistory = Array.isArray(task.promptHistory) ? task.promptHistory : [];
  task.promptHistory = mergePromptHistory([...persistedPromptHistory, ...outputPromptHistory(task)]);
  const storedStepCount = normalizeTaskMetric(task.stepCount) ?? normalizeTaskMetric(task.llmStepCount);
  const stepCount = storedStepCount ?? taskRetainedTraceStepCount(task);
  task.stepCount = stepCount;
  task.llmStepCount = stepCount;
  task.outputMaxSeq = Math.max(normalizeTaskMetric(task.outputMaxSeq) ?? 0, taskViewOutputMaxSeq(task));
  task.schedulerHeartbeat = normalizeSchedulerHeartbeat(task);
  if (!taskCanHaveActiveTurn(task.status)) task.schedulerHeartbeat = null;
  pruneTaskHotState(task);
  return task;
}

function persistState(markAllDatabaseTasks = false): void {
  persistDirty = false;
  if (persistTimer !== null) {
    clearTimeout(persistTimer);
    persistTimer = null;
  }
  for (const task of state.tasks) pruneTaskHotState(task);
  state.updatedAt = nowIso();
  if (markAllDatabaseTasks) markAllDatabaseTasksDirty();
  scheduleDatabaseFlush();
}

function schedulePersistState(delayMs = 1000): void {
  persistDirty = true;
  if (persistTimer !== null) return;
  persistTimer = setTimeout(() => {
    persistTimer = null;
    if (persistDirty) persistState(false);
  }, delayMs);
}

function redactDatabaseUrl(value: string): string {
  try {
    const url = new URL(value);
    if (url.password) url.password = "***";
    return url.toString();
  } catch {
    return "<invalid-url>";
  }
}

function publishTaskOaEvent(task: QueueTask, reason: string, options: { onlyStepChange?: boolean; stepChanged?: boolean } = {}): void {
  const stepCount = taskListStepCount(task);
  const outputMaxSeq = taskOutputMaxSeq(task);
  if (options.onlyStepChange === true && options.stepChanged !== true) return;
  const queueId = queueIdOf(task);
  publishCodeQueueTaskUpdated(task, queueId, reason, stepCount, outputMaxSeq);
  publishCodeQueueTraceStatsSnapshot(task, queueId, reason, stepCount, outputMaxSeq);
}

function publishQueueEvent(reason: string, queueId = ""): void {
  publishCodeQueueQueueUpdated(queueId, reason);
}

function refreshSchedulerHeartbeat(task: QueueTask, options: { observedAgentEvent?: boolean; persistedTrace?: boolean } = {}): void {
  if (!taskCanHaveActiveTurn(task.status)) {
    task.schedulerHeartbeat = null;
    return;
  }
  const run = activeRunForTask(task);
  const now = nowIso();
  if (run !== null) {
    run.lastLocalHeartbeatAt = now;
    if (options.observedAgentEvent === true) run.lastObservedAgentEventAt = now;
    if (options.persistedTrace === true) run.lastPersistedTraceAt = now;
  }
  const previous = normalizeSchedulerHeartbeat(task);
  task.schedulerHeartbeat = buildSchedulerHeartbeat(task, run, {
    now,
    owner: config.instanceId,
    schedulerInstance: config.instanceId,
    agentPort: codeAgentPortForModel(task.model),
    lastObservedAgentEventAt: options.observedAgentEvent === true ? now : run?.lastObservedAgentEventAt ?? previous?.lastObservedAgentEventAt ?? null,
    lastPersistedTraceAt: options.persistedTrace === true ? now : run?.lastPersistedTraceAt ?? previous?.lastPersistedTraceAt ?? null,
  });
}

function isOpenCodeStepBoundaryMethod(method: string | undefined): boolean {
  return method === "opencode/step-start" || method === "opencode/step-finish";
}

function outputCanChangeStepCount(output: LiveOutput): boolean {
  if (output.channel === "user" && output.method === "enqueue") return false;
  if (output.channel === "system") return false;
  return !isOpenCodeStepBoundaryMethod(output.method);
}

function commandStartedBeforeIn(outputs: LiveOutput[], output: LiveOutput): boolean {
  if (typeof output.itemId !== "string") return false;
  return outputs.some((item) => item !== output && item.itemId === output.itemId && item.channel === "command" && item.method === "item/started");
}

function outputStartsTraceStepInHistory(outputs: LiveOutput[], output: LiveOutput): boolean {
  if (output.channel === "user" && output.method === "enqueue") return false;
  if (isOpenCodeStepBoundaryMethod(output.method)) return false;
  if (output.channel === "system") return false;
  if (codexToolLifecycleStartedBeforeIn(outputs, output)) return false;
  if (output.channel === "diff" || output.channel === "tool" || output.channel === "error" || output.channel === "assistant" || output.channel === "reasoning") return true;
  if (output.channel === "user") return true;
  if (output.channel !== "command") return true;
  const method = String(output.method || "");
  if (method === "item/started") return true;
  if (method === "item/commandExecution/outputDelta") return false;
  if (method === "item/completed") return !commandStartedBeforeIn(outputs, output);
  return true;
}

function outputStartsTraceStep(task: QueueTask, output: LiveOutput): boolean {
  return outputStartsTraceStepInHistory(task.output, output);
}

function traceStatsFromOutputs(outputs: LiveOutput[]): { stepCount: number; readCount: number; editCount: number; runCount: number; errorCount: number } {
  const visibleOutputs = outputs.filter((output) => outputStartsTraceStepInHistory(outputs, output));
  const stats = { stepCount: visibleOutputs.length, readCount: 0, editCount: 0, runCount: 0, errorCount: 0 };
  for (const output of visibleOutputs) {
    const kind = outputTraceKind(output);
    if (kind === "read") stats.readCount += 1;
    if (kind === "edit") stats.editCount += 1;
    if (kind === "run") stats.runCount += 1;
    if (kind === "error") stats.errorCount += 1;
  }
  return stats;
}

function attemptIndexFromOutput(output: LiveOutput): number | null {
  const match = String(output.text || "").match(/\battempt\s+(\d+)\s*\/\s*\d+\b/iu);
  const value = Number(match?.[1] ?? NaN);
  return Number.isInteger(value) && value > 0 ? value : null;
}

function outputAttemptIndexMap(outputs: LiveOutput[]): Map<number, number> {
  const result = new Map<number, number>();
  let currentAttempt = 0;
  for (const output of outputs.slice().sort((left, right) => Number(left.seq) - Number(right.seq))) {
    const startedAttempt = attemptIndexFromOutput(output);
    if (startedAttempt !== null) currentAttempt = startedAttempt;
    if (currentAttempt > 0 && outputStartsTraceStepInHistory(outputs, output)) result.set(output.seq, currentAttempt);
  }
  return result;
}

function traceAttemptIndexesForTask(task: QueueTask): number[] {
  const indexes = new Set<number>();
  for (const attempt of task.attempts) {
    const index = Number(attempt.index);
    if (Number.isInteger(index) && index > 0) indexes.add(index);
  }
  const currentAttempt = Number(task.currentAttempt || 0);
  const maxAttempt = Math.max(currentAttempt, ...Array.from(indexes), 0);
  for (let index = 1; index <= maxAttempt; index += 1) indexes.add(index);
  return Array.from(indexes).sort((left, right) => left - right);
}

function recordTaskOutputMetrics(task: QueueTask, output: LiveOutput, op: "set" | "append"): boolean {
  task.outputMaxSeq = Math.max(taskOutputMaxSeq(task), Number.isFinite(Number(output.seq)) ? Number(output.seq) : 0);
  if (op === "append" || !outputStartsTraceStep(task, output)) return false;
  const storedStepCount = Number(task.stepCount ?? task.llmStepCount);
  const nextStepCount = Number.isFinite(storedStepCount) && storedStepCount >= 0
    ? Math.floor(storedStepCount) + 1
    : task.output.reduce((count, item) => count + (outputStartsTraceStep(task, item) ? 1 : 0), 0);
  task.stepCount = nextStepCount;
  task.llmStepCount = nextStepCount;
  return true;
}

function outputUpdatesExistingTraceStep(output: LiveOutput): boolean {
  if (output.channel === "assistant" || output.channel === "reasoning" || output.channel === "diff") return true;
  if (isCodexToolLifecycleOutput(output) && output.method === "item/completed") return true;
  return false;
}

function traceStepOutputForProjection(task: QueueTask, output: LiveOutput): LiveOutput {
  if (!isCodexToolLifecycleOutput(output) || output.method !== "item/completed" || typeof output.itemId !== "string") return output;
  const started = taskFullOutput(task)
    .filter((item) => item !== output && isCodexToolLifecycleOutput(item) && item.itemId === output.itemId && item.method === "item/started")
    .sort((left, right) => Number(left.seq) - Number(right.seq))[0];
  return started === undefined ? output : { ...output, seq: started.seq, at: output.at, itemId: output.itemId, rawSeqs: [started.seq, output.seq] } as LiveOutput;
}

function errorToJson(error: unknown): JsonValue {
  if (error instanceof Error) return { name: error.name, message: error.message, stack: error.stack ?? null };
  return String(error);
}

function transientDatabaseRetryDelayMs(attempt: number): number {
  return Math.min(2000, 150 * Math.max(1, attempt));
}

function scheduleDatabaseFlushRetry(delayMs: number): void {
  if (serviceRoleReadOnly(config.serviceRole)) return;
  if (!databaseReady || shutdownRequested) return;
  if (dirtyDatabaseTaskIds.size === 0 && dirtyDatabaseQueueIds.size === 0) return;
  if (databaseFlushTimer !== null) return;
  const boundedDelayMs = Math.max(100, Math.min(30_000, Math.floor(delayMs)));
  scheduleDatabaseFlush(boundedDelayMs, { retry: true });
}

async function rotateSqlClientAfterTransientError(reason: string, error: unknown): Promise<void> {
  setDatabaseLastError(error);
  if (sqlRotationInFlight !== null) return sqlRotationInFlight;
  const previous = sql;
  sql = createSqlClient();
  databaseClientRotationCount += 1;
  databaseLastClientRotationAt = nowIso();
  databaseLastClientRotationReason = reason;
  sqlRotationInFlight = previous.end({ timeout: 1 })
    .catch((closeError) => logger("warn", "database_client_close_failed", { reason, error: errorToJson(closeError) }))
    .finally(() => {
      sqlRotationInFlight = null;
    });
  logger("warn", "database_client_rotated", { reason, error: errorToJson(error) });
  return sqlRotationInFlight;
}

async function withTransientDatabaseRetries<T>(operation: string, action: () => Promise<T>, attempts = 3): Promise<T> {
  let lastError: unknown = null;
  for (let attempt = 1; attempt <= attempts; attempt += 1) {
    try {
      return await action();
    } catch (error) {
      lastError = error;
      const classification = classifyTransientDatabaseError(error);
      if (!classification.transient || attempt >= attempts) throw error;
      logger("warn", "database_transient_retry", { operation, attempt, attempts, delayMs: transientDatabaseRetryDelayMs(attempt), error: errorToJson(error) });
      await rotateSqlClientAfterTransientError(operation, error);
      await Bun.sleep(transientDatabaseRetryDelayMs(attempt));
    }
  }
  throw lastError;
}

function judgeFailureDetailsForOutput(judge: JudgeResult): string {
  const details = judge.failureDetails;
  if (details === undefined || details === null) return "";
  return [
    "",
    `MiniMax failure details: stage=${details.stage}`,
    `timedOut=${details.timedOut}`,
    `durationMs=${details.durationMs}`,
    `timeoutMs=${details.timeoutMs}`,
    details.promptChars === undefined ? "" : `promptChars=${details.promptChars}`,
    details.promptLines === undefined ? "" : `promptLines=${details.promptLines}`,
    details.payloadBytes === undefined ? "" : `payloadBytes=${details.payloadBytes}`,
    details.responseStatus === undefined || details.responseStatus === null ? "" : `responseStatus=${details.responseStatus}`,
    `errorName=${details.errorName}`,
    `error=${safePreview(details.errorMessage, 220)}`,
  ].filter((part) => part.length > 0).join(" ");
}

function databaseErrorMessage(error: unknown): string {
  if (error instanceof Error) return error.message;
  return String(error);
}

function setDatabaseLastError(error: unknown): void {
  databaseLastError = databaseErrorMessage(error);
  databaseLastErrorDetail = error;
}

function clearDatabaseLastError(): void {
  databaseLastError = null;
  databaseLastErrorDetail = null;
}

function markTaskDirty(taskId: string): void {
  dirtyDatabaseTaskIds.add(taskId);
  scheduleDatabaseFlush();
}

function persistTaskState(task: QueueTask): void {
  if (taskCanHaveActiveTurn(task.status)) refreshSchedulerHeartbeat(task);
  else task.schedulerHeartbeat = null;
  markTaskDirty(task.id);
  persistState(false);
  publishTaskOaEvent(task, "persist");
}

function markQueueDirty(queueId: string): void {
  dirtyDatabaseQueueIds.add(queueId);
  scheduleDatabaseFlush();
}

function markAllDatabaseTasksDirty(): void {
  for (const task of state.tasks) dirtyDatabaseTaskIds.add(task.id);
  for (const queue of state.queues) dirtyDatabaseQueueIds.add(queue.id);
}

function runGarbageCollection(): void {
  const gc = (Bun as typeof Bun & { gc?: (force?: boolean) => void }).gc;
  if (typeof gc === "function") gc(true);
}

function scheduleDatabaseFlush(delayMs = config.databaseFlushIntervalMs, options: { retry?: boolean } = {}): void {
  if (serviceRoleReadOnly(config.serviceRole)) return;
  if (!databaseReady || (dirtyDatabaseTaskIds.size === 0 && dirtyDatabaseQueueIds.size === 0) || shutdownRequested) return;
  if (databaseFlushTimer !== null) return;
  if (options.retry === true) databaseNextFlushRetryAt = new Date(Date.now() + Math.max(0, delayMs)).toISOString();
  databaseFlushTimer = setTimeout(() => {
    databaseFlushTimer = null;
    void flushDirtyTasksToDatabase().catch((error) => {
      setDatabaseLastError(error);
      logger("error", "database_flush_failed", { error: errorToJson(error), dirtyTaskCount: dirtyDatabaseTaskIds.size });
    });
  }, delayMs);
}

function taskTimestamp(value: string | null): string | null {
  if (value === null || value.length === 0) return null;
  const time = Date.parse(value);
  return Number.isFinite(time) ? new Date(time).toISOString() : null;
}

function lastOutputSeq(task: QueueTask): number {
  return taskOutputMaxSeq(task);
}

function updateNextSeqFromTasks(): void {
  let nextSeq = Math.max(1, Number.isFinite(state.nextSeq) ? Math.floor(state.nextSeq) : 1);
  for (const task of state.tasks) nextSeq = Math.max(nextSeq, lastOutputSeq(task) + 1);
  state.nextSeq = nextSeq;
}

interface UpsertTaskOptions {
  claimQueueId?: string | null;
}

async function upsertTaskToDatabase(client: SqlExecutor, task: QueueTask, options: UpsertTaskOptions = {}): Promise<boolean> {
  const claimQueueId = options.claimQueueId ?? null;
  const rows = await client<Array<{ read_at: Date | string | null }>>`
    INSERT INTO unidesk_code_queue_tasks (
      id,
      queue_id,
      status,
      provider_id,
      execution_mode,
      model,
      cwd,
      prompt,
      base_prompt,
      reference_task_ids,
      reference_injection,
      reasoning_effort,
      max_attempts,
      current_attempt,
      current_mode,
      codex_thread_id,
      active_turn_id,
      created_at,
      updated_at,
      started_at,
      finished_at,
      read_at,
      last_error,
      last_judge,
      output_count,
      event_count,
      attempt_count,
      last_output_seq,
      task_json
    ) VALUES (
      ${task.id},
      ${queueIdOf(task)},
      ${task.status},
      ${task.providerId},
      ${task.executionMode},
      ${task.model},
      ${task.cwd},
      ${task.prompt},
      ${task.basePrompt},
      ${client.json(task.referenceTaskIds as unknown as postgres.JSONValue)},
      ${task.referenceInjection === null ? null : client.json(task.referenceInjection as unknown as postgres.JSONValue)},
      ${task.reasoningEffort},
      ${task.maxAttempts},
      ${task.currentAttempt},
      ${task.currentMode},
      ${task.codexThreadId},
      ${task.activeTurnId},
      ${taskTimestamp(task.createdAt) ?? nowIso()},
      ${taskTimestamp(task.updatedAt) ?? nowIso()},
      ${taskTimestamp(task.startedAt)},
      ${taskTimestamp(task.finishedAt)},
      ${taskTimestamp(task.readAt)},
      ${task.lastError},
      ${task.lastJudge === null ? null : client.json(task.lastJudge as unknown as postgres.JSONValue)},
      ${task.output.length},
      ${task.events.length},
      ${task.attempts.length},
      ${lastOutputSeq(task)},
      ${client.json(task as unknown as postgres.JSONValue)}
    )
    ON CONFLICT (id) DO UPDATE SET
      status = EXCLUDED.status,
      queue_id = EXCLUDED.queue_id,
      provider_id = EXCLUDED.provider_id,
      execution_mode = EXCLUDED.execution_mode,
      model = EXCLUDED.model,
      cwd = EXCLUDED.cwd,
      prompt = EXCLUDED.prompt,
      base_prompt = EXCLUDED.base_prompt,
      reference_task_ids = EXCLUDED.reference_task_ids,
      reference_injection = EXCLUDED.reference_injection,
      reasoning_effort = EXCLUDED.reasoning_effort,
      max_attempts = EXCLUDED.max_attempts,
      current_attempt = EXCLUDED.current_attempt,
      current_mode = EXCLUDED.current_mode,
      codex_thread_id = EXCLUDED.codex_thread_id,
      active_turn_id = EXCLUDED.active_turn_id,
      created_at = EXCLUDED.created_at,
      updated_at = EXCLUDED.updated_at,
      started_at = EXCLUDED.started_at,
      finished_at = EXCLUDED.finished_at,
      read_at = CASE
        WHEN EXCLUDED.status IN ('queued', 'running', 'judging', 'retry_wait') THEN NULL
        WHEN unidesk_code_queue_tasks.read_at IS NOT NULL AND EXCLUDED.read_at IS NOT NULL THEN GREATEST(unidesk_code_queue_tasks.read_at, EXCLUDED.read_at)
        WHEN unidesk_code_queue_tasks.read_at IS NOT NULL THEN unidesk_code_queue_tasks.read_at
        ELSE EXCLUDED.read_at
      END,
      last_error = EXCLUDED.last_error,
      last_judge = EXCLUDED.last_judge,
      output_count = EXCLUDED.output_count,
      event_count = EXCLUDED.event_count,
      attempt_count = EXCLUDED.attempt_count,
      last_output_seq = EXCLUDED.last_output_seq,
      task_json = jsonb_set(
        EXCLUDED.task_json,
        '{readAt}',
        CASE
          WHEN EXCLUDED.status IN ('queued', 'running', 'judging', 'retry_wait') THEN 'null'::jsonb
          WHEN unidesk_code_queue_tasks.read_at IS NOT NULL AND EXCLUDED.read_at IS NOT NULL THEN to_jsonb(GREATEST(unidesk_code_queue_tasks.read_at, EXCLUDED.read_at))
          WHEN unidesk_code_queue_tasks.read_at IS NOT NULL THEN to_jsonb(unidesk_code_queue_tasks.read_at)
          ELSE COALESCE(to_jsonb(EXCLUDED.read_at), 'null'::jsonb)
        END,
        true
      )
    WHERE (
        ${claimQueueId === null}
        OR (
          unidesk_code_queue_tasks.queue_id = ${claimQueueId}
          AND (
            (
              unidesk_code_queue_tasks.status = 'queued'
              AND unidesk_code_queue_tasks.started_at IS NULL
              AND unidesk_code_queue_tasks.current_attempt = 0
              AND unidesk_code_queue_tasks.codex_thread_id IS NULL
              AND unidesk_code_queue_tasks.active_turn_id IS NULL
            )
            OR (
              unidesk_code_queue_tasks.status = 'retry_wait'
              AND unidesk_code_queue_tasks.active_turn_id IS NULL
            )
          )
        )
      )
      AND NOT (
        EXCLUDED.status IN ('queued', 'retry_wait')
        AND EXCLUDED.started_at IS NULL
        AND EXCLUDED.current_attempt = 0
        AND EXCLUDED.codex_thread_id IS NULL
        AND EXCLUDED.active_turn_id IS NULL
        AND (
          unidesk_code_queue_tasks.status IN ('running', 'judging')
          OR unidesk_code_queue_tasks.started_at IS NOT NULL
          OR unidesk_code_queue_tasks.current_attempt > 0
          OR unidesk_code_queue_tasks.codex_thread_id IS NOT NULL
          OR unidesk_code_queue_tasks.active_turn_id IS NOT NULL
        )
      )
    RETURNING read_at
  `;
  if (rows.length === 0) {
    const current = await client<DatabaseTaskStatusRow[]>`
      SELECT id, queue_id, status, started_at, current_attempt, codex_thread_id, active_turn_id
      FROM unidesk_code_queue_tasks
      WHERE id = ${task.id}
      LIMIT 1
    `;
    logger("warn", "database_task_stale_unclaimed_write_rejected", {
      taskId: task.id,
      attemptedQueueId: queueIdOf(task),
      attemptedStatus: task.status,
      attemptedStartedAt: task.startedAt,
      attemptedCurrentAttempt: task.currentAttempt,
      attemptedCodexThreadId: task.codexThreadId,
      attemptedActiveTurnId: task.activeTurnId,
      current: databaseStatusRowJson(current[0] ?? null),
    });
    return false;
  }
  task.readAt = timestampToIso(rows[0]?.read_at ?? null);
  return true;
}

async function upsertQueueToDatabase(client: SqlExecutor, queue: QueueRecord): Promise<void> {
  await client`
    INSERT INTO unidesk_code_queue_queues (
      id,
      name,
      created_at,
      updated_at
    ) VALUES (
      ${queue.id},
      ${safeQueueName(queue.name, queue.id)},
      ${taskTimestamp(queue.createdAt) ?? nowIso()},
      ${taskTimestamp(queue.updatedAt) ?? nowIso()}
    )
    ON CONFLICT (id) DO UPDATE SET
      name = EXCLUDED.name,
      updated_at = EXCLUDED.updated_at
  `;
}

async function upsertWorkdirsToDatabase(records: WorkdirRecord[]): Promise<void> {
  if (records.length === 0) return;
  for (const record of records) {
    await sql`
      INSERT INTO unidesk_code_queue_workdirs (
        provider_id,
        execution_mode,
        path,
        created_at,
        updated_at
      ) VALUES (
        ${record.providerId},
        ${record.executionMode},
        ${record.path},
        ${taskTimestamp(record.createdAt) ?? nowIso()},
        ${taskTimestamp(record.updatedAt) ?? nowIso()}
      )
      ON CONFLICT (provider_id, execution_mode, path) DO UPDATE SET
        updated_at = EXCLUDED.updated_at
    `;
  }
}

interface DatabaseTaskRow {
  id: string;
  queue_id: string;
  updated_at: Date | string;
  status: TaskStatus;
  read_at: Date | string | null;
  current_attempt: number | string | null;
  current_mode: RunMode | null;
  codex_thread_id: string | null;
  active_turn_id: string | null;
  started_at: Date | string | null;
  finished_at: Date | string | null;
  task_json: unknown;
}

interface DatabaseTaskStatusRow {
  id: string;
  queue_id: string;
  status: TaskStatus;
  started_at: Date | string | null;
  current_attempt: number | string | null;
  codex_thread_id: string | null;
  active_turn_id: string | null;
}

interface DatabaseQueueRow {
  id: string;
  name: string;
  created_at: Date | string;
  updated_at: Date | string;
}

interface DatabaseTaskIdRow {
  id: string;
}

function normalizeDatabaseTaskRows(rows: DatabaseTaskRow[], source: string): QueueTask[] {
  const tasks: QueueTask[] = [];
  for (const row of rows) {
    try {
      const taskJson = row.task_json;
      if (typeof taskJson !== "object" || taskJson === null || Array.isArray(taskJson)) throw new Error("task_json is not an object");
      tasks.push(normalizeTask({
        ...(taskJson as Record<string, unknown>),
        id: row.id,
        queueId: safeQueueId(row.queue_id),
        status: row.status,
        currentAttempt: Number(row.current_attempt ?? 0),
        currentMode: row.current_mode,
        codexThreadId: row.codex_thread_id,
        activeTurnId: row.active_turn_id,
        updatedAt: timestampToIso(row.updated_at) ?? nowIso(),
        startedAt: timestampToIso(row.started_at),
        finishedAt: timestampToIso(row.finished_at),
        readAt: timestampToIso(row.read_at),
      } as unknown as QueueTask));
    } catch (error) {
      logger("warn", "database_task_row_ignored", { source, id: String(row.id), error: errorToJson(error) });
    }
  }
  return tasks.sort((left, right) => (timestampMs(left.createdAt) ?? 0) - (timestampMs(right.createdAt) ?? 0) || left.id.localeCompare(right.id));
}

function databaseStatusRowJson(row: DatabaseTaskStatusRow | null): JsonValue {
  if (row === null) return null;
  return {
    id: row.id,
    queueId: safeQueueId(row.queue_id),
    status: row.status,
    startedAt: timestampToIso(row.started_at),
    currentAttempt: Number(row.current_attempt ?? 0),
    codexThreadId: row.codex_thread_id,
    activeTurnId: row.active_turn_id,
  };
}

function taskIsUnclaimedMovable(task: QueueTask): boolean {
  return (task.status === "queued" || task.status === "retry_wait")
    && task.startedAt === null
    && task.currentAttempt === 0
    && task.codexThreadId === null
    && task.activeTurnId === null;
}

function databaseTaskMoveBlocker(row: DatabaseTaskStatusRow | null): string {
  if (row === null) return "task not found";
  if (row.status !== "queued" && row.status !== "retry_wait") return `status=${row.status}`;
  if (row.started_at !== null) return "task already has started_at";
  if (Number(row.current_attempt ?? 0) !== 0) return `task already has current_attempt=${Number(row.current_attempt ?? 0)}`;
  if (row.codex_thread_id !== null) return "task already has codex_thread_id";
  if (row.active_turn_id !== null) return "task already has active_turn_id";
  return "";
}

function taskMoveBlocker(task: QueueTask): string {
  if (activeRunForTask(task) !== null) return "task has an active agent run";
  if (processingQueues.has(queueIdOf(task))) return "queue processor is currently active";
  if (activeRunSlotReservations.has(queueIdOf(task))) return "queue is reserving an active run slot";
  if (activeRunSlotWaiters.some((waiter) => waiter.taskId === task.id || waiter.queueId === queueIdOf(task))) return "queue is waiting for an active run slot";
  if (task.status !== "queued" && task.status !== "retry_wait") return `status=${task.status}`;
  if (!taskIsUnclaimedMovable(task)) return "task has already been claimed";
  return "";
}

function reconcileHotTaskFromDatabase(task: QueueTask): QueueTask {
  const existing = findTask(task.id);
  if (existing === null) return rememberHotTask(task);
  if (activeRunForTask(existing) !== null) return existing;
  Object.assign(existing, task);
  return existing;
}

function taskHasClaimMarkers(task: QueueTask): boolean {
  return task.status === "running"
    || task.status === "judging"
    || task.startedAt !== null
    || task.currentAttempt > 0
    || task.codexThreadId !== null
    || task.activeTurnId !== null;
}

function shouldPreferHotTaskOverDatabase(hotTask: QueueTask, databaseTask: QueueTask): boolean {
  if (activeRunForTask(hotTask) !== null) return true;
  if (taskIsUnclaimedMovable(hotTask) && taskHasClaimMarkers(databaseTask)) return false;
  const hotUpdatedAt = timestampMs(hotTask.updatedAt) ?? 0;
  const databaseUpdatedAt = timestampMs(databaseTask.updatedAt) ?? 0;
  return hotUpdatedAt >= databaseUpdatedAt;
}

async function deleteTaskFromDatabase(taskId: string): Promise<void> {
  if (!databaseReady) return;
  await sql`
    DELETE FROM unidesk_code_queue_tasks
    WHERE id = ${taskId}
  `;
}

async function claimTaskInDatabase(task: QueueTask, expectedQueueId: string): Promise<boolean> {
  if (!databaseReady) return true;
  const timeout = `${config.databaseClaimTimeoutMs}ms`;
  const claimed = await withTransientDatabaseRetries("claim-task", async () => sql.begin(async (client) => {
    await client`SELECT set_config('statement_timeout', ${timeout}, true), set_config('lock_timeout', ${timeout}, true)`;
    return await upsertTaskToDatabase(client, task, { claimQueueId: expectedQueueId });
  }));
  if (claimed) return true;
  const databaseTask = await loadTaskFromDatabase(task.id);
  if (databaseTask !== null) reconcileHotTaskFromDatabase(databaseTask);
  logger("warn", "task_claim_conflict", {
    taskId: task.id,
    expectedQueueId,
    attemptedQueueId: queueIdOf(task),
    attemptedStatus: task.status,
    attemptedCurrentAttempt: task.currentAttempt,
  });
  return false;
}

async function runDatabaseClaimMoveSelfTest(): Promise<JsonValue | null> {
  if (!databaseReady) return null;
  const suffix = String(Date.now());
  const taskId = `codex_claim_move_db_${suffix}`;
  const queuedAt = nowIso();
  const sourceQueueId = `claim_move_db_source_${suffix}`;
  const targetQueueId = `claim_move_db_target_${suffix}`;
  const before = state.tasks.slice();
  const beforeQueues = state.queues.slice();
  await deleteTaskFromDatabase(taskId);
  try {
    const queuedTask = normalizeTask({
      ...createTask({ prompt: "claim/move DB race self-test", queueId: sourceQueueId }),
      id: taskId,
      queueId: sourceQueueId,
      queueEnteredAt: queuedAt,
      createdAt: queuedAt,
      updatedAt: queuedAt,
      output: [],
    });
    await sql.begin(async (client) => {
      await upsertQueueToDatabase(client, { id: sourceQueueId, name: sourceQueueId, createdAt: queuedAt, updatedAt: queuedAt });
      await upsertTaskToDatabase(client, queuedTask);
    });
    const staleHotTask = normalizeTask(JSON.parse(JSON.stringify(queuedTask)) as QueueTask);
    const claimedTask = normalizeTask(JSON.parse(JSON.stringify(queuedTask)) as QueueTask);
    const claimedAt = nowIso();
    claimedTask.status = "running";
    claimedTask.startedAt = claimedAt;
    claimedTask.currentAttempt = 1;
    claimedTask.currentMode = "initial";
    claimedTask.updatedAt = claimedAt;
    const claimed = await claimTaskInDatabase(claimedTask, sourceQueueId);
    if (!claimed) throw new Error("database claim self-test failed to claim queued task");
    state.tasks.splice(0, state.tasks.length, staleHotTask);
    const response = await moveTaskToQueue(staleHotTask, new Request(`http://code-queue.local/api/tasks/${taskId}/move`, {
      method: "POST",
      body: JSON.stringify({ queueId: targetQueueId }),
      headers: { "content-type": "application/json" },
    }), { bypassRoleCheck: true });
    const after = await loadTaskFromDatabase(taskId);
    const body = await response.json() as Record<string, JsonValue>;
    if (response.status !== 409) throw new Error(`database stale move should return 409, got ${response.status}`);
    if (after === null) throw new Error("database self-test task disappeared after stale move");
    if (after.status !== "running") throw new Error(`database self-test task status changed to ${after.status}`);
    if (queueIdOf(after) !== sourceQueueId) throw new Error(`database self-test task queue changed to ${queueIdOf(after)}`);
    if (after.currentAttempt !== 1 || after.startedAt === null) throw new Error("database self-test task claim markers were lost");
    return {
      ok: true,
      taskId,
      moveStatus: response.status,
      databaseStatus: after.status,
      databaseQueueId: queueIdOf(after),
      currentAttempt: after.currentAttempt,
      startedAt: after.startedAt,
      response: body as JsonValue,
    } as unknown as JsonValue;
  } finally {
    await deleteTaskFromDatabase(taskId);
    await deleteDatabaseQueues([sourceQueueId, targetQueueId]);
    state.tasks.splice(0, state.tasks.length, ...before);
    state.queues.splice(0, state.queues.length, ...beforeQueues);
  }
}

async function loadPrunedDatabaseTaskRows(where: "all" | "hot"): Promise<DatabaseTaskRow[]> {
  if (where === "hot") {
    return await sql<DatabaseTaskRow[]>`
      SELECT
        id,
        queue_id,
        updated_at,
        status,
        read_at,
        current_attempt,
        current_mode,
        codex_thread_id,
        active_turn_id,
        started_at,
        finished_at,
        task_json - 'output' - 'events' AS task_json
      FROM unidesk_code_queue_tasks
      WHERE status IN ('queued', 'running', 'judging', 'retry_wait')
      ORDER BY created_at ASC, id ASC
    `;
  }
  return await sql<DatabaseTaskRow[]>`
    SELECT id, queue_id, updated_at, status, read_at, current_attempt, current_mode, codex_thread_id, active_turn_id, started_at, finished_at, task_json
    FROM (
      SELECT
        id,
        queue_id,
        updated_at,
        status,
        read_at,
        current_attempt,
        current_mode,
        codex_thread_id,
        active_turn_id,
        started_at,
        finished_at,
        jsonb_set(
          jsonb_set(
            task_json,
            '{output}',
            CASE
              WHEN ${config.maxInMemoryOutputRecords} > 0 THEN COALESCE((
                SELECT jsonb_agg(value ORDER BY ord)
                FROM (
                  SELECT value, ord
                  FROM jsonb_array_elements(
                    CASE
                      WHEN jsonb_typeof(task_json->'output') = 'array' THEN task_json->'output'
                      ELSE '[]'::jsonb
                    END
                  ) WITH ORDINALITY AS output_items(value, ord)
                  ORDER BY ord DESC
                  LIMIT ${config.maxInMemoryOutputRecords}
                ) AS kept_output
              ), '[]'::jsonb)
              ELSE CASE
                WHEN jsonb_typeof(task_json->'output') = 'array' THEN task_json->'output'
                ELSE '[]'::jsonb
              END
            END,
            true
          ),
          '{events}',
          CASE
            WHEN ${config.maxInMemoryEventRecords} > 0 THEN COALESCE((
              SELECT jsonb_agg(value ORDER BY ord)
              FROM (
                SELECT value, ord
                FROM jsonb_array_elements(
                  CASE
                    WHEN jsonb_typeof(task_json->'events') = 'array' THEN task_json->'events'
                    ELSE '[]'::jsonb
                  END
                ) WITH ORDINALITY AS event_items(value, ord)
                ORDER BY ord DESC
                LIMIT ${config.maxInMemoryEventRecords}
              ) AS kept_events
            ), '[]'::jsonb)
            ELSE CASE
              WHEN jsonb_typeof(task_json->'events') = 'array' THEN task_json->'events'
              ELSE '[]'::jsonb
            END
          END,
          true
        ) AS task_json,
        created_at
      FROM unidesk_code_queue_tasks
    ) AS pruned_tasks
    WHERE ${where === "all"} OR status IN ('queued', 'running', 'judging', 'retry_wait')
    ORDER BY created_at ASC, id ASC
  `;
}

async function loadTasksFromDatabase(where: "all" | "hot" = "all"): Promise<QueueTask[]> {
  return normalizeDatabaseTaskRows(await loadPrunedDatabaseTaskRows(where), where);
}

function databaseQueueRowsToRecords(rows: DatabaseQueueRow[]): QueueRecord[] {
  const queueMap = new Map<string, QueueRecord>();
  for (const row of rows) {
    const id = safeQueueId(row.id);
    queueMap.set(id, {
      id,
      name: safeQueueName(row.name, id),
      createdAt: taskTimestamp(String(row.created_at)) ?? nowIso(),
      updatedAt: taskTimestamp(String(row.updated_at)) ?? nowIso(),
    });
  }
  if (!queueMap.has(defaultQueueId)) queueMap.set(defaultQueueId, { id: defaultQueueId, name: defaultQueueId, createdAt: state.updatedAt, updatedAt: state.updatedAt });
  return Array.from(queueMap.values()).sort((left, right) => left.id.localeCompare(right.id));
}

async function loadQueuesFromDatabase(): Promise<QueueRecord[]> {
  if (!databaseReady) return state.queues;
  const queueRows = await sql<DatabaseQueueRow[]>`
    SELECT id, name, created_at, updated_at
    FROM unidesk_code_queue_queues
    ORDER BY id ASC
  `;
  return databaseQueueRowsToRecords(queueRows);
}

async function loadTasksFromDatabaseByIds(taskIds: string[]): Promise<QueueTask[]> {
  const ids = Array.from(new Set(taskIds.map((id) => id.trim()).filter(Boolean)));
  if (ids.length === 0) return [];
  const rows = await sql<DatabaseTaskRow[]>`
    SELECT id, queue_id, updated_at, status, read_at, current_attempt, current_mode, codex_thread_id, active_turn_id, started_at, finished_at, task_json
    FROM (
      SELECT
        id,
        queue_id,
        updated_at,
        status,
        read_at,
        current_attempt,
        current_mode,
        codex_thread_id,
        active_turn_id,
        started_at,
        finished_at,
        task_json - 'output' - 'events' - 'attempts' - 'promptHistory' AS task_json
      FROM unidesk_code_queue_tasks
      WHERE id IN ${sql(ids)}
    ) AS lite_tasks
  `;
  return normalizeDatabaseTaskRows(rows, "by_ids");
}

async function loadTaskFromDatabase(taskId: string): Promise<QueueTask | null> {
  const rows = await sql<DatabaseTaskRow[]>`
    SELECT id, queue_id, updated_at, status, read_at, current_attempt, current_mode, codex_thread_id, active_turn_id, started_at, finished_at, task_json
    FROM (
      SELECT
        id,
        queue_id,
        updated_at,
        status,
        read_at,
        current_attempt,
        current_mode,
        codex_thread_id,
        active_turn_id,
        started_at,
        finished_at,
        jsonb_set(
          jsonb_set(
            task_json,
            '{output}',
            CASE
              WHEN ${config.maxInMemoryOutputRecords} > 0 THEN COALESCE((
                SELECT jsonb_agg(value ORDER BY ord)
                FROM (
                  SELECT value, ord
                  FROM jsonb_array_elements(
                    CASE
                      WHEN jsonb_typeof(task_json->'output') = 'array' THEN task_json->'output'
                      ELSE '[]'::jsonb
                    END
                  ) WITH ORDINALITY AS output_items(value, ord)
                  ORDER BY ord DESC
                  LIMIT ${config.maxInMemoryOutputRecords}
                ) AS kept_output
              ), '[]'::jsonb)
              ELSE CASE
                WHEN jsonb_typeof(task_json->'output') = 'array' THEN task_json->'output'
                ELSE '[]'::jsonb
              END
            END,
            true
          ),
          '{events}',
          CASE
            WHEN ${config.maxInMemoryEventRecords} > 0 THEN COALESCE((
              SELECT jsonb_agg(value ORDER BY ord)
              FROM (
                SELECT value, ord
                FROM jsonb_array_elements(
                  CASE
                    WHEN jsonb_typeof(task_json->'events') = 'array' THEN task_json->'events'
                    ELSE '[]'::jsonb
                  END
                ) WITH ORDINALITY AS event_items(value, ord)
                ORDER BY ord DESC
                LIMIT ${config.maxInMemoryEventRecords}
              ) AS kept_events
            ), '[]'::jsonb)
            ELSE CASE
              WHEN jsonb_typeof(task_json->'events') = 'array' THEN task_json->'events'
              ELSE '[]'::jsonb
            END
          END,
          true
        ) AS task_json
      FROM unidesk_code_queue_tasks
      WHERE id = ${taskId}
    ) AS pruned_tasks
    LIMIT 1
  `;
  return normalizeDatabaseTaskRows(rows, "single")[0] ?? null;
}

async function warmDatabaseOverviewQueries(): Promise<void> {
  if (!databaseReady) return;
  const started = performance.now();
  try {
    const [recentRows, unreadRows, activeRows] = await Promise.all([
      sql<DatabaseTaskIdRow[]>`
        SELECT id
        FROM unidesk_code_queue_tasks
        ORDER BY created_at DESC, id DESC
        LIMIT 48
      `,
      sql<DatabaseTaskIdRow[]>`
        SELECT id
        FROM unidesk_code_queue_tasks
        WHERE read_at IS NULL
          AND status IN ('succeeded', 'failed', 'canceled')
        ORDER BY updated_at DESC, id DESC
        LIMIT 100
      `,
      sql<DatabaseTaskIdRow[]>`
        SELECT id
        FROM unidesk_code_queue_tasks
        WHERE status IN ('running', 'judging', 'retry_wait')
        ORDER BY updated_at DESC, id DESC
        LIMIT 48
      `,
    ]);
    const ids = Array.from(new Set([...unreadRows, ...activeRows, ...recentRows].map((row) => row.id)));
    await Promise.all([
      queueSummaryForHealth(false),
      loadTasksFromDatabaseByIds(ids),
    ]);
    logger("info", "database_overview_warm_complete", {
      taskIdCount: ids.length,
      durationMs: Math.round(performance.now() - started),
    });
  } catch (error) {
    logger("warn", "database_overview_warm_failed", { error: errorToJson(error) });
  }
}

async function warmCodeQueueFirstPaintOverview(): Promise<void> {
  if (!databaseReady) return;
  const started = performance.now();
  try {
    const response = await tasksOverviewResponse(new URL(firstPaintOverviewWarmUrl));
    await response.arrayBuffer();
    logger("info", "database_first_paint_overview_warm_complete", {
      status: response.status,
      durationMs: Math.round(performance.now() - started),
    });
  } catch (error) {
    logger("warn", "database_first_paint_overview_warm_failed", { error: errorToJson(error) });
  }
}

async function warmCodeQueueFirstPaintOverviewDeduped(): Promise<void> {
  if (firstPaintOverviewWarmInFlight !== null) return firstPaintOverviewWarmInFlight;
  firstPaintOverviewWarmInFlight = warmCodeQueueFirstPaintOverview()
    .finally(() => {
      firstPaintOverviewWarmInFlight = null;
    });
  return firstPaintOverviewWarmInFlight;
}

async function ensureDatabaseIndexes(): Promise<void> {
  logger("info", "database_index_maintenance_start", {});
  const started = performance.now();
  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_status_updated ON unidesk_code_queue_tasks(status, updated_at DESC)`;
  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_queue_status_updated ON unidesk_code_queue_tasks(queue_id, status, updated_at DESC)`;
  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_provider_updated ON unidesk_code_queue_tasks(provider_id, updated_at DESC)`;
  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_execution_mode_updated ON unidesk_code_queue_tasks(execution_mode, updated_at DESC)`;
  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_created ON unidesk_code_queue_tasks(created_at DESC)`;
  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_queue_created ON unidesk_code_queue_tasks(queue_id, created_at DESC, id DESC)`;
  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_unread_terminal ON unidesk_code_queue_tasks(queue_id, updated_at DESC) WHERE read_at IS NULL AND status IN ('succeeded', 'failed', 'canceled')`;
  await sql`CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_unidesk_code_queue_tasks_model_updated ON unidesk_code_queue_tasks(model, updated_at DESC)`;
  logger("info", "database_index_maintenance_complete", { durationMs: Math.round(performance.now() - started) });
}

function scheduleStartupDatabaseMaintenance(): void {
  setTimeout(() => {
    void warmCodeQueueFirstPaintOverviewDeduped();
  }, 100).unref?.();
  setTimeout(() => {
    void (async () => {
      const started = performance.now();
      logger("info", "database_startup_maintenance_start", {
        queueCount: state.queues.length,
      });
      await warmCodeQueueFirstPaintOverviewDeduped();
      for (const queue of state.queues) dirtyDatabaseQueueIds.add(queue.id);
      await flushDirtyTasksToDatabase(true);
      await loadClaudeQqNotificationOutboxFromDatabase();
      await ensureDatabaseIndexes();
      runGarbageCollection();
      await warmDatabaseOverviewQueries();
      await warmCodeQueueFirstPaintOverviewDeduped();
      logger("info", "database_startup_maintenance_complete", {
        databaseNotificationCount: claudeQqNotificationOutboxItemCount(),
        durationMs: Math.round(performance.now() - started),
      });
    })().catch((error) => {
      setDatabaseLastError(error);
      logger("warn", "database_startup_maintenance_failed", { error: errorToJson(error) });
    });
  }, 1000).unref?.();
}

function rememberHotTask(task: QueueTask): QueueTask {
  const existing = findTask(task.id);
  if (existing !== null) return existing;
  state.tasks.push(task);
  state.tasks.sort((left, right) => (timestampMs(left.createdAt) ?? 0) - (timestampMs(right.createdAt) ?? 0) || left.id.localeCompare(right.id));
  updateNextSeqFromTasks();
  return task;
}

async function findTaskForRead(taskId: string): Promise<QueueTask | null> {
  const hotTask = findTask(taskId);
  if (!databaseReady) return hotTask;
  try {
    const databaseTask = await loadTaskFromDatabase(taskId);
    if (hotTask === null) return databaseTask;
    if (databaseTask === null) return hotTask;
    return shouldPreferHotTaskOverDatabase(hotTask, databaseTask) ? hotTask : databaseTask;
  } catch (error) {
    setDatabaseLastError(error);
    logger("warn", "read_database_fallback", { taskId, error: errorToJson(error) });
    return hotTask;
  }
}

async function findTaskForMutation(taskId: string): Promise<QueueTask | null> {
  const hotTask = findTask(taskId);
  if (!databaseReady) return hotTask;
  try {
    const databaseTask = await loadTaskFromDatabase(taskId);
    if (databaseTask === null) return hotTask;
    if (hotTask === null) return rememberHotTask(databaseTask);
    return shouldPreferHotTaskOverDatabase(hotTask, databaseTask) ? hotTask : reconcileHotTaskFromDatabase(databaseTask);
  } catch (error) {
    setDatabaseLastError(error);
    logger("warn", "mutation_database_fallback", { taskId, error: errorToJson(error) });
    return hotTask;
  }
}

async function loadNextSeqFromDatabase(): Promise<number> {
  const rows = await sql<Array<{ next_seq: string | number | null }>>`
    SELECT COALESCE(MAX(last_output_seq), 0) + 1 AS next_seq
    FROM unidesk_code_queue_tasks
  `;
  const value = Number(rows[0]?.next_seq ?? 1);
  return Number.isFinite(value) && value > 0 ? Math.floor(value) : 1;
}

async function flushDirtyTasksToDatabase(force = false): Promise<void> {
  if (serviceRoleReadOnly(config.serviceRole)) {
    dirtyDatabaseTaskIds.clear();
    dirtyDatabaseQueueIds.clear();
    return;
  }
  if (!databaseReady) return;
  if (databaseFlushInFlight && !force) {
    scheduleDatabaseFlush();
    return;
  }
  const batch = takeDirtyFlushBatch(dirtyDatabaseTaskIds, dirtyDatabaseQueueIds, nowIso());
  const ids = batch.taskIds;
  const queueIds = batch.queueIds;
  if (ids.length === 0 && queueIds.length === 0) return;
  databaseFlushInFlight = true;
  let rejectedTaskIds: string[] = [];
  try {
    rejectedTaskIds = await withTransientDatabaseRetries("flush-dirty-tasks", async () => sql.begin(async (client) => {
      const currentRejectedTaskIds: string[] = [];
      for (const id of queueIds) {
        const queue = state.queues.find((item) => item.id === id);
        if (queue !== undefined) await upsertQueueToDatabase(client, queue);
      }
      for (const id of ids) {
        const task = state.tasks.find((item) => item.id === id);
        if (task !== undefined && !await upsertTaskToDatabase(client, task)) currentRejectedTaskIds.push(id);
      }
      return currentRejectedTaskIds;
    }));
    clearDatabaseLastError();
    databaseFlushFailureCount = 0;
    databaseLastFlushFailureAt = null;
    databaseNextFlushRetryAt = null;
  } catch (error) {
    restoreDirtyFlushBatch(batch, dirtyDatabaseTaskIds, dirtyDatabaseQueueIds);
    setDatabaseLastError(error);
    databaseFlushFailureCount += 1;
    databaseLastFlushFailureAt = nowIso();
    const classification = classifyTransientDatabaseError(error);
    if (classification.transient) {
      await rotateSqlClientAfterTransientError("flush-dirty-tasks-final", error);
      const retryDelayMs = Math.min(30_000, Math.max(config.databaseFlushIntervalMs, transientDatabaseRetryDelayMs(databaseFlushFailureCount + 1) * Math.max(1, databaseFlushFailureCount)));
      logger("error", "database_flush_degraded_retry_scheduled", {
        operation: "flush-dirty-tasks",
        force,
        failureCount: databaseFlushFailureCount,
        retryDelayMs,
        dirtyTaskCount: dirtyDatabaseTaskIds.size,
        dirtyQueueCount: dirtyDatabaseQueueIds.size,
        classification: classification as unknown as JsonValue,
        error: errorToJson(error),
      });
      scheduleDatabaseFlushRetry(retryDelayMs);
      return;
    }
    throw error;
  } finally {
    databaseFlushInFlight = false;
    if (dirtyDatabaseTaskIds.size > 0 || dirtyDatabaseQueueIds.size > 0) scheduleDatabaseFlush();
  }
  for (const id of rejectedTaskIds) {
    const databaseTask = await loadTaskFromDatabase(id);
    if (databaseTask !== null) reconcileHotTaskFromDatabase(databaseTask);
  }
}

async function initDatabasePersistence(): Promise<void> {
  logger("info", "database_persistence_init_start", { databaseUrl: redactDatabaseUrl(config.databaseUrl) });
  if (!serviceRoleReadOnly(config.serviceRole)) {
    await sql`
      CREATE TABLE IF NOT EXISTS unidesk_code_queue_tasks (
        id TEXT PRIMARY KEY,
        queue_id TEXT NOT NULL DEFAULT 'default',
        status TEXT NOT NULL,
        provider_id TEXT NOT NULL DEFAULT 'main-server',
        execution_mode TEXT NOT NULL DEFAULT 'default',
        model TEXT NOT NULL,
        cwd TEXT NOT NULL,
        prompt TEXT NOT NULL,
        base_prompt TEXT NOT NULL DEFAULT '',
        reference_task_ids JSONB NOT NULL DEFAULT '[]'::jsonb,
        reference_injection JSONB,
        reasoning_effort TEXT,
        max_attempts INTEGER NOT NULL,
        current_attempt INTEGER NOT NULL DEFAULT 0,
        current_mode TEXT,
        codex_thread_id TEXT,
        active_turn_id TEXT,
        created_at TIMESTAMPTZ NOT NULL,
        updated_at TIMESTAMPTZ NOT NULL,
        started_at TIMESTAMPTZ,
        finished_at TIMESTAMPTZ,
        read_at TIMESTAMPTZ,
        last_error TEXT,
        last_judge JSONB,
        output_count INTEGER NOT NULL DEFAULT 0,
        event_count INTEGER NOT NULL DEFAULT 0,
        attempt_count INTEGER NOT NULL DEFAULT 0,
        last_output_seq BIGINT NOT NULL DEFAULT 0,
        task_json JSONB NOT NULL
      )
    `;
    await sql`
      CREATE TABLE IF NOT EXISTS unidesk_code_queue_queues (
        id TEXT PRIMARY KEY,
        name TEXT NOT NULL DEFAULT '',
        created_at TIMESTAMPTZ NOT NULL,
        updated_at TIMESTAMPTZ NOT NULL
      )
    `;
    await sql`
      CREATE TABLE IF NOT EXISTS unidesk_code_queue_workdirs (
        provider_id TEXT NOT NULL,
        execution_mode TEXT NOT NULL DEFAULT 'default',
        path TEXT NOT NULL,
        created_at TIMESTAMPTZ NOT NULL,
        updated_at TIMESTAMPTZ NOT NULL,
        PRIMARY KEY (provider_id, execution_mode, path)
      )
    `;
    await sql`
      CREATE TABLE IF NOT EXISTS unidesk_code_queue_notifications (
        id TEXT PRIMARY KEY,
        kind TEXT NOT NULL,
        dedup_key TEXT NOT NULL,
        target TEXT NOT NULL,
        message TEXT NOT NULL,
        created_at TIMESTAMPTZ NOT NULL,
        updated_at TIMESTAMPTZ NOT NULL,
        attempts INTEGER NOT NULL DEFAULT 0,
        next_attempt_at TIMESTAMPTZ NOT NULL,
        last_error TEXT,
        sent_at TIMESTAMPTZ
      )
    `;
    await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS queue_id TEXT NOT NULL DEFAULT 'default'`;
    await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS provider_id TEXT NOT NULL DEFAULT 'main-server'`;
    await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
    await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS base_prompt TEXT NOT NULL DEFAULT ''`;
    await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_task_ids JSONB NOT NULL DEFAULT '[]'::jsonb`;
    await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS reference_injection JSONB`;
    await sql`ALTER TABLE unidesk_code_queue_tasks ADD COLUMN IF NOT EXISTS read_at TIMESTAMPTZ`;
    await sql`
      UPDATE unidesk_code_queue_tasks
      SET read_at = NULLIF(task_json->>'readAt', '')::timestamptz
      WHERE read_at IS NULL
        AND status IN ('succeeded', 'failed', 'canceled')
        AND COALESCE(task_json->>'readAt', '') <> ''
        AND (task_json->>'readAt') ~ '^\\d{4}-\\d{2}-\\d{2}T'
    `;
    await sql`
      UPDATE unidesk_code_queue_tasks
      SET
        active_turn_id = NULL,
        task_json = jsonb_set(task_json, '{activeTurnId}', 'null'::jsonb, true)
      WHERE status NOT IN ('running', 'judging')
        AND active_turn_id IS NOT NULL
    `;
    await sql`ALTER TABLE unidesk_code_queue_queues ADD COLUMN IF NOT EXISTS name TEXT NOT NULL DEFAULT ''`;
    await sql`ALTER TABLE unidesk_code_queue_workdirs ADD COLUMN IF NOT EXISTS execution_mode TEXT NOT NULL DEFAULT 'default'`;
  }

  const countRows = await sql<Array<{ count: string | number }>>`SELECT COUNT(*) AS count FROM unidesk_code_queue_tasks`;
  const hotTasks = await loadTasksFromDatabase("hot");
  state.tasks.splice(0, state.tasks.length, ...hotTasks);
  state.nextSeq = await loadNextSeqFromDatabase();
  state.updatedAt = nowIso();
  logger("info", "database_task_rows_loaded", {
    databaseTaskCount: Number(countRows[0]?.count ?? hotTasks.length),
    hotTaskCount: hotTasks.length,
    inMemoryOutputRecords: config.maxInMemoryOutputRecords,
    inMemoryEventRecords: config.maxInMemoryEventRecords,
  });
  const queueRows = await sql<DatabaseQueueRow[]>`
    SELECT id, name, created_at, updated_at
    FROM unidesk_code_queue_queues
    ORDER BY id ASC
  `;
  runGarbageCollection();
  const queueMap = new Map(databaseQueueRowsToRecords(queueRows).map((queue) => [queue.id, queue]));
  if (!queueMap.has(defaultQueueId)) queueMap.set(defaultQueueId, { id: defaultQueueId, name: defaultQueueId, createdAt: state.updatedAt, updatedAt: state.updatedAt });
  for (const task of state.tasks) {
    const id = queueIdOf(task);
    const existing = queueMap.get(id);
    if (existing === undefined) {
      queueMap.set(id, { id, name: id, createdAt: task.createdAt, updatedAt: task.updatedAt });
    } else if ((timestampMs(task.updatedAt) ?? 0) > (timestampMs(existing.updatedAt) ?? 0)) {
      existing.updatedAt = task.updatedAt;
    }
  }
  state.queues.splice(0, state.queues.length, ...Array.from(queueMap.values()).sort((left, right) => left.id.localeCompare(right.id)));
  const workdirRows = await sql<Array<{ provider_id: string; execution_mode: string; path: string; created_at: Date | string; updated_at: Date | string }>>`
    SELECT provider_id, execution_mode, path, created_at, updated_at
    FROM unidesk_code_queue_workdirs
    ORDER BY provider_id ASC, execution_mode ASC, path ASC
  `;
  for (const row of workdirRows) {
    rememberWorkdir(row.provider_id, row.execution_mode, row.path, taskTimestamp(String(row.updated_at)) ?? nowIso());
    const record = workdirRecords.get(workdirRecordKey(normalizeTaskProviderId(row.provider_id), normalizeCodeExecutionMode(row.execution_mode), normalizeWorkdirPath(row.path, normalizeTaskProviderId(row.provider_id))));
    if (record !== undefined) {
      record.createdAt = taskTimestamp(String(row.created_at)) ?? record.createdAt;
      record.updatedAt = taskTimestamp(String(row.updated_at)) ?? record.updatedAt;
    }
  }
  ensureDefaultWorkdirRecords();
  if (!serviceRoleReadOnly(config.serviceRole)) await upsertWorkdirsToDatabase(sortedWorkdirRecords());
  databaseReady = true;
  if (config.serviceRole === "combined" || config.serviceRole === "scheduler") scheduleStartupDatabaseMaintenance();
  runGarbageCollection();
  logger("info", "database_persistence_init_complete", {
    databaseTaskCount: Number(countRows[0]?.count ?? hotTasks.length),
    hotTaskCount: state.tasks.length,
    databaseQueueCount: queueRows.length,
    taskCount: state.tasks.length,
    queueCount: state.queues.length,
    maintenanceMode: "background",
  });
}

async function initDatabasePersistenceWithRetry(): Promise<void> {
  const started = Date.now();
  let attempt = 0;
  let reportedLongWait = false;
  while (!databaseReady) {
    attempt += 1;
    try {
      await initDatabasePersistence();
      return;
    } catch (error) {
      setDatabaseLastError(error);
      const elapsedMs = Date.now() - started;
      logger("warn", "database_persistence_init_retry", { attempt, elapsedMs, error: errorToJson(error) });
      if (elapsedMs > 90_000 && !reportedLongWait) {
        reportedLongWait = true;
        logger("error", "database_persistence_required_still_waiting", { attempt, elapsedMs, error: errorToJson(error) });
      }
      await Bun.sleep(Math.min(1000 * attempt, 10_000));
    }
  }
}

function prepareCodexHome(): void {
  mkdirSync(config.codexHome, { recursive: true });
  if (existsSync(config.sourceCodexConfig)) {
    copyFileSync(config.sourceCodexConfig, resolve(config.codexHome, "config.toml"));
  } else {
    logger("warn", "codex_config_source_missing", { sourceCodexConfig: config.sourceCodexConfig });
  }
  const sourceCodexAuth = resolve(dirname(config.sourceCodexConfig), "auth.json");
  if (existsSync(sourceCodexAuth)) {
    copyFileSync(sourceCodexAuth, resolve(config.codexHome, "auth.json"));
  } else {
    logger("warn", "codex_auth_source_missing", { sourceCodexAuth });
  }
}

function openCodeXdgEnv(root = config.opencodeXdgDir): Record<string, string> {
  return {
    XDG_DATA_HOME: resolve(root, "data"),
    XDG_CONFIG_HOME: resolve(root, "config"),
    XDG_CACHE_HOME: resolve(root, "cache"),
    XDG_STATE_HOME: resolve(root, "state"),
  };
}

function prepareOpenCodeHome(): void {
  for (const dir of Object.values(openCodeXdgEnv())) mkdirSync(dir, { recursive: true });
}

function commandPath(command: string): string | null {
  const result = spawnSync("sh", ["-lc", `command -v ${shellQuote(command)}`], { encoding: "utf8", timeout: 2_000 });
  if (result.status !== 0) return null;
  const stdout = typeof result.stdout === "string" ? result.stdout.trim() : "";
  return stdout.length > 0 ? stdout.split(/\r?\n/u)[0] ?? null : null;
}

function runProbe(command: string, args: string[], timeout = 3_000): { ok: boolean; output: string } {
  const result = spawnSync(command, args, { encoding: "utf8", timeout });
  const stdout = typeof result.stdout === "string" ? result.stdout : "";
  const stderr = typeof result.stderr === "string" ? result.stderr : "";
  const output = safePreview(`${stdout}\n${stderr}`, 600);
  return { ok: result.status === 0, output };
}

function currentSkillAvailability() {
  return collectSkillAvailability({ target: config.skillsPath });
}

function collectSkillsStatus(): JsonValue {
  return skillAvailabilityJson(currentSkillAvailability());
}

function collectSkillsSyncPreflight(): JsonValue {
  return skillSyncPreflightJson(collectSkillSyncPreflight({ target: config.skillsPath }));
}

function resolvedRunnerSkillsPath(): string {
  return config.skillsPath;
}

function runnerSkillsBlocker(): Record<string, JsonValue> | null {
  const skills = currentSkillAvailability();
  if (skills.contractOk) return null;
  const pathSpelling = {
    expectedTarget: skills.pathSpelling.expectedTarget,
    forbiddenPathChecked: skills.pathSpelling.forbiddenPathChecked,
    forbiddenPathExists: skills.pathSpelling.forbiddenPathExists,
    forbiddenPathConfigured: skills.pathSpelling.forbiddenPathConfigured,
    forbiddenPathRoles: skills.pathSpelling.forbiddenPathRoles,
    forbiddenPathMustNotBeUsed: skills.pathSpelling.forbiddenPathMustNotBeUsed,
  };
  return {
    ok: false,
    runnerDisposition: "infra-blocked",
    failureKind: skills.blocker ?? "runner-skills-unavailable",
    degradedReason: skills.degradedReason ?? skills.blocker ?? "runner-skills-unavailable",
    message: "Code Queue runner skills are unavailable; refusing to start a code agent without the controlled skills projection.",
    checkedAt: skills.checkedAt,
    source: skills.source,
    target: skills.target,
    resolvedPath: skills.resolvedPath,
    resolvedPathSource: skills.resolvedPathSource,
    requiredSkills: skills.requiredSkills,
    missingSkills: skills.missingSkills,
    counts: {
      sourceSkills: skills.sourceSkillCount,
      targetSkills: skills.targetSkillCount,
      missingSourceSkills: skills.sourceMissingSkills.length,
      missingTargetSkills: skills.targetMissingSkills.length,
    },
    version: skills.version,
    pathSpelling,
    repairHint: skills.repairHint,
    valuesPrinted: false,
  };
}

function collectDevReady(): JsonValue {
  const now = Date.now();
  if (devReadyCache !== null && now - devReadyCache.checkedAtMs < 30_000) return devReadyCache.value;
  const requiredTools = [
    "bash",
    "bun",
    "node",
    "npm",
    "npx",
    "codex",
    "opencode",
    "git",
    "rg",
    "curl",
    "python3",
    "pip3",
    "docker",
    "docker-compose",
    "gh",
    "jq",
    "ssh",
    "rsync",
    "make",
    "gcc",
    "g++",
    "tar",
    "gzip",
    "unzip",
  ];
  const tools = requiredTools.map((name) => {
    const path = commandPath(name);
    return { name, ok: path !== null, path };
  });
  const missingTools = tools.filter((tool) => !tool.ok).map((tool) => tool.name);
  const dockerProbe = runProbe("docker", ["version", "--format", "{{.Client.Version}} client / {{.Server.Version}} server"]);
  const composeProbe = runProbe("docker", ["compose", "version"]);
  const workdirExists = existsSync(config.defaultWorkdir);
  const dockerSocketExists = existsSync("/var/run/docker.sock");
  const homeCodexConfig = resolve(config.codexHome, "config.toml");
  const sourceCodexAuth = resolve(dirname(config.sourceCodexConfig), "auth.json");
  const homeCodexAuth = resolve(config.codexHome, "auth.json");
  const codexConfigReady = existsSync(config.sourceCodexConfig) || existsSync(homeCodexConfig);
  const sshKeyProbe = runProbe("sh", ["-lc", "test -d /root/.ssh && find /root/.ssh -maxdepth 1 -type f \\( -name 'id_*' ! -name '*.pub' \\) -perm -400 -print -quit"]);
  const githubKnownHostProbe = runProbe("ssh-keygen", ["-F", "github.com", "-f", "/root/.ssh/known_hosts"]);
  const sshSharedReady = existsSync("/root/.ssh") && sshKeyProbe.ok && sshKeyProbe.output.trim().length > 0;
  const skills = collectSkillsStatus() as Record<string, JsonValue>;
  const skillsSync = collectSkillsSyncPreflight() as Record<string, JsonValue>;
  const skillsReady = skills.contractOk === true;
  const runtimePreflight = runtimePreflightJson(collectRuntimePreflight({ includeRemote: false, includePushDryRun: false }));
  const ok = missingTools.length === 0 && dockerProbe.ok && composeProbe.ok && workdirExists && dockerSocketExists && codexConfigReady && sshSharedReady && skillsReady;
  const value: JsonValue = {
    ok,
    missingTools,
    tools: tools as unknown as JsonValue,
    workdir: { path: config.defaultWorkdir, exists: workdirExists },
    docker: {
      socketPath: "/var/run/docker.sock",
      socketExists: dockerSocketExists,
      versionOk: dockerProbe.ok,
      version: dockerProbe.output,
      composeOk: composeProbe.ok,
      composeVersion: composeProbe.output,
    },
    codexConfig: {
      sourcePath: config.sourceCodexConfig,
      sourceExists: existsSync(config.sourceCodexConfig),
      homeConfigPath: homeCodexConfig,
      homeConfigExists: existsSync(homeCodexConfig),
      sourceAuthPath: sourceCodexAuth,
      sourceAuthExists: existsSync(sourceCodexAuth),
      homeAuthPath: homeCodexAuth,
      homeAuthExists: existsSync(homeCodexAuth),
      ready: codexConfigReady,
    },
    ssh: {
      rootSshPath: "/root/.ssh",
      rootSshExists: existsSync("/root/.ssh"),
      privateKeyPresent: sshSharedReady,
      githubKnownHostPresent: githubKnownHostProbe.ok,
      ready: sshSharedReady,
    },
    skills,
    skillsSync,
    runtimePreflight,
  };
  devReadyCache = { checkedAtMs: now, value };
  return value;
}

function makeTaskId(): string {
  return `codex_${Date.now()}_${Math.random().toString(16).slice(2, 8)}`;
}

function isCodexTaskId(value: string): boolean {
  return /^codex_\d+_[A-Za-z0-9_-]+$/u.test(value.trim());
}

function addUniqueTaskId(ids: string[], value: string): void {
  const id = value.trim();
  if (isCodexTaskId(id) && !ids.includes(id)) ids.push(id);
}

function collectTaskIdsFromValue(value: unknown, ids: string[]): void {
  if (typeof value === "string") {
    for (const part of value.split(/[\s,，;；]+/u)) addUniqueTaskId(ids, part);
    return;
  }
  if (Array.isArray(value)) {
    for (const item of value) collectTaskIdsFromValue(item, ids);
  }
}

function referenceTaskIdsFromPrompt(prompt: string): string[] {
  const ids: string[] = [];
  const patterns = [
    /引用\s+Code Queue\s+任务\s+(codex_\d+_[A-Za-z0-9_-]+)/giu,
    /\bcodex\s+task\s+(codex_\d+_[A-Za-z0-9_-]+)/giu,
    /(?:引用|上下文|context|reference)[^\n]{0,160}\b(codex_\d+_[A-Za-z0-9_-]+)/giu,
  ];
  for (const pattern of patterns) {
    for (const match of prompt.matchAll(pattern)) addUniqueTaskId(ids, String(match[1] ?? ""));
  }
  return ids;
}

function collectReferenceTaskIds(record: Record<string, unknown>, prompt: string): string[] {
  const ids: string[] = [];
  collectTaskIdsFromValue(record.referenceTaskId, ids);
  collectTaskIdsFromValue(record.referenceTaskIds, ids);
  for (const id of referenceTaskIdsFromPrompt(prompt)) addUniqueTaskId(ids, id);
  return ids;
}

function normalizeRequest(value: unknown): QueueTaskRequest {
  if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error("request body must be an object");
  const record = value as Record<string, unknown>;
  if (typeof record.prompt !== "string" || record.prompt.trim().length === 0) throw new Error("prompt is required");
  const request: QueueTaskRequest = { prompt: record.prompt };
  if (typeof record.queueId === "string" && record.queueId.trim().length > 0) request.queueId = normalizeQueueId(record.queueId);
  const providerId = normalizeProviderId(record.providerId);
  if (providerId !== null) request.providerId = providerId;
  if (typeof record.cwd === "string" && record.cwd.length > 0) request.cwd = record.cwd;
  if (typeof record.model === "string" && record.model.length > 0) request.model = record.model;
  if (typeof record.reasoningEffort === "string" && record.reasoningEffort.length > 0) request.reasoningEffort = record.reasoningEffort;
  if (typeof record.executionMode === "string" && record.executionMode.length > 0) {
    const requestedExecutionMode = normalizeRequestedCodeExecutionMode(record.executionMode);
    if (requestedExecutionMode === null) throw new Error("executionMode must be a short mode identifier");
    request.requestedExecutionMode = requestedExecutionMode;
    request.executionMode = normalizeCodeExecutionMode(requestedExecutionMode);
  }
  if (typeof record.maxAttempts === "number" && Number.isInteger(record.maxAttempts) && record.maxAttempts > 0) request.maxAttempts = clampTaskAttempts(record.maxAttempts);
  const referenceTaskIds = collectReferenceTaskIds(record, record.prompt);
  if (referenceTaskIds.length > 0) request.referenceTaskIds = referenceTaskIds;
  return request;
}

function validateExecutionModeForTask(providerId: string, cwd: string, model: string, executionMode: ReturnType<typeof normalizeCodeExecutionMode>): void {
  if (executionMode !== "windows-native") return;
  if (providerIsMain(providerId)) throw new Error("windows-native executionMode requires a non-main WSL provider");
  if (codeAgentPortForModel(model) !== "codex") throw new Error("windows-native executionMode only supports Codex models");
  if (!cwd.startsWith("/mnt/")) throw new Error("windows-native executionMode requires cwd under /mnt/<drive>");
}

function createTask(request: QueueTaskRequest): QueueTask {
  const at = nowIso();
  const basePrompt = request.basePrompt ?? userPromptForDisplay(request.prompt);
  const referenceTaskIds = request.referenceTaskIds ?? [];
  const providerId = normalizeTaskProviderId(request.providerId);
  const model = normalizeCodeModel(request.model ?? config.defaultModel);
  const executionMode = normalizeCodeExecutionMode(request.executionMode);
  const requestedExecutionMode = normalizeRequestedCodeExecutionMode(request.requestedExecutionMode);
  const cwd = resolveTaskCwd(providerId, request.cwd);
  validateExecutionModeForTask(providerId, cwd, model, executionMode);
  rememberWorkdir(providerId, executionMode, cwd, at);
  const queueId = normalizeQueueId(request.queueId);
  ensureQueue(queueId);
  return {
    id: makeTaskId(),
    queueId,
    queueEnteredAt: at,
    prompt: request.prompt,
    basePrompt,
    referenceTaskIds,
    referenceInjection: request.referenceInjection ?? null,
    providerId,
    cwd,
    model,
    reasoningEffort: resolveReasoningEffort(model, request.reasoningEffort),
    executionMode,
    requestedExecutionMode,
    maxAttempts: request.maxAttempts ?? config.defaultMaxAttempts,
    status: "queued",
    createdAt: at,
    updatedAt: at,
    startedAt: null,
    finishedAt: null,
    readAt: null,
    currentAttempt: 0,
    currentMode: null,
    codexThreadId: null,
    activeTurnId: null,
    finalResponse: "",
    lastError: null,
    lastJudge: null,
    judgeFailCount: 0,
    promptHistory: [],
    output: [],
    events: [],
    attempts: [],
    cancelRequested: false,
    nextPrompt: null,
    nextMode: null,
  };
}

function appendPromptHistory(task: QueueTask, output: LiveOutput | null, method: PromptHistoryItem["method"], text: string, options: { steerId?: string } = {}): void {
  if (output === null) return;
  task.promptHistory = mergePromptHistory([...(Array.isArray(task.promptHistory) ? task.promptHistory : []), {
    seq: output.seq,
    at: output.at,
    method,
    text,
    ...(options.steerId === undefined ? {} : { steerId: options.steerId }),
  }]);
  markTaskDirty(task.id);
  schedulePersistState();
}

function addEvent(task: QueueTask, event: CodexEventSummary): void {
  task.events.push(event);
  if (config.maxInMemoryEventRecords > 0 && task.events.length > config.maxInMemoryEventRecords) task.events.splice(0, task.events.length - config.maxInMemoryEventRecords);
  refreshSchedulerHeartbeat(task, { observedAgentEvent: true });
  markTaskDirty(task.id);
  publishTaskOaEvent(task, "agent-event", { onlyStepChange: true });
}

function taskReferencesEqual(left: string[], right: string[]): boolean {
  if (left.length !== right.length) return false;
  return left.every((value, index) => value === right[index]);
}

function queuedTaskPromptEditable(task: QueueTask): boolean {
  return task.status === "queued"
    && task.startedAt === null
    && task.currentAttempt === 0
    && task.codexThreadId === null
    && task.nextMode === null
    && task.nextPrompt === null
    && task.attempts.length === 0;
}

function normalizePromptEditRequest(task: QueueTask, value: unknown): QueueTaskRequest {
  if (typeof value !== "object" || value === null || Array.isArray(value)) throw new Error("request body must be an object");
  const record = value as Record<string, unknown>;
  if (typeof record.prompt !== "string" || record.prompt.trim().length === 0) throw new Error("prompt is required");
  const referenceTaskIds: string[] = [];
  const hasExplicitReferenceIds = Object.prototype.hasOwnProperty.call(record, "referenceTaskId")
    || Object.prototype.hasOwnProperty.call(record, "referenceTaskIds");
  if (hasExplicitReferenceIds) {
    collectTaskIdsFromValue(record.referenceTaskId, referenceTaskIds);
    collectTaskIdsFromValue(record.referenceTaskIds, referenceTaskIds);
  } else {
    for (const id of task.referenceTaskIds ?? []) addUniqueTaskId(referenceTaskIds, id);
  }
  for (const id of referenceTaskIdsFromPrompt(record.prompt)) addUniqueTaskId(referenceTaskIds, id);
  if (referenceTaskIds.includes(task.id)) throw new Error("a task cannot reference itself while editing prompt");
  return {
    prompt: record.prompt,
    basePrompt: userPromptForDisplay(record.prompt),
    referenceTaskIds,
    queueId: queueIdOf(task),
    providerId: task.providerId,
    cwd: task.cwd,
    model: task.model,
    reasoningEffort: task.reasoningEffort ?? undefined,
    maxAttempts: task.maxAttempts,
  };
}

async function buildQueuedPromptUpdate(task: QueueTask, body: unknown): Promise<QueueTaskRequest> {
  return injectCodeQueueEnvironmentHint(await injectReferencedTaskContext(normalizePromptEditRequest(task, body)));
}

function rewriteEnqueueOutput(task: QueueTask): void {
  const text = `${task.prompt}\n`;
  const output = taskFullOutput(task).find((item) => item.channel === "user" && item.method === "enqueue") ?? null;
  if (output === null) return;
  const nextOutput = { ...output, at: task.updatedAt, text };
  const inMemory = task.output.find((item) => item.seq === output.seq);
  if (inMemory !== undefined) {
    inMemory.at = nextOutput.at;
    inMemory.text = nextOutput.text;
  }
  appendOutputArchive(task, nextOutput, "set", text);
}

function truthyParam(url: URL, name: string): boolean {
  const value = url.searchParams.get(name);
  return value === "1" || value === "true" || value === "yes";
}

function boundedStringSetParam(url: URL, name: string, maxItems = 50): Set<string> | undefined {
  const rawValues = url.searchParams.getAll(name).flatMap((value) => value.split(","));
  const values = Array.from(new Set(rawValues.map((value) => value.trim()).filter(Boolean))).slice(0, maxItems);
  return values.length === 0 ? undefined : new Set(values);
}

function parseSeqParam(url: URL, name: string, defaultValue: number | null): number | null {
  const raw = url.searchParams.get(name);
  if (raw === null) return defaultValue;
  const value = Number(raw);
  return Number.isFinite(value) ? value : defaultValue;
}

function parseTextLimit(url: URL): number {
  const value = Number(url.searchParams.get("maxTextChars") ?? 12_000);
  return Number.isInteger(value) && value > 0 ? Math.min(500_000, value) : 12_000;
}

function pageBySeq<T extends { seq: number }>(items: T[], url: URL, limit: number): {
  mode: "tail" | "after" | "before";
  afterSeq: number;
  beforeSeq: number | null;
  nextAfterSeq: number;
  previousBeforeSeq: number | null;
  hasMore: boolean;
  hasBefore: boolean;
  chunk: T[];
} {
  const beforeSeq = parseSeqParam(url, "beforeSeq", null);
  const afterSeq = parseSeqParam(url, "afterSeq", 0) ?? 0;
  const mode = beforeSeq !== null ? "before" : truthyParam(url, "tail") ? "tail" : "after";
  const boundedBeforeSeq = beforeSeq ?? 0;
  const chunk = mode === "before"
    ? items.filter((item) => Number(item.seq) < boundedBeforeSeq).slice(-limit)
    : mode === "tail"
      ? items.slice(-limit)
      : items.filter((item) => Number(item.seq) > afterSeq).slice(0, limit);
  const firstSeq = chunk[0]?.seq;
  const lastSeq = chunk.at(-1)?.seq;
  return {
    mode,
    afterSeq,
    beforeSeq,
    nextAfterSeq: lastSeq ?? afterSeq,
    previousBeforeSeq: firstSeq ?? beforeSeq,
    hasMore: lastSeq !== undefined && items.some((item) => Number(item.seq) > Number(lastSeq)),
    hasBefore: firstSeq !== undefined && items.some((item) => Number(item.seq) < Number(firstSeq)),
    chunk,
  };
}

function terminalTask(task: QueueTask): boolean {
  return task.status === "succeeded" || task.status === "failed" || task.status === "canceled";
}

function terminalTaskUnread(task: QueueTask): boolean {
  return terminalTask(task) && task.readAt === null;
}

function durationMsBetween(startAt: string | null | undefined, endAt: string | null | undefined): number | null {
  const startMs = timestampMs(startAt);
  const endMs = timestampMs(endAt);
  return nonNegativeElapsed(startMs, endMs);
}

function formatDurationMs(value: number | null): string {
  if (value === null) return "-";
  const totalSeconds = Math.max(0, Math.floor(value / 1000));
  const days = Math.floor(totalSeconds / 86_400);
  const hours = Math.floor((totalSeconds % 86_400) / 3600);
  const minutes = Math.floor((totalSeconds % 3600) / 60);
  const seconds = totalSeconds % 60;
  const parts: string[] = [];
  if (days > 0) parts.push(`${days}d`);
  if (hours > 0 || parts.length > 0) parts.push(`${hours}h`);
  if (minutes > 0 || parts.length > 0) parts.push(`${minutes}m`);
  parts.push(`${seconds}s`);
  return parts.join(" ");
}

function openCodeFreshRecoveryPrompt(task: QueueTask, prompt: string, reason: string): string {
  const previous = safePreview(task.finalResponse, 2000);
  return [
    "OpenCode port recovery：上一个 OpenCode session 无法恢复；Code Queue 将为同一任务开启新的 OpenCode session，避免对缺失 session 无限重试。",
    `恢复原因：${judgeReasonForPrompt(reason)}`,
    "原始任务摘要/按需查询：",
    compactRetryTaskContext(task),
    previous.length > 0 ? `上一轮可见 assistant response 摘要：\n${previous}` : "",
    "本轮 recovery continuation prompt：",
    prompt,
  ].filter((line) => line.length > 0).join("\n\n");
}

function codexFreshRecoveryPrompt(task: QueueTask, prompt: string, reason: string): string {
  const previous = safePreview(task.finalResponse, 2000);
  return [
    "Codex port recovery：上一个 Codex threadId 缺失；Code Queue 将为同一任务开启新的 Codex thread，避免对缺失 thread 无限重试。",
    `恢复原因：${judgeReasonForPrompt(reason)}`,
    "原始任务摘要/按需查询：",
    compactRetryTaskContext(task),
    previous.length > 0 ? `上一轮可见 assistant response 摘要：\n${previous}` : "",
    "本轮 recovery continuation prompt：",
    prompt,
  ].filter((line) => line.length > 0).join("\n\n");
}


async function runCodeAgentTurn(task: QueueTask, prompt: string): Promise<CodexRunResult> {
  return codeAgentPortForModel(task.model) === "opencode" ? runOpenCodeTurn(task, prompt) : runCodexTurn(task, prompt);
}



configureProviderRuntime({
  config,
  safePreview,
});
ensureDefaultWorkdirRecords();

configureTaskOutput({
  config,
  allocateSeq: () => state.nextSeq++,
  errorToJson,
  logger,
  markTaskDirty,
  nowIso,
  onOutputAppended: (task, output, op) => {
    refreshSchedulerHeartbeat(task, { persistedTrace: true });
    const archiveOp = op === "append" ? "append" : "set";
    const stepChanged = recordTaskOutputMetrics(task, output, archiveOp);
    const projectionOutput = traceStepOutputForProjection(task, output);
    if (stepChanged) publishCodeQueueTraceStep(task, queueIdOf(task), projectionOutput, taskOutputMaxSeq(task));
    else if ((archiveOp === "append" || output.method === "item/completed") && outputUpdatesExistingTraceStep(output)) publishCodeQueueTraceStep(task, queueIdOf(task), projectionOutput, taskOutputMaxSeq(task), null, String(output.text || "").length);
    if (archiveOp === "append" && !outputCanChangeStepCount(output)) return;
    publishTaskOaEvent(task, "output", { onlyStepChange: archiveOp === "append", stepChanged });
  },
  schedulePersistState,
});

configureTaskView({
  config,
  errorToJson,
  jsonResponse,
  logger,
  mergePromptHistory,
  nowIso,
  outputPromptHistory,
  pageBySeq,
  parseLimit,
  parseSeqParam,
  queueIdOf,
  queuedStatusReason,
  queuedTaskPromptEditable,
  taskQueueEnteredAt,
});

configureNotifications({
  config,
  activeRunCount: () => activeRuns.size,
  activeRunSlotCount,
  activeRunTaskIds: () => Array.from(activeRuns.values()).map((run) => run.taskId),
  databaseReady: () => databaseReady,
  errorToJson,
  hasRunnableTask,
  lastAssistantMessage: (task) => lastAssistantMessage(task),
  loadAllTasksForRead,
  logger,
  nonNegativeElapsed,
  nowIso,
  processingQueueCount: () => processingQueues.size,
  queueCount: () => perQueueSummaries().length,
  queueIdOf,
  safePreview,
  shutdownRequested: () => shutdownRequested,
  sql: () => sql,
  taskTimestamp,
  tasks: () => state.tasks,
  timestampMs,
});

configureQueueApi({
  config,
  activeRunSlotQueueIds,
  activeRunSlotWaiterSummaries,
  activeRuns,
  codexSqliteLogExporter: () => codexSqliteLogExporter as unknown as Record<string, JsonValue>,
  collectDevReady,
  compactJsonResponse,
  databaseLastError: () => databaseLastError,
  databaseReady: () => databaseReady,
  defaultQueueId,
  dirtyDatabaseQueueCount: () => dirtyDatabaseQueueIds.size,
  dirtyDatabaseTaskCount: () => dirtyDatabaseTaskIds.size,
  databaseStorageHealth: () => databaseStorageHealth({
    postgresReady: databaseReady,
    dirtyTaskCount: dirtyDatabaseTaskIds.size,
    dirtyQueueCount: dirtyDatabaseQueueIds.size,
    lastError: databaseLastErrorDetail ?? databaseLastError,
    flushInFlight: databaseFlushInFlight,
    clientRotationCount: databaseClientRotationCount,
    lastClientRotationAt: databaseLastClientRotationAt,
    lastClientRotationReason: databaseLastClientRotationReason,
    transientUncaughtSuppressedCount: databaseTransientUncaughtSuppressedCount,
    lastTransientUncaughtSuppressedAt: databaseLastTransientUncaughtSuppressedAt,
    consecutiveFlushFailures: databaseFlushFailureCount,
    lastFlushFailureAt: databaseLastFlushFailureAt,
    nextFlushRetryAt: databaseNextFlushRetryAt,
  }),
  jsonResponse,
  judgeFailRetryLimit,
  loadQueuesFromDatabase,
  loadTaskFromDatabase,
  loadTasksFromDatabase,
  loadTasksFromDatabaseByIds,
  pageBySeq,
  parseLimit,
  parseTextLimit,
  processing: () => processing,
  processingQueues,
  oaEventPublisherStatus: () => oaEventPublisherStatus() as JsonValue,
  queueHeadTask,
  queueIdOf,
  queues: () => state.queues,
  queueTaskIsRunnable,
  queuedStatusReason,
  queuedTaskPromptEditable,
  reconcileStatus: schedulerReconcileStatus,
  runGarbageCollection,
  safeQueueId,
  safeQueueName,
  sql: () => sql,
  taskQueueEnteredAt,
  traceStatsForTasks: readOaTraceStatsForTasks,
  tasks: () => state.tasks,
  truthyParam,
});

configureReferences({
  addUniqueTaskId,
  findTask: findTaskForRead,
  nowIso,
  referenceInjectionMaxRounds,
  referenceTaskIdsFromPrompt,
});

configureSelfTests({
  config,
  activeRunSlotCount,
  activeRunSlotReservations,
  activeRunSlotWaiters,
  availableQueueStartSlotsFor,
  defaultQueueId,
  enqueueActiveRunSlotWaiter,
  injectReferencedTaskContext,
  moveTaskToQueueForTest: (task, req) => moveTaskToQueue(task, req, { bypassRoleCheck: true }),
  nextRunnableTaskFrom,
  normalizeTask,
  nowIso,
  processingQueues,
  queueActiveTasksForRestartRetry,
  queueHeadTask,
  queuedStatusReason,
  removeActiveRunSlotWaiter,
  resolveReasoningEffort,
  runDatabaseClaimMoveSelfTest,
  staleActiveRecoverySelfTest: runSchedulerStaleActiveRecoverySelfTest,
  taskIsRecoverableOrphanedActive,
  tasks: () => state.tasks,
  updateProcessingFlag,
});

configureJudge({
  config,
  logger,
  safePreview,
  userPromptForDisplay,
  taskFullOutput,
  taskReferenceIds,
  extractRecord,
  extractString,
  promptLineCount,
  judgeFailRetryLimit,
  publishJudgeEvent: (task, type, payload) => publishCodeQueueJudgeEvent(task, queueIdOf(task), type, payload),
});

configureCodexPort({
  config,
  activeRuns,
  appendOutput,
  addEvent,
  ensureTaskExecutionContainer,
  formatCommandOutput,
  logger,
  persistTaskState,
  providerIsMain,
  queueIdOf,
  recordNumberField,
  recordStringField,
  refreshSchedulerHeartbeat,
  remoteAppServerCommand,
  windowsNativeAppServerCommand,
  resolveReasoningEffort,
  safePreview,
  nowIso,
});

configureOpenCodePort({
  config,
  activeRuns,
  addEvent,
  appendOutput,
  buildDevContainerPlan,
  compactRetryTaskContext,
  ensureTaskExecutionContainer,
  judgeReasonForPrompt,
  logger,
  nowIso,
  openCodeFreshRecoveryPrompt,
  openCodeXdgEnv,
  persistTaskState,
  providerIsMain,
  queueIdOf,
  recordStringField,
  refreshSchedulerHeartbeat,
  remoteHostWorkdirForTask,
  safePreview,
  shellQuote,
  shutdownRequested: () => shutdownRequested,
});

configureDevContainers({
  config,
  appendOutput,
  buildDevContainerPlan,
  containerTunnelStartScript,
  devContainerEnsurePromises,
  devContainerPingScript,
  errorToJson,
  extractRecord,
  jsonResponse,
  logger,
  masterKeyReadScript,
  masterKeySetupScript,
  masterProxyEvidenceScript,
  masterProxyFinishScript,
  masterProxyPrepareScript,
  normalizeProviderId,
  providerIsMain,
  readJson,
  remoteCodexConfigInstallScript,
  remoteCodexRuntimePrepareScript,
  remoteContainerStartScript,
  remoteHostWorkdirForTask,
  remoteKeyInstallScript,
  runCodeQueueSsh,
  safePreview,
  shellQuote,
  throwIfCommandFailed,
});

function outputForProbe(item: { channel: OutputChannel; text: string; method?: string }, index: number): LiveOutput {
  return {
    seq: index + 1,
    at: nowIso(),
    channel: item.channel,
    text: item.text,
    method: item.method,
  };
}

function taskForJudgeProbe(probe: JudgeProbeCase): QueueTask {
  const at = nowIso();
  return {
    id: `judge_probe_${probe.id}`,
    queueId: defaultQueueId,
    queueEnteredAt: at,
    prompt: probe.prompt,
    basePrompt: probe.prompt,
    referenceTaskIds: [],
    referenceInjection: null,
    providerId: config.mainProviderId,
    cwd: config.defaultWorkdir,
    model: config.defaultModel,
    reasoningEffort: resolveReasoningEffort(config.defaultModel, config.defaultReasoningEffort),
    executionMode: "default",
    maxAttempts: 3,
    status: "judging",
    createdAt: at,
    updatedAt: at,
    startedAt: at,
    finishedAt: null,
    readAt: null,
    currentAttempt: 1,
    currentMode: "initial",
    codexThreadId: "judge-probe-thread",
    activeTurnId: null,
    finalResponse: probe.finalResponse,
    lastError: null,
    lastJudge: null,
    judgeFailCount: 0,
    promptHistory: [],
    output: (probe.outputs ?? []).map(outputForProbe),
    events: probe.events ?? [],
    attempts: [],
    cancelRequested: probe.cancelRequested ?? false,
    nextPrompt: null,
    nextMode: null,
  };
}

function resultForJudgeProbe(probe: JudgeProbeCase, task: QueueTask): CodexRunResult {
  return {
    threadId: "judge-probe-thread",
    turnId: "judge-probe-turn",
    finalResponse: probe.finalResponse,
    terminalStatus: probe.terminalStatus,
    terminalError: probe.terminalError ?? null,
    transportClosedBeforeTerminal: probe.transportClosedBeforeTerminal ?? false,
    appServerExit: {
      code: probe.transportClosedBeforeTerminal ? 1 : 0,
      signal: null,
      stderrTail: probe.stderrTail ?? "",
    },
    events: task.events,
  };
}

async function runJudgeProbe(): Promise<Response> {
  const results = await Promise.all(defaultJudgeProbeCases.map(async (probe) => {
    const task = taskForJudgeProbe(probe);
    const result = resultForJudgeProbe(probe, task);
    const startedAt = nowIso();
    const finishedAt = nowIso();
    task.attempts.push(attemptFromResult(task, "initial", startedAt, finishedAt, result));
    const judge = await judgeTask(task, result);
    const expectedContinuePromptIncludes = probe.expectedContinuePromptIncludes ?? [];
    const expectedContinuePromptExcludes = probe.expectedContinuePromptExcludes ?? [];
    const continuePrompt = judge.continuePrompt ?? "";
    const continuePromptHit = expectedContinuePromptIncludes.every((text) => continuePrompt.includes(text));
    const continuePromptExclusionHit = expectedContinuePromptExcludes.every((text) => !continuePrompt.includes(text));
    const continuePromptMaxCharsHit = probe.expectedContinuePromptMaxChars === undefined || continuePrompt.length <= probe.expectedContinuePromptMaxChars;
    const continuePromptMaxLinesHit = probe.expectedContinuePromptMaxLines === undefined || promptLineCount(continuePrompt) <= probe.expectedContinuePromptMaxLines;
    return {
      id: probe.id,
      expected: probe.expected,
      decision: judge.decision,
      hit: judge.decision === probe.expected && continuePromptHit && continuePromptExclusionHit && continuePromptMaxCharsHit && continuePromptMaxLinesHit,
      continuePromptHit,
      continuePromptExclusionHit,
      continuePromptMaxCharsHit,
      continuePromptMaxLinesHit,
      expectedContinuePromptIncludes,
      expectedContinuePromptExcludes,
      expectedContinuePromptMaxChars: probe.expectedContinuePromptMaxChars ?? null,
      expectedContinuePromptMaxLines: probe.expectedContinuePromptMaxLines ?? null,
      confidence: judge.confidence,
      source: judge.source,
      reason: judge.reason,
      continuePrompt: continuePrompt.length > 0 ? continuePrompt : null,
    };
  }));
  const hits = results.filter((result) => result.hit).length;
  const hitRate = results.length === 0 ? 0 : hits / results.length;
  logger("info", "judge_probe_completed", { configured: config.minimaxApiKey.length > 0, model: config.minimaxModel, hits, total: results.length, hitRate });
  return jsonResponse({
    ok: true,
    behaviorVersion: judgeProbeBehaviorVersion,
    behavior: {
      version: judgeProbeBehaviorVersion,
      contract: "defaultJudgeProbeCases decision/hit contract",
    },
    configured: config.minimaxApiKey.length > 0,
    model: config.minimaxModel,
    hits,
    total: results.length,
    hitRate,
    results,
  });
}

function cloneJson<T>(value: T): T {
  return JSON.parse(JSON.stringify(value)) as T;
}

function numberOrNull(value: unknown): number | null {
  const parsed = Number(value);
  return Number.isFinite(parsed) ? parsed : null;
}

function maxOutputSeqBefore(output: LiveOutput[], exclusiveSeq: number): number | null {
  let maxSeq: number | null = null;
  for (const item of output) {
    const seq = numberOrNull(item.seq);
    if (seq === null || seq >= exclusiveSeq) continue;
    maxSeq = maxSeq === null ? seq : Math.max(maxSeq, seq);
  }
  return maxSeq;
}

function outputSeqAtOrBeforeTime(output: LiveOutput[], at: string | null): number | null {
  const atMs = timestampMs(at);
  if (atMs === null) return null;
  let maxSeq: number | null = null;
  for (const item of output) {
    const itemMs = timestampMs(item.at);
    if (itemMs === null || itemMs > atMs) continue;
    const seq = numberOrNull(item.seq);
    if (seq === null) continue;
    maxSeq = maxSeq === null ? seq : Math.max(maxSeq, seq);
  }
  return maxSeq;
}

function preJudgeOutputEndSeq(attempt: AttemptSummary, output: LiveOutput[]): { endSeq: number | null; source: string } {
  const judgeSeq = numberOrNull(attempt.judgeSeq);
  if (judgeSeq !== null) return { endSeq: maxOutputSeqBefore(output, judgeSeq), source: "before-judgeSeq" };
  const explicitEnd = numberOrNull(attempt.outputEndSeq);
  if (explicitEnd !== null) {
    const explicitItem = output.find((item) => numberOrNull(item.seq) === explicitEnd);
    if (explicitItem?.method === "judge") return { endSeq: maxOutputSeqBefore(output, explicitEnd), source: "before-outputEndSeq-judge" };
    return { endSeq: explicitEnd, source: "outputEndSeq" };
  }
  const byFinishedAt = outputSeqAtOrBeforeTime(output, attempt.finishedAt);
  return { endSeq: byFinishedAt, source: byFinishedAt === null ? "unbounded" : "finishedAt" };
}

function outputBeforeJudge(output: LiveOutput[], endSeq: number | null): LiveOutput[] {
  if (endSeq === null) return output;
  return output.filter((item) => numberOrNull(item.seq) !== null && Number(item.seq) <= endSeq);
}

function currentAttemptOutputCountForReplay(attempt: AttemptSummary, output: LiveOutput[], hotOutput: LiveOutput[], endSeq: number | null): number {
  const startSeq = numberOrNull(attempt.outputStartSeq);
  if (startSeq === null) return hotOutput.slice(-80).length;
  return output.filter((item) => {
    const seq = numberOrNull(item.seq);
    return seq !== null && seq >= startSeq && (endSeq === null || seq <= endSeq);
  }).length;
}

function finalResponseForTaskBeforeJudge(attempts: AttemptSummary[]): string {
  for (let index = attempts.length - 1; index >= 0; index -= 1) {
    const finalResponse = String(attempts[index]?.finalResponse ?? "");
    if (finalResponse.trim().length > 0) return finalResponse;
  }
  return "";
}

function retainedEventsForAttempt(task: QueueTask, attempt: AttemptSummary): CodexRunResult["events"] {
  const startedMs = timestampMs(attempt.startedAt);
  const finishedMs = timestampMs(attempt.finishedAt);
  return task.events.filter((event) => {
    const eventMs = timestampMs(event.at);
    if (eventMs === null) return false;
    if (startedMs !== null && eventMs < startedMs) return false;
    if (finishedMs !== null && eventMs > finishedMs + 1000) return false;
    return true;
  });
}

function attemptEventsForReplay(task: QueueTask, attempt: AttemptSummary): { events: CodexRunResult["events"]; source: string; exact: boolean } {
  if (Array.isArray(attempt.events)) return { events: cloneJson(attempt.events), source: "attempt.events", exact: true };
  return { events: cloneJson(retainedEventsForAttempt(task, attempt)), source: "retained-task-events", exact: false };
}

function replayAttemptIndexFromUrl(task: QueueTask, url: URL): number {
  const raw = url.searchParams.get("attempt") ?? url.searchParams.get("attemptId") ?? url.searchParams.get("attemptIndex");
  if (raw !== null) {
    const value = Number(raw);
    if (!Number.isInteger(value) || value <= 0) throw new Error("judge replay attempt must be a positive integer");
    return value;
  }
  const latest = task.attempts.at(-1)?.index ?? task.currentAttempt;
  if (!Number.isInteger(latest) || latest <= 0) throw new Error("task has no completed attempt to judge");
  return latest;
}

function buildJudgeReplay(task: QueueTask, attemptIndex: number): {
  task: QueueTask;
  result: CodexRunResult;
  attempt: AttemptSummary;
  replay: Record<string, JsonValue>;
} {
  const attempt = task.attempts.find((item) => item.index === attemptIndex);
  if (attempt === undefined) throw new Error(`attempt ${attemptIndex} not found on task ${task.id}`);
  const fullOutput = taskFullOutput(task);
  const preJudge = preJudgeOutputEndSeq(attempt, fullOutput);
  const beforeJudge = outputBeforeJudge(fullOutput, preJudge.endSeq);
  const hotOutput = config.maxInMemoryOutputRecords > 0 ? beforeJudge.slice(-config.maxInMemoryOutputRecords) : beforeJudge;
  const attempts = task.attempts
    .filter((item) => item.index <= attempt.index)
    .map((item) => cloneJson(item));
  const latestAttempt = attempts[attempts.length - 1];
  if (latestAttempt === undefined) throw new Error(`attempt ${attemptIndex} could not be prepared`);
  latestAttempt.judge = null;
  latestAttempt.judgeAt = null;
  latestAttempt.judgeSeq = null;
  latestAttempt.outputEndSeq = preJudge.endSeq;
  const eventReplay = attemptEventsForReplay(task, attempt);
  const previousAttempts = attempts.filter((item) => item.index < attempt.index);
  const result: CodexRunResult = {
    threadId: task.codexThreadId,
    turnId: null,
    finalResponse: String(attempt.finalResponse ?? ""),
    terminalStatus: attempt.terminalStatus,
    terminalError: attempt.error,
    transportClosedBeforeTerminal: attempt.transportClosedBeforeTerminal,
    appServerExit: {
      code: attempt.appServerExitCode,
      signal: attempt.appServerSignal,
      stderrTail: String(attempt.stderrTail ?? ""),
    },
    events: eventReplay.events,
  };
  const replayTask: QueueTask = {
    ...cloneJson(task),
    status: "judging",
    currentAttempt: attempt.index,
    currentMode: attempt.mode,
    attempts,
    output: numberOrNull(attempt.outputStartSeq) === null ? hotOutput : beforeJudge,
    events: eventReplay.events,
    finalResponse: finalResponseForTaskBeforeJudge(attempts),
    lastJudge: previousAttempts.at(-1)?.judge ?? null,
    judgeFailCount: previousAttempts.filter((item) => item.judge?.decision === "fail").length,
    activeTurnId: null,
    finishedAt: null,
    readAt: null,
    nextPrompt: null,
    nextMode: null,
  };
  return {
    task: replayTask,
    result,
    attempt,
    replay: {
      sameJudgeCodePath: true,
      replayExact: eventReplay.exact,
      taskId: task.id,
      attemptIndex: attempt.index,
      attemptMode: attempt.mode,
      outputStartSeq: attempt.outputStartSeq ?? null,
      storedOutputEndSeq: attempt.outputEndSeq ?? null,
      replayOutputEndSeq: preJudge.endSeq,
      replayOutputEndSeqSource: preJudge.source,
      judgeSeq: attempt.judgeSeq ?? null,
      fullOutputBeforeJudgeCount: beforeJudge.length,
      hotOutputCount: hotOutput.length,
      currentAttemptOutputCount: currentAttemptOutputCountForReplay(attempt, beforeJudge, hotOutput, preJudge.endSeq),
      eventSource: eventReplay.source,
      eventReplayExact: eventReplay.exact,
      eventCount: eventReplay.events.length,
      note: eventReplay.exact
        ? "Replay uses attempt-stored event summaries plus the same judgeTask context builder and MiniMax call path."
        : "Historical attempt did not store per-attempt events; replay falls back to retained task.events. Output, final response, terminal status, stderr, attempt window, prompt compaction, and MiniMax call path are reconstructed from persisted task/output archives.",
    },
  };
}

async function runSingleTaskJudge(task: QueueTask, url: URL): Promise<Response> {
  const attemptIndex = replayAttemptIndexFromUrl(task, url);
  const dryRun = truthyParam(url, "dryRun") || truthyParam(url, "noCall");
  const includePrompt = truthyParam(url, "includePrompt");
  const replay = buildJudgeReplay(task, attemptIndex);
  const context = judgeTaskInputDiagnostics(replay.task, replay.result, includePrompt);
  const startedAt = Date.now();
  const judge = dryRun ? null : await judgeTask(replay.task, replay.result);
  const durationMs = Date.now() - startedAt;
  const storedJudge = replay.attempt.judge ?? null;
  const storedFailure = storedJudge?.failureDetails ?? null;
  const promptChars = Number(context.promptChars);
  const payloadBytes = Number(context.payloadBytes);
  return jsonResponse({
    ok: true,
    dryRun,
    configured: config.minimaxApiKey.length > 0,
    model: config.minimaxModel,
    taskId: task.id,
    attempt: {
      index: replay.attempt.index,
      mode: replay.attempt.mode,
      terminalStatus: replay.attempt.terminalStatus,
      finalResponseChars: replay.attempt.finalResponseChars ?? String(replay.attempt.finalResponse ?? "").length,
      startedAt: replay.attempt.startedAt,
      finishedAt: replay.attempt.finishedAt,
    },
    replay: replay.replay,
    context,
    judge,
    durationMs,
    storedJudge,
    comparison: {
      storedDecision: storedJudge?.decision ?? null,
      storedSource: storedJudge?.source ?? null,
      decisionMatchesStored: judge === null || storedJudge === null ? null : judge.decision === storedJudge.decision,
      storedFailureTimedOut: storedFailure?.timedOut ?? null,
      storedFailureStage: storedFailure?.stage ?? null,
      storedPromptChars: storedFailure?.promptChars ?? null,
      promptCharsDeltaFromStored: storedFailure?.promptChars === undefined || !Number.isFinite(promptChars) ? null : promptChars - storedFailure.promptChars,
      storedPayloadBytes: storedFailure?.payloadBytes ?? null,
      payloadBytesDeltaFromStored: storedFailure?.payloadBytes === undefined || !Number.isFinite(payloadBytes) ? null : payloadBytes - storedFailure.payloadBytes,
    },
  });
}

function attemptFromResult(task: QueueTask, mode: RunMode, startedAt: string, finishedAt: string, result: CodexRunResult, outputStartSeq: number | null = null, outputEndSeq: number | null = null, inputPrompt: string | null = null): AttemptSummary {
  const finalResponse = result.finalResponse;
  const attempt: AttemptSummary = {
    index: task.currentAttempt,
    mode,
    startedAt,
    finishedAt,
    providerId: task.providerId,
    executionMode: task.executionMode,
    terminalStatus: result.terminalStatus,
    transportClosedBeforeTerminal: result.transportClosedBeforeTerminal,
    appServerExitCode: result.appServerExit.code,
    appServerSignal: result.appServerExit.signal,
    error: result.terminalError,
    events: result.events,
    finalResponse,
    finalResponsePreview: safePreview(finalResponse, 3000),
    finalResponseChars: finalResponse.length,
    judge: null,
    judgeAt: null,
    judgeSeq: null,
    stderrTail: safePreview(result.appServerExit.stderrTail, 3000),
    outputStartSeq,
    outputEndSeq,
    runnerErrorClassification: result.runnerErrorClassification ?? null,
  };
  if (inputPrompt !== null) setAttemptInputPrompt(attempt, inputPrompt);
  return attempt;
}

function retryBackoffMs(completedAttempts: number): number {
  const retryIndex = Math.max(1, Math.floor(completedAttempts));
  const exponent = Math.min(20, retryIndex - 1);
  return Math.min(retryBackoffMaxMs, retryBackoffBaseMs * (2 ** exponent));
}

function stableRetryJitterMs(taskId: string, completedAttempts: number, jitterWindowMs: number): number {
  const boundedWindow = Math.max(0, Math.floor(jitterWindowMs));
  if (boundedWindow <= 0) return 0;
  let hash = 2166136261;
  const input = `${taskId}:${Math.max(1, Math.floor(completedAttempts))}`;
  for (let index = 0; index < input.length; index += 1) {
    hash ^= input.charCodeAt(index);
    hash = Math.imul(hash, 16777619);
  }
  return (hash >>> 0) % (boundedWindow + 1);
}

function retryBackoffWithJitterMs(task: QueueTask, completedAttempts: number, minimumDelayMs = 0): { baseDelayMs: number; jitterMs: number; delayMs: number } {
  const baseDelayMs = Math.min(retryBackoffMaxMs, Math.max(minimumDelayMs, retryBackoffMs(completedAttempts)));
  const jitterWindowMs = Math.min(30_000, Math.max(1_000, Math.floor(baseDelayMs * 0.3)));
  const jitterMs = stableRetryJitterMs(task.id, completedAttempts, jitterWindowMs);
  return {
    baseDelayMs,
    jitterMs,
    delayMs: Math.min(retryBackoffMaxMs, baseDelayMs + jitterMs),
  };
}

function judgeProviderBackoffMs(task: QueueTask): number {
  const fallbackCount = Math.max(1, fallbackJudgeRetryCount(task));
  return Math.min(5 * 60_000, Math.max(30_000, retryBackoffMs(fallbackCount)));
}

function latestAttemptRunnerErrorClassification(task: QueueTask): Record<string, JsonValue> | null {
  const latest = task.attempts.at(-1)?.runnerErrorClassification;
  return typeof latest === "object" && latest !== null && !Array.isArray(latest) ? latest as Record<string, JsonValue> : null;
}

function latestAttemptExternalProvider429(task: QueueTask): boolean {
  return latestAttemptRunnerErrorClassification(task)?.externalProvider429 === true;
}

function latestAttemptStaleBadResume(task: QueueTask): boolean {
  return latestAttemptRunnerErrorClassification(task)?.staleBadResume === true;
}

function clearStaleBadResumeThread(task: QueueTask): boolean {
  if (!latestAttemptStaleBadResume(task) || task.codexThreadId === null) return false;
  const previousThreadId = task.codexThreadId;
  task.codexThreadId = null;
  task.activeTurnId = null;
  appendOutput(task, "system", `stale bad-resume thread ${previousThreadId} superseded; next retry will start a fresh Codex thread because app-server reported no rollout found for thread id\n`, "thread/resume-superseded");
  logger("warn", "stale_bad_resume_thread_superseded", { taskId: task.id, previousThreadId });
  return true;
}

function retryBackoffEvidence(task: QueueTask, completedAttempts: number, reason: string): JsonValue {
  const runnerClassification = latestAttemptRunnerErrorClassification(task);
  const externalProvider429 = runnerClassification?.externalProvider429 === true;
  const staleBadResume = runnerClassification?.staleBadResume === true;
  const baseDelayMs = retryBackoffMs(completedAttempts);
  const jittered = externalProvider429
    ? retryBackoffWithJitterMs(task, completedAttempts, 30_000)
    : { baseDelayMs, jitterMs: 0, delayMs: baseDelayMs };
  return {
    strategy: externalProvider429 ? "exponential-jitter" : "exponential",
    reason,
    completedAttempts,
    baseDelayMs: jittered.baseDelayMs,
    jitterMs: jittered.jitterMs,
    delayMs: jittered.delayMs,
    maxDelayMs: retryBackoffMaxMs,
    externalProvider429,
    staleBadResume,
    failureKind: runnerClassification?.failureKind ?? null,
    runnerDisposition: runnerClassification?.disposition ?? null,
    commanderBenefit: staleBadResume
      ? "stale resume thread is superseded once and retried on a fresh Codex thread"
      : "fewer failed runners and clearer retry/backoff evidence",
  };
}

function judgeProviderBackoffEvidence(task: QueueTask): JsonValue {
  const fallbackCount = Math.max(1, fallbackJudgeRetryCount(task));
  const baseDelayMs = judgeProviderBackoffMs(task);
  const jitterMs = stableRetryJitterMs(task.id, fallbackCount, Math.min(30_000, Math.max(1_000, Math.floor(baseDelayMs * 0.3))));
  return {
    strategy: "exponential-jitter",
    reason: "judge-provider-transient-retry",
    completedAttempts: task.attempts.length,
    baseDelayMs,
    jitterMs,
    delayMs: Math.min(5 * 60_000, baseDelayMs + jitterMs),
    maxDelayMs: 5 * 60_000,
    externalProvider429: false,
    failureKind: "judge-provider-transient",
    runnerDisposition: "retryable-transient",
    commanderBenefit: "fewer failed runners and clearer retry/backoff evidence",
  };
}

function appendRetryBackoffEvidence(task: QueueTask, evidence: JsonValue, continuation: string): void {
  const record = typeof evidence === "object" && evidence !== null && !Array.isArray(evidence) ? evidence as Record<string, JsonValue> : {};
  const delaySeconds = Math.round(Number(record.delayMs ?? 0) / 1000);
  const baseSeconds = Math.round(Number(record.baseDelayMs ?? 0) / 1000);
  const jitterSeconds = Math.round(Number(record.jitterMs ?? 0) / 1000);
  const strategy = String(record.strategy ?? "exponential");
  const external = record.externalProvider429 === true ? " externalProvider429=true" : "";
  appendOutput(task, "system", `retry backoff ${delaySeconds}s (base=${baseSeconds}s jitter=${jitterSeconds}s strategy=${strategy}${external}) before ${continuation}\n`, "queue/backoff");
}

async function withTimeout<T>(promise: Promise<T>, timeoutMs: number, message: string): Promise<T> {
  let timer: ReturnType<typeof setTimeout> | null = null;
  const timeout = new Promise<never>((_, reject) => {
    timer = setTimeout(() => reject(new Error(message)), timeoutMs);
  });
  try {
    return await Promise.race([promise, timeout]);
  } finally {
    if (timer !== null) clearTimeout(timer);
  }
}

function fallbackJudgeFromWatchdog(task: QueueTask, result: CodexRunResult, error: unknown, startedAt: number): JudgeResult {
  const durationMs = Date.now() - startedAt;
  const message = error instanceof Error ? error.message : String(error);
  const detail: JudgeFailureDetails = {
    provider: "minimax",
    stage: "unknown",
    model: config.minimaxModel,
    apiBase: config.minimaxApiBase,
    occurredAt: nowIso(),
    durationMs,
    timeoutMs: config.judgeTimeoutMs,
    timedOut: durationMs >= config.judgeTimeoutMs,
    errorName: error instanceof Error ? error.name : "Error",
    errorMessage: message,
  };
  logger("warn", "judge_watchdog_fallback", {
    taskId: task.id,
    attempt: task.currentAttempt,
    durationMs,
    timeoutMs: config.judgeTimeoutMs,
    error: safePreview(message, 1000),
  });
  return {
    decision: result.terminalStatus === "completed" ? "complete" : "retry",
    confidence: 0.3,
    reason: `MiniMax judge watchdog fallback: ${message}`,
    source: "fallback",
    failureDetails: detail,
    raw: { minimaxFailure: detail as unknown as JsonValue },
  };
}

async function judgeTaskWithWatchdog(task: QueueTask, result: CodexRunResult): Promise<JudgeResult> {
  const startedAt = Date.now();
  try {
    return await withTimeout(
      judgeTask(task, result),
      config.judgeTimeoutMs + 5_000,
      `judge task timed out after ${config.judgeTimeoutMs + 5_000}ms`,
    );
  } catch (error) {
    return fallbackJudgeFromWatchdog(task, result, error, startedAt);
  }
}

async function sleepForRetryBackoff(task: QueueTask, delayMs: number): Promise<void> {
  let remaining = delayMs;
  while (remaining > 0 && !task.cancelRequested && !shutdownRequested) {
    const chunk = Math.min(1000, remaining);
    await Bun.sleep(chunk);
    remaining -= chunk;
  }
}

function taskHasLocalExecutionRecoveryClaim(task: QueueTask): boolean {
  if (task.status === "judging") return true;
  if (activeRunForTask(task) !== null) return true;
  return activeRunSlotReservations.has(queueIdOf(task));
}

function activeTaskHasLocalRunSlotOrWaiter(task: QueueTask): boolean {
  return activeRunForTask(task) !== null
    || activeRunSlotWaiters.some((waiter) => waiter.taskId === task.id);
}

function orphanedActiveTaskAgeMs(task: QueueTask): number {
  const updatedAt = timestampMs(task.updatedAt) ?? timestampMs(task.startedAt) ?? timestampMs(task.createdAt) ?? Date.now();
  return Math.max(0, Date.now() - updatedAt);
}

function taskIsRecoverableOrphanedActive(task: QueueTask, staleMs = orphanedActiveTaskRecoveryStaleMs): boolean {
  if (task.status !== "running" && task.status !== "judging") return false;
  if (activeTaskHasLocalRunSlotOrWaiter(task)) return false;
  const effectiveStaleMs = normalizeSchedulerHeartbeatStaleMs(staleMs, {
    min: orphanedActiveTaskRecoveryStaleMs,
    max: schedulerHeartbeatStaleMsMax,
  });
  const decision = staleRecoveryCandidate({ task, localActive: false, heartbeatStaleMs: effectiveStaleMs, now: nowIso() });
  return decision.allowed;
}

function recoverableOrphanedActiveTaskIds(tasks: QueueTask[] = state.tasks, staleMs = orphanedActiveTaskRecoveryStaleMs, filterIds?: Set<string>): string[] {
  return tasks
    .filter((task) => filterIds === undefined || filterIds.has(task.id))
    .filter((task) => taskIsRecoverableOrphanedActive(task, staleMs))
    .map((task) => task.id)
    .sort();
}

function schedulerReconcileCandidateTasks(databaseTasks: QueueTask[]): QueueTask[] {
  const mergedById = new Map<string, QueueTask>();
  for (const task of state.tasks) mergedById.set(task.id, task);
  for (const task of databaseTasks) {
    if (task.status !== "queued" && task.status !== "retry_wait" && task.status !== "running" && task.status !== "judging") continue;
    const existing = mergedById.get(task.id);
    if (existing === undefined) {
      mergedById.set(task.id, task);
      continue;
    }
    const existingUpdatedAt = timestampMs(existing.updatedAt) ?? 0;
    const taskUpdatedAt = timestampMs(task.updatedAt) ?? 0;
    if (taskUpdatedAt > existingUpdatedAt && !activeRunForTask(existing)) mergedById.set(task.id, task);
  }
  return Array.from(mergedById.values());
}

function schedulerReconcileStatus(tasks: QueueTask[] = state.tasks): JsonValue {
  const orphanedActiveTasks = tasks
    .filter((task) => (task.status === "running" || task.status === "judging") && !activeTaskHasLocalRunSlotOrWaiter(task))
    .sort((left, right) => left.id.localeCompare(right.id));
  const recoverableOrphanedActiveTasks = orphanedActiveTasks
    .filter((task) => taskIsRecoverableOrphanedActive(task))
    .sort((left, right) => left.id.localeCompare(right.id));
  const retryWaitTaskIds = tasks
    .filter((task) => task.status === "retry_wait")
    .map((task) => task.id)
    .sort();
  return {
    ok: recoverableOrphanedActiveTasks.length === 0,
    schedulerEnabled: config.schedulerEnabled,
    pollIntervalMs: config.schedulerPollIntervalMs,
    staleActiveRecoveryStaleMs: orphanedActiveTaskRecoveryStaleMs,
    orphanedActiveTaskCount: orphanedActiveTasks.length,
    orphanedActiveTaskIds: orphanedActiveTasks.map((task) => task.id),
    recoverableOrphanedActiveTaskCount: recoverableOrphanedActiveTasks.length,
    recoverableOrphanedActiveTaskIds: recoverableOrphanedActiveTasks.map((task) => task.id),
    retryWaitTaskCount: retryWaitTaskIds.length,
    retryWaitTaskIds,
    lastRunAt: schedulerReconcileAudit.lastRunAt,
    lastReason: schedulerReconcileAudit.lastReason,
    lastLoaded: schedulerReconcileAudit.lastLoaded,
    lastChanged: schedulerReconcileAudit.lastChanged,
    lastRecovered: schedulerReconcileAudit.lastRecovered,
    lastError: schedulerReconcileAudit.lastError,
  };
}

function clearInactiveActiveTurnIds(): number {
  let changed = 0;
  for (const task of state.tasks) {
    if (taskCanHaveActiveTurn(task.status) || task.activeTurnId === null) continue;
    task.activeTurnId = null;
    task.updatedAt = nowIso();
    markTaskDirty(task.id);
    changed += 1;
  }
  return changed;
}

function queueActiveTasksForRestartRetry(reason: string, method: string, options: { onlyActiveRuns?: boolean; taskIds?: Set<string> } = {}): number {
  if (!config.schedulerEnabled || !serviceRoleAllowsScheduler(config.serviceRole)) return 0;
  let recovered = 0;
  const recoveredTaskIds: string[] = [];
  for (const task of state.tasks) {
    if (task.status !== "running" && task.status !== "judging") continue;
    if (options.taskIds !== undefined && !options.taskIds.has(task.id)) continue;
    if (options.onlyActiveRuns === true && !taskHasLocalExecutionRecoveryClaim(task)) continue;
    task.status = "retry_wait";
    task.finishedAt = null;
    task.readAt = null;
    task.activeTurnId = null;
    task.lastError = reason;
    task.nextMode = "retry";
    task.nextPrompt = queueRecoveryRetryPrompt(task, reason);
    if (!reserveNextAttemptBudget(task)) {
      task.status = "failed";
      task.finishedAt = nowIso();
      task.nextMode = null;
      task.nextPrompt = null;
      task.lastError = `Max attempts reached (${maxTaskAttempts}) before restart recovery. ${reason}`;
      appendOutput(task, "error", `${task.lastError}\n`, method);
      recoveredTaskIds.push(task.id);
      continue;
    }
    setAttemptFeedbackPrompt(task.attempts.at(-1), task.nextPrompt, "queue-recovery-retry", task.attempts.length + 1);
    task.updatedAt = nowIso();
    appendOutput(task, "system", `${reason}; task queued for retry\n`, method);
    recoveredTaskIds.push(task.id);
    recovered += 1;
  }
  for (const taskId of recoveredTaskIds) markTaskDirty(taskId);
  if (recovered > 0) armIdleNotification();
  return recovered;
}

async function recoverOrphanedActiveTasks(reason: string, method: string, options: { taskIds?: Set<string>; staleMs?: number } = {}): Promise<number> {
  schedulerReconcileAudit.lastRunAt = nowIso();
  schedulerReconcileAudit.lastReason = reason;
  const staleMs = normalizeSchedulerHeartbeatStaleMs(options.staleMs, {
    min: orphanedActiveTaskRecoveryStaleMs,
    max: schedulerHeartbeatStaleMsMax,
  });
  const staleOrphanTaskIds = new Set(recoverableOrphanedActiveTaskIds(state.tasks, staleMs, options.taskIds));
  if (staleOrphanTaskIds.size === 0) {
    schedulerReconcileAudit.lastRecovered = 0;
    schedulerReconcileAudit.lastError = null;
    return 0;
  }
  const recovered = queueActiveTasksForRestartRetry(reason, method, { taskIds: staleOrphanTaskIds });
  schedulerReconcileAudit.lastRecovered = recovered;
  if (recovered === 0) return 0;
  logger("warn", "scheduler_orphaned_active_tasks_requeued", {
    reason,
    recovered,
    taskIds: Array.from(staleOrphanTaskIds).sort(),
    dirtyTaskCount: dirtyDatabaseTaskIds.size,
  });
  await flushDirtyTasksToDatabase(true);
  schedulerReconcileAudit.lastError = null;
  logger("info", "scheduler_orphaned_active_tasks_flushed", { reason, recovered });
  scheduleQueue();
  return recovered;
}

function schedulerRecoveryStaleMsFromUrl(url: URL): {
  staleMs: number;
  requestedStaleMs: number | null;
  requestedStaleMsRaw: string | null;
  staleMsAdjusted: boolean;
  staleMsAdjustmentReason: "default" | "accepted" | "invalid" | "below-safe-minimum" | "above-maximum";
  minStaleMs: number;
  maxStaleMs: number;
} {
  const requestedStaleMsRaw = url.searchParams.get("staleMs");
  const minStaleMs = orphanedActiveTaskRecoveryStaleMs;
  const maxStaleMs = schedulerHeartbeatStaleMsMax;
  if (requestedStaleMsRaw === null || requestedStaleMsRaw.trim().length === 0) {
    return {
      staleMs: minStaleMs,
      requestedStaleMs: null,
      requestedStaleMsRaw,
      staleMsAdjusted: false,
      staleMsAdjustmentReason: "default",
      minStaleMs,
      maxStaleMs,
    };
  }
  const requestedStaleMs = Number(requestedStaleMsRaw);
  const staleMs = normalizeSchedulerHeartbeatStaleMs(requestedStaleMs, { min: minStaleMs, max: maxStaleMs });
  const requestedFloor = Math.floor(requestedStaleMs);
  const staleMsAdjustmentReason = !Number.isFinite(requestedStaleMs)
    ? "invalid"
    : requestedFloor < minStaleMs
      ? "below-safe-minimum"
      : requestedFloor > maxStaleMs
        ? "above-maximum"
        : "accepted";
  return {
    staleMs,
    requestedStaleMs: Number.isFinite(requestedStaleMs) ? requestedStaleMs : null,
    requestedStaleMsRaw,
    staleMsAdjusted: staleMsAdjustmentReason !== "accepted",
    staleMsAdjustmentReason,
    minStaleMs,
    maxStaleMs,
  };
}

async function reconcileStaleActiveTasksFromApi(url: URL, req: Request): Promise<Response> {
  if (!serviceRoleAllowsScheduler(config.serviceRole)) return schedulerOnlyRejectResponse(req.method, "/api/scheduler/reconcile");
  const notReady = requireDatabaseReadyForWrite(req.method, "/api/scheduler/reconcile");
  if (notReady !== null) return notReady;
  const recoverRequested = req.method === "POST" && truthyParam(url, "recover");
  const dryRun = !recoverRequested || truthyParam(url, "dryRun");
  const taskIds = boundedStringSetParam(url, "taskId");
  const staleThreshold = schedulerRecoveryStaleMsFromUrl(url);
  const staleMs = staleThreshold.staleMs;
  let loaded = 0;
  let candidateTasks = state.tasks;
  if (dryRun) {
    const databaseTasks = await loadTasksFromDatabase("hot");
    loaded = databaseTasks.length;
    candidateTasks = schedulerReconcileCandidateTasks(databaseTasks);
  } else {
    await refreshSchedulerTasksFromDatabase("api-reconcile", { recover: false });
    loaded = schedulerReconcileAudit.lastLoaded;
  }
  const candidateTaskIds = recoverableOrphanedActiveTaskIds(candidateTasks, staleMs, taskIds);
  let recovered = 0;
  if (!dryRun && candidateTaskIds.length > 0) {
    recovered = await recoverOrphanedActiveTasks("Scheduler API recovered expired database active task without local run", "scheduler/api-reconcile-recovery", { taskIds: new Set(candidateTaskIds), staleMs });
  }
  return jsonResponse({
    ok: true,
    dryRun,
    recoverRequested,
    loaded,
    staleMs,
    requestedStaleMs: staleThreshold.requestedStaleMs,
    requestedStaleMsRaw: staleThreshold.requestedStaleMsRaw,
    staleMsAdjusted: staleThreshold.staleMsAdjusted,
    staleMsAdjustmentReason: staleThreshold.staleMsAdjustmentReason,
    minStaleMs: staleThreshold.minStaleMs,
    maxStaleMs: staleThreshold.maxStaleMs,
    candidateTaskIds,
    candidateCount: candidateTaskIds.length,
    recovered,
    reconcile: schedulerReconcileStatus(dryRun ? candidateTasks : state.tasks),
    executionDiagnostics: executionDiagnosticsForTasks(dryRun ? candidateTasks : state.tasks),
    oaEventPublisher: oaEventPublisherStatus(),
  });
}

async function runSchedulerStaleActiveRecoverySelfTest(): Promise<JsonValue> {
  const staleAt = new Date(Date.now() - orphanedActiveTaskRecoveryStaleMs - 60_000).toISOString();
  const createdAt = new Date(Date.now() - orphanedActiveTaskRecoveryStaleMs - 120_000).toISOString();
  const task = normalizeTask({
    id: "codex_stale_active_recovery_self_test",
    queueId: "stale_active_recovery_self_test",
    queueEnteredAt: createdAt,
    prompt: "stale active recovery self-test",
    basePrompt: "stale active recovery self-test",
    referenceTaskIds: [],
    referenceInjection: null,
    providerId: config.mainProviderId,
    cwd: config.defaultWorkdir,
    model: config.defaultModel,
    reasoningEffort: resolveReasoningEffort(config.defaultModel, config.defaultReasoningEffort),
    executionMode: "default",
    maxAttempts: 1,
    status: "running",
    createdAt,
    updatedAt: staleAt,
    startedAt: createdAt,
    finishedAt: null,
    readAt: null,
    currentAttempt: 1,
    currentMode: "initial",
    codexThreadId: "thread_stale_active_recovery_self_test",
    activeTurnId: "turn_stale_active_recovery_self_test",
    finalResponse: "",
    lastError: null,
    lastJudge: null,
    judgeFailCount: 0,
    promptHistory: [],
    output: [],
    events: [],
    attempts: [],
    cancelRequested: false,
    nextPrompt: null,
    nextMode: null,
  });
  task.schedulerHeartbeat = {
    taskId: task.id,
    queueId: task.queueId,
    attempt: 1,
    activeTurnId: task.activeTurnId,
    codexThreadId: task.codexThreadId,
    owner: "self-test",
    schedulerInstance: "self-test",
    executionPlane: "scheduler-execution-plane",
    agentPort: "codex",
    status: task.status,
    lastLocalHeartbeatAt: staleAt,
    lastObservedAgentEventAt: staleAt,
    lastPersistedTraceAt: staleAt,
    outputMaxSeq: 0,
    source: "scheduler",
  };
  const before = state.tasks.slice();
  try {
    state.tasks.push(task);
    const candidatesBefore = recoverableOrphanedActiveTaskIds(state.tasks, 0, new Set([task.id]));
    const recovered = queueActiveTasksForRestartRetry("stale active recovery self-test", "self-test/stale-active-recovery", { taskIds: new Set(candidatesBefore) });
    const candidatesAfter = recoverableOrphanedActiveTaskIds(state.tasks, 0, new Set([task.id]));
    if (!candidatesBefore.includes(task.id)) throw new Error("expired active self-test task was not recovery-eligible");
    if (recovered !== 1) throw new Error(`expected one recovered stale active task, got ${recovered}`);
    if (task.status !== "retry_wait") throw new Error(`expected retry_wait after recovery, got ${task.status}`);
    if (task.activeTurnId !== null) throw new Error("recovery should clear activeTurnId");
    if (candidatesAfter.length !== 0) throw new Error("recovered self-test task should no longer be stale-active eligible");
    return {
      ok: true,
      taskId: task.id,
      recovered,
      status: task.status,
      candidatesBefore,
      candidatesAfter,
      lastError: task.lastError,
      nextMode: task.nextMode,
    } as unknown as JsonValue;
  } finally {
    state.tasks.splice(0, state.tasks.length, ...before);
    dirtyDatabaseTaskIds.delete(task.id);
  }
}

function failTaskForFallbackRetryLimit(task: QueueTask, judge: JudgeResult | null): void {
  const count = fallbackJudgeRetryCount(task);
  const reason = `Fallback/non-LLM judge retry limit reached (${count}/${fallbackJudgeRetryLimit}). ${judge?.reason ?? task.lastJudge?.reason ?? "MiniMax judge was unavailable."}`;
  task.status = "failed";
  task.finishedAt = nowIso();
  task.updatedAt = task.finishedAt;
  task.activeTurnId = null;
  task.nextPrompt = null;
  task.nextMode = null;
  task.lastError = safePreview(reason, 2000);
  appendOutput(task, "error", `${reason}\n`, "queue");
  persistTaskState(task);
  logger("warn", "task_failed_by_fallback_retry_limit", {
    taskId: task.id,
    fallbackRetryCount: count,
    fallbackJudgeRetryLimit,
    reason: safePreview(reason, 500),
  });
  void notifyTaskTerminal(task);
}

async function runTask(task: QueueTask): Promise<void> {
  const claimQueueId = queueIdOf(task);
  logger("info", "task_processor_start", { taskId: task.id, queueId: claimQueueId, providerId: task.providerId, executionMode: task.executionMode, cwd: task.cwd, maxAttempts: task.maxAttempts, model: task.model, agentPort: codeAgentPortForModel(task.model), promptPreview: safePreview(task.prompt, 240) });
  const skillsBlocker = runnerSkillsBlocker();
  if (skillsBlocker !== null) {
    const finishedAt = nowIso();
    const message = String(skillsBlocker.message);
    task.status = "failed";
    task.startedAt ??= finishedAt;
    task.finishedAt = finishedAt;
    task.updatedAt = finishedAt;
    task.activeTurnId = null;
    task.lastError = message;
    appendOutput(task, "error", `${message}\n${JSON.stringify(skillsBlocker)}\n`, "runner-skills/blocker");
    task.attempts.push({
      index: task.attempts.length + 1,
      mode: task.nextMode ?? "initial",
      startedAt: finishedAt,
      finishedAt,
      providerId: task.providerId,
      executionMode: task.executionMode,
      terminalStatus: "failed",
      transportClosedBeforeTerminal: false,
      appServerExitCode: null,
      appServerSignal: null,
      error: message,
      events: [],
      finalResponse: "",
      finalResponsePreview: "",
      finalResponseChars: 0,
      judge: null,
      judgeAt: null,
      judgeSeq: null,
      stderrTail: "",
      runnerErrorClassification: skillsBlocker as JsonValue,
    });
    persistTaskState(task);
    logger("error", "task_blocked_by_runner_skills", { taskId: task.id, queueId: claimQueueId, skills: skillsBlocker as JsonValue });
    void notifyTaskTerminal(task);
    return;
  }
  if (
    task.status === "retry_wait"
    && task.lastJudge?.source === "fallback"
    && task.lastJudge.decision === "retry"
    && !fallbackJudgeIsProviderTransient(task.lastJudge)
    && fallbackJudgeRetryCount(task) >= fallbackJudgeRetryLimit
  ) {
    failTaskForFallbackRetryLimit(task, task.lastJudge);
    return;
  }
  armIdleNotification();
  task.maxAttempts = clampTaskAttempts(task.maxAttempts || config.defaultMaxAttempts);
  task.startedAt ??= nowIso();
  task.lastError = null;
  while (task.attempts.length < task.maxAttempts && !task.cancelRequested && !shutdownRequested) {
    const mode = task.nextMode ?? (task.attempts.length === 0 ? "initial" : "retry");
    const rawPrompt = task.nextPrompt ?? task.prompt;
    const needsFreshRecoveryPrompt = mode === "retry" && task.codexThreadId === null && task.attempts.length > 0;
    const recoveryPrompt = needsFreshRecoveryPrompt
      ? codeAgentPortForModel(task.model) === "opencode"
        ? openCodeFreshRecoveryPrompt(task, rawPrompt, "retry_wait task has no persisted OpenCode session id")
        : codexFreshRecoveryPrompt(task, rawPrompt, "retry_wait task has no persisted Codex thread id")
      : rawPrompt;
    const prompt = promptWithCodeQueueEnvironmentHint(recoveryPrompt);
    if (needsFreshRecoveryPrompt) {
      appendOutput(task, "system", "retry has no persisted thread/session id; starting a fresh agent thread with compact recovery context\n", "thread/recovery");
    }
    const releaseRunSlot = await acquireActiveRunSlot(task);
    if (releaseRunSlot === null) break;
    const startedAt = nowIso();
    task.currentAttempt = task.attempts.length + 1;
    task.currentMode = mode;
    task.status = "running";
    task.readAt = null;
    task.finishedAt = null;
    task.updatedAt = startedAt;
    refreshSchedulerHeartbeat(task);
    let claimed = false;
    try {
      claimed = await claimTaskInDatabase(task, claimQueueId);
    } catch (error) {
      releaseRunSlot();
      task.status = "retry_wait";
      task.activeTurnId = null;
      task.finishedAt = null;
      task.readAt = null;
      task.updatedAt = nowIso();
      task.lastError = `Database claim failed before agent start: ${error instanceof Error ? error.message : String(error)}`;
      appendOutput(task, "error", `${task.lastError}\n`, "database/claim");
      persistTaskState(task);
      logger("error", "task_claim_failed", { taskId: task.id, queueId: claimQueueId, error: errorToJson(error) });
      await sleepForRetryBackoff(task, retryBackoffMs(Math.max(1, task.attempts.length)));
      return;
    }
    if (!claimed) {
      releaseRunSlot();
      await recoverOrphanedActiveTasks("Database claim found stale active task without local run", "database/claim-recovery", { taskIds: new Set([task.id]), staleMs: 0 });
      return;
    }
    publishTaskOaEvent(task, "claim");
    logger("info", "task_run_start", { taskId: task.id, queueId: queueIdOf(task), attempt: task.currentAttempt, mode, providerId: task.providerId, executionMode: task.executionMode, cwd: task.cwd, maxAttempts: task.maxAttempts, model: task.model, agentPort: codeAgentPortForModel(task.model), freshRecovery: needsFreshRecoveryPrompt });
    const attemptStartOutput = appendOutput(task, "system", `attempt ${task.currentAttempt}/${task.maxAttempts} queue=${queueIdOf(task)} provider=${task.providerId} executionMode=${task.executionMode} cwd=${task.cwd} mode=${mode} model=${task.model} port=${codeAgentPortForModel(task.model)}\n`, "queue");

    let result: CodexRunResult;
    try {
      result = await runCodeAgentTurn(task, prompt);
    } finally {
      releaseRunSlot();
    }
    const finishedAt = nowIso();
    task.finalResponse = result.finalResponse || task.finalResponse;
    task.attempts.push(attemptFromResult(task, mode, startedAt, finishedAt, result, attemptStartOutput?.seq ?? null, taskOutputMaxSeq(task), recoveryPrompt));
    task.status = "judging";
    task.activeTurnId = null;
    task.updatedAt = nowIso();
    persistTaskState(task);

    if (task.cancelRequested) break;
    const judge = await judgeTaskWithWatchdog(task, result);
    task.lastJudge = judge;
    const judgeOutput = appendOutput(task, judge.decision === "complete" ? "system" : judge.decision === "fail" ? "error" : "system", `judge=${judge.decision} confidence=${judge.confidence.toFixed(2)} source=${judge.source}: ${judge.reason}${judgeFailureDetailsForOutput(judge)}\n`, "judge");
    const latestAttempt = task.attempts.at(-1);
    if (latestAttempt !== undefined && latestAttempt.index === task.currentAttempt) {
      latestAttempt.judge = judge;
      latestAttempt.judgeAt = judgeOutput?.at ?? nowIso();
      latestAttempt.judgeSeq = judgeOutput?.seq ?? null;
      latestAttempt.outputEndSeq = judgeOutput?.seq ?? latestAttempt.outputEndSeq ?? null;
    }
    logger("info", "task_judged", { taskId: task.id, attempt: task.currentAttempt, decision: judge.decision, confidence: judge.confidence, source: judge.source, reason: safePreview(judge.reason, 500), failureDetails: judge.failureDetails ?? null } as unknown as JsonValue);

    if (judge.decision === "complete") {
      task.status = "succeeded";
      task.finishedAt = nowIso();
      task.activeTurnId = null;
      task.nextPrompt = null;
      task.nextMode = null;
      persistTaskState(task);
      logger("info", "task_succeeded", { taskId: task.id, attempts: task.attempts.length });
      void notifyTaskTerminal(task);
      return;
    }
    if (judge.decision === "fail") {
      task.judgeFailCount += 1;
      if (!explicitUserInterrupt(task, result) && task.judgeFailCount < judgeFailRetryLimit && reserveNextAttemptBudget(task)) {
        task.status = "retry_wait";
        task.finishedAt = null;
        task.readAt = null;
        const nextPrompt = judgeFailContinuationPrompt(task, judge, task.judgeFailCount);
        task.nextPrompt = nextPrompt;
        task.nextMode = "retry";
        clearStaleBadResumeThread(task);
        setAttemptFeedbackPrompt(latestAttempt, nextPrompt, "judge-fail-retry", task.attempts.length + 1);
        task.updatedAt = nowIso();
        appendOutput(task, "system", `judge=fail treated as retry (${task.judgeFailCount}/${judgeFailRetryLimit}); appending continuation prompt to existing session\n`, "queue");
        persistTaskState(task);
        await flushDirtyTasksToDatabase(true);
        logger("warn", "task_judge_fail_treated_as_retry", {
          taskId: task.id,
          attempt: task.currentAttempt,
          judgeFailCount: task.judgeFailCount,
          judgeFailRetryLimit,
          reason: safePreview(judge.reason, 500),
        });
        const backoff = retryBackoffEvidence(task, task.attempts.length, "judge-fail-treated-as-retry");
        appendRetryBackoffEvidence(task, backoff, "appending continuation to existing session");
        logger("info", "task_retry_backoff", { taskId: task.id, attempt: task.currentAttempt, backoff });
        await sleepForRetryBackoff(task, Number((backoff as Record<string, JsonValue>).delayMs ?? 0));
        continue;
      }
      task.status = "failed";
      task.finishedAt = nowIso();
      task.updatedAt = task.finishedAt;
      task.activeTurnId = null;
      task.lastError = judge.reason;
      appendOutput(task, "system", `judge=fail reached terminal threshold (${task.judgeFailCount}/${judgeFailRetryLimit}); queue will continue to the next queued task\n`, "queue");
      persistTaskState(task);
      logger("warn", "task_failed_by_judge_queue_continues", { taskId: task.id, judgeFailCount: task.judgeFailCount, judgeFailRetryLimit, reason: safePreview(judge.reason, 500) });
      void notifyTaskTerminal(task);
      return;
    }
    if (fallbackJudgeIsProviderTransient(judge)) {
      task.status = "retry_wait";
      task.finishedAt = null;
      task.readAt = null;
      task.nextPrompt = retryPrompt(task, judge);
      task.nextMode = "retry";
      setAttemptFeedbackPrompt(latestAttempt, task.nextPrompt, "judge-provider-transient-retry", task.attempts.length + 1);
      task.updatedAt = nowIso();
      const backoff = judgeProviderBackoffEvidence(task);
      const providerDelayMs = Number((backoff as Record<string, JsonValue>).delayMs ?? 0);
      appendRetryBackoffEvidence(task, backoff, "continuing existing session after judge provider transient failure");
      persistTaskState(task);
      await flushDirtyTasksToDatabase(true);
      logger("warn", "task_judge_provider_transient_retry", {
        taskId: task.id,
        fallbackRetryCount: fallbackJudgeRetryCount(task),
        delayMs: providerDelayMs,
        backoff,
        reason: safePreview(judge.reason, 500),
      });
      await sleepForRetryBackoff(task, providerDelayMs);
      continue;
    }
    if (judge.source === "fallback" && fallbackJudgeRetryCount(task) >= fallbackJudgeRetryLimit) {
      failTaskForFallbackRetryLimit(task, judge);
      return;
    }
    if (task.attempts.length >= task.maxAttempts) break;
    task.status = "retry_wait";
    task.finishedAt = null;
    task.readAt = null;
    const nextPrompt = retryPrompt(task, judge);
    task.nextPrompt = nextPrompt;
    task.nextMode = "retry";
    clearStaleBadResumeThread(task);
    setAttemptFeedbackPrompt(latestAttempt, nextPrompt, judge.continuePrompt?.trim() ? "judge-continue-prompt" : "judge-retry-generated", task.attempts.length + 1);
    task.updatedAt = nowIso();
    persistTaskState(task);
    await flushDirtyTasksToDatabase(true);
    const externalProvider429 = latestAttemptExternalProvider429(task);
    const backoff = retryBackoffEvidence(task, task.attempts.length, externalProvider429 ? "external-provider-429" : "judge-retry");
    appendRetryBackoffEvidence(task, backoff, "appending continuation to existing session");
    logger(externalProvider429 ? "warn" : "info", "task_retry_backoff", { taskId: task.id, attempt: task.currentAttempt, backoff });
    await sleepForRetryBackoff(task, Number((backoff as Record<string, JsonValue>).delayMs ?? 0));
  }

  if (shutdownRequested) {
    if (task.status === "running" || task.status === "judging") {
      queueActiveTasksForRestartRetry("Service stopping while task was active", "shutdown");
      persistState();
    }
    return;
  }

  if (task.cancelRequested) {
    task.status = "canceled";
    task.finishedAt = nowIso();
    task.lastError = "Task canceled by request.";
  } else {
    task.status = "failed";
    task.finishedAt = nowIso();
    task.lastError = `Max attempts reached (${task.maxAttempts}).`;
  }
  task.activeTurnId = null;
  persistTaskState(task);
  logger(task.status === "canceled" ? "warn" : "error", "task_terminal", { taskId: task.id, status: task.status, attempts: task.attempts.length, error: task.lastError ?? "" });
  void notifyTaskTerminal(task);
}

function updateProcessingFlag(): void {
  processing = processingQueues.size > 0;
}

function activeRunSlotQueueIds(): string[] {
  return Array.from(new Set([
    ...Array.from(activeRuns.keys()),
    ...Array.from(activeRunSlotReservations),
  ])).sort((left, right) => left.localeCompare(right));
}

function activeRunSlotCount(): number {
  return activeRunSlotQueueIds().length;
}

function enqueueActiveRunSlotWaiter(task: QueueTask): ActiveRunSlotWaiter {
  const waiter = {
    id: nextActiveRunSlotWaiterId,
    taskId: task.id,
    queueId: queueIdOf(task),
    enqueuedAt: nowIso(),
  };
  nextActiveRunSlotWaiterId += 1;
  activeRunSlotWaiters.push(waiter);
  return waiter;
}

function removeActiveRunSlotWaiter(waiter: ActiveRunSlotWaiter): void {
  const index = activeRunSlotWaiters.findIndex((item) => item.id === waiter.id);
  if (index >= 0) activeRunSlotWaiters.splice(index, 1);
}

function firstActiveRunSlotWaiter(): ActiveRunSlotWaiter | null {
  return activeRunSlotWaiters[0] ?? null;
}

function activeRunSlotWaiterSummaries(): JsonValue[] {
  return activeRunSlotWaiters.map((waiter, index) => ({
    position: index + 1,
    taskId: waiter.taskId,
    queueId: waiter.queueId,
    enqueuedAt: waiter.enqueuedAt,
  }));
}

async function acquireActiveRunSlot(task: QueueTask): Promise<(() => void) | null> {
  const queueId = queueIdOf(task);
  const waiter = enqueueActiveRunSlotWaiter(task);
  let lastWaitLogAt = 0;
  try {
    while (!shutdownRequested && !task.cancelRequested && queueTaskIsRunnable(task)) {
      const memoryPressure = activeRunMemoryPressure();
      const atHead = firstActiveRunSlotWaiter()?.id === waiter.id;
      if (atHead && availableQueueStartSlots() > 0 && memoryPressure === null) {
        removeActiveRunSlotWaiter(waiter);
        activeRunSlotReservations.add(queueId);
        let released = false;
        return () => {
          if (released) return;
          released = true;
          activeRunSlotReservations.delete(queueId);
          if (!shutdownRequested) scheduleQueue();
        };
      }
      if (Date.now() - lastWaitLogAt > 30_000) {
        lastWaitLogAt = Date.now();
        const head = firstActiveRunSlotWaiter();
        logger(memoryPressure === null ? "info" : "warn", "active_run_slot_waiting", {
          taskId: task.id,
          queueId,
          waitPosition: activeRunSlotWaiters.findIndex((item) => item.id === waiter.id) + 1,
          headTaskId: head?.taskId ?? null,
          headQueueId: head?.queueId ?? null,
          currentBytes: memoryPressure?.currentBytes ?? null,
          inactiveFileBytes: memoryPressure?.inactiveFileBytes ?? null,
          workingSetBytes: memoryPressure?.workingSetBytes ?? null,
          swapCurrentBytes: memoryPressure?.swapCurrentBytes ?? null,
          swapMaxBytes: memoryPressure?.swapMaxBytes ?? null,
          thresholdBytes: memoryPressure?.thresholdBytes ?? memoryWatchdogThreshold(),
          activeRunSlotCount: activeRunSlotCount(),
        });
      }
      await Bun.sleep(memoryPressure === null ? 500 : 2000);
    }
    return null;
  } finally {
    removeActiveRunSlotWaiter(waiter);
  }
}

function taskQueueOrderMs(task: QueueTask): number {
  return timestampMs(taskQueueEnteredAt(task)) ?? timestampMs(task.createdAt) ?? timestampMs(task.updatedAt) ?? 0;
}

function compareTaskQueueOrder(left: QueueTask, right: QueueTask): number {
  const queueDelta = taskQueueOrderMs(left) - taskQueueOrderMs(right);
  if (queueDelta !== 0) return queueDelta;
  const createdDelta = (timestampMs(left.createdAt) ?? 0) - (timestampMs(right.createdAt) ?? 0);
  if (createdDelta !== 0) return createdDelta;
  return left.id.localeCompare(right.id);
}

function queueTaskIsRunnable(task: QueueTask): boolean {
  return task.status === "queued" || task.status === "retry_wait";
}

function queueTaskBlocksFollowing(task: QueueTask): boolean {
  return !terminalTask(task);
}

function queueTaskRows(queueId: string, tasks: QueueTask[] = state.tasks): QueueTask[] {
  return tasks
    .filter((task) => queueIdOf(task) === queueId)
    .sort(compareTaskQueueOrder);
}

function queueHeadTask(queueId: string, tasks: QueueTask[] = state.tasks): QueueTask | null {
  return queueTaskRows(queueId, tasks).find(queueTaskBlocksFollowing) ?? null;
}

function nextRunnableTaskFrom(queueId: string, tasks: QueueTask[] = state.tasks): QueueTask | null {
  const head = queueHeadTask(queueId, tasks);
  return head !== null && queueTaskIsRunnable(head) ? head : null;
}

function queueIdsForTasks(tasks: QueueTask[] = state.tasks): string[] {
  const ids = new Set<string>(tasks === state.tasks ? state.queues.map((queue) => queue.id) : []);
  for (const task of tasks) ids.add(queueIdOf(task));
  return Array.from(ids).sort((left, right) => left.localeCompare(right));
}

function runnableQueueIds(): string[] {
  return queueIdsForTasks().filter((queueId) => nextRunnableTaskFrom(queueId) !== null);
}

function availableQueueStartSlotsFor(activeSlotCount: number, maxActiveQueues = config.maxActiveQueues): number {
  if (maxActiveQueues <= 0) return Number.POSITIVE_INFINITY;
  return Math.max(0, maxActiveQueues - activeSlotCount);
}

function availableQueueStartSlots(): number {
  return availableQueueStartSlotsFor(activeRunSlotCount());
}

function queuedReason(code: string, label: string, message: string, extra: Omit<QueuedStatusReason, "code" | "label" | "message"> = {}): QueuedStatusReason {
  return { code, label, message, ...extra };
}

function memoryPressureReasonPayload(usage: CgroupMemoryUsage & { thresholdBytes: number }): JsonValue {
  return {
    currentBytes: usage.currentBytes,
    inactiveFileBytes: usage.inactiveFileBytes,
    workingSetBytes: usage.workingSetBytes,
    thresholdBytes: usage.thresholdBytes,
    swapCurrentBytes: usage.swapCurrentBytes,
    swapMaxBytes: usage.swapMaxBytes,
  };
}

function queuedStatusReason(task: QueueTask, tasks: QueueTask[] = state.tasks): QueuedStatusReason | null {
  if (task.status !== "queued") return null;
  const queueId = queueIdOf(task);
  const sourceTasks = tasks.some((item) => item.id === task.id) ? tasks : [...tasks, task];
  const rows = queueTaskRows(queueId, sourceTasks);
  const head = rows.find(queueTaskBlocksFollowing) ?? null;
  if (head !== null && head.id !== task.id) {
    return queuedReason(
      "prev_task",
      "PREV TASK",
      `Waiting for previous task ${head.id} in queue ${queueId} to finish first.`,
      { blockerTaskId: head.id, blockerQueueId: queueId },
    );
  }
  if (shutdownRequested) {
    return queuedReason("shutdown", "SHUTDOWN", "Code Queue is shutting down; queued work will resume after restart.");
  }
  if (!serviceReady) {
    return queuedReason("service", "SERVICE", "Code Queue is still starting and has not enabled scheduling yet.");
  }
  if (!config.schedulerEnabled && config.serviceRole !== "read") {
    return queuedReason("scheduler_disabled", "STANDBY", "This Code Queue instance is a k3s-managed standby and does not start queued work.");
  }
  if (mergingQueues.has(queueId)) {
    return queuedReason("queue_merge", "MERGING", "Queue merge is rewriting queue membership; scheduling will resume immediately after it finishes.");
  }

  const memoryPressure = activeRunMemoryPressure();
  if (memoryPressure !== null) {
    return queuedReason(
      "mem_limit",
      "MEM LIMIT",
      "Waiting for cgroup memory working set to fall below the configured start threshold.",
      { memory: memoryPressureReasonPayload(memoryPressure), activeRunSlotCount: activeRunSlotCount(), maxActiveQueues: config.maxActiveQueues },
    );
  }

  const waitPosition = activeRunSlotWaiters.findIndex((waiter) => waiter.taskId === task.id);
  if (config.maxActiveQueues > 0 && availableQueueStartSlots() <= 0) {
    return queuedReason(
      "active_limit",
      "ACTIVE LIMIT",
      `Waiting for an active run slot (${activeRunSlotCount()}/${config.maxActiveQueues} in use).`,
      { waitPosition: waitPosition >= 0 ? waitPosition + 1 : null, activeRunSlotCount: activeRunSlotCount(), maxActiveQueues: config.maxActiveQueues },
    );
  }
  if (waitPosition > 0) {
    const blocker = firstActiveRunSlotWaiter();
    return queuedReason(
      "slot_fifo",
      "SLOT FIFO",
      "Waiting behind an older runnable queue for the next active run slot.",
      { blockerTaskId: blocker?.taskId ?? null, blockerQueueId: blocker?.queueId ?? null, waitPosition: waitPosition + 1, activeRunSlotCount: activeRunSlotCount(), maxActiveQueues: config.maxActiveQueues },
    );
  }
  if (processingQueues.has(queueId) || waitPosition === 0) {
    return queuedReason(
      "starting",
      "STARTING",
      "Queue processor has selected this task and is starting the agent run.",
      { waitPosition: waitPosition >= 0 ? waitPosition + 1 : null, activeRunSlotCount: activeRunSlotCount(), maxActiveQueues: config.maxActiveQueues },
    );
  }
  return queuedReason("ready", "READY", "Task is the head of its queue and ready to start.");
}

function nextRunnableTask(queueId: string): QueueTask | null {
  return nextRunnableTaskFrom(queueId);
}

async function processQueue(queueId: string): Promise<void> {
  if (!serviceReady || !config.schedulerEnabled || processingQueues.has(queueId) || shutdownRequested) return;
  processingQueues.add(queueId);
  updateProcessingFlag();
  try {
    while (true) {
      if (shutdownRequested) break;
      const task = nextRunnableTask(queueId);
      if (task === null) break;
      try {
        await runTask(task);
      } catch (error) {
        const message = error instanceof Error ? error.stack ?? error.message : String(error);
        appendOutput(task, "error", `${message}\n`, "queue-loop");
        task.status = "failed";
        task.finishedAt = nowIso();
        task.activeTurnId = null;
        task.lastError = safePreview(message, 2000);
        task.updatedAt = nowIso();
        persistTaskState(task);
        logger("error", "task_failed_by_queue_exception", { taskId: task.id, error: safePreview(message, 1000) });
        void notifyTaskTerminal(task);
      }
    }
  } finally {
    processingQueues.delete(queueId);
    updateProcessingFlag();
    persistState();
    if (!shutdownRequested && nextRunnableTask(queueId) !== null) scheduleQueue(queueId);
    if (!shutdownRequested) scheduleQueue();
    void maybeNotifyQueueIdle().catch((error) => logger("warn", "claudeqq_idle_notify_schedule_failed", { error: errorToJson(error) }));
  }
}

function scheduleQueue(queueId?: string): void {
  if (!serviceReady || !config.schedulerEnabled || shutdownRequested) return;
  const ids = queueId === undefined ? runnableQueueIds() : [queueId];
  for (const id of ids) {
    if (mergingQueues.has(id)) continue;
    if (processingQueues.has(id)) continue;
    void processQueue(id).catch((error) => {
      logger("error", "queue_loop_failed", { queueId: id, error: error instanceof Error ? error.stack ?? error.message : String(error) });
      processingQueues.delete(id);
      activeRunSlotReservations.delete(id);
      for (let index = activeRunSlotWaiters.length - 1; index >= 0; index -= 1) {
        if (activeRunSlotWaiters[index]?.queueId === id) activeRunSlotWaiters.splice(index, 1);
      }
      updateProcessingFlag();
      const run = activeRuns.get(id);
      if (run !== undefined) {
        run.app.stop();
        activeRuns.delete(id);
      }
      void maybeNotifyQueueIdle().catch((idleError) => logger("warn", "claudeqq_idle_notify_schedule_failed", { error: errorToJson(idleError) }));
    });
  }
}

function hasRunnableTask(): boolean {
  return state.tasks.some((task) => task.status === "queued" || task.status === "retry_wait");
}

function shouldPollSchedulerDatabase(): boolean {
  return config.schedulerEnabled && serviceRoleAllowsScheduler(config.serviceRole);
}

function mergeSchedulerDatabaseTasks(tasks: QueueTask[]): number {
  let changed = 0;
  for (const task of tasks) {
    if (task.status !== "queued" && task.status !== "retry_wait" && task.status !== "running" && task.status !== "judging") continue;
    const existing = findTask(task.id);
    if (existing === null) {
      state.tasks.push(task);
      changed += 1;
      continue;
    }
    const existingUpdatedAt = timestampMs(existing.updatedAt) ?? 0;
    const taskUpdatedAt = timestampMs(task.updatedAt) ?? 0;
    if (taskUpdatedAt > existingUpdatedAt && !activeRunForTask(existing)) {
      Object.assign(existing, task);
      changed += 1;
    }
  }
  if (changed > 0) {
    state.tasks.sort((left, right) => (timestampMs(left.createdAt) ?? 0) - (timestampMs(right.createdAt) ?? 0) || left.id.localeCompare(right.id));
    updateNextSeqFromTasks();
  }
  return changed;
}

async function refreshSchedulerTasksFromDatabase(reason: string, options: { recover?: boolean } = {}): Promise<number> {
  if (!databaseReady || !config.schedulerEnabled) return 0;
  schedulerReconcileAudit.lastRunAt = nowIso();
  schedulerReconcileAudit.lastReason = reason;
  try {
    const tasks = await loadTasksFromDatabase("hot");
    const changed = mergeSchedulerDatabaseTasks(tasks);
    const recovered = options.recover === false ? 0 : await recoverOrphanedActiveTasks("Scheduler recovered database active task without local run", `scheduler/${reason}-recovery`);
    schedulerReconcileAudit.lastLoaded = tasks.length;
    schedulerReconcileAudit.lastChanged = changed;
    schedulerReconcileAudit.lastRecovered = recovered;
    schedulerReconcileAudit.lastError = null;
    if (changed > 0) {
      logger("info", "scheduler_database_hot_tasks_refreshed", { reason, changed, loaded: tasks.length });
      scheduleQueue();
    }
    return changed + recovered;
  } catch (error) {
    schedulerReconcileAudit.lastError = databaseErrorMessage(error);
    throw error;
  }
}

function startSchedulerDatabasePoller(): void {
  if (!shouldPollSchedulerDatabase()) return;
  const interval = setInterval(() => {
    if (!serviceReady || shutdownRequested) return;
    void refreshSchedulerTasksFromDatabase("poll").catch((error) => {
      setDatabaseLastError(error);
      logger("warn", "scheduler_database_poll_failed", { error: errorToJson(error) });
    });
  }, config.schedulerPollIntervalMs);
  interval.unref?.();
}

function activeRunForTask(task: QueueTask): ActiveRun | null {
  const queueRun = activeRuns.get(queueIdOf(task));
  if (queueRun?.taskId === task.id) return queueRun;
  return Array.from(activeRuns.values()).find((run) => run.taskId === task.id) ?? null;
}

function executionDiagnosticsForTasks(tasks: QueueTask[] = state.tasks): JsonValue {
  return buildExecutionDiagnostics({
    controlPlane: "D601-code-queue-scheduler",
    executionStateSource: "scheduler-execution-plane",
    tasks,
    activeRuns: activeRuns.values(),
    activeRunSlotCount: activeRunSlotCount(),
    activeQueueIds: activeRunSlotQueueIds(),
    processingQueueIds: Array.from(processingQueues),
    orphanedActiveTaskIds: tasks
      .filter((task) => (task.status === "running" || task.status === "judging") && activeRunForTask(task) === null)
      .map((task) => task.id),
    oaPublisher: oaEventPublisherStatus() as JsonValue,
    heartbeatStaleMs: schedulerHeartbeatStaleMs,
  }) as unknown as JsonValue;
}

function installShutdownHandlers(): void {
  const stop = (signal: NodeJS.Signals): void => {
    if (shutdownRequested) process.exit(0);
    shutdownRequested = true;
    const recovered = queueActiveTasksForRestartRetry("Service stopping while task was active", "shutdown", { onlyActiveRuns: true });
    for (const run of activeRuns.values()) run.app.stop();
    activeRuns.clear();
    activeRunSlotReservations.clear();
    activeRunSlotWaiters.splice(0, activeRunSlotWaiters.length);
    processingQueues.clear();
    updateProcessingFlag();
    persistState();
    logger("warn", "service_shutdown_requeued_active_tasks", { signal, recovered });
    const forceExit = setTimeout(() => process.exit(1), 8_000);
    forceExit.unref?.();
    void flushDirtyTasksToDatabase(true)
      .then(() => process.exit(0))
      .catch((error) => {
        logger("error", "service_shutdown_database_flush_failed", { signal, error: errorToJson(error) });
        process.exit(1);
      });
  };
  process.once("SIGTERM", stop);
  process.once("SIGINT", stop);
  process.on("uncaughtException", (error) => {
    if (isTransientDatabaseError(error)) {
      databaseTransientUncaughtSuppressedCount += 1;
      databaseLastTransientUncaughtSuppressedAt = nowIso();
      void rotateSqlClientAfterTransientError("uncaught-exception", error).catch((closeError) => logger("warn", "database_client_rotate_uncaught_failed", { error: errorToJson(closeError) }));
      if (dirtyDatabaseTaskIds.size > 0 || dirtyDatabaseQueueIds.size > 0) scheduleDatabaseFlushRetry(config.databaseFlushIntervalMs);
      logger("error", "transient_database_uncaught_exception_suppressed", {
        suppressedCount: databaseTransientUncaughtSuppressedCount,
        dirtyTaskCount: dirtyDatabaseTaskIds.size,
        dirtyQueueCount: dirtyDatabaseQueueIds.size,
        classification: classifyTransientDatabaseError(error) as unknown as JsonValue,
        error: errorToJson(error),
      });
      return;
    }
    logger("error", "uncaught_exception", { error: errorToJson(error) });
    throw error;
  });
  process.on("unhandledRejection", (reason) => {
    if (isTransientDatabaseError(reason)) {
      databaseTransientUncaughtSuppressedCount += 1;
      databaseLastTransientUncaughtSuppressedAt = nowIso();
      void rotateSqlClientAfterTransientError("unhandled-rejection", reason).catch((error) => logger("warn", "database_client_rotate_rejection_failed", { error: errorToJson(error) }));
      if (dirtyDatabaseTaskIds.size > 0 || dirtyDatabaseQueueIds.size > 0) scheduleDatabaseFlushRetry(config.databaseFlushIntervalMs);
      logger("error", "transient_database_unhandled_rejection_suppressed", {
        suppressedCount: databaseTransientUncaughtSuppressedCount,
        dirtyTaskCount: dirtyDatabaseTaskIds.size,
        dirtyQueueCount: dirtyDatabaseQueueIds.size,
        classification: classifyTransientDatabaseError(reason) as unknown as JsonValue,
        error: errorToJson(reason),
      });
      return;
    }
    logger("error", "unhandled_rejection", { error: errorToJson(reason) });
  });
}

setInterval(() => {
  if (!serviceReady) return;
  const pendingQueues = runnableQueueIds().filter((queueId) => !processingQueues.has(queueId));
  if (pendingQueues.length > 0) {
    logger("warn", "queue_watchdog_rescheduled", { runnable: state.tasks.filter((task) => task.status === "queued" || task.status === "retry_wait").length, pendingQueues });
    scheduleQueue();
  }
}, 5000).unref?.();

function jsonResponse(body: unknown, status = 200): Response {
  return new Response(JSON.stringify(body, null, 2), {
    status,
    headers: {
      "content-type": "application/json; charset=utf-8",
      "access-control-allow-origin": "*",
      "access-control-allow-methods": "GET,HEAD,POST,PATCH,DELETE,OPTIONS",
      "access-control-allow-headers": "content-type",
    },
  });
}

function compactJsonResponse(body: unknown, status = 200): Response {
  return new Response(JSON.stringify(body), {
    status,
    headers: {
      "content-type": "application/json; charset=utf-8",
      "access-control-allow-origin": "*",
      "access-control-allow-methods": "GET,HEAD,POST,PATCH,DELETE,OPTIONS",
      "access-control-allow-headers": "content-type",
    },
  });
}

async function readJson(req: Request): Promise<unknown> {
  const text = await req.text();
  if (text.trim().length === 0) return {};
  return JSON.parse(text) as unknown;
}

function requestErrorResponse(error: unknown): Response | null {
  if (error instanceof ReferenceTaskLookupError) {
    return jsonResponse({
      ok: false,
      error: error.message,
      missingReferenceTaskIds: error.missingIds,
    }, 400);
  }
  if (error instanceof SyntaxError && /json/iu.test(error.message)) {
    return jsonResponse({
      ok: false,
      error: "invalid JSON request body",
      detail: error.message,
    }, 400);
  }
  if (error instanceof Error && (
    error.message === "request body must be an object"
    || error.message === "prompt is required"
    || error.message === "a task cannot reference itself while editing prompt"
    || error.message === "sourceQueueId is required"
    || error.message === "source queue must be different from target queue"
    || error.message === "workdir path is required"
    || error.message === "workdir path contains an invalid character"
    || error.message.startsWith("referenceTaskIds supports at most ")
    || error.message.startsWith("queueId must match ")
    || error.message.startsWith("queue name must be ")
    || error.message.startsWith("workdir path must be ")
    || error.message.startsWith("windows-native executionMode ")
  )) {
    return jsonResponse({ ok: false, error: error.message }, 400);
  }
  return null;
}

function readOnlyRejectResponse(method: string, targetPath: string): Response {
  return jsonResponse({
    ok: false,
    error: "Code Queue read service is read-only",
    serviceRole: config.serviceRole,
    method,
    path: targetPath,
  }, 405);
}

function legacyQueueFrozenResponse(method: string, targetPath: string, action: string): Response {
  return jsonResponse({
    ok: false,
    frozen: true,
    mutation: false,
    error: "Legacy UniDesk Code Queue is frozen for new task and execution mutations; use AgentRun Queue for new commander work.",
    reason: "legacy-code-queue-frozen",
    serviceRole: config.serviceRole,
    method,
    path: targetPath,
    action,
    replacement: {
      queueCommander: "bun scripts/cli.ts agentrun v01 queue commander --reader-id cli",
      queueSubmit: "bun scripts/cli.ts agentrun v01 queue submit --json-stdin <<'JSON'",
      queueDispatch: "bun scripts/cli.ts agentrun v01 queue dispatch <taskId> --json-stdin <<'JSON'",
      sessionsTrace: "bun scripts/cli.ts agentrun v01 sessions trace <sessionId> --after-seq 0 --limit 100",
      sessionsOutput: "bun scripts/cli.ts agentrun v01 sessions output <sessionId> --after-seq 0 --limit 100",
      sessionsSteer: "bun scripts/cli.ts agentrun v01 sessions steer <sessionId> --prompt-stdin <<'EOF'",
      sessionsRead: "bun scripts/cli.ts agentrun v01 sessions read <sessionId> --reader-id cli",
      queueCancel: "bun scripts/cli.ts agentrun v01 queue cancel <taskId> --reason <text>",
    },
    legacy: {
      retainedFor: "historical archive/read-only troubleshooting",
      allowedReadPaths: [
        "GET /api/tasks",
        "GET /api/tasks/overview",
        "GET /api/tasks/:id",
        "GET /api/tasks/:id/output",
        "GET /api/tasks/:id/trace-summary",
        "GET /api/tasks/:id/trace-steps",
        "GET /api/queues",
      ],
      cancellationPath: "POST /api/tasks/:id/interrupt remains available only to stop residual old runs",
      noDoubleWrite: true,
    },
  }, 410);
}

function legacyQueueWritesFrozen(): boolean {
  return true;
}

function databaseNotReadyResponse(method: string, targetPath: string): Response {
  return jsonResponse({
    ok: false,
    error: "Code Queue PostgreSQL storage is not ready",
    serviceRole: config.serviceRole,
    method,
    path: targetPath,
    databaseReady,
    databaseLastError,
  }, 503);
}

function requireDatabaseReadyForWrite(method: string, targetPath: string): Response | null {
  return databaseReady ? null : databaseNotReadyResponse(method, targetPath);
}

function schedulerOnlyRejectResponse(method: string, targetPath: string): Response {
  return jsonResponse({
    ok: false,
    error: "Code Queue write service does not own active scheduler control for this endpoint",
    serviceRole: config.serviceRole,
    method,
    path: targetPath,
    hint: "Route active-run control to the code-queue-scheduler service.",
  }, 409);
}

function findTask(id: string): QueueTask | null {
  return state.tasks.find((task) => task.id === id) ?? null;
}

function parseLimit(url: URL): number {
  const value = Number(url.searchParams.get("limit") ?? 100);
  return Number.isInteger(value) && value > 0 ? Math.min(500, value) : 100;
}

function workdirRowsForResponse(providerIdValue: string | null, executionModeValue: string | null): WorkdirRecord[] {
  const providerId = normalizeProviderId(providerIdValue) ?? null;
  const executionMode = executionModeValue === null ? null : normalizeCodeExecutionMode(executionModeValue);
  return sortedWorkdirRecords().filter((record) => {
    if (providerId !== null && record.providerId !== providerId) return false;
    if (executionMode !== null && record.executionMode !== executionMode) return false;
    return true;
  });
}

async function listWorkdirs(url: URL): Promise<Response> {
  ensureDefaultWorkdirRecords();
  return jsonResponse({
    ok: true,
    workdirs: workdirRowsForResponse(url.searchParams.get("providerId"), url.searchParams.get("executionMode")),
    defaultProviderId: config.mainProviderId,
    defaultWorkdir: config.defaultWorkdir,
    remoteDefaultWorkdir: config.remoteDefaultWorkdir,
    windowsNativeCodexDefaultWorkdir: config.windowsNativeCodexDefaultWorkdir,
  });
}

async function createWorkdir(req: Request): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, "/api/workdirs", "workdir-create");
  const notReady = requireDatabaseReadyForWrite(req.method, "/api/workdirs");
  if (notReady !== null) return notReady;
  const body = await readJson(req);
  const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
  const providerId = normalizeTaskProviderId(record.providerId);
  const executionMode = normalizeCodeExecutionMode(record.executionMode);
  const path = normalizeWorkdirPath(record.path ?? record.cwd ?? record.workdir, providerId);
  validateExecutionModeForTask(providerId, path, config.defaultModel, executionMode);
  const previous = workdirRecords.get(workdirRecordKey(providerId, executionMode, path));
  let ensureResult: JsonValue = { ok: false, skipped: true, reason: "remote provider workdirs are created when a task starts" };
  if (providerIsMain(providerId)) {
    ensureResult = ensureLocalWorkdir(path) as unknown as JsonValue;
  } else if (record.ensure === true || record.createOnProvider === true) {
    const command = await runCodeQueueSsh(providerId, `set -euo pipefail\nmkdir -p ${shellQuote(path)}\ntest -d ${shellQuote(path)}\nprintf 'workdir_ready path=%s\\n' ${shellQuote(path)}`, 30_000, "workdir-create");
    ensureResult = {
      ok: command.exitCode === 0,
      exitCode: command.exitCode,
      stdout: safePreview(command.stdout, 800),
      stderr: safePreview(command.stderr, 800),
      durationMs: command.durationMs,
    } as unknown as JsonValue;
    if (command.exitCode !== 0) return jsonResponse({ ok: false, error: `failed to create workdir on provider ${providerId}`, ensure: ensureResult }, 502);
  }
  const workdir = rememberWorkdir(providerId, executionMode, path);
  await upsertWorkdirsToDatabase([workdir]);
  logger("info", "workdir_saved", { providerId, executionMode, path, existed: previous !== undefined, ensure: ensureResult });
  return jsonResponse({ ok: true, workdir, workdirs: sortedWorkdirRecords(), ensure: ensureResult }, previous === undefined ? 201 : 200);
}

async function deleteWorkdir(providerIdValue: string, executionModeValue: string, pathValue: string): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse("DELETE", `/api/workdirs/${providerIdValue}/${executionModeValue}/${pathValue}`, "workdir-delete");
  const notReady = requireDatabaseReadyForWrite("DELETE", `/api/workdirs/${providerIdValue}/${executionModeValue}/${pathValue}`);
  if (notReady !== null) return notReady;
  const providerId = normalizeTaskProviderId(providerIdValue);
  const executionMode = normalizeCodeExecutionMode(executionModeValue);
  const path = normalizeWorkdirPath(pathValue, providerId);
  const key = workdirRecordKey(providerId, executionMode, path);
  const existing = workdirRecords.get(key) ?? null;
  if (existing === null) return jsonResponse({ ok: false, error: "workdir not found" }, 404);
  workdirRecords.delete(key);
  if (databaseReady) {
    await sql`
      DELETE FROM unidesk_code_queue_workdirs
      WHERE provider_id = ${providerId}
        AND execution_mode = ${executionMode}
        AND path = ${path}
    `;
  }
  logger("info", "workdir_deleted", { providerId, executionMode, path });
  return jsonResponse({ ok: true, deleted: existing, workdirs: sortedWorkdirRecords() });
}

async function createTasks(req: Request): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, "/api/tasks", "task-submit");
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, "/api/tasks");
  const notReady = requireDatabaseReadyForWrite(req.method, "/api/tasks");
  if (notReady !== null) return notReady;
  const body = await readJson(req);
  const batchRecord = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
  const batchQueueId = typeof batchRecord.queueId === "string" && batchRecord.queueId.trim().length > 0 ? normalizeQueueId(batchRecord.queueId) : undefined;
  const records = typeof body === "object" && body !== null && !Array.isArray(body) && Array.isArray((body as Record<string, unknown>).tasks)
    ? (body as Record<string, unknown>).tasks as unknown[]
    : [body];
  const tasks = await Promise.all(records.map(async (record) => {
    const normalized = normalizeRequest(record);
    if (normalized.queueId === undefined && batchQueueId !== undefined) normalized.queueId = batchQueueId;
    return createTask(injectCodeQueueEnvironmentHint(await injectReferencedTaskContext(normalized)));
  }));
  for (const task of tasks) appendOutput(task, "user", `${task.prompt}\n`, "enqueue");
  state.tasks.push(...tasks);
  if (tasks.length > 0) armIdleNotification();
  for (const task of tasks) publishTaskOaEvent(task, "enqueue");
  for (const id of new Set(tasks.map(queueIdOf))) publishQueueEvent("enqueue", id);
  persistState();
  logger("info", "tasks_enqueued", { count: tasks.length, ids: tasks.map((task) => task.id), queueIds: Array.from(new Set(tasks.map(queueIdOf))), providerIds: Array.from(new Set(tasks.map((task) => task.providerId))), executionModes: Array.from(new Set(tasks.map((task) => task.executionMode))) });
  scheduleQueue();
  await flushDirtyTasksToDatabase(true);
  await upsertWorkdirsToDatabase(sortedWorkdirRecords());
  return jsonResponse({ ok: true, tasks: tasks.map((task) => taskForResponse(task)), queue: await queueSummaryForResponse() }, 202);
}

async function steerTask(task: QueueTask, req: Request): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, `/api/tasks/${task.id}/steer`, "task-steer");
  if (!serviceRoleAllowsScheduler(config.serviceRole)) return schedulerOnlyRejectResponse(req.method, `/api/tasks/${task.id}/steer`);
  const body = await readJson(req);
  const bodyRecord = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
  const prompt = typeof bodyRecord.prompt === "string" ? String(bodyRecord.prompt) : "";
  if (prompt.trim().length === 0) return jsonResponse({ ok: false, error: "prompt is required" }, 400);
  const steerId = normalizeSteerId(bodyRecord.steerId) ?? createSteerId(task.id, prompt);
  const duplicateDecision = steerDuplicateDecision(task, steerId, prompt, taskFullOutput(task));
  if (duplicateDecision.duplicate) {
    await flushDirtyTasksToDatabase(true);
    return jsonResponse({
      ok: true,
      accepted: true,
      duplicateSuppressed: true,
      deliveryState: "accepted",
      steerId,
      traceConfirmation: steerTraceConfirmationJson(duplicateDecision.confirmation),
      task: taskForResponse(task),
      queue: await queueSummaryForResponse(),
    });
  }
  if (duplicateDecision.conflict) {
    return jsonResponse({
      ok: false,
      error: "steerId already exists with a different prompt hash",
      accepted: false,
      deliveryState: "not_accepted",
      steerId,
      existingPromptHash: duplicateDecision.existingPromptHash,
      requestedPromptHash: duplicateDecision.requestedPromptHash,
      traceConfirmation: steerTraceConfirmationJson(duplicateDecision.confirmation),
      task: taskForResponse(task),
    }, 409);
  }
  const activeRun = activeRunForTask(task);
  if (activeRun === null || activeRun.threadId === null || activeRun.turnId === null || typeof activeRun.app.steer !== "function") {
    return jsonResponse({
      ok: false,
      error: "task does not have an active steerable turn",
      accepted: false,
      deliveryState: "not_accepted",
      steerId,
      traceConfirmation: steerTraceConfirmationJson(duplicateDecision.confirmation),
      task: taskForResponse(task),
    }, 409);
  }
  const output = appendOutput(task, "user", steerTraceText(steerId, prompt), "turn/steer", steerId);
  appendPromptHistory(task, output, "turn/steer", prompt, { steerId });
  await activeRun.app.steer(activeRun.threadId, activeRun.turnId, prompt);
  await flushDirtyTasksToDatabase(true);
  const traceConfirmation = findSteerTraceConfirmation(task, steerId, taskFullOutput(task));
  return jsonResponse({
    ok: true,
    accepted: true,
    duplicateSuppressed: false,
    deliveryState: traceConfirmation.found ? "accepted" : "accepted_response_timeout",
    steerId,
    promptHash: steerPromptHash(prompt),
    traceConfirmation: steerTraceConfirmationJson(traceConfirmation),
    task: taskForResponse(task),
    queue: await queueSummaryForResponse(),
  });
}

async function resumeTask(task: QueueTask, req: Request): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, `/api/tasks/${task.id}/resume`, "task-resume");
  if (!serviceRoleAllowsScheduler(config.serviceRole)) return schedulerOnlyRejectResponse(req.method, `/api/tasks/${task.id}/resume`);
  const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${task.id}/resume`);
  if (notReady !== null) return notReady;
  const body = await readJson(req);
  const bodyRecord = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
  const prompt = typeof bodyRecord.prompt === "string" ? String(bodyRecord.prompt) : "";
  if (prompt.trim().length === 0) return jsonResponse({ ok: false, error: "prompt is required" }, 400);
  const resumeId = normalizeResumeId(bodyRecord.resumeId) ?? createResumeId(task.id, prompt);
  const originalCodexThreadId = task.codexThreadId;
  const duplicateDecision = resumeDuplicateDecision(task, resumeId, prompt, taskFullOutput(task));
  if (duplicateDecision.duplicate) {
    await flushDirtyTasksToDatabase(true);
    return jsonResponse({
      ok: true,
      accepted: true,
      duplicateSuppressed: true,
      deliveryState: "duplicate_suppressed",
      resumeId,
      reuseOriginalThread: originalCodexThreadId !== null,
      originalCodexThreadId,
      codexThreadId: task.codexThreadId,
      traceConfirmation: resumeTraceConfirmationJson(duplicateDecision.confirmation),
      task: taskForResponse(task),
      queue: await queueSummaryForResponse(),
    });
  }
  if (duplicateDecision.conflict) {
    return jsonResponse({
      ok: false,
      error: "resumeId already exists with a different prompt hash",
      accepted: false,
      deliveryState: "not_accepted",
      resumeId,
      existingPromptHash: duplicateDecision.existingPromptHash,
      requestedPromptHash: duplicateDecision.requestedPromptHash,
      traceConfirmation: resumeTraceConfirmationJson(duplicateDecision.confirmation),
      task: taskForResponse(task),
    }, 409);
  }
  if (task.status === "running" || task.status === "judging") {
    return jsonResponse({
      ok: false,
      error: `task is active: ${task.status}`,
      accepted: false,
      deliveryState: "not_accepted",
      disposition: "use-steer-for-active-task",
      resumeId,
      task: taskForResponse(task),
      commands: {
        steer: `bun scripts/cli.ts codex steer ${task.id} --prompt-file <path>`,
        show: `bun scripts/cli.ts codex task ${task.id}`,
      },
    }, 409);
  }
  if (task.status !== "failed" && task.status !== "canceled" && task.status !== "succeeded") {
    return jsonResponse({
      ok: false,
      error: `task is not resumable: ${task.status}`,
      accepted: false,
      deliveryState: "not_accepted",
      disposition: "not-terminal",
      resumeId,
      task: taskForResponse(task),
    }, 409);
  }
  const traceOutput = appendOutput(task, "user", resumeTraceText(resumeId, prompt), "turn/resume", resumeId);
  task.status = "queued";
  task.finishedAt = null;
  task.readAt = null;
  task.cancelRequested = false;
  task.lastError = null;
  task.maxAttempts = Math.max(task.maxAttempts, task.attempts.length + 1);
  task.nextMode = "retry";
  task.nextPrompt = prompt;
  setAttemptFeedbackPrompt(task.attempts.at(-1), task.nextPrompt, "manual-resume", task.attempts.length + 1);
  task.updatedAt = nowIso();
  task.queueEnteredAt = task.updatedAt;
  appendOutput(task, "system", `resume queued resumeId=${resumeId} reuseOriginalThread=${originalCodexThreadId !== null ? "true" : "false"}\n`, "manual-resume", resumeId);
  armIdleNotification();
  persistState();
  scheduleQueue(queueIdOf(task));
  await flushDirtyTasksToDatabase(true);
  const traceConfirmation = findResumeTraceConfirmation(task, resumeId, taskFullOutput(task));
  return jsonResponse({
    ok: true,
    accepted: true,
    duplicateSuppressed: false,
    deliveryState: traceConfirmation.found ? "queued_for_existing_thread" : "accepted_response_timeout",
    resumeId,
    turnId: traceOutput?.seq ?? null,
    promptHash: resumePromptHash(prompt),
    reuseOriginalThread: originalCodexThreadId !== null,
    originalCodexThreadId,
    codexThreadId: task.codexThreadId,
    traceConfirmation: resumeTraceConfirmationJson(traceConfirmation),
    task: taskForResponse(task),
    queue: await queueSummaryForResponse(),
  }, 202);
}

async function editQueuedTaskPrompt(task: QueueTask, req: Request): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, `/api/tasks/${task.id}/edit`, "queued-prompt-edit");
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/edit`);
  const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${task.id}/edit`);
  if (notReady !== null) return notReady;
  if (!queuedTaskPromptEditable(task)) {
    return jsonResponse({
      ok: false,
      error: `task prompt can only be edited before first run while status=queued; current status=${task.status}`,
      editable: false,
      task: taskForResponse(task),
    }, 409);
  }
  let update: QueueTaskRequest;
  try {
    update = await buildQueuedPromptUpdate(task, await readJson(req));
  } catch (error) {
    return jsonResponse({ ok: false, error: error instanceof Error ? error.message : String(error) }, 400);
  }
  const nextBasePrompt = update.basePrompt ?? userPromptForDisplay(update.prompt);
  const nextReferenceTaskIds = update.referenceTaskIds ?? [];
  if (task.prompt === update.prompt && task.basePrompt === nextBasePrompt && taskReferencesEqual(task.referenceTaskIds, nextReferenceTaskIds)) {
    return jsonResponse({ ok: true, changed: false, editable: true, task: taskForResponse(task), queue: await queueSummaryForResponse(false) });
  }
  const previousPromptChars = task.prompt.length;
  const previousBasePromptChars = task.basePrompt.length;
  task.prompt = update.prompt;
  task.basePrompt = nextBasePrompt;
  task.referenceTaskIds = nextReferenceTaskIds;
  task.referenceInjection = update.referenceInjection ?? null;
  task.updatedAt = nowIso();
  rewriteEnqueueOutput(task);
  appendOutput(task, "system", `queued prompt edited before first run; base ${previousBasePromptChars}->${task.basePrompt.length} chars; final ${previousPromptChars}->${task.prompt.length} chars\n`, "prompt/edit");
  persistState();
  logger("info", "queued_task_prompt_edited", {
    taskId: task.id,
    queueId: queueIdOf(task),
    promptChars: task.prompt.length,
    basePromptChars: task.basePrompt.length,
    referenceTaskIds: task.referenceTaskIds,
  });
  scheduleQueue(queueIdOf(task));
  await flushDirtyTasksToDatabase(true);
  return jsonResponse({ ok: true, changed: true, editable: true, task: taskForResponse(task), queue: await queueSummaryForResponse() });
}

async function interruptTask(task: QueueTask, method = "POST"): Promise<Response> {
  if (!serviceRoleAllowsScheduler(config.serviceRole)) return schedulerOnlyRejectResponse(method, `/api/tasks/${task.id}/interrupt`);
  if (task.status === "succeeded" || task.status === "failed" || task.status === "canceled") {
    return jsonResponse({ ok: false, error: `task is already terminal: ${task.status}`, task: taskForResponse(task) }, 409);
  }
  task.cancelRequested = true;
  task.updatedAt = nowIso();
  appendOutput(task, "system", "interrupt requested\n", "turn/interrupt");
  const activeRun = activeRunForTask(task);
  if (activeRun !== null) {
    if (activeRun.threadId !== null && activeRun.turnId !== null && typeof activeRun.app.interrupt === "function") {
      await activeRun.app.interrupt(activeRun.threadId, activeRun.turnId).catch((error) => {
        appendOutput(task, "error", `interrupt request failed: ${error instanceof Error ? error.message : String(error)}\n`, "turn/interrupt");
      });
    } else {
      activeRun.app.stop();
    }
  }
  if (task.status === "queued" || task.status === "retry_wait") {
    task.status = "canceled";
    task.finishedAt = nowIso();
  }
  persistTaskState(task);
  if (terminalTask(task)) {
    void notifyTaskTerminal(task).then(() => maybeNotifyQueueIdle(task.id)).catch((error) => logger("warn", "claudeqq_interrupt_notify_failed", { taskId: task.id, error: errorToJson(error) }));
  }
  await flushDirtyTasksToDatabase(true);
  return jsonResponse({ ok: true, task: taskForResponse(task), queue: await queueSummaryForResponse() });
}

async function manualRetry(task: QueueTask, req: Request): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, `/api/tasks/${task.id}/retry`, "task-retry");
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/retry`);
  const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${task.id}/retry`);
  if (notReady !== null) return notReady;
  if (task.status !== "failed" && task.status !== "canceled" && task.status !== "succeeded") {
    return jsonResponse({ ok: false, error: `task is not terminal: ${task.status}`, task: taskForResponse(task) }, 409);
  }
  const body = await readJson(req);
  const record = extractRecord(body) ?? {};
  const explicitPrompt = typeof record.prompt === "string" ? record.prompt.trim() : typeof record.continuePrompt === "string" ? record.continuePrompt.trim() : "";
  task.status = "queued";
  task.finishedAt = null;
  task.readAt = null;
  task.cancelRequested = false;
  task.lastError = null;
  task.maxAttempts = Math.max(task.maxAttempts, task.attempts.length + 1);
  task.nextMode = "retry";
  task.nextPrompt = explicitPrompt.length > 0 ? explicitPrompt : retryPrompt(task, { decision: "retry", confidence: 1, reason: "Manual retry", source: "fallback" });
  setAttemptFeedbackPrompt(task.attempts.at(-1), task.nextPrompt, explicitPrompt.length > 0 ? "manual-retry-explicit" : "manual-retry-generated", task.attempts.length + 1);
  task.updatedAt = nowIso();
  task.queueEnteredAt = task.updatedAt;
  appendOutput(task, "system", explicitPrompt.length > 0 ? "manual retry queued with explicit continuation prompt\n" : "manual retry queued\n", "manual-retry");
  armIdleNotification();
  persistState();
  scheduleQueue(queueIdOf(task));
  await flushDirtyTasksToDatabase(true);
  return jsonResponse({ ok: true, task: taskForResponse(task), queue: await queueSummaryForResponse() }, 202);
}

async function markTaskRead(task: QueueTask): Promise<Response> {
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse("POST", `/api/tasks/${task.id}/read`);
  const notReady = requireDatabaseReadyForWrite("POST", `/api/tasks/${task.id}/read`);
  if (notReady !== null) return notReady;
  if (!terminalTask(task)) {
    return jsonResponse({ ok: false, error: `task is not terminal: ${task.status}`, task: taskForResponse(task) }, 409);
  }
  if (task.readAt === null) {
    task.readAt = nowIso();
    markTaskDirty(task.id);
    persistState(false);
    publishTaskOaEvent(task, "read");
    logger("info", "task_marked_read", { taskId: task.id, queueId: queueIdOf(task), status: task.status });
  }
  await flushDirtyTasksToDatabase(true);
  return compactJsonResponse({ ok: true, task: taskForListResponse(task, true), queue: await queueSummaryForResponse(false) });
}

function timestampToIso(value: Date | string | null): string | null {
  if (value === null) return null;
  const date = value instanceof Date ? value : new Date(value);
  return Number.isNaN(date.getTime()) ? String(value) : date.toISOString();
}

async function markTaskReadById(taskId: string): Promise<Response> {
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse("POST", `/api/tasks/${taskId}/read`);
  const notReady = requireDatabaseReadyForWrite("POST", `/api/tasks/${taskId}/read`);
  if (notReady !== null) return notReady;
  if (!databaseReady) {
    const task = await findTaskForMutation(taskId);
    return task === null ? jsonResponse({ ok: false, error: "task not found" }, 404) : markTaskRead(task);
  }
  const rows = await sql<Array<{ id: string; queue_id: string; status: TaskStatus; read_at: Date | string | null }>>`
    SELECT id, queue_id, status, read_at
    FROM unidesk_code_queue_tasks
    WHERE id = ${taskId}
    LIMIT 1
  `;
  const row = rows[0] ?? null;
  if (row === null) {
    const task = findTask(taskId);
    return task === null ? jsonResponse({ ok: false, error: "task not found" }, 404) : markTaskRead(task);
  }
  if (!(row.status === "succeeded" || row.status === "failed" || row.status === "canceled")) {
    return compactJsonResponse({ ok: false, error: `task is not terminal: ${row.status}`, task: { id: taskId, queueId: safeQueueId(row.queue_id), status: row.status } }, 409);
  }
  const readAt = timestampToIso(row.read_at) ?? nowIso();
  if (row.read_at === null) {
    await sql`
      UPDATE unidesk_code_queue_tasks
      SET
        read_at = ${readAt},
        task_json = jsonb_set(task_json, '{readAt}', to_jsonb(${readAt}::text), true)
      WHERE id = ${taskId}
        AND read_at IS NULL
    `;
    const hotTask = findTask(taskId);
    if (hotTask !== null) {
      hotTask.readAt = readAt;
      publishTaskOaEvent(hotTask, "read");
    }
    logger("info", "task_marked_read", { taskId, queueId: safeQueueId(row.queue_id), status: row.status });
  }
  return compactJsonResponse({
    ok: true,
    task: {
      id: taskId,
      queueId: safeQueueId(row.queue_id),
      status: row.status,
      readAt,
      terminalUnread: false,
    },
    queue: await queueSummaryForResponse(false),
  });
}

async function markTerminalTasksRead(url: URL): Promise<Response> {
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse("POST", "/api/tasks/read-all");
  const notReady = requireDatabaseReadyForWrite("POST", "/api/tasks/read-all");
  if (notReady !== null) return notReady;
  const queueFilter = url.searchParams.get("queueId");
  const queueId = queueFilter === null || queueFilter.length === 0 ? null : safeQueueId(queueFilter);
  const readAt = nowIso();
  if (databaseReady) {
    const rows = queueId === null
      ? await sql<Array<{ id: string }>>`
        UPDATE unidesk_code_queue_tasks
        SET
          read_at = ${readAt},
          task_json = jsonb_set(task_json, '{readAt}', to_jsonb(${readAt}::text), true)
        WHERE status IN ('succeeded', 'failed', 'canceled')
          AND read_at IS NULL
        RETURNING id
      `
      : await sql<Array<{ id: string }>>`
        UPDATE unidesk_code_queue_tasks
        SET
          read_at = ${readAt},
          task_json = jsonb_set(task_json, '{readAt}', to_jsonb(${readAt}::text), true)
        WHERE queue_id = ${queueId}
          AND status IN ('succeeded', 'failed', 'canceled')
          AND read_at IS NULL
        RETURNING id
      `;
    const ids = new Set(rows.map((row) => row.id));
    for (const task of state.tasks) {
      if (!ids.has(task.id)) continue;
      task.readAt = readAt;
      publishTaskOaEvent(task, "read-all");
    }
    if (ids.size > 0) {
      logger("info", "terminal_tasks_marked_read", { count: ids.size, queueId });
    }
    return jsonResponse({ ok: true, count: ids.size, readAt, queue: await queueSummaryForResponse(false) });
  }
  let count = 0;
  for (const task of state.tasks) {
    if (queueId !== null && queueIdOf(task) !== queueId) continue;
    if (!terminalTaskUnread(task)) continue;
    task.readAt = readAt;
    markTaskDirty(task.id);
    publishTaskOaEvent(task, "read-all");
    count += 1;
  }
  if (count > 0) {
    persistState(false);
    logger("info", "terminal_tasks_marked_read", { count, queueId });
  }
  await flushDirtyTasksToDatabase(true);
  return jsonResponse({ ok: true, count, readAt, queue: await queueSummaryForResponse(false) });
}

async function createQueue(req: Request): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, "/api/queues", "queue-create");
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, "/api/queues");
  const notReady = requireDatabaseReadyForWrite(req.method, "/api/queues");
  if (notReady !== null) return notReady;
  const body = await readJson(req);
  const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
  const queueId = normalizeQueueId(record.queueId ?? record.id);
  const beforeCount = state.queues.length;
  const queue = ensureQueue(queueId, record.name);
  queue.updatedAt = nowIso();
  markQueueDirty(queue.id);
  persistState(false);
  publishQueueEvent("queue-created", queue.id);
  logger("info", "queue_created", { queueId, name: queue.name, existed: beforeCount === state.queues.length });
  await flushDirtyTasksToDatabase(true);
  const tasks = await loadAllTasksForRead();
  return jsonResponse({ ok: true, queue, queues: perQueueSummaries(tasks), summary: queueSummary(false, tasks) }, beforeCount === state.queues.length ? 200 : 201);
}

async function updateQueue(queueIdValue: string, req: Request): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, `/api/queues/${queueIdValue}`, "queue-update");
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/queues/${queueIdValue}`);
  const notReady = requireDatabaseReadyForWrite(req.method, `/api/queues/${queueIdValue}`);
  if (notReady !== null) return notReady;
  const queueId = normalizeQueueId(queueIdValue);
  const queue = state.queues.find((item) => item.id === queueId);
  if (queue === undefined) return jsonResponse({ ok: false, error: "queue not found" }, 404);
  const body = await readJson(req);
  const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
  if (!Object.prototype.hasOwnProperty.call(record, "name")) {
    return jsonResponse({ ok: false, error: "name is required" }, 400);
  }
  const previousName = safeQueueName(queue.name, queue.id);
  queue.name = normalizeQueueName(record.name, queue.id);
  queue.updatedAt = nowIso();
  markQueueDirty(queue.id);
  persistState(false);
  publishQueueEvent("queue-updated", queue.id);
  logger("info", "queue_updated", { queueId, previousName, name: queue.name });
  await flushDirtyTasksToDatabase(true);
  const tasks = await loadAllTasksForRead();
  return jsonResponse({ ok: true, queue, queues: perQueueSummaries(tasks), summary: queueSummary(false, tasks) });
}

function knownQueue(queueId: string): QueueRecord | null {
  return state.queues.find((queue) => queue.id === queueId) ?? null;
}

function queueHasKnownTasks(queueId: string): boolean {
  return state.tasks.some((task) => queueIdOf(task) === queueId);
}

function queueMergeBlocker(queueId: string): string | null {
  if (processingQueues.has(queueId)) return "queue processor is currently active";
  if (activeRuns.has(queueId)) return "queue has an active agent run";
  if (activeRunSlotReservations.has(queueId)) return "queue is reserving an active run slot";
  if (activeRunSlotWaiters.some((waiter) => waiter.queueId === queueId)) return "queue is waiting for an active run slot";
  const activeTask = state.tasks.find((task) => queueIdOf(task) === queueId && (task.status === "running" || task.status === "judging"));
  if (activeTask !== undefined) return `task ${activeTask.id} is ${activeTask.status}`;
  const claimedPendingTask = state.tasks.find((task) => queueIdOf(task) === queueId && (task.status === "queued" || task.status === "retry_wait") && !taskIsUnclaimedMovable(task));
  return claimedPendingTask === undefined ? null : `task ${claimedPendingTask.id} has already been claimed`;
}

function parseSourceQueueIds(record: Record<string, unknown>, targetQueueId: string): string[] {
  const raw = record.sourceQueueIds ?? record.sources ?? record.sourceQueueId ?? record.fromQueueId ?? record.from ?? record.queueId ?? record.id;
  const values = Array.isArray(raw)
    ? raw
    : typeof raw === "string"
      ? raw.split(/[,\s]+/u)
      : [];
  const ids: string[] = [];
  const seen = new Set<string>();
  for (const value of values) {
    const id = normalizeQueueId(value);
    if (id === targetQueueId) throw new Error("source queue must be different from target queue");
    if (seen.has(id)) continue;
    seen.add(id);
    ids.push(id);
  }
  if (ids.length === 0) throw new Error("sourceQueueId is required");
  return ids;
}

async function mergeDatabaseQueueTasks(sourceQueueIds: string[], targetQueueId: string, mergedAt: string): Promise<{ movedTaskIds: string[]; blocker: DatabaseTaskStatusRow | null }> {
  if (!databaseReady || sourceQueueIds.length === 0) return { movedTaskIds: [], blocker: null };
  return await sql.begin(async (client) => {
    const mergeQueueIds = Array.from(new Set([targetQueueId, ...sourceQueueIds]));
    const lockedRows = await client<DatabaseTaskStatusRow[]>`
      SELECT id, queue_id, status, started_at, current_attempt, codex_thread_id, active_turn_id
      FROM unidesk_code_queue_tasks
      WHERE queue_id IN ${client(mergeQueueIds)}
      ORDER BY updated_at DESC, id DESC
      FOR UPDATE
    `;
    const blocker = lockedRows.find((row) => {
      return row.status === "running"
        || row.status === "judging"
        || (
          (row.status === "queued" || row.status === "retry_wait")
          && (
            row.started_at !== null
            || Number(row.current_attempt ?? 0) > 0
            || row.codex_thread_id !== null
            || row.active_turn_id !== null
          )
        );
    }) ?? null;
    if (blocker !== null) return { movedTaskIds: [], blocker };
    const rows = await client<Array<{ id: string }>>`
      UPDATE unidesk_code_queue_tasks
      SET
        queue_id = ${targetQueueId},
        updated_at = ${mergedAt},
        task_json = jsonb_set(
          jsonb_set(
            jsonb_set(
              task_json,
              '{queueId}',
              to_jsonb(${targetQueueId}::text),
              true
            ),
            '{queueEnteredAt}',
            to_jsonb(COALESCE(NULLIF(task_json->>'queueEnteredAt', ''), to_char(created_at AT TIME ZONE 'UTC', 'YYYY-MM-DD"T"HH24:MI:SS.MS"Z"'))::text),
            true
          ),
          '{updatedAt}',
          to_jsonb(${mergedAt}::text),
          true
        )
      WHERE queue_id IN ${client(sourceQueueIds)}
        AND (
          status IN ('succeeded', 'failed', 'canceled')
          OR (
            status IN ('queued', 'retry_wait')
            AND started_at IS NULL
            AND current_attempt = 0
            AND codex_thread_id IS NULL
            AND active_turn_id IS NULL
          )
        )
      RETURNING id
    `;
    return { movedTaskIds: rows.map((row) => row.id), blocker: null };
  });
}

async function moveDatabaseTaskToQueue(taskId: string, targetQueueId: string, movedAt: string): Promise<{ ok: boolean; row: DatabaseTaskStatusRow | null; previousQueueId: string | null; blocker: string }> {
  if (!databaseReady) return { ok: true, row: null, previousQueueId: null, blocker: "" };
  return await sql.begin(async (client) => {
    const rows = await client<DatabaseTaskStatusRow[]>`
      SELECT id, queue_id, status, started_at, current_attempt, codex_thread_id, active_turn_id
      FROM unidesk_code_queue_tasks
      WHERE id = ${taskId}
      LIMIT 1
      FOR UPDATE
    `;
    const row = rows[0] ?? null;
    const blocker = databaseTaskMoveBlocker(row);
    if (blocker.length > 0) return { ok: false, row, previousQueueId: row === null ? null : safeQueueId(row.queue_id), blocker };
    const previousQueueId = safeQueueId(row?.queue_id);
    const updated = await client<DatabaseTaskStatusRow[]>`
      UPDATE unidesk_code_queue_tasks
      SET
        queue_id = ${targetQueueId},
        updated_at = ${movedAt},
        task_json = jsonb_set(
          jsonb_set(
            jsonb_set(
              task_json,
              '{queueId}',
              to_jsonb(${targetQueueId}::text),
              true
            ),
            '{queueEnteredAt}',
            to_jsonb(${movedAt}::text),
            true
          ),
          '{updatedAt}',
          to_jsonb(${movedAt}::text),
          true
        )
      WHERE id = ${taskId}
        AND status IN ('queued', 'retry_wait')
        AND started_at IS NULL
        AND current_attempt = 0
        AND codex_thread_id IS NULL
        AND active_turn_id IS NULL
      RETURNING id, queue_id, status, started_at, current_attempt, codex_thread_id, active_turn_id
    `;
    const updatedRow = updated[0] ?? null;
    return updatedRow === null
      ? { ok: false, row, previousQueueId, blocker: "conditional update matched no rows" }
      : { ok: true, row: updatedRow, previousQueueId, blocker: "" };
  });
}

async function deleteDatabaseQueues(queueIds: string[]): Promise<string[]> {
  if (!databaseReady || queueIds.length === 0) return [];
  const rows = await sql<Array<{ id: string }>>`
    DELETE FROM unidesk_code_queue_queues
    WHERE id IN ${sql(queueIds)}
    RETURNING id
  `;
  return rows.map((row) => row.id);
}

function queueSnapshot(queueId: string, timestamp: string): QueueRecord {
  const queue = knownQueue(queueId);
  return queue === null
    ? { id: queueId, name: queueId, createdAt: timestamp, updatedAt: timestamp }
    : { ...queue };
}

function deleteQueuesFromState(queueIds: string[]): QueueRecord[] {
  const queueIdSet = new Set(queueIds);
  const deletedQueues: QueueRecord[] = [];
  const retainedQueues: QueueRecord[] = [];
  for (const queue of state.queues) {
    if (queueIdSet.has(queue.id)) {
      deletedQueues.push({ ...queue });
      dirtyDatabaseQueueIds.delete(queue.id);
    } else {
      retainedQueues.push(queue);
    }
  }
  state.queues.splice(0, state.queues.length, ...retainedQueues);
  for (const queueId of queueIds) dirtyDatabaseQueueIds.delete(queueId);
  return deletedQueues;
}

async function mergeQueues(targetQueueIdValue: string | null, req: Request): Promise<Response> {
  if (legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, targetQueueIdValue === null ? "/api/queues/merge" : `/api/queues/${targetQueueIdValue}/merge`, "queue-merge");
  if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, targetQueueIdValue === null ? "/api/queues/merge" : `/api/queues/${targetQueueIdValue}/merge`);
  const mergePath = targetQueueIdValue === null ? "/api/queues/merge" : `/api/queues/${targetQueueIdValue}/merge`;
  const notReady = requireDatabaseReadyForWrite(req.method, mergePath);
  if (notReady !== null) return notReady;
  const body = await readJson(req);
  const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
  const targetQueueId = normalizeQueueId(targetQueueIdValue ?? record.targetQueueId ?? record.intoQueueId ?? record.into);
  const sourceQueueIds = parseSourceQueueIds(record, targetQueueId);
  const missingSources = sourceQueueIds.filter((id) => knownQueue(id) === null && !queueHasKnownTasks(id));
  if (missingSources.length > 0) return jsonResponse({ ok: false, error: `source queue not found: ${missingSources.join(", ")}` }, 404);

  const mergeQueueIds = Array.from(new Set([targetQueueId, ...sourceQueueIds]));
  for (const id of mergeQueueIds) mergingQueues.add(id);
  try {
    for (const id of mergeQueueIds) {
      const blocker = queueMergeBlocker(id);
      if (blocker !== null) {
        return jsonResponse({ ok: false, error: `cannot merge queue ${id}: ${blocker}` }, 409);
      }
    }

    const mergedAt = nowIso();
    const databaseMerge = await mergeDatabaseQueueTasks(sourceQueueIds, targetQueueId, mergedAt);
    if (databaseMerge.blocker !== null) {
      const blockerQueueId = safeQueueId(databaseMerge.blocker.queue_id);
      const databaseTask = await loadTaskFromDatabase(databaseMerge.blocker.id);
      if (databaseTask !== null) reconcileHotTaskFromDatabase(databaseTask);
      return jsonResponse({
        ok: false,
        error: `cannot merge queue ${blockerQueueId}: task ${databaseMerge.blocker.id} is already claimed (${databaseTaskMoveBlocker(databaseMerge.blocker) || databaseMerge.blocker.status})`,
        blocker: databaseStatusRowJson(databaseMerge.blocker),
      }, 409);
    }
    const targetQueue = ensureQueue(targetQueueId);
    const sourceQueues = sourceQueueIds.map((id) => queueSnapshot(id, mergedAt));
    targetQueue.updatedAt = mergedAt;
    markQueueDirty(targetQueue.id);

    const sourceSet = new Set(sourceQueueIds);
    const hotMovedTasks: QueueTask[] = [];
    for (const task of state.tasks) {
      const previousQueueId = queueIdOf(task);
      if (!sourceSet.has(previousQueueId)) continue;
      task.queueEnteredAt = taskQueueEnteredAt(task);
      task.queueId = targetQueueId;
      task.updatedAt = mergedAt;
      hotMovedTasks.push(task);
      markTaskDirty(task.id);
      publishTaskOaEvent(task, "queue-merged");
    }
    const deletedSourceQueues = deleteQueuesFromState(sourceQueueIds);
    const databaseDeletedQueueIds = await deleteDatabaseQueues(sourceQueueIds);
    persistState(false);
    publishQueueEvent("queue-merged", targetQueueId);
    for (const sourceQueueId of sourceQueueIds) publishQueueEvent("queue-deleted-after-merge", sourceQueueId);
    if (hotMovedTasks.some((task) => task.status === "queued" || task.status === "retry_wait")) armIdleNotification();
    logger("info", "queues_merged", {
      targetQueueId,
      sourceQueueIds,
      deletedSourceQueueIds: deletedSourceQueues.map((queue) => queue.id),
      hotMovedTaskCount: hotMovedTasks.length,
      databaseMovedTaskCount: databaseReady ? databaseMerge.movedTaskIds.length : null,
      databaseDeletedQueueIds: databaseReady ? databaseDeletedQueueIds : null,
    });
    for (const id of mergeQueueIds) mergingQueues.delete(id);
    scheduleQueue(targetQueueId);
    await flushDirtyTasksToDatabase(true);
    const tasks = await loadAllTasksForRead();
    const movedIdSet = new Set(databaseReady ? databaseMerge.movedTaskIds : hotMovedTasks.map((task) => task.id));
    const orderedMovedTaskIds = tasks
      .filter((task) => movedIdSet.has(task.id))
      .sort(compareTaskQueueOrder)
      .map((task) => task.id);
    const targetTaskOrder = tasks
      .filter((task) => queueIdOf(task) === targetQueueId)
      .sort(compareTaskQueueOrder)
      .map((task) => ({ id: task.id, queueEnteredAt: taskQueueEnteredAt(task), createdAt: task.createdAt, status: task.status }));
    return jsonResponse({
      ok: true,
      targetQueueId,
      sourceQueueIds,
      mergedTaskCount: databaseReady ? databaseMerge.movedTaskIds.length : hotMovedTasks.length,
      movedTaskIds: orderedMovedTaskIds.slice(0, 500),
      targetTaskOrder: targetTaskOrder.slice(0, 500),
      order: "merged tasks keep their original queueEnteredAt/createdAt ordering; source queue records are deleted after merge",
      targetQueue,
      sourceQueues,
      deletedSourceQueues: deletedSourceQueues.length > 0 ? deletedSourceQueues : sourceQueues,
      deletedSourceQueueIds: sourceQueueIds,
      queues: perQueueSummaries(tasks),
      summary: queueSummary(false, tasks),
    }, 202);
  } finally {
    for (const id of mergeQueueIds) mergingQueues.delete(id);
  }
}

async function moveTaskToQueue(task: QueueTask, req: Request, options: { bypassRoleCheck?: boolean } = {}): Promise<Response> {
  if (options.bypassRoleCheck !== true && legacyQueueWritesFrozen()) return legacyQueueFrozenResponse(req.method, `/api/tasks/${task.id}/move`, "task-move");
  if (options.bypassRoleCheck !== true && !serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${task.id}/move`);
  const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${task.id}/move`);
  if (notReady !== null) return notReady;
  const body = await readJson(req);
  const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
  const queueId = normalizeQueueId(record.queueId ?? record.id);
  const movedAt = nowIso();
  const hotBlocker = taskMoveBlocker(task);
  if (hotBlocker.length > 0) {
    const databaseTask = databaseReady ? await loadTaskFromDatabase(task.id) : null;
    if (databaseTask !== null) task = reconcileHotTaskFromDatabase(databaseTask);
    return jsonResponse({
      ok: false,
      error: `cannot move task ${task.id}: ${hotBlocker}`,
      task: taskForResponse(task),
      databaseTask: databaseTask === null ? null : taskForResponse(databaseTask),
    }, 409);
  }
  const databaseMove = await moveDatabaseTaskToQueue(task.id, queueId, movedAt);
  if (!databaseMove.ok) {
    const databaseTask = databaseReady ? await loadTaskFromDatabase(task.id) : null;
    if (databaseTask !== null) task = reconcileHotTaskFromDatabase(databaseTask);
    return jsonResponse({
      ok: false,
      error: `cannot move task ${task.id}: ${databaseMove.blocker}`,
      blocker: databaseStatusRowJson(databaseMove.row),
      task: taskForResponse(task),
      databaseTask: databaseTask === null ? null : taskForResponse(databaseTask),
    }, databaseMove.row === null ? 404 : 409);
  }
  const previousQueueId = databaseMove.previousQueueId ?? queueIdOf(task);
  const queue = ensureQueue(queueId);
  queue.updatedAt = movedAt;
  markQueueDirty(queue.id);
  task.queueId = queueId;
  task.queueEnteredAt = movedAt;
  task.updatedAt = movedAt;
  appendOutput(task, "system", `moved from queue=${previousQueueId} to queue=${queueId}\n`, "queue/move");
  if (task.status === "queued" || task.status === "retry_wait") armIdleNotification();
  persistState();
  publishQueueEvent("task-moved", previousQueueId);
  publishQueueEvent("task-moved", queueId);
  logger("info", "task_moved_queue", { taskId: task.id, previousQueueId, queueId, status: task.status });
  if (task.status === "queued" || task.status === "retry_wait") scheduleQueue(queueId);
  await flushDirtyTasksToDatabase(true);
  return jsonResponse({ ok: true, task: taskForResponse(task), queue: await queueSummaryForResponse() }, 202);
}

async function backfillOaTraceStats(url: URL): Promise<JsonValue> {
  const limitRaw = Number(url.searchParams.get("limit") ?? 2000);
  const limit = Number.isInteger(limitRaw) && limitRaw > 0 ? Math.min(20_000, limitRaw) : 2000;
  const taskId = url.searchParams.get("taskId")?.trim() ?? "";
  const tasks = taskId.length > 0
    ? [await findTaskForRead(taskId)].filter((task): task is QueueTask => task !== null)
    : (await loadAllTasksForRead()).slice(-limit);
  const includeSteps = url.searchParams.get("steps") !== "0";
  let stepEventCount = 0;
  for (const task of tasks) {
    const queueId = queueIdOf(task);
    const outputMaxSeq = taskOutputMaxSeq(task);
    const output = taskFullOutput(task);
    const traceStats = traceStatsFromOutputs(output);
    const attemptBySeq = outputAttemptIndexMap(output);
    if (includeSteps) {
      for (const item of output) {
        const projectionOutput = traceStepOutputForProjection(task, item);
        if (outputStartsTraceStepInHistory(output, item)) {
          publishCodeQueueTraceStep(task, queueId, projectionOutput, outputMaxSeq, attemptBySeq.get(item.seq) ?? null);
          stepEventCount += 1;
        } else if (outputUpdatesExistingTraceStep(item)) {
          publishCodeQueueTraceStep(task, queueId, projectionOutput, outputMaxSeq, attemptBySeq.get(projectionOutput.seq) ?? attemptBySeq.get(item.seq) ?? null, String(item.text || "").length);
          stepEventCount += 1;
        }
      }
    }
    publishCodeQueueTraceStatsSnapshot(task, queueId, "backfill", traceStats.stepCount, outputMaxSeq, traceStats);
  }
  logger("info", "oa_trace_stats_backfill_enqueued", { taskCount: tasks.length, limit, taskId: taskId || null, includeSteps, stepEventCount });
  return { ok: true, taskCount: tasks.length, limit, taskId: taskId || null, includeSteps, stepEventCount, eventFlowBaseUrl: config.oaEventFlowBaseUrl } as unknown as JsonValue;
}

async function route(req: Request): Promise<Response> {
  const url = new URL(req.url);
  if (req.method === "OPTIONS") return jsonResponse({ ok: true });
  try {
    if (url.pathname === "/live") return jsonResponse({
      ok: true,
      service: "code-queue",
      instanceId: config.instanceId,
      role: config.serviceRole,
      ...runtimeIdentity(),
      deploy: {
        commit: config.deployCommit,
        requestedCommit: config.deployRequestedCommit,
      },
      databaseReady,
      serviceReady,
      startedAt: serviceStartedAt,
    });
    if (url.pathname === "/" || url.pathname === "/health") {
      if (!databaseReady) return jsonResponse({
        ok: false,
        service: "code-queue",
        instanceId: config.instanceId,
        role: config.serviceRole,
        ...runtimeIdentity(),
        deploy: {
          commit: config.deployCommit,
          requestedCommit: config.deployRequestedCommit,
        },
        status: "starting",
        databaseReady,
        databaseLastError,
        modelProviderConfig: codeModelProviderSourceContract(config),
        skills: collectSkillsStatus(),
        skillsSync: collectSkillsSyncPreflight(),
        startedAt: serviceStartedAt,
      }, 503);
      return jsonResponse({
        ok: true,
        service: "code-queue",
        instanceId: config.instanceId,
        role: config.serviceRole,
        ...runtimeIdentity(),
        deploy: {
          commit: config.deployCommit,
          requestedCommit: config.deployRequestedCommit,
        },
        schedulerEnabled: config.schedulerEnabled,
        schedulerPollIntervalMs: config.schedulerPollIntervalMs,
        queue: queueSummary(false, state.tasks),
        modelProviderConfig: codeModelProviderSourceContract(config),
        executionDiagnostics: executionDiagnosticsForTasks(state.tasks),
        skills: collectSkillsStatus(),
        skillsSync: collectSkillsSyncPreflight(),
        egressProxy: await providerGatewayEgressProxyStatus(),
        oaEventPublisher: oaEventPublisherStatus(),
        startedAt: serviceStartedAt,
      });
    }
    if (url.pathname === "/logs") return jsonResponse({ ok: true, logs: recentLogs.slice(-parseLimit(url)) });
    if (url.pathname === "/api/events" && req.method === "GET") return jsonResponse({ ok: false, error: "Code Queue private SSE was removed; subscribe to oa-event-flow /api/events/stream with service:code-queue tags." }, 410);
    if (url.pathname === "/api/dev-ready" && req.method === "GET") return jsonResponse({ ok: true, devReady: collectDevReady() });
    if (url.pathname === "/api/skills-sync" && req.method === "GET") {
      const dryRun = url.searchParams.get("dryRun") === "1" || url.searchParams.get("dry-run") === "1";
      if (!dryRun) return jsonResponse({
        ok: false,
        error: "skills sync is dry-run only from the runtime API; pass dryRun=1",
        mutation: false,
        valuesPrinted: false,
      }, 400);
      return jsonResponse({ ok: true, skillsSync: collectSkillsSyncPreflight() });
    }
    if (url.pathname === "/api/runtime-preflight" && req.method === "GET") {
      return jsonResponse({
        ok: true,
        runtimePreflight: runtimePreflightJson(collectRuntimePreflight({
          includeRemote: url.searchParams.get("remote") === "1",
          includePushDryRun: url.searchParams.get("pushDryRun") === "1",
          pushDryRunRef: url.searchParams.get("pushDryRunRef") ?? undefined,
          includePrCreateDryRun: url.searchParams.get("prCreateDryRun") === "1",
          prCreateDryRunHead: url.searchParams.get("prCreateDryRunHead") ?? undefined,
          issueNumber: Number(url.searchParams.get("issue") ?? url.searchParams.get("issueNumber") ?? NaN),
        })),
      });
    }
    if (url.pathname === "/api/dev-containers" && req.method === "GET") {
      const plan = buildDevContainerPlan(normalizeProviderId(config.devContainerDefaultProviderId) ?? "D601", {});
      return jsonResponse({
        ok: true,
        defaultProviderId: plan.providerId,
        startEndpoint: `/api/dev-containers/${encodeURIComponent(plan.providerId)}/start`,
        statusEndpoint: `/api/dev-containers/${encodeURIComponent(plan.providerId)}/status`,
        masterProxyMode: "ssh-tun-nat-sealed",
        defaultPlan: {
          containerName: plan.containerName,
          image: plan.image,
          workdir: plan.workdir,
          masterHost: plan.masterHost,
          tunName: plan.tunName,
          serverIp: plan.serverIp,
          clientIp: plan.clientIp,
          natChain: plan.natChain,
          egressFirewallChain: plan.egressFirewallChain,
        },
      });
    }
    const devContainerStartMatch = url.pathname.match(/^\/api\/dev-containers(?:\/([^/]+))?\/start$/u);
    if (devContainerStartMatch !== null && req.method === "POST") return await startDevContainer(req, devContainerStartMatch[1] === undefined ? null : decodeURIComponent(devContainerStartMatch[1]));
    const devContainerStatusMatch = url.pathname.match(/^\/api\/dev-containers(?:\/([^/]+))?\/status$/u);
    if (devContainerStatusMatch !== null && req.method === "GET") return await devContainerStatus(devContainerStatusMatch[1] === undefined ? null : decodeURIComponent(devContainerStatusMatch[1]));
    if (url.pathname === "/api/judge/probe" && (req.method === "GET" || req.method === "POST")) return await runJudgeProbe();
    if (url.pathname === "/api/judge/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runJudgeInfraSelfTest());
    if (url.pathname === "/api/queue-order/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runQueueOrderingSelfTest());
    if (url.pathname === "/api/queue-claim-move/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(await runQueueClaimMoveSelfTest());
    if (url.pathname === "/api/reference-injection/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(await runReferenceInjectionSelfTest());
    if (url.pathname === "/api/trace-port/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runTracePortSelfTest());
    if (url.pathname === "/api/trace-summary-contract/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(runTraceSummaryContractSelfTest());
    if (url.pathname === "/api/stale-active-recovery/self-test" && (req.method === "GET" || req.method === "POST")) return jsonResponse(await runStaleActiveRecoverySelfTest());
    if (url.pathname === "/api/scheduler/reconcile" && (req.method === "GET" || req.method === "POST")) return await reconcileStaleActiveTasksFromApi(url, req);
    if (url.pathname === "/api/oa/backfill" && (req.method === "GET" || req.method === "POST")) return jsonResponse(await backfillOaTraceStats(url));
    if (url.pathname === "/api/notifications/claudeqq" && req.method === "GET") {
      await loadClaudeQqNotificationOutboxFromDatabase();
      const limit = parseLimit(url);
      return jsonResponse({ ok: true, stats: claudeQqNotificationOutboxStats(), items: claudeQqNotificationItems(limit) });
    }
    if (url.pathname === "/api/notifications/claudeqq/drain" && req.method === "POST") {
      return jsonResponse(await drainClaudeQqNotificationOutbox("api"));
    }
    if (url.pathname === "/api/notifications/claudeqq/backfill" && req.method === "POST") {
      const body = await readJson(req);
      const record = typeof body === "object" && body !== null && !Array.isArray(body) ? body as Record<string, unknown> : {};
      const since = typeof record.since === "string" ? record.since : url.searchParams.get("since");
      const dryRun = record.dryRun === true || url.searchParams.get("dryRun") === "1";
      const rawLimit = Number(record.limit ?? url.searchParams.get("limit") ?? 100);
      const limit = Number.isInteger(rawLimit) && rawLimit > 0 ? Math.min(500, rawLimit) : 100;
      return jsonResponse(await backfillClaudeQqTaskNotifications(since, limit, dryRun));
    }
    if (url.pathname === "/api/queues" && req.method === "GET") {
      const tasks = await loadAllTasksForRead();
      const queueRecords = await loadQueuesFromDatabase();
      return jsonResponse({ ok: true, queues: perQueueSummaries(tasks, queueRecords), queue: await queueSummaryForResponse(false, tasks, queueRecords) });
    }
    if (url.pathname === "/api/workdirs" && req.method === "GET") return await listWorkdirs(url);
    if (url.pathname === "/api/workdirs" && req.method === "POST") return await createWorkdir(req);
    const workdirMatch = url.pathname.match(/^\/api\/workdirs\/([^/]+)\/([^/]+)\/(.+)$/u);
    if (workdirMatch !== null && req.method === "DELETE") {
      return await deleteWorkdir(
        decodeURIComponent(workdirMatch[1] ?? ""),
        decodeURIComponent(workdirMatch[2] ?? ""),
        decodeURIComponent(workdirMatch[3] ?? ""),
      );
    }
    if (url.pathname === "/api/queues" && req.method === "POST") return await createQueue(req);
    if (url.pathname === "/api/queues/merge" && req.method === "POST") return await mergeQueues(null, req);
    const queueMergeMatch = url.pathname.match(/^\/api\/queues\/([^/]+)\/merge$/u);
    if (queueMergeMatch !== null && req.method === "POST") return await mergeQueues(decodeURIComponent(queueMergeMatch[1] ?? ""), req);
    const queueMatch = url.pathname.match(/^\/api\/queues\/([^/]+)$/u);
    if (queueMatch !== null && (req.method === "PATCH" || req.method === "PUT" || req.method === "POST")) return await updateQueue(decodeURIComponent(queueMatch[1] ?? ""), req);
    if (url.pathname === "/api/tasks/read-all" && req.method === "POST") return await markTerminalTasksRead(url);
    if (url.pathname === "/api/tasks/stats" && req.method === "GET") {
      const queueId = url.searchParams.get("queueId");
      const allTasks = await loadAllTasksForRead();
      const statsTasks = queueId === null ? allTasks : allTasks.filter((task) => queueIdOf(task) === safeQueueId(queueId));
      return jsonResponse({ ok: true, statistics: taskStatisticsSummary(statsTasks, statsDaysFromUrl(url)), queue: queueSummary(false, allTasks) });
    }
    if (url.pathname === "/api/tasks/overview" && req.method === "GET") return await tasksOverviewResponse(url);
    if (url.pathname === "/api/tasks" && req.method === "GET") {
      const status = url.searchParams.get("status");
      const queueId = url.searchParams.get("queueId");
      const lite = url.searchParams.get("lite") === "1";
      const includeDevReady = url.searchParams.get("devReady") !== "0" && !lite;
      const searchTerms = taskSearchTerms(url);
      const allTasks = await loadAllTasksForRead();
      const queueFilteredTasks = queueId === null ? allTasks : allTasks.filter((task) => queueIdOf(task) === safeQueueId(queueId));
      const filteredTasks = allTasks
        .filter((task) => status === null || task.status === status)
        .filter((task) => queueId === null || queueIdOf(task) === safeQueueId(queueId))
        .filter((task) => taskMatchesSearch(task, searchTerms));
      const limit = parseLimit(url);
      const page = taskPageRows(filteredTasks, url, limit);
      const traceStats = await readOaTraceStatsForTasks(page.rows.map((task) => task.id));
      const tasks = page.rows.map((task) => applyOaTraceStatsToTaskJson(taskForListResponse(task, lite, allTasks), traceStats.get(`task:${task.id}`) ?? null));
      return jsonResponse({
        ok: true,
        queue: queueSummary(includeDevReady, allTasks),
        statistics: taskStatisticsSummary(queueFilteredTasks, statsDaysFromUrl(url)),
        tasks,
        pagination: {
          limit,
          returned: page.returned,
          total: page.total,
          hasMore: page.hasMore,
          nextBeforeId: page.nextBeforeId,
          beforeId: page.beforeId,
          includeActive: page.includeActive,
        },
      });
    }
    if ((url.pathname === "/api/tasks" || url.pathname === "/api/tasks/batch") && req.method === "POST") return await createTasks(req);
    const outputMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/output$/u);
    if (outputMatch !== null && req.method === "GET") {
      const task = await findTaskForRead(decodeURIComponent(outputMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      return await outputChunkResponse(task, url);
    }
    const transcriptMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/transcript$/u);
    if (transcriptMatch !== null && req.method === "GET") {
      const task = await findTaskForRead(decodeURIComponent(transcriptMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      return await transcriptChunkResponse(task, url);
    }
    const promptMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/prompt$/u);
    if (promptMatch !== null && req.method === "GET") {
      const task = await findTaskForRead(decodeURIComponent(promptMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      return await taskPromptDetailResponse(task, url);
    }
    if (promptMatch !== null && req.method === "PATCH") {
      const task = await findTaskForMutation(decodeURIComponent(promptMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      return await editQueuedTaskPrompt(task, req);
    }
    const traceSummaryMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/trace-summary$/u);
    if (traceSummaryMatch !== null && req.method === "GET") {
      const task = await findTaskForRead(decodeURIComponent(traceSummaryMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      const attemptIndexes = traceAttemptIndexesForTask(task);
      const [traceStats, allOaSteps, attemptOaStepsEntries] = await Promise.all([
        readOaTraceStatsForTaskAttempts(task.id, attemptIndexes),
        readOaTraceStepsForTask(task.id, null).catch((error) => {
          logger("warn", "oa_trace_summary_steps_read_failed", { taskId: task.id, error: errorToJson(error) });
          return [];
        }),
        Promise.all(attemptIndexes.map(async (attemptIndex) => {
          const steps = await readOaTraceStepsForTask(task.id, attemptIndex).catch((error) => {
            logger("warn", "oa_trace_summary_attempt_steps_read_failed", { taskId: task.id, attemptIndex, error: errorToJson(error) });
            return [];
          });
          return [attemptIndex, steps] as const;
        })),
      ]);
      const attemptOaSteps = new Map(attemptOaStepsEntries);
      return jsonResponse({ ok: true, summary: taskTraceSummaryResponse(task, traceStats.get(`task:${task.id}`) ?? null, traceStats, { allSteps: allOaSteps, attemptSteps: attemptOaSteps }) });
    }
    const traceStepsMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/trace-steps$/u);
    if (traceStepsMatch !== null && req.method === "GET") {
      const task = await findTaskForRead(decodeURIComponent(traceStepsMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      return await taskTraceStepsResponse(task, url);
    }
    const traceStepMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/trace-step$/u);
    if (traceStepMatch !== null && req.method === "GET") {
      const task = await findTaskForRead(decodeURIComponent(traceStepMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      return await taskTraceStepDetailResponse(task, url);
    }
    const steerConfirmationMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/steer-confirmation$/u);
    if (steerConfirmationMatch !== null && req.method === "GET") {
      const task = await findTaskForRead(decodeURIComponent(steerConfirmationMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      const steerId = normalizeSteerId(url.searchParams.get("steerId"));
      if (steerId === null) return jsonResponse({ ok: false, error: "steerId is required" }, 400);
      return jsonResponse({
        ok: true,
        confirmation: steerTraceConfirmationJson(findSteerTraceConfirmation(task, steerId, taskFullOutput(task))),
      });
    }
    const judgeTaskMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/judge$/u);
    if (judgeTaskMatch !== null && (req.method === "GET" || req.method === "POST")) {
      const task = await findTaskForRead(decodeURIComponent(judgeTaskMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      return await runSingleTaskJudge(task, url);
    }
    const summaryMatch = url.pathname.match(/^\/api\/tasks\/([^/]+)\/summary$/u);
    if (summaryMatch !== null && req.method === "GET") {
      const task = await findTaskForRead(decodeURIComponent(summaryMatch[1] ?? ""));
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      return jsonResponse({ ok: true, summary: taskSummaryResponse(task, url) });
    }
    const match = url.pathname.match(/^\/api\/tasks\/([^/]+)(?:\/(retry|resume|steer|interrupt|move|read|edit))?$/u);
    if (match !== null) {
      const action = match[2];
      const taskId = decodeURIComponent(match[1] ?? "");
      if ((action === "retry" || action === "move" || action === "read" || action === "edit") && req.method !== "GET") {
        if (!serviceRoleAllowsWrite(config.serviceRole)) return readOnlyRejectResponse(req.method, `/api/tasks/${taskId}/${action}`);
        const notReady = requireDatabaseReadyForWrite(req.method, `/api/tasks/${taskId}/${action}`);
        if (notReady !== null) return notReady;
      }
      if (action === "read" && req.method === "POST") return await markTaskReadById(taskId);
      const task = action === undefined && req.method === "GET"
        ? await findTaskForRead(taskId)
        : await findTaskForMutation(taskId);
      if (task === null) return jsonResponse({ ok: false, error: "task not found" }, 404);
      if (action === "retry" && req.method === "POST") return await manualRetry(task, req);
      if (action === "resume" && req.method === "POST") return await resumeTask(task, req);
      if (action === "steer" && req.method === "POST") return await steerTask(task, req);
      if (action === "interrupt" && req.method === "POST") return await interruptTask(task, req.method);
      if (action === "move" && req.method === "POST") return await moveTaskToQueue(task, req);
      if (action === "edit" && (req.method === "POST" || req.method === "PATCH")) return await editQueuedTaskPrompt(task, req);
      if (action !== undefined) return jsonResponse({ ok: false, error: "not found" }, 404);
      if (req.method === "GET") {
        const traceStats = await readOaTraceStatsForTask(task.id);
        if (url.searchParams.get("meta") === "1") return jsonResponse({ ok: true, task: applyOaTraceStatsToTaskJson(taskForMetaResponse(task), traceStats) });
        const includeRaw = url.searchParams.get("raw") === "1" || url.searchParams.get("full") === "1";
        return jsonResponse({ ok: true, task: applyOaTraceStatsToTaskJson(taskForResponse(task, true, includeRaw), traceStats) });
      }
      if (req.method === "DELETE") return await interruptTask(task, req.method);
      return jsonResponse({ ok: false, error: "method not allowed" }, 405);
    }
    return jsonResponse({ ok: false, error: "not found", path: url.pathname }, 404);
  } catch (error) {
    const requestError = requestErrorResponse(error);
    if (requestError !== null) {
      logger("warn", "request_rejected", { path: url.pathname, error: errorToJson(error) });
      return requestError;
    }
    logger("error", "request_failed", { path: url.pathname, error: error instanceof Error ? error.stack ?? error.message : String(error) });
    return jsonResponse({ ok: false, error: error instanceof Error ? error.message : String(error) }, 500);
  }
}

installShutdownHandlers();
prepareCodexHome();
prepareOpenCodeHome();
startCodexSqliteLogExporter();
startMemoryWatchdog();
Bun.serve({ hostname: config.host, port: config.port, idleTimeout: 120, fetch: route });
logger("info", "service_listening", { port: config.port, instanceId: config.instanceId, role: config.serviceRole, schedulerEnabled: config.schedulerEnabled, schedulerPollIntervalMs: config.schedulerPollIntervalMs, workdir: config.defaultWorkdir, defaultModel: config.defaultModel, judgeConfigured: config.minimaxApiKey.length > 0, storage: "postgres", databaseReady });

async function startDatabaseBackedRuntime(): Promise<void> {
  if (serviceReady) return;
  logger("info", "service_started", { port: config.port, instanceId: config.instanceId, role: config.serviceRole, schedulerEnabled: config.schedulerEnabled, schedulerPollIntervalMs: config.schedulerPollIntervalMs, workdir: config.defaultWorkdir, defaultModel: config.defaultModel, judgeConfigured: config.minimaxApiKey.length > 0, storage: "postgres" });
  const devReady = collectDevReady() as Record<string, JsonValue>;
  logger(devReady.ok === true ? "info" : "warn", "dev_ready_check", devReady);
  const inactiveActiveTurnsCleared = clearInactiveActiveTurnIds();
  const startupRecovered = config.schedulerEnabled ? queueActiveTasksForRestartRetry("Service restarted while task was active", "startup") : 0;
  const startupRecoveryDirtyTaskCount = dirtyDatabaseTaskIds.size;
  if (startupRecovered > 0 || startupRecoveryDirtyTaskCount > 0) {
    logger("warn", "startup_requeued_active_tasks", { recovered: startupRecovered, inactiveActiveTurnsCleared, dirtyTaskCount: startupRecoveryDirtyTaskCount });
    await flushDirtyTasksToDatabase(true);
    logger("info", "startup_requeued_active_tasks_flushed", { recovered: startupRecovered, inactiveActiveTurnsCleared, dirtyTaskCount: startupRecoveryDirtyTaskCount });
  }
  persistState();
  serviceReady = true;
  void refreshSchedulerTasksFromDatabase("startup").catch((error) => {
    setDatabaseLastError(error);
    logger("warn", "scheduler_database_startup_refresh_failed", { error: errorToJson(error) });
  });
  startSchedulerDatabasePoller();
  if (config.startupOaBackfillEnabled) {
    setTimeout(() => { void backfillOaTraceStats(new URL("http://code-queue.local/api/oa/backfill?limit=2000")).catch((error) => logger("warn", "oa_trace_stats_startup_backfill_failed", { error: errorToJson(error) })); }, 1000).unref?.();
  }
  scheduleQueue();
  scheduleClaudeQqNotificationDrain(1000);
}

void initDatabasePersistenceWithRetry()
  .then(() => startDatabaseBackedRuntime())
  .catch((error) => {
    setDatabaseLastError(error);
    logger("error", "database_persistence_init_failed", { error: errorToJson(error) });
  });
