// 重构前 index.ts 只读参考：commit 6a04144d3f5103014f75b637d7e6bc2f45bf007f，blob 56e590c1a6b5ca7ad128bf2c992f60e46c355a58；可用 `git show 6a04144d3f5103014f75b637d7e6bc2f45bf007f:src/components/microservices/code-queue/src/index.ts` 查看。

import postgres from "postgres";
import type { ClaudeQqNotificationItem, ClaudeQqNotificationOutboxState, ClaudeQqNotificationRow, JsonValue, QueueTask, RuntimeConfig } from "./types";

type SqlExecutor = postgres.Sql | postgres.TransactionSql;

export interface NotificationsContext {
  config: Pick<RuntimeConfig,
    "notifyClaudeQqBaseUrl" | "notifyClaudeQqEnabled" | "notifyClaudeQqGroupId" | "notifyClaudeQqMaxOutboxItems" | "notifyClaudeQqMaxResponseChars" | "notifyClaudeQqRetryIntervalMs" | "notifyClaudeQqSendAttempts" | "notifyClaudeQqTargetType" | "notifyClaudeQqTimeoutMs" | "notifyClaudeQqUserId"
  >;
  activeRunCount: () => number;
  activeRunSlotCount: () => number;
  activeRunTaskIds: () => string[];
  errorToJson: (error: unknown) => JsonValue;
  hasRunnableTask: () => boolean;
  lastAssistantMessage: (task: QueueTask) => JsonValue;
  loadAllTasksForRead: () => Promise<QueueTask[]>;
  logger: (level: "debug" | "info" | "warn" | "error", message: string, data?: JsonValue) => void;
  nonNegativeElapsed: (startMs: number | null, endMs: number | null) => number | null;
  nowIso: () => string;
  processingQueueCount: () => number;
  queueCount: () => number;
  queueIdOf: (task: QueueTask) => string;
  safePreview: (value: string, max?: number) => string;
  shutdownRequested: () => boolean;
  sql: () => postgres.Sql;
  taskTimestamp: (value: string | null) => string | null;
  tasks: () => QueueTask[];
  timestampMs: (value: string | null | undefined) => number | null;
  databaseReady: () => boolean;
}

let context: NotificationsContext | null = null;
let claudeQqNotificationOutbox: ClaudeQqNotificationOutboxState = { version: 1, updatedAt: new Date().toISOString(), items: [] };
const sentTaskNotificationKeys = new Set<string>();
const inFlightTaskNotificationKeys = new Set<string>();
let idleNotificationSent = true;
let idleNotificationInFlight = false;
let claudeQqNotificationDrainTimer: ReturnType<typeof setTimeout> | null = null;
let claudeQqNotificationDrainInFlight = false;

export function configureNotifications(runtimeContext: NotificationsContext): void {
  context = runtimeContext;
}

function ctx(): NotificationsContext {
  if (context === null) throw new Error("notifications module is not configured");
  return context;
}

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

function durationMsBetween(startAt: string | null | undefined, endAt: string | null | undefined): number | null {
  return ctx().nonNegativeElapsed(ctx().timestampMs(startAt), ctx().timestampMs(endAt));
}

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 emptyClaudeQqNotificationOutbox(): ClaudeQqNotificationOutboxState {
  return { version: 1, updatedAt: ctx().nowIso(), items: [] };
}

function notificationItemFromRow(row: ClaudeQqNotificationRow): ClaudeQqNotificationItem {
  const createdAt = ctx().taskTimestamp(String(row.created_at)) ?? ctx().nowIso();
  const updatedAt = ctx().taskTimestamp(String(row.updated_at)) ?? createdAt;
  return {
    id: row.id,
    kind: row.kind || "unknown",
    dedupKey: row.dedup_key || row.id,
    target: row.target || "-",
    message: row.message || "",
    createdAt,
    updatedAt,
    attempts: Number.isInteger(row.attempts) && row.attempts >= 0 ? row.attempts : 0,
    nextAttemptAt: ctx().taskTimestamp(String(row.next_attempt_at)) ?? updatedAt,
    lastError: typeof row.last_error === "string" && row.last_error.length > 0 ? row.last_error : null,
    sentAt: row.sent_at === null ? null : ctx().taskTimestamp(String(row.sent_at)),
  };
}

async function loadClaudeQqNotificationOutboxFromDatabase(client: SqlExecutor = ctx().sql()): Promise<void> {
  const rows = await client<ClaudeQqNotificationRow[]>`
    SELECT id, kind, dedup_key, target, message, created_at, updated_at, attempts, next_attempt_at, last_error, sent_at
    FROM unidesk_code_queue_notifications
    ORDER BY created_at ASC, id ASC
  `;
  claudeQqNotificationOutbox = {
    version: 1,
    updatedAt: ctx().nowIso(),
    items: rows.map(notificationItemFromRow).filter((item) => item.id.length > 0 && item.message.length > 0),
  };
  pruneClaudeQqNotificationOutbox();
}

async function upsertClaudeQqNotificationToDatabase(client: SqlExecutor, item: ClaudeQqNotificationItem): Promise<void> {
  await client`
    INSERT INTO unidesk_code_queue_notifications (
      id,
      kind,
      dedup_key,
      target,
      message,
      created_at,
      updated_at,
      attempts,
      next_attempt_at,
      last_error,
      sent_at
    ) VALUES (
      ${item.id},
      ${item.kind},
      ${item.dedupKey},
      ${item.target},
      ${item.message},
      ${ctx().taskTimestamp(item.createdAt) ?? ctx().nowIso()},
      ${ctx().taskTimestamp(item.updatedAt) ?? ctx().nowIso()},
      ${item.attempts},
      ${ctx().taskTimestamp(item.nextAttemptAt) ?? ctx().nowIso()},
      ${item.lastError},
      ${ctx().taskTimestamp(item.sentAt)}
    )
    ON CONFLICT (id) DO UPDATE SET
      kind = EXCLUDED.kind,
      dedup_key = EXCLUDED.dedup_key,
      target = EXCLUDED.target,
      message = EXCLUDED.message,
      updated_at = EXCLUDED.updated_at,
      attempts = EXCLUDED.attempts,
      next_attempt_at = EXCLUDED.next_attempt_at,
      last_error = EXCLUDED.last_error,
      sent_at = EXCLUDED.sent_at
  `;
}

async function persistClaudeQqNotificationItem(item: ClaudeQqNotificationItem): Promise<void> {
  if (!ctx().databaseReady()) throw new Error("PostgreSQL is not ready for ClaudeQQ notification outbox");
  claudeQqNotificationOutbox.updatedAt = ctx().nowIso();
  const deletedIds = pruneClaudeQqNotificationOutbox();
  const stillPresent = claudeQqNotificationOutbox.items.some((candidate) => candidate.id === item.id);
  await ctx().sql().begin(async (client) => {
    if (stillPresent) await upsertClaudeQqNotificationToDatabase(client, item);
    for (const id of deletedIds) await client`DELETE FROM unidesk_code_queue_notifications WHERE id = ${id}`;
  });
  if (stillPresent && !claudeQqNotificationOutbox.items.some((candidate) => candidate.id === item.id)) {
    claudeQqNotificationOutbox.items.push(item);
    pruneClaudeQqNotificationOutbox();
  }
}

async function persistClaudeQqNotificationOutbox(): Promise<void> {
  if (!ctx().databaseReady()) throw new Error("PostgreSQL is not ready for ClaudeQQ notification outbox");
  claudeQqNotificationOutbox.updatedAt = ctx().nowIso();
  const deletedIds = pruneClaudeQqNotificationOutbox();
  await ctx().sql().begin(async (client) => {
    for (const item of claudeQqNotificationOutbox.items) await upsertClaudeQqNotificationToDatabase(client, item);
    for (const id of deletedIds) await client`DELETE FROM unidesk_code_queue_notifications WHERE id = ${id}`;
  });
}


function queueNotificationStats(): Record<string, JsonValue> {
  const counts = ctx().tasks().reduce<Record<string, number>>((memo, task) => {
    memo[task.status] = (memo[task.status] ?? 0) + 1;
    return memo;
  }, {});
  const activeTaskIds = Array.from(new Set([
    ...ctx().activeRunTaskIds(),
    ...ctx().tasks().filter((task) => task.status === "running" || task.status === "judging").map((task) => task.id),
  ])).sort();
  const activeTask = activeTaskIds.length > 0 ? ctx().tasks().find((task) => task.id === activeTaskIds[0]) ?? null : null;
  const queued = (counts.queued ?? 0) + (counts.retry_wait ?? 0);
  const running = (counts.running ?? 0) + (counts.judging ?? 0);
  return {
    running,
    queued,
    retryWait: counts.retry_wait ?? 0,
    judging: counts.judging ?? 0,
    total: ctx().tasks().length,
    queueCount: ctx().queueCount(),
    processingQueueCount: ctx().processingQueueCount(),
    activeRunCount: ctx().activeRunCount(),
    activeRunSlotCount: ctx().activeRunSlotCount(),
    activeTaskIds,
    activeTaskElapsed: activeTask === null ? "-" : formatDurationMs(durationMsBetween(activeTask.startedAt ?? activeTask.createdAt, ctx().nowIso())),
  };
}

function notificationTargetConfigured(): boolean {
  if (!ctx().config.notifyClaudeQqEnabled) return false;
  return ctx().config.notifyClaudeQqTargetType === "group"
    ? ctx().config.notifyClaudeQqGroupId.length > 0
    : ctx().config.notifyClaudeQqUserId.length > 0;
}

function claudeQqTargetPayload(message: string): Record<string, string> {
  if (ctx().config.notifyClaudeQqTargetType === "group") {
    return { targetType: "group", groupId: ctx().config.notifyClaudeQqGroupId, message };
  }
  return { targetType: "private", userId: ctx().config.notifyClaudeQqUserId, message };
}

function notificationTargetLabel(): string {
  return ctx().config.notifyClaudeQqTargetType === "group"
    ? `group:${ctx().config.notifyClaudeQqGroupId || "-"}`
    : `private:${ctx().config.notifyClaudeQqUserId || "-"}`;
}

function claudeQqSendPayload(message: string): Record<string, string> {
  return ctx().config.notifyClaudeQqTargetType === "group"
    ? { groupId: ctx().config.notifyClaudeQqGroupId, message }
    : { userId: ctx().config.notifyClaudeQqUserId, message };
}

function truncateNotificationText(value: string, maxChars: number): string {
  if (value.length <= maxChars) return value;
  return `${value.slice(0, maxChars)}\n\n...[Code Queue notification truncated: ${value.length - maxChars} chars omitted; use CLI/WebUI for the full trace]`;
}

function taskFinalResponseForNotification(task: QueueTask): string {
  const last = ctx().lastAssistantMessage(task) as Record<string, unknown>;
  const text = typeof last.text === "string" ? last.text.trimEnd() : "";
  if (text.trim().length > 0) return text;
  if (typeof task.lastError === "string" && task.lastError.trim().length > 0) return `（没有最终 assistant response；lastError: ${task.lastError.trim()}）`;
  return "（没有最终 assistant response）";
}

function taskNotificationKey(task: QueueTask): string {
  return `${task.id}:${task.status}:${task.finishedAt ?? task.updatedAt}:${task.attempts.length}`;
}

function rememberTaskNotificationKey(key: string): void {
  sentTaskNotificationKeys.add(key);
  while (sentTaskNotificationKeys.size > 1000) {
    const oldest = sentTaskNotificationKeys.values().next().value as string | undefined;
    if (oldest === undefined) break;
    sentTaskNotificationKeys.delete(oldest);
  }
}

function claudeQqNotificationId(kind: string, dedupKey: string): string {
  return `${kind}:${dedupKey}`;
}

function claudeQqNotificationOutboxStats(): Record<string, JsonValue> {
  const pending = claudeQqNotificationOutbox.items.filter((item) => item.sentAt === null);
  const failed = pending.filter((item) => item.lastError !== null);
  const oldestPending = pending.reduce<string | null>((oldest, item) => {
    if (oldest === null) return item.createdAt;
    return (ctx().timestampMs(item.createdAt) ?? 0) < (ctx().timestampMs(oldest) ?? 0) ? item.createdAt : oldest;
  }, null);
  return {
    storage: "postgres",
    total: claudeQqNotificationOutbox.items.length,
    pending: pending.length,
    failed: failed.length,
    sent: claudeQqNotificationOutbox.items.length - pending.length,
    inFlight: claudeQqNotificationDrainInFlight,
    nextDueAt: pending
      .map((item) => item.nextAttemptAt)
      .sort((left, right) => (ctx().timestampMs(left) ?? 0) - (ctx().timestampMs(right) ?? 0))[0] ?? null,
    oldestPendingAt: oldestPending,
  };
}

function pruneClaudeQqNotificationOutbox(): string[] {
  const beforeIds = new Set(claudeQqNotificationOutbox.items.map((item) => item.id));
  const pending = claudeQqNotificationOutbox.items.filter((item) => item.sentAt === null);
  const sent = claudeQqNotificationOutbox.items
    .filter((item) => item.sentAt !== null)
    .sort((left, right) => (ctx().timestampMs(right.sentAt) ?? 0) - (ctx().timestampMs(left.sentAt) ?? 0));
  const sentBudget = Math.max(0, ctx().config.notifyClaudeQqMaxOutboxItems - pending.length);
  claudeQqNotificationOutbox.items = [...pending, ...sent.slice(0, sentBudget)]
    .sort((left, right) => (ctx().timestampMs(left.createdAt) ?? 0) - (ctx().timestampMs(right.createdAt) ?? 0));
  const afterIds = new Set(claudeQqNotificationOutbox.items.map((item) => item.id));
  return Array.from(beforeIds).filter((id) => !afterIds.has(id));
}

function claudeQqNotificationRetryDelayMs(attempts: number): number {
  const exponent = Math.max(0, Math.min(8, attempts - 1));
  return Math.min(30 * 60_000, ctx().config.notifyClaudeQqRetryIntervalMs * (2 ** exponent));
}

function scheduleClaudeQqNotificationDrain(delayMs = ctx().config.notifyClaudeQqRetryIntervalMs): void {
  if (!notificationTargetConfigured() || ctx().shutdownRequested()) return;
  if (claudeQqNotificationOutbox.items.every((item) => item.sentAt !== null)) return;
  if (claudeQqNotificationDrainTimer !== null) return;
  claudeQqNotificationDrainTimer = setTimeout(() => {
    claudeQqNotificationDrainTimer = null;
    void drainClaudeQqNotificationOutbox("timer").catch((error) => ctx().logger("warn", "claudeqq_notify_outbox_drain_failed", { trigger: "timer", error: ctx().errorToJson(error) }));
  }, Math.max(1000, delayMs));
}

async function enqueueClaudeQqNotification(kind: string, dedupKey: string, message: string): Promise<boolean> {
  if (!notificationTargetConfigured()) return false;
  const id = claudeQqNotificationId(kind, dedupKey);
  const existing = claudeQqNotificationOutbox.items.find((item) => item.id === id);
  if (existing !== undefined && existing.sentAt !== null) return false;
  const at = ctx().nowIso();
  let item: ClaudeQqNotificationItem;
  if (existing !== undefined) {
    existing.message = message;
    existing.target = notificationTargetLabel();
    existing.updatedAt = at;
    existing.nextAttemptAt = at;
    existing.lastError = null;
    item = existing;
  } else {
    item = {
      id,
      kind,
      dedupKey,
      target: notificationTargetLabel(),
      message,
      createdAt: at,
      updatedAt: at,
      attempts: 0,
      nextAttemptAt: at,
      lastError: null,
      sentAt: null,
    };
    claudeQqNotificationOutbox.items.push(item);
  }
  await persistClaudeQqNotificationItem(item);
  void drainClaudeQqNotificationOutbox(`enqueue:${kind}`).catch((error) => ctx().logger("warn", "claudeqq_notify_outbox_drain_failed", { trigger: `enqueue:${kind}`, error: ctx().errorToJson(error) }));
  return true;
}

function dueClaudeQqNotificationItems(limit = 3): ClaudeQqNotificationItem[] {
  const nowMs = Date.now();
  return claudeQqNotificationOutbox.items
    .filter((item) => item.sentAt === null && (ctx().timestampMs(item.nextAttemptAt) ?? 0) <= nowMs)
    .sort((left, right) => (ctx().timestampMs(left.nextAttemptAt) ?? 0) - (ctx().timestampMs(right.nextAttemptAt) ?? 0))
    .slice(0, limit);
}

async function drainClaudeQqNotificationOutbox(trigger = "manual"): Promise<Record<string, JsonValue>> {
  if (!notificationTargetConfigured()) return { ok: true, trigger, skipped: "not_configured" };
  if (claudeQqNotificationDrainInFlight) return { ok: true, trigger, skipped: "in_flight" };
  claudeQqNotificationDrainInFlight = true;
  let sent = 0;
  let failed = 0;
  try {
    await loadClaudeQqNotificationOutboxFromDatabase();
    const due = dueClaudeQqNotificationItems();
    for (const item of due) {
      item.attempts += 1;
      item.updatedAt = ctx().nowIso();
      await persistClaudeQqNotificationItem(item);
      try {
        await postClaudeQqText(item.kind, item.message);
        item.sentAt = ctx().nowIso();
        item.updatedAt = item.sentAt;
        item.lastError = null;
        sent += 1;
        if (item.kind === "task_terminal") rememberTaskNotificationKey(item.dedupKey);
        ctx().logger("info", "claudeqq_notify_outbox_sent", { id: item.id, kind: item.kind, target: item.target, attempts: item.attempts, trigger });
      } catch (error) {
        failed += 1;
        const message = error instanceof Error ? error.message : String(error);
        item.lastError = ctx().safePreview(message, 1000);
        item.updatedAt = ctx().nowIso();
        item.nextAttemptAt = new Date(Date.now() + claudeQqNotificationRetryDelayMs(item.attempts)).toISOString();
        ctx().logger("warn", "claudeqq_notify_outbox_retry_scheduled", {
          id: item.id,
          kind: item.kind,
          target: item.target,
          attempts: item.attempts,
          nextAttemptAt: item.nextAttemptAt,
          trigger,
          error: ctx().errorToJson(error),
        });
      } finally {
        await persistClaudeQqNotificationItem(item);
      }
    }
  } finally {
    claudeQqNotificationDrainInFlight = false;
    await persistClaudeQqNotificationOutbox();
    if (claudeQqNotificationOutbox.items.some((item) => item.sentAt === null)) scheduleClaudeQqNotificationDrain();
  }
  return { ok: true, trigger, sent, failed, outbox: claudeQqNotificationOutboxStats() };
}

async function postClaudeQqText(kind: string, message: string): Promise<void> {
  if (!notificationTargetConfigured()) return;
  const primaryUrl = `${ctx().config.notifyClaudeQqBaseUrl}/api/push/text`;
  const legacyUrl = `${ctx().config.notifyClaudeQqBaseUrl}/api/send/text`;
  let lastError: unknown = null;
  for (let attempt = 1; attempt <= ctx().config.notifyClaudeQqSendAttempts; attempt += 1) {
    const controller = new AbortController();
    const timer = setTimeout(() => controller.abort(), ctx().config.notifyClaudeQqTimeoutMs);
    let responseText = "";
    try {
      let response = await fetch(primaryUrl, {
        method: "POST",
        headers: { "content-type": "application/json" },
        body: JSON.stringify(claudeQqTargetPayload(message)),
        signal: controller.signal,
      });
      responseText = await response.text();
      if (response.status === 404 || response.status === 405) {
        response = await fetch(legacyUrl, {
          method: "POST",
          headers: { "content-type": "application/json" },
          body: JSON.stringify(claudeQqSendPayload(message)),
          signal: controller.signal,
        });
        responseText = await response.text();
      }
      if (!response.ok) throw new Error(`ClaudeQQ proxy returned HTTP ${response.status}: ${ctx().safePreview(responseText, 500)}`);
      try {
        const parsed = JSON.parse(responseText) as Record<string, unknown>;
        if (parsed.ok === false || parsed.success === false || parsed.status === "napcat_offline") {
          throw new Error(`ClaudeQQ push failed: ${ctx().safePreview(responseText, 500)}`);
        }
      } catch (error) {
        if (error instanceof SyntaxError) {
          // Some deployments return plain text; HTTP 2xx is still accepted.
        } else {
          throw error;
        }
      }
      ctx().logger("info", "claudeqq_notify_sent", { kind, target: notificationTargetLabel(), attempt, chars: message.length, responsePreview: ctx().safePreview(responseText, 500) });
      return;
    } catch (error) {
      lastError = error;
      if (attempt >= ctx().config.notifyClaudeQqSendAttempts) break;
      const delayMs = Math.min(30_000, 1000 * (2 ** (attempt - 1)));
      ctx().logger("warn", "claudeqq_notify_retry", { kind, target: notificationTargetLabel(), attempt, nextDelayMs: delayMs, error: ctx().errorToJson(error) });
      await Bun.sleep(delayMs);
    } finally {
      clearTimeout(timer);
    }
  }
  if (lastError !== null) {
    throw lastError instanceof Error ? lastError : new Error(String(lastError));
  }
}

function taskTerminalNotificationMessage(task: QueueTask): string {
  const stats = queueNotificationStats();
  const totalElapsed = formatDurationMs(durationMsBetween(task.createdAt, task.finishedAt ?? task.updatedAt));
  const runElapsed = formatDurationMs(durationMsBetween(task.startedAt ?? task.createdAt, task.finishedAt ?? task.updatedAt));
  const response = truncateNotificationText(taskFinalResponseForNotification(task), ctx().config.notifyClaudeQqMaxResponseChars);
  return [
    "Code Queue 任务结束",
    `task: ${task.id}`,
    `queue: ${ctx().queueIdOf(task)}`,
    `status: ${task.status}`,
    `provider: ${task.providerId}`,
    `cwd: ${task.cwd}`,
    `model: ${task.model}${task.reasoningEffort ? ` (${task.reasoningEffort})` : ""}`,
    `attempts: ${task.attempts.length}/${task.maxAttempts}`,
    `elapsed: total=${totalElapsed}, run=${runElapsed}`,
    `current queue: running=${stats.running}, queued=${stats.queued}, retry_wait=${stats.retryWait}`,
    "",
    "Final response:",
    response,
  ].join("\n");
}

async function notifyTaskTerminal(task: QueueTask): Promise<void> {
  if (!terminalTask(task) || !notificationTargetConfigured()) return;
  const key = taskNotificationKey(task);
  if (sentTaskNotificationKeys.has(key) || inFlightTaskNotificationKeys.has(key)) return;
  inFlightTaskNotificationKeys.add(key);
  try {
    await enqueueClaudeQqNotification("task_terminal", key, taskTerminalNotificationMessage(task));
  } catch (error) {
    ctx().logger("warn", "claudeqq_task_notify_failed", { taskId: task.id, status: task.status, target: notificationTargetLabel(), error: ctx().errorToJson(error) });
  } finally {
    inFlightTaskNotificationKeys.delete(key);
  }
}

async function backfillClaudeQqTaskNotifications(since: string | null, limit: number, dryRun: boolean): Promise<Record<string, JsonValue>> {
  if (!notificationTargetConfigured()) return { ok: true, skipped: "not_configured" };
  await loadClaudeQqNotificationOutboxFromDatabase();
  const sinceMs = ctx().timestampMs(since);
  const candidates = ctx().tasks()
    .filter((task) => terminalTask(task))
    .filter((task) => sinceMs === null || (ctx().timestampMs(task.finishedAt ?? task.updatedAt) ?? 0) > sinceMs)
    .sort((left, right) => (ctx().timestampMs(left.finishedAt ?? left.updatedAt) ?? 0) - (ctx().timestampMs(right.finishedAt ?? right.updatedAt) ?? 0))
    .slice(0, limit);
  const items = candidates.map((task) => ({
    taskId: task.id,
    status: task.status,
    queueId: ctx().queueIdOf(task),
    finishedAt: task.finishedAt ?? task.updatedAt,
    dedupKey: taskNotificationKey(task),
    alreadyInOutbox: claudeQqNotificationOutbox.items.some((item) => item.id === claudeQqNotificationId("task_terminal", taskNotificationKey(task))),
  }));
  if (dryRun) return { ok: true, dryRun, since: since ?? null, scanned: candidates.length, enqueued: 0, items };
  let enqueued = 0;
  for (const task of candidates) {
    if (await enqueueClaudeQqNotification("task_terminal", taskNotificationKey(task), taskTerminalNotificationMessage(task))) enqueued += 1;
  }
  return { ok: true, dryRun, since: since ?? null, scanned: candidates.length, enqueued, outbox: claudeQqNotificationOutboxStats(), items };
}

function armIdleNotification(): void {
  if (notificationTargetConfigured()) idleNotificationSent = false;
}

async function maybeNotifyQueueIdle(triggerTaskId: string | null = null): Promise<void> {
  if (!notificationTargetConfigured() || idleNotificationSent || idleNotificationInFlight) return;
  const stats = queueNotificationStats();
  if (Number(stats.running) !== 0 || Number(stats.queued) !== 0 || ctx().processingQueueCount() !== 0 || ctx().activeRunCount() !== 0) return;
  idleNotificationInFlight = true;
  try {
    const message = [
      "Code Queue 已空闲",
      "running=0, queued=0",
      `total tasks=${stats.total}, queues=${stats.queueCount}`,
      triggerTaskId === null ? "" : `last task=${triggerTaskId}`,
    ].filter((line) => line.length > 0).join("\n");
    await enqueueClaudeQqNotification("queue_idle", `queue_idle:${triggerTaskId ?? "unknown"}:${stats.total}`, message);
    idleNotificationSent = true;
  } catch (error) {
    ctx().logger("warn", "claudeqq_idle_notify_failed", { triggerTaskId: triggerTaskId ?? "", target: notificationTargetLabel(), error: ctx().errorToJson(error) });
  } finally {
    idleNotificationInFlight = false;
  }
}







function claudeQqNotificationOutboxItemCount(): number {
  return claudeQqNotificationOutbox.items.length;
}

function claudeQqNotificationItems(limit: number): JsonValue[] {
  return claudeQqNotificationOutbox.items.slice(-limit).map((item) => ({
    id: item.id,
    kind: item.kind,
    dedupKey: item.dedupKey,
    target: item.target,
    createdAt: item.createdAt,
    updatedAt: item.updatedAt,
    attempts: item.attempts,
    nextAttemptAt: item.nextAttemptAt,
    lastError: item.lastError,
    sentAt: item.sentAt,
    messagePreview: ctx().safePreview(item.message, 500),
    messageChars: item.message.length,
  })) as unknown as JsonValue[];
}

export {
  armIdleNotification,
  backfillClaudeQqTaskNotifications,
  claudeQqNotificationItems,
  claudeQqNotificationOutboxItemCount,
  claudeQqNotificationOutboxStats,
  drainClaudeQqNotificationOutbox,
  loadClaudeQqNotificationOutboxFromDatabase,
  maybeNotifyQueueIdle,
  notificationTargetConfigured,
  notificationTargetLabel,
  notifyTaskTerminal,
  persistClaudeQqNotificationOutbox,
  scheduleClaudeQqNotificationDrain,
};
