import { randomUUID } from 'node:crypto';
import { hostname } from 'node:os';
import { pathToFileURL } from 'node:url';
import { fileURLToPath } from 'node:url';
import { resolve } from 'node:path';
import Fastify, { LogController, type FastifyReply, type FastifyRequest } from 'fastify';
import cookie from '@fastify/cookie';
import helmet from '@fastify/helmet';
import rateLimit from '@fastify/rate-limit';
import fastifyStatic from '@fastify/static';
import pino, { type DestinationStream } from 'pino';
import { z } from 'zod';
import { loadConfig, type AppConfig } from './config.js';
import { assertDatabaseSchema, closePool, databaseReadiness, databaseReady, getPool } from './db.js';
import { AuthError, AuthService, type ActiveSession } from './auth.js';
import {
  AgentBusManager,
  AgentBusChannelService,
  mergeRuntimeChannelStatuses
} from './agentbus-channels.js';
import { LeaderNotificationService } from './leader-notification-service.js';
import {
  TaskError,
  TaskService,
  canUseTaskDataPlane,
  canViewOperationsDashboard,
  type ParseDecisionInput,
  type ParseTaskClaim,
  type TaskBrowserCommand,
  type TaskContext,
  type TaskEvent
} from './task-service.js';
import { ParserOrchestrator, type AiParser as ExternalParser } from './parser-orchestrator.js';
import { BUSINESS_ROUTES, businessRouteById } from './business-routes.js';
import {
  decodeInlineInputAttachment,
  InputAttachmentError,
  type EncodedTaskInputAttachment,
  type TaskInputAttachmentInput
} from './input-attachment.js';
import {
  diagnosticDurationMs,
  diagnosticError,
  diagnosticMetadataKeys,
  diagnosticRequestPath,
  normalizeRequestId,
  writeEmergencyDiagnostic
} from './diagnostics.js';

export function aiServiceConnected(databaseIsReady: boolean, probe: Record<string, unknown>): boolean {
  return databaseIsReady && probe.configured === true && probe.reachable === true;
}

const loginSchema = z.object({
  username: z.string().min(1).max(160),
  password: z.string().min(1)
});

const changePasswordSchema = z.object({
  current_password: z.string().min(1),
  new_password: z.string().min(1)
});

const accountCreateSchema = z.object({
  username: z.string().min(1).max(160),
  password: z.string().min(1),
  role: z.enum(['admin', 'team_lead', 'user']).default('user'),
  erp_account: z.string().trim().max(200).optional(),
  business_route_ids: z.array(
    z.string().trim().refine((routeId) => Boolean(businessRouteById(routeId)), '业务类型不存在。')
  ).max(BUSINESS_ROUTES.length).default([])
    .refine((routeIds) => new Set(routeIds).size === routeIds.length, '业务类型不能重复。')
});

const accountUpdateSchema = z.object({
  role: z.enum(['admin', 'team_lead', 'user']).optional(),
  is_active: z.boolean().optional(),
  erp_account: z.string().trim().max(200).nullable().optional()
}).refine((body) => body.role !== undefined || body.is_active !== undefined || body.erp_account !== undefined, {
  message: '至少提供一个账号更新字段。'
});

const accountPasswordResetSchema = z.object({
  password: z.string().min(1)
});

const accountBusinessAuthorizationsSchema = z.object({
  business_route_ids: z.array(
    z.string().trim().refine((routeId) => Boolean(businessRouteById(routeId)), '业务类型不存在。')
  ).max(BUSINESS_ROUTES.length)
    .refine((routeIds) => new Set(routeIds).size === routeIds.length, '业务类型不能重复。'),
  expected_revision: z.number().int().min(0)
});

const encodedInputAttachmentSchema = z.object({
  name: z.string().min(1).max(200),
  content_type: z.string().max(200).optional(),
  size: z.number().int().positive().max(50_000_000).optional(),
  sha256: z.string().regex(/^[a-f0-9]{64}$/i).optional(),
  content_base64: z.string().min(4).max(70_000_000)
});

const createTaskSchema = z.object({
  raw_text: z.string().min(1).max(200_000),
  idempotency_key: z.string().min(8).max(200).optional(),
  conversation_id: z.string().min(1).max(200).optional(),
  attachments: z.array(encodedInputAttachmentSchema).max(1).optional()
});

const taskMessageSchema = z.object({
  message: z.string().max(200_000).default(''),
  task_id: z.string().min(1).max(200).optional(),
  conversation_id: z.string().min(1).max(200).optional(),
  idempotency_key: z.string().min(8).max(200).optional(),
  attachments: z.array(encodedInputAttachmentSchema).max(1).optional()
}).refine(
  (body) => Boolean(body.message.trim()) || Boolean(body.attachments?.length),
  { message: '补充信息或名单附件至少提供一项。' }
);

const claimSchema = z.object({ connection_id: z.string().min(1).max(200) });
const resultSchema = z.object({
  connection_id: z.string().min(1).max(200),
  execution_id: z.string().uuid(),
  result: z.record(z.unknown())
});
const heartbeatSchema = z.object({
  connection_id: z.string().min(1).max(200),
  extension_version: z.string().max(80).optional(),
  erp_account: z.string().trim().max(200).optional(),
  erp_account_matched: z.boolean().optional(),
  metadata: z.record(z.unknown()).optional()
});
const automationSettingsSchema = z.object({ enabled: z.boolean() });
const parserRoutingUpdateSchema = z.object({
  mode: z.enum(['ai', 'shadow', 'auto', 'program']),
  expected_revision: z.number().int().min(0)
});
const parserEmergencyAiSchema = z.object({ reason: z.string().trim().min(1).max(1_000) });
const parserReparseSchema = z.object({ engine: z.literal('ai'), reason: z.string().trim().min(1).max(1_000) });
const parserDecisionReviewSchema = z.object({
  verdict: z.enum(['equivalent', 'program_correct', 'ai_correct', 'both_wrong']),
  note: z.string().trim().max(2_000).optional()
});
const channelCreateSchema = z.object({
  display_name: z.string().min(1).max(120),
  owner_user_id: z.string().uuid(),
  external_user_ref: z.string().max(200).optional(),
  agentbus_key: z.string().min(1).max(4_000),
  bot_address: z.string().max(200).optional(),
  enabled: z.boolean().optional()
});
const channelUpdateSchema = z.object({
  display_name: z.string().min(1).max(120).optional(),
  owner_user_id: z.string().uuid().optional(),
  external_user_ref: z.string().max(200).optional(),
  bot_address: z.string().max(200).optional(),
  enabled: z.boolean().optional()
});
const channelRotateKeySchema = z.object({ agentbus_key: z.string().min(1).max(4_000) });
const listTasksQuerySchema = z.object({
  status: z.string().max(80).optional(),
  search: z.string().max(200).optional(),
  limit: z.coerce.number().int().min(1).max(200).default(200),
  offset: z.coerce.number().int().min(0).max(1_000_000).default(0),
  archive: z.enum(['active', 'archived', 'all']).default('active'),
  executable_by: z.enum(['me']).optional(),
  include_total: z.preprocess(
    (value) => value === undefined ? true : String(value).toLowerCase() !== 'false',
    z.boolean()
  ).default(true)
});
const taskEventsQuerySchema = z.object({
  since: z.coerce.number().int().min(0).default(0),
  executable_by: z.literal('me').default('me')
});
const taskIdListSchema = z.array(z.string().trim().min(1).max(200)).min(1).max(100);
const taskBulkDeleteSchema = z.object({
  task_ids: taskIdListSchema
}).refine(
  (body) => new Set(body.task_ids).size === body.task_ids.length,
  { message: '任务编号不能重复。', path: ['task_ids'] }
);
const taskBulkArchiveSchema = z.object({
  task_ids: taskIdListSchema,
  reason: z.string().trim().max(500).optional()
}).refine(
  (body) => new Set(body.task_ids).size === body.task_ids.length,
  { message: '任务编号不能重复。', path: ['task_ids'] }
);

const taskArchiveSchema = z.object({ reason: z.string().trim().max(500).optional() });

const auditQuerySchema = z.object({
  event_type: z.string().trim().max(200).optional(),
  actor_user_id: z.string().uuid().optional(),
  entity_type: z.string().trim().max(120).optional(),
  limit: z.coerce.number().int().min(1).max(200).default(100),
  offset: z.coerce.number().int().min(0).max(1_000_000).default(0)
});

const operationsDashboardQuerySchema = z.object({
  from: z.string().datetime({ offset: true }).optional(),
  to: z.string().datetime({ offset: true }).optional(),
  actor_user_id: z.string().uuid().optional(),
  business_route_id: z.string().trim().max(120).optional(),
  status: z.enum(['all', 'active', 'completed', 'attention', 'failed', 'cancelled', 'archived']).default('all'),
  search: z.string().trim().max(200).optional(),
  limit: z.coerce.number().int().min(1).max(100).default(20),
  offset: z.coerce.number().int().min(0).max(1_000_000).default(0)
});

const LOG_REDACTION_PATHS = [
  'req.headers.authorization',
  'req.headers.cookie',
  'req.headers["x-csrf-token"]',
  'request.headers.authorization',
  'request.headers.cookie',
  'authorization',
  'cookie',
  'password',
  'api_key',
  'token',
  'access_key',
  '*.authorization',
  '*.cookie',
  '*.password',
  '*.api_key',
  '*.token',
  '*.access_key'
];

export function createControlPlaneLogger(config: AppConfig, destination?: DestinationStream) {
  const options = {
    level: config.LOG_LEVEL,
    base: {
      pid: process.pid,
      hostname: hostname(),
      service: 'ltjt-control-plane',
      environment: config.NODE_ENV,
      deployment_revision: config.DEPLOYMENT_REVISION
    },
    redact: {
      paths: LOG_REDACTION_PATHS,
      censor: '[REDACTED]'
    }
  };
  return destination ? pino(options, destination) : pino(options);
}

function agentBusDiagnosticMetadata(metadata: Record<string, unknown>): Record<string, unknown> {
  const rawEvent = String(metadata.agentbus_event || 'event');
  const event = /^[a-z0-9_]{1,80}$/u.test(rawEvent) ? rawEvent : 'event';
  return {
    ...metadata,
    diagnostic_event: `agentbus.${event}`,
    diagnostic_stage: 'agentbus'
  };
}

function requestId(request: FastifyRequest): string {
  return String(request.id);
}

function requestTaskId(request: FastifyRequest): string | undefined {
  const params = request.params && typeof request.params === 'object'
    ? request.params as Record<string, unknown>
    : {};
  const taskId = String(params.taskId || '').trim();
  return taskId ? taskId.slice(0, 200) : undefined;
}

function clientAddress(request: FastifyRequest): string {
  return String(request.headers['x-forwarded-for'] || request.ip || '').split(',')[0].trim();
}

function userAgent(request: FastifyRequest): string {
  return String(request.headers['user-agent'] || '').slice(0, 500);
}

function workerErrorCode(error: unknown): string {
  const code = error && typeof error === 'object' ? String((error as { code?: unknown }).code || '') : '';
  if (/^[A-Za-z0-9_]{2,80}$/.test(code)) return `error_${code.toLowerCase()}`;
  return 'parser_exception';
}

function workerFailureResult(errorCode: string, stage: string) {
  return {
    status: 'agent_parse_blocked',
    blockers: ['解析 Worker 未能在规定时间内取得确定结果，任务已阻断，请检查解析服务后手动重试。'],
    operation: null,
    error_code: errorCode,
    external_request: {
      service: 'ltjt-control-plane',
      transport: 'internal',
      stage,
      event_count: 0,
      session_id_present: false,
      error_code: errorCode
    }
  };
}

function decodeInputAttachments(
  values: EncodedTaskInputAttachment[] | undefined,
  config: AppConfig,
  source: TaskInputAttachmentInput['source'] = 'manual'
): TaskInputAttachmentInput[] {
  try {
    return (values || []).map((value) => decodeInlineInputAttachment(value, config.ARTIFACT_MAX_BYTES, source));
  } catch (error) {
    if (error instanceof InputAttachmentError) {
      throw new TaskError(
        error.code,
        error.message,
        error.code === 'roster_file_too_large' ? 413 : 400,
        error.details
      );
    }
    throw error;
  }
}

function waitMs(milliseconds: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, milliseconds));
}

function requireSameOrigin(config: AppConfig, request: FastifyRequest): void {
  const origin = request.headers.origin;
  if (origin && origin !== config.APP_ORIGIN) {
    throw new AuthError('origin_mismatch', '请求来源不受信任。', 403);
  }
}

function setAuthNoStore(reply: FastifyReply): void {
  reply.header('Cache-Control', 'no-store, no-cache, must-revalidate, max-age=0');
  reply.header('Pragma', 'no-cache');
}

export function isTaskDataPlaneRoute(routePath: unknown): boolean {
  const path = String(routePath || '');
  return path === '/api/messages'
    || path === '/api/connections/heartbeat'
    || path === '/api/events'
    || path === '/api/tasks'
    || path.startsWith('/api/tasks/')
    || path.startsWith('/api/parser-decisions/')
    || path === '/api/operations-dashboard'
    || path.startsWith('/api/operations-dashboard/');
}

function attachmentContentDisposition(fileName: string): string {
  const safeName = String(fileName || 'team-file.bin')
    .replace(/[\\"\r\n\u0000-\u001f\u007f]/g, '_')
    .trim()
    .slice(0, 200) || 'team-file.bin';
  const asciiFallback = safeName.replace(/[^\x20-\x7e]/g, '_') || 'team-file.bin';
  return `attachment; filename="${asciiFallback}"; filename*=UTF-8''${encodeURIComponent(safeName)}`;
}

const PERSISTENT_SESSION_COOKIE_MAX_AGE_SECONDS = 2_147_483_647;

function sessionCookieOptions(config: AppConfig) {
  const maxAge = PERSISTENT_SESSION_COOKIE_MAX_AGE_SECONDS;
  return {
    httpOnly: true,
    secure: config.NODE_ENV === 'production',
    sameSite: 'lax' as const,
    path: '/',
    maxAge,
    // The server session has no automatic expiry. The long-lived cookie keeps
    // the browser logged in across restarts until the user logs out or the
    // server explicitly revokes the session.
    expires: new Date(Date.now() + maxAge * 1_000)
  };
}

function canonicalStaticRedirect(config: AppConfig, request: FastifyRequest): string | null {
  if (!['GET', 'HEAD'].includes(request.method)) return null;
  const configuredOrigin = new URL(config.APP_ORIGIN);
  const forwardedHost = request.headers['x-forwarded-host'];
  const requestHost = String(forwardedHost || request.headers.host || '').split(',')[0].trim();
  if (!requestHost || requestHost.toLowerCase() === configuredOrigin.host.toLowerCase()) return null;

  const incomingUrl = new URL(request.raw.url || '/', 'http://local.invalid');
  const target = new URL(config.APP_ORIGIN);
  target.pathname = incomingUrl.pathname || '/';
  target.search = incomingUrl.search;
  return target.toString();
}

function publicUser(session: ActiveSession) {
  return {
    id: session.user.id,
    username: session.user.username,
    role: session.user.role,
    erp_account: session.user.erpAccount
  };
}

async function loadExternalParser(): Promise<ExternalParser> {
  const modulePath = resolve(process.cwd(), 'LianSyn-platform/server.mjs');
  const module = await import(pathToFileURL(modulePath).href) as {
    createExternalParserFromEnv: () => ExternalParser;
  };
  return module.createExternalParserFromEnv();
}

export async function buildServer({
  config = loadConfig(),
  parser,
  startParserLoop = true,
  loggerDestination
}: {
  config?: AppConfig;
  parser?: ExternalParser;
  startParserLoop?: boolean;
  loggerDestination?: DestinationStream;
} = {}) {
  const requestStartedAt = new WeakMap<FastifyRequest, bigint>();
  const app = Fastify({
    loggerInstance: createControlPlaneLogger(config, loggerDestination),
    logController: new LogController({ disableRequestLogging: true }),
    genReqId: (rawRequest) => normalizeRequestId(rawRequest.headers['x-request-id']),
    trustProxy: true,
    bodyLimit: Math.min(75_000_000, Math.max(2_000_000, Math.ceil(config.ARTIFACT_MAX_BYTES * 1.4) + 1_000_000))
  });
  await app.register(cookie);
  await app.register(helmet, { contentSecurityPolicy: false });
  await app.register(rateLimit, { global: false });
  await app.register(fastifyStatic, {
    root: resolve(process.cwd(), 'LianSyn-platform'),
    prefix: '/'
  });

  app.log.info({
    diagnostic_event: 'service.initialized',
    diagnostic_stage: 'startup',
    host: config.HOST,
    port: config.PORT,
    log_level: config.LOG_LEVEL,
    database_schema: config.DATABASE_SCHEMA,
    database_ssl: config.DATABASE_SSL,
    artifact_storage_backend: config.ARTIFACT_STORAGE_BACKEND,
    agentbus_enabled: config.agentBusEnabled,
    leader_summary_webhook_enabled: config.leaderSummaryWebhookEnabled,
    leader_summary_webhook_configuration_error: config.leaderSummaryWebhookConfigurationError,
    parser_loop_enabled: startParserLoop,
    data_retention_enabled: config.DATA_RETENTION_ENABLED,
    raw_payload_logging: config.AGENTBUS_LOG_PAYLOADS
  }, 'control plane initialized');

  app.addHook('onRequest', async (request, reply) => {
    requestStartedAt.set(request, process.hrtime.bigint());
    reply.header('x-request-id', requestId(request));
    const contentLength = Number(request.headers['content-length']);
    request.log.info({
      diagnostic_event: 'http.request.started',
      diagnostic_stage: 'http',
      request_id: requestId(request),
      method: request.method,
      path: diagnosticRequestPath(request.url),
      ...(Number.isSafeInteger(contentLength) && contentLength >= 0 ? { content_length: contentLength } : {})
    }, 'HTTP request started');
  });

  app.addHook('onResponse', async (request, reply) => {
    const startedAt = requestStartedAt.get(request);
    requestStartedAt.delete(request);
    const metadata = {
      diagnostic_event: 'http.request.completed',
      diagnostic_stage: 'http',
      request_id: requestId(request),
      method: request.method,
      path: diagnosticRequestPath(request.routeOptions.url || request.url),
      status_code: reply.statusCode,
      ...(startedAt ? { duration_ms: diagnosticDurationMs(startedAt) } : {})
    };
    if (reply.statusCode >= 500) request.log.error(metadata, 'HTTP request completed with server error');
    else if (reply.statusCode >= 400) request.log.warn(metadata, 'HTTP request completed with client error');
    else request.log.info(metadata, 'HTTP request completed');
  });

  app.get('/history', async (_request, reply) => {
    reply.header('Cache-Control', 'no-store');
    return reply.sendFile('index.html');
  });

  app.get('/channels', async (_request, reply) => {
    reply.header('Cache-Control', 'no-store');
    return reply.sendFile('index.html');
  });

  app.get('/parser-routing', async (_request, reply) => {
    reply.header('Cache-Control', 'no-store');
    return reply.sendFile('index.html');
  });

  app.get('/accounts', async (_request, reply) => {
    reply.header('Cache-Control', 'no-store');
    return reply.sendFile('index.html');
  });

  app.get('/audit', async (_request, reply) => {
    reply.header('Cache-Control', 'no-store');
    return reply.sendFile('index.html');
  });

  app.get('/operations-dashboard', async (_request, reply) => {
    reply.header('Cache-Control', 'no-store');
    return reply.sendFile('index.html');
  });

  app.addHook('onRequest', async (request, reply) => {
    if (request.url.startsWith('/api/') || request.url.startsWith('/health/')) return;
    const redirectUrl = canonicalStaticRedirect(config, request);
    if (redirectUrl) return reply.redirect(redirectUrl, 308);
  });

  const auth = new AuthService(config, {
    info: (metadata, message) => app.log.info(metadata, message),
    warn: (metadata, message) => app.log.warn(metadata, message),
    error: (metadata, message) => app.log.error(metadata, message)
  });
  const tasks = new TaskService(config, undefined, {
    info: (metadata, message) => app.log.info(metadata, message),
    warn: (metadata, message) => app.log.warn(metadata, message),
    error: (metadata, message) => app.log.error(metadata, message)
  });
  const externalParser = parser || await loadExternalParser();
  const parserOrchestrator = new ParserOrchestrator(externalParser);
  const activeParseWorkers = new Map<string, string>();
  let parseQueueInFlight: Promise<void> | null = null;
  let lastParseQueueErrorAt = 0;
  let aiProbeValue: unknown;
  let aiProbeExpiresAt = 0;
  let aiProbeInFlight: Promise<unknown> | null = null;
  const channelService = new AgentBusChannelService(config, {
    info: (metadata, message) => app.log.info(agentBusDiagnosticMetadata(metadata), message),
    warn: (metadata, message) => app.log.warn(agentBusDiagnosticMetadata(metadata), message),
    error: (metadata, message) => app.log.error(agentBusDiagnosticMetadata(metadata), message)
  });
  const leaderNotificationService = new LeaderNotificationService(config, {
    info: (metadata, message) => app.log.info(metadata, message),
    warn: (metadata, message) => app.log.warn(metadata, message),
    error: (metadata, message) => app.log.error(metadata, message)
  });
  let agentBus: AgentBusManager | null = null;

  const getSession = async (request: FastifyRequest): Promise<ActiveSession> => {
    const session = await auth.getActiveSession(request.cookies[config.SESSION_COOKIE_NAME]);
    if (!session) throw new AuthError('authentication_required', '请先登录。', 401);
    return session;
  };

  const requireAdmin = (session: ActiveSession): ActiveSession => {
    if (session.user.role !== 'admin') throw new AuthError('admin_required', '需要管理员权限。', 403);
    return session;
  };

  const requireLeadership = (session: ActiveSession): ActiveSession => {
    if (!canViewOperationsDashboard(session.user.role)) {
      throw new AuthError('leadership_required', '需要组长权限。', 403);
    }
    return session;
  };

  const requireTaskDataPlane = (session: ActiveSession): ActiveSession => {
    if (!canUseTaskDataPlane(session.user.role)) {
      throw new AuthError('task_access_forbidden', '管理员账号仅用于平台管理，不能访问业务任务。', 403);
    }
    return session;
  };

  const contextFor = (session: ActiveSession, request: FastifyRequest): TaskContext => ({
    organizationId: session.user.organizationId,
    userId: session.user.id,
    requestId: requestId(request),
    role: session.user.role,
    source: 'manual'
  });

  const requireAuthenticatedMutationSession = async (request: FastifyRequest): Promise<ActiveSession> => {
    requireSameOrigin(config, request);
    const session = await getSession(request);
    const csrf = String(request.headers['x-csrf-token'] || '');
    if (!(await auth.verifyCsrf(session, csrf))) {
      throw new AuthError('csrf_failed', '请求校验失败，请刷新页面后重试。', 403);
    }
    return session;
  };

  const requireMutationSession = requireAuthenticatedMutationSession;

  const requireTaskSession = async (request: FastifyRequest): Promise<ActiveSession> => (
    requireTaskDataPlane(await getSession(request))
  );

  const requireTaskMutationSession = async (request: FastifyRequest): Promise<ActiveSession> => (
    requireTaskDataPlane(await requireMutationSession(request))
  );

  const requireAdminSession = async (request: FastifyRequest): Promise<ActiveSession> => (
    requireAdmin(await getSession(request))
  );

  const requireAdminMutationSession = async (request: FastifyRequest): Promise<ActiveSession> => (
    requireAdmin(await requireMutationSession(request))
  );

  const requireLeadershipSession = async (request: FastifyRequest): Promise<ActiveSession> => (
    requireLeadership(await getSession(request))
  );

  app.addHook('preHandler', async (request) => {
    if (!isTaskDataPlaneRoute(request.routeOptions.url)) return;
    requireTaskDataPlane(await getSession(request));
  });

  async function persistParseOutcome(
    claim: ParseTaskClaim,
    result: unknown,
    decision?: ParseDecisionInput
  ): Promise<void> {
    const startedAt = process.hrtime.bigint();
    const context: TaskContext = {
      organizationId: claim.task.organization_id,
      userId: '',
      requestId: `parse:${claim.task.task_id}:attempt:${claim.attemptNo}`
    };
    let lastError: unknown;
    for (let attempt = 0; attempt < 3; attempt += 1) {
      try {
        const finalized = await tasks.applyParseResult(
          context,
          claim.task.task_id,
          result,
          claim.workerId,
          claim.attemptNo,
          decision
        );
        app.log.info({
          diagnostic_event: 'parser.outcome.persisted',
          diagnostic_stage: 'parser_persistence',
          request_id: context.requestId,
          task_id: finalized.task_id,
          attempt_no: claim.attemptNo,
          status: finalized.status,
          duration_ms: diagnosticDurationMs(startedAt)
        }, 'parse task finalized');
        return;
      } catch (error) {
        if (error instanceof TaskError && ['stale_parse_result', 'task_cancelled', 'task_not_found'].includes(error.code)) {
          app.log.info({
            diagnostic_event: 'parser.outcome.ignored',
            diagnostic_stage: 'parser_persistence',
            request_id: context.requestId,
            task_id: claim.task.task_id,
            attempt_no: claim.attemptNo,
            error_code: error.code,
            duration_ms: diagnosticDurationMs(startedAt)
          }, 'late parse outcome ignored');
          return;
        }
        lastError = error;
        app.log.warn({
          diagnostic_event: 'parser.outcome.persist_retry',
          diagnostic_stage: 'parser_persistence',
          request_id: context.requestId,
          task_id: claim.task.task_id,
          attempt_no: claim.attemptNo,
          persistence_attempt: attempt + 1,
          retrying: attempt < 2,
          ...diagnosticError(error, workerErrorCode(error))
        }, 'parse outcome persistence attempt failed');
        if (attempt < 2) await waitMs(250 * (attempt + 1));
      }
    }
    app.log.error({
      diagnostic_event: 'parser.outcome.persist_failed',
      diagnostic_stage: 'parser_persistence',
      request_id: context.requestId,
      task_id: claim.task.task_id,
      attempt_no: claim.attemptNo,
      duration_ms: diagnosticDurationMs(startedAt),
      ...diagnosticError(lastError, workerErrorCode(lastError))
    }, 'parse outcome persistence failed after retries');
  }

  async function runParseTask(claim: ParseTaskClaim): Promise<void> {
    const startedAt = process.hrtime.bigint();
    const parseRequestId = `parse:${claim.task.task_id}:attempt:${claim.attemptNo}`;
    app.log.info({
      diagnostic_event: 'parser.worker.started',
      diagnostic_stage: 'parser_execution',
      request_id: parseRequestId,
      task_id: claim.task.task_id,
      attempt_no: claim.attemptNo,
      parser_mode: claim.task.parser.configured_mode,
      business_route_id: claim.task.parser.route_id
    }, 'parse worker started');
    const controller = new AbortController();
    let timeout: NodeJS.Timeout | undefined;
    const parserPromise = Promise.resolve().then(() => parserOrchestrator.parse(claim, controller.signal));
    // A parser implementation may still reject after the watchdog has already
    // finalized the task. Attach a handler so that late failures are never
    // reported as unhandled process-level errors.
    parserPromise.catch(() => undefined);
    try {
      const outcome = await Promise.race([
        parserPromise.then((result) => ({ timedOut: false as const, result })),
        new Promise<{ timedOut: true }>((resolve) => {
          timeout = setTimeout(() => {
            controller.abort();
            resolve({ timedOut: true });
          }, config.PARSE_WORKER_TIMEOUT_MS);
        })
      ]);
      if (outcome.timedOut) {
        app.log.error({
          diagnostic_event: 'parser.worker.timeout',
          diagnostic_stage: 'parser_execution',
          request_id: parseRequestId,
          task_id: claim.task.task_id,
          attempt_no: claim.attemptNo,
          timeout_ms: config.PARSE_WORKER_TIMEOUT_MS,
          duration_ms: diagnosticDurationMs(startedAt),
          error_code: 'parse_worker_timeout'
        }, 'parse worker timed out');
        await persistParseOutcome(
          claim,
          workerFailureResult('parse_worker_timeout', 'worker_watchdog_timeout')
        );
        return;
      }
      await persistParseOutcome(claim, outcome.result.result, outcome.result.decision);
      app.log.info({
        diagnostic_event: 'parser.worker.completed',
        diagnostic_stage: 'parser_execution',
        request_id: parseRequestId,
        task_id: claim.task.task_id,
        attempt_no: claim.attemptNo,
        duration_ms: diagnosticDurationMs(startedAt)
      }, 'parse worker completed');
    } catch (error) {
      const errorCode = workerErrorCode(error);
      app.log.error({
        diagnostic_event: 'parser.worker.failed',
        diagnostic_stage: 'parser_execution',
        request_id: parseRequestId,
        task_id: claim.task.task_id,
        attempt_no: claim.attemptNo,
        duration_ms: diagnosticDurationMs(startedAt),
        ...diagnosticError(error, errorCode)
      }, 'parse worker failed');
      await persistParseOutcome(claim, workerFailureResult(errorCode, 'worker_exception'));
    } finally {
      if (timeout) clearTimeout(timeout);
    }
  }

  async function processParseQueue(): Promise<void> {
    const recovered = await tasks.recoverExpiredParseTasks();
    if (recovered > 0) {
      app.log.warn({
        diagnostic_event: 'parser.queue.expired_tasks_recovered',
        diagnostic_stage: 'parser_queue',
        recovered_tasks: recovered
      }, 'expired parse tasks were durably blocked');
    }
    const recoveredExecutions = await tasks.recoverExpiredExecutionTasks();
    if (recoveredExecutions > 0) {
      app.log.warn({
        diagnostic_event: 'erp.execution.expired_tasks_recovered',
        diagnostic_stage: 'erp_execution',
        recovered_tasks: recoveredExecutions
      }, 'expired ERP executions were automatically failed and released');
    }
    const staleReconciliations = await tasks.maintainStaleReconciliationTasks();
    if (staleReconciliations > 0) {
      app.log.warn({
        diagnostic_event: 'erp.reconciliation.stale_tasks_recovered',
        diagnostic_stage: 'erp_reconciliation',
        stale_tasks: staleReconciliations
      }, 'stale reconciliation tasks were automatically failed and released');
    }
    while (activeParseWorkers.size < 2) {
      const workerId = `parse:${process.pid}:${randomUUID()}`;
      const claim = await tasks.claimNextParseTask(workerId, [...activeParseWorkers.keys()]);
      if (!claim) break;
      activeParseWorkers.set(claim.task.task_id, workerId);
      app.log.info({
        diagnostic_event: 'parser.queue.claimed',
        diagnostic_stage: 'parser_queue',
        task_id: claim.task.task_id,
        attempt_no: claim.attemptNo,
        active_workers: activeParseWorkers.size
      }, 'parse queue task claimed');
      void runParseTask(claim)
        .catch((error) => {
          app.log.error({
            diagnostic_event: 'parser.runner.crashed',
            diagnostic_stage: 'parser_execution',
            task_id: claim.task.task_id,
            attempt_no: claim.attemptNo,
            ...diagnosticError(error, workerErrorCode(error))
          }, 'parse task runner crashed');
        })
        .finally(() => {
          activeParseWorkers.delete(claim.task.task_id);
          void scheduleParseQueue();
        })
        .catch(() => undefined);
    }
  }

  function scheduleParseQueue(): Promise<void> {
    if (parseQueueInFlight) return parseQueueInFlight;
    parseQueueInFlight = processParseQueue()
      .catch((error: unknown) => {
        const now = Date.now();
        if (now - lastParseQueueErrorAt < 60_000) return;
        lastParseQueueErrorAt = now;
        app.log.error({
          diagnostic_event: 'parser.queue.tick_failed',
          diagnostic_stage: 'parser_queue',
          ...diagnosticError(error, workerErrorCode(error))
        }, 'parse queue tick failed');
      })
      .finally(() => {
        parseQueueInFlight = null;
      });
    return parseQueueInFlight;
  }

  if (config.agentBusEnabled) {
    if (!startParserLoop) {
      throw new Error('AgentBus requires the parser loop to be enabled.');
    }
    const organization = await auth.getOrganization();
    if (!organization) {
      throw new Error(`AgentBus organization ${config.ORG_SLUG} was not found; bootstrap the administrator first.`);
    }
    agentBus = new AgentBusManager({
      config,
      tasks,
      organizationId: organization.id,
      scheduleParseQueue,
      logger: {
        info: (metadata, message) => app.log.info(agentBusDiagnosticMetadata(metadata), message),
        warn: (metadata, message) => app.log.warn(agentBusDiagnosticMetadata(metadata), message),
        error: (metadata, message) => app.log.error(agentBusDiagnosticMetadata(metadata), message)
      }
    });
    await agentBus.start();
  }
  if (config.leaderSummaryWebhookEnabled) {
    try {
      const organization = await auth.getOrganization();
      if (organization) {
        leaderNotificationService.startProjector(organization.id);
      } else {
        app.log.warn({
          diagnostic_event: 'leader_summary.webhook_initialization_skipped',
          notification_event: 'webhook_initialization_skipped',
          error_code: 'leader_summary_webhook_organization_not_found'
        }, 'Leader summary webhook initialization skipped; normal task services remain available');
      }
    } catch (error) {
      app.log.warn({
        diagnostic_event: 'leader_summary.webhook_initialization_failed',
        notification_event: 'webhook_initialization_failed',
        ...diagnosticError(error, 'leader_summary_webhook_initialization_failed')
      }, 'Leader summary webhook initialization failed; normal task services remain available');
    }
  }

  async function getAiProbe(): Promise<unknown> {
    const now = Date.now();
    if (aiProbeValue !== undefined && aiProbeExpiresAt > now) return aiProbeValue;
    if (aiProbeInFlight) return aiProbeInFlight;
    aiProbeInFlight = (async () => {
      if (typeof externalParser.checkConnection !== 'function') {
        return { configured: Boolean(config.DEERFLOW_OPEN_API_KEY), reachable: false, ok: false, error_code: 'status_probe_unavailable' };
      }
      try {
        return await externalParser.checkConnection();
      } catch (error) {
        app.log.warn({
          diagnostic_event: 'parser.status_probe.failed',
          diagnostic_stage: 'parser_probe',
          ...diagnosticError(error, 'probe_failed')
        }, 'external parser status probe failed');
        return { configured: Boolean(config.DEERFLOW_OPEN_API_KEY), reachable: false, ok: false, error_code: 'probe_failed' };
      }
    })();
    try {
      aiProbeValue = await aiProbeInFlight;
      aiProbeExpiresAt = Date.now() + 30_000;
      return aiProbeValue;
    } finally {
      aiProbeInFlight = null;
    }
  }

  app.get('/health/live', async () => ({ ok: true, service: 'ltjt-control-plane' }));
  app.get('/health/ready', async (_request, reply) => {
    const readiness = await databaseReadiness(config);
    if (!readiness.ready) {
      app.log.warn({
        diagnostic_event: 'health.readiness.failed',
        diagnostic_stage: 'health',
        database_ready: readiness.database,
        schema_ready: readiness.schema,
        required_migration: readiness.requiredMigration,
        error_code: readiness.errorCode || (readiness.database ? 'database_schema_outdated' : 'database_unavailable')
      }, 'control plane readiness check failed');
      return reply.code(503).send({
        ok: false,
        database: readiness.database,
        schema: readiness.schema,
        required_migration: readiness.requiredMigration,
        ...(readiness.errorCode ? { error_code: readiness.errorCode } : {})
      });
    }
    app.log.info({
      diagnostic_event: 'health.readiness.passed',
      diagnostic_stage: 'health',
      database_ready: true,
      schema_ready: true,
      required_migration: readiness.requiredMigration,
      agentbus_connected: agentBus?.status().connected || false
    }, 'control plane readiness check passed');
    return {
      ok: true,
      database: true,
      schema: true,
      required_migration: readiness.requiredMigration,
      agentbus: agentBus?.status() || {
        enabled: false,
        connected: false,
        session_ready: false,
        address: null,
        channels: []
      }
    };
  });

  app.get('/api/status', async () => {
    const ready = await databaseReady(config);
    const [aiProbe, parserEvidence] = await Promise.all([
      getAiProbe(),
      tasks.parserConnectionEvidence()
    ]);
    const probe = aiProbe && typeof aiProbe === 'object' ? aiProbe as Record<string, unknown> : {};
    const authenticated = typeof probe.authenticated === 'boolean'
      ? probe.authenticated
      : parserEvidence.authenticated;
    const enrichedProbe = {
      ...probe,
      authenticated,
      authentication_checked: probe.authentication_checked === true || authenticated !== null,
      authentication_evidence: parserEvidence.source,
      last_success_at: parserEvidence.lastSuccessAt,
      last_auth_failure_at: parserEvidence.lastAuthFailureAt,
      last_auth_failure_code: parserEvidence.lastAuthFailureCode
    };
    const configured = probe.configured === true;
    const reachable = probe.reachable === true;
    const connected = aiServiceConnected(ready, probe);
    return {
      ok: true,
      service: 'ltjt-control-plane',
      checked_at: new Date().toISOString(),
      database_ready: ready,
      ai_configured: configured,
      ai_reachable: reachable,
      ai_authenticated: authenticated,
      ai_connected: connected,
      ai_connection_basis: 'service_reachability',
      ai_probe: enrichedProbe,
      agentbus: agentBus?.status() || {
        enabled: false,
        connected: false,
        session_ready: false,
        address: null,
        channels: []
      }
    };
  });

  app.post('/api/auth/login', { config: { rateLimit: { max: 10, timeWindow: '1 minute' } } }, async (request, reply) => {
    setAuthNoStore(reply);
    requireSameOrigin(config, request);
    const body = loginSchema.parse(request.body);
    const session = await auth.authenticate(body.username, body.password, clientAddress(request), userAgent(request));
    reply.setCookie(config.SESSION_COOKIE_NAME, session.token, sessionCookieOptions(config));
    await auth.recordAudit(session.user.organizationId, session.user.id, 'login', requestId(request), {
      ip: clientAddress(request),
      user_agent_present: Boolean(request.headers['user-agent'])
    });
    return { ok: true, user: publicUser({ id: session.id, user: session.user, csrfTokenHash: Buffer.alloc(0) }), csrf_token: session.csrfToken };
  });

  app.get('/api/auth/me', async (request, reply) => {
    setAuthNoStore(reply);
    const session = await getSession(request);
    return { ok: true, user: publicUser(session) };
  });

  app.get('/api/auth/csrf', async (request, reply) => {
    setAuthNoStore(reply);
    const session = await getSession(request);
    return { ok: true, csrf_token: await auth.rotateCsrf(session.id) };
  });

  app.put('/api/auth/password', async (request) => {
    const session = await requireAuthenticatedMutationSession(request);
    const body = changePasswordSchema.parse(request.body);
    await auth.changeOwnPassword(session, body.current_password, body.new_password, requestId(request));
    return { ok: true, password_changed: true };
  });

  app.get('/api/accounts', async (request) => {
    const session = await requireAdminSession(request);
    return {
      ok: true,
      accounts: await auth.listAccounts(session.user),
      task_types: BUSINESS_ROUTES.map((route, displayOrder) => ({
        route_id: route.routeId,
        directive: route.directive,
        action: route.action,
        display_order: displayOrder + 1
      }))
    };
  });

  app.post('/api/accounts', { config: { rateLimit: { max: 20, timeWindow: '1 minute' } } }, async (request) => {
    const session = await requireAdminMutationSession(request);
    const body = accountCreateSchema.parse(request.body);
    const account = await auth.createAccount(session.user, {
      username: body.username,
      password: body.password,
      role: body.role,
      erpAccount: body.erp_account,
      businessRouteIds: body.business_route_ids
    }, requestId(request));
    return { ok: true, account };
  });

  app.patch('/api/accounts/:userId', async (request) => {
    const session = await requireAdminMutationSession(request);
    const params = request.params as { userId: string };
    const userId = z.string().uuid().parse(params.userId);
    const body = accountUpdateSchema.parse(request.body);
    const account = await auth.updateAccount(session.user, userId, {
      role: body.role,
      isActive: body.is_active,
      erpAccount: body.erp_account
    }, requestId(request));
    await agentBus?.reload();
    return { ok: true, account };
  });

  app.put('/api/accounts/:userId/business-authorizations', async (request) => {
    const session = await requireAdminMutationSession(request);
    const params = request.params as { userId: string };
    const userId = z.string().uuid().parse(params.userId);
    const body = accountBusinessAuthorizationsSchema.parse(request.body);
    const account = await auth.setBusinessRouteAuthorizations(
      session.user,
      userId,
      body.business_route_ids,
      body.expected_revision,
      requestId(request)
    );
    return { ok: true, account };
  });

  app.post('/api/accounts/:userId/reset-password', { config: { rateLimit: { max: 20, timeWindow: '1 minute' } } }, async (request) => {
    const session = await requireAdminMutationSession(request);
    const params = request.params as { userId: string };
    const userId = z.string().uuid().parse(params.userId);
    const body = accountPasswordResetSchema.parse(request.body);
    await auth.resetAccountPassword(
      session.user,
      userId,
      body.password,
      requestId(request)
    );
    return { ok: true, password_reset: true, sessions_revoked: true };
  });

  app.post('/api/accounts/:userId/revoke-sessions', async (request) => {
    const session = await requireAdminMutationSession(request);
    const params = request.params as { userId: string };
    const userId = z.string().uuid().parse(params.userId);
    const revoked = await auth.revokeAccountSessions(session.user, userId, requestId(request));
    return { ok: true, sessions_revoked: revoked };
  });

  app.get('/api/settings/automation', async (request) => {
    const session = await requireAdminSession(request);
    return { ok: true, settings: await tasks.getAutomationSettings(session.user.organizationId) };
  });

  app.put('/api/settings/automation', async (request) => {
    const session = await requireAdminMutationSession(request);
    const body = automationSettingsSchema.parse(request.body);
    return {
      ok: true,
      settings: await tasks.setAutomationEnabled(contextFor(session, request), body.enabled)
    };
  });

  app.get('/api/settings/parser-routing', async (request) => {
    const session = await requireAdminSession(request);
    return { ok: true, routes: await tasks.getParserRoutingSettings(session.user.organizationId) };
  });

  app.put('/api/settings/parser-routing/:routeId', async (request) => {
    const session = await requireAdminMutationSession(request);
    const routeId = String((request.params as { routeId?: string }).routeId || '');
    const body = parserRoutingUpdateSchema.parse(request.body);
    return {
      ok: true,
      route: await tasks.setParserRoutingMode(
        contextFor(session, request),
        routeId,
        body.mode,
        body.expected_revision
      )
    };
  });

  app.post('/api/settings/parser-routing/emergency-ai', async (request) => {
    const session = await requireAdminMutationSession(request);
    const body = parserEmergencyAiSchema.parse(request.body);
    return {
      ok: true,
      routes: await tasks.emergencySetAllParserRoutesToAi(contextFor(session, request), body.reason)
    };
  });

  app.post('/api/tasks/:taskId/reparse', async (request) => {
    const session = requireTaskDataPlane(await requireAdminMutationSession(request));
    const taskId = String((request.params as { taskId?: string }).taskId || '');
    const body = parserReparseSchema.parse(request.body);
    const task = await tasks.reparseTaskWithAi(contextFor(session, request), taskId, body.reason);
    void scheduleParseQueue();
    return {
      ok: true,
      task
    };
  });

  app.put('/api/parser-decisions/:decisionId/review', async (request) => {
    const session = requireTaskDataPlane(await requireAdminMutationSession(request));
    const decisionId = String((request.params as { decisionId?: string }).decisionId || '');
    const body = parserDecisionReviewSchema.parse(request.body);
    return {
      ok: true,
      decision: await tasks.reviewParserDecision(
        contextFor(session, request), decisionId, body.verdict, body.note
      )
    };
  });

  app.get('/api/channels', async (request) => {
    const session = await requireAdminSession(request);
    const channels = await channelService.list(session.user.organizationId);
    return {
      ok: true,
      channels: mergeRuntimeChannelStatuses(channels, agentBus?.status().channels || [])
    };
  });

  app.post('/api/channels', async (request) => {
    const session = await requireAdminMutationSession(request);
    const body = channelCreateSchema.parse(request.body);
    const channel = await channelService.create(contextFor(session, request), {
      displayName: body.display_name,
      ownerUserId: body.owner_user_id,
      externalUserRef: body.external_user_ref,
      agentbusKey: body.agentbus_key,
      botAddress: body.bot_address,
      enabled: body.enabled
    });
    await agentBus?.reload();
    return { ok: true, channel };
  });

  app.patch('/api/channels/:channelId', async (request) => {
    const session = await requireAdminMutationSession(request);
    const params = request.params as { channelId: string };
    const body = channelUpdateSchema.parse(request.body);
    const channel = await channelService.update(contextFor(session, request), params.channelId, {
      displayName: body.display_name,
      ownerUserId: body.owner_user_id,
      externalUserRef: body.external_user_ref,
      botAddress: body.bot_address,
      enabled: body.enabled
    });
    await agentBus?.reload();
    return { ok: true, channel };
  });

  app.post('/api/channels/:channelId/rotate-key', async (request) => {
    const session = await requireAdminMutationSession(request);
    const params = request.params as { channelId: string };
    const body = channelRotateKeySchema.parse(request.body);
    const channel = await channelService.rotateKey(
      contextFor(session, request),
      params.channelId,
      body.agentbus_key
    );
    await agentBus?.reload();
    return { ok: true, channel };
  });

  app.delete('/api/channels/:channelId', async (request) => {
    const session = await requireAdminMutationSession(request);
    const params = request.params as { channelId: string };
    const result = await channelService.delete(contextFor(session, request), params.channelId);
    await agentBus?.reload();
    return { ok: true, ...result };
  });

  app.get('/api/settings/leader-summary-webhook', async (request) => {
    const session = await requireAdminSession(request);
    return {
      ok: true,
      status: await leaderNotificationService.getWebhookStatus(session.user.organizationId)
    };
  });

  app.post('/api/auth/logout', async (request, reply) => {
    setAuthNoStore(reply);
    const session = await requireAuthenticatedMutationSession(request);
    await auth.revokeSession(request.cookies[config.SESSION_COOKIE_NAME]);
    reply.clearCookie(config.SESSION_COOKIE_NAME, { path: '/' });
    await auth.recordAudit(session.user.organizationId, session.user.id, 'logout', requestId(request));
    app.log.info({ user_id: session.user.id, request_id: requestId(request) }, 'user logged out');
    return { ok: true };
  });

  app.get('/api/tasks', async (request) => {
    const session = await requireTaskSession(request);
    const query = listTasksQuerySchema.parse(request.query || {});
    const page = await tasks.listTasksPage(session.user.organizationId, {
      status: query.status || undefined,
      search: query.search || undefined,
      limit: query.limit,
      offset: query.offset,
      includeTotal: query.include_total,
      archive: query.archive,
      assignedToUserId: query.executable_by === 'me' ? session.user.id : undefined,
      access: contextFor(session, request)
    });
    return {
      ok: true,
      tasks: page.tasks,
      pagination: {
        total: page.total,
        offset: page.offset,
        limit: page.limit,
        has_more: page.has_more
      }
    };
  });

  app.post('/api/tasks', async (request) => {
    const session = await requireTaskMutationSession(request);
    const body = createTaskSchema.parse(request.body);
    const task = await tasks.createTask(
      contextFor(session, request),
      body.raw_text,
      body.idempotency_key,
      body.conversation_id,
      decodeInputAttachments(body.attachments, config)
    );
    if (task.status === 'parse_queued') void scheduleParseQueue();
    return { ok: true, task };
  });

  app.post('/api/messages', async (request) => {
    const session = await requireTaskMutationSession(request);
    const body = taskMessageSchema.parse(request.body);
    const result = await tasks.ingestMessage(contextFor(session, request), {
      message: body.message,
      taskId: body.task_id,
      conversationId: body.conversation_id,
      idempotencyKey: body.idempotency_key,
      attachments: decodeInputAttachments(body.attachments, config)
    });
    if (result.task.status === 'parse_queued') void scheduleParseQueue();
    return { ok: true, ...result };
  });

  app.get('/api/tasks/:taskId', async (request) => {
    const session = await requireTaskSession(request);
    const params = request.params as { taskId: string };
    return { ok: true, task: await tasks.getTask(session.user.organizationId, params.taskId, contextFor(session, request)) };
  });

  app.get('/api/tasks/:taskId/input-history', async (request) => {
    const session = await requireTaskSession(request);
    const params = request.params as { taskId: string };
    return { ok: true, ...(await tasks.getTaskInputHistory(contextFor(session, request), params.taskId)) };
  });

  app.get('/api/tasks/:taskId/artifacts/:artifactId', async (request, reply) => {
    const session = await requireTaskSession(request);
    const params = request.params as { taskId: string; artifactId: string };
    const artifactId = z.string().uuid().safeParse(params.artifactId);
    if (!artifactId.success) throw new TaskError('artifact_not_found', '附件不存在或无权访问。', 404);
    const artifact = await tasks.getTaskArtifact(
      session.user.organizationId,
      params.taskId,
      artifactId.data,
      contextFor(session, request)
    );
    if (artifact.storage_backend === 'oss' && artifact.public_url) {
      reply.header('Cache-Control', 'no-store');
      return reply.redirect(artifact.public_url, 302);
    }
    reply.header('Cache-Control', 'private, no-store');
    reply.header('Content-Type', artifact.content_type || 'application/octet-stream');
    reply.header('Content-Length', String(artifact.content.byteLength));
    reply.header('Content-Disposition', attachmentContentDisposition(artifact.file_name));
    reply.header('X-Content-Type-Options', 'nosniff');
    return reply.send(artifact.content);
  });

  app.post('/api/tasks/:taskId/confirm', async (request) => {
    const session = await requireTaskMutationSession(request);
    const params = request.params as { taskId: string };
    return { ok: true, task: await tasks.confirmTask(contextFor(session, request), params.taskId) };
  });

  app.post('/api/tasks/:taskId/claim', async (request) => {
    const session = await requireTaskMutationSession(request);
    const body = claimSchema.parse(request.body);
    const params = request.params as { taskId: string };
    const claim = await tasks.claimForBrowser(contextFor(session, request), params.taskId, body.connection_id);
    return {
      ok: true,
      task: claim.task,
      claimed: claim.claimed,
      execution_id: claim.executionId,
      queue_status: claim.queueStatus,
      queue_position: claim.queuePosition,
      active_task_id: claim.activeTaskId
    };
  });

  app.post('/api/tasks/:taskId/result', async (request) => {
    const session = await requireTaskMutationSession(request);
    const body = resultSchema.parse(request.body);
    const params = request.params as { taskId: string };
    return {
      ok: true,
      task: await tasks.recordExecutionResult(
        contextFor(session, request),
        params.taskId,
        body.result,
        body.connection_id,
        body.execution_id
      )
    };
  });

  app.post('/api/tasks/:taskId/cancel', async (request) => {
    const session = await requireTaskMutationSession(request);
    const params = request.params as { taskId: string };
    return { ok: true, task: await tasks.cancelTask(contextFor(session, request), params.taskId) };
  });

  app.post('/api/tasks/bulk-delete', async (request) => {
    const session = await requireTaskMutationSession(request);
    const body = taskBulkDeleteSchema.parse(request.body);
    return {
      ok: true,
      deleted: true,
      ...(await tasks.hardDeleteTasks(contextFor(session, request), body.task_ids))
    };
  });

  app.post('/api/tasks/bulk-archive', async (request) => {
    const session = await requireTaskMutationSession(request);
    const body = taskBulkArchiveSchema.parse(request.body);
    return {
      ok: true,
      archived: true,
      ...(await tasks.archiveTasks(contextFor(session, request), body.task_ids, body.reason))
    };
  });

  app.delete('/api/tasks/:taskId', async (request) => {
    const session = await requireTaskMutationSession(request);
    const params = request.params as { taskId: string };
    return {
      ok: true,
      ...(await tasks.hardDeleteTask(contextFor(session, request), params.taskId))
    };
  });

  app.post('/api/tasks/:taskId/archive', async (request) => {
    const session = await requireTaskMutationSession(request);
    const params = request.params as { taskId: string };
    const body = taskArchiveSchema.parse(request.body || {});
    return {
      ok: true,
      archived: true,
      task: await tasks.archiveTask(contextFor(session, request), params.taskId, body.reason)
    };
  });

  app.post('/api/tasks/:taskId/restore', async (request) => {
    const session = await requireTaskMutationSession(request);
    const params = request.params as { taskId: string };
    return { ok: true, restored: true, task: await tasks.restoreTask(contextFor(session, request), params.taskId) };
  });

  app.post('/api/connections/heartbeat', async (request) => {
    const session = await requireTaskMutationSession(request);
    const body = heartbeatSchema.parse(request.body);
    const worker = await tasks.heartbeat(
      contextFor(session, request),
      body.connection_id,
      body.extension_version || '',
      body.metadata || {},
      {
        erpAccount: body.erp_account || '',
        erpAccountMatched: body.erp_account_matched === true
      }
    );
    return { ok: true, connected: true, ...worker };
  });

  app.get('/api/audit', async (request) => {
    const session = await requireAdminSession(request);
    const query = auditQuerySchema.parse(request.query || {});
    return {
      ok: true,
      ...(await tasks.listAuditEvents(contextFor(session, request), {
        eventType: query.event_type,
        actorUserId: query.actor_user_id,
        entityType: query.entity_type,
        limit: query.limit,
        offset: query.offset
      }))
    };
  });

  app.get('/api/operations-dashboard', async (request, reply) => {
    setAuthNoStore(reply);
    const session = await requireLeadershipSession(request);
    const query = operationsDashboardQuerySchema.parse(request.query || {});
    const controller = new AbortController();
    const abortRequest = () => controller.abort();
    const abortClosedReply = () => {
      if (!reply.raw.writableEnded) abortRequest();
    };
    request.raw.once('aborted', abortRequest);
    reply.raw.once('close', abortClosedReply);
    if (request.raw.aborted) abortRequest();
    try {
      return {
        ok: true,
        ...(await tasks.listOperationsDashboard(contextFor(session, request), {
          from: query.from,
          to: query.to,
          actorUserId: query.actor_user_id,
          businessRouteId: query.business_route_id,
          status: query.status,
          search: query.search,
          limit: query.limit,
          offset: query.offset,
          signal: controller.signal
        }))
      };
    } finally {
      request.raw.off('aborted', abortRequest);
      reply.raw.off('close', abortClosedReply);
    }
  });

  app.get('/api/operations-dashboard/tasks/:taskId', async (request, reply) => {
    setAuthNoStore(reply);
    const session = await requireLeadershipSession(request);
    const params = request.params as { taskId: string };
    return {
      ok: true,
      read_only: true,
      ...(await tasks.getOperationsDashboardTask(contextFor(session, request), params.taskId))
    };
  });

  app.get('/api/events', async (request, reply) => {
    const session = await requireTaskSession(request);
    const query = taskEventsQuerySchema.parse(request.query || {});
    const querySince = query.since;
    const reconnectSince = Number(request.headers['last-event-id'] || 0);
    const since = Math.max(
      Number.isFinite(querySince) ? querySince : 0,
      Number.isFinite(reconnectSince) ? reconnectSince : 0
    );
    reply.hijack();
    const response = reply.raw;
    response.writeHead(200, {
      'Content-Type': 'text/event-stream; charset=utf-8',
      'Cache-Control': 'no-cache, no-transform',
      Connection: 'keep-alive',
      'X-Accel-Buffering': 'no'
    });
    const send = (event: TaskEvent) => {
      if (event.organization_id !== session.user.organizationId) return;
      // This is an employee execution feed. Administrators are rejected
      // before the stream is opened, and workers receive only assigned work.
      if (event.owner_user_id !== session.user.id) return;
      const publicEvent = {
        id: event.id,
        organization_id: event.organization_id,
        task_id: event.task_id,
        status: event.status,
        stage: event.stage,
        message: event.message,
        payload: event.payload,
        created_at: event.created_at
      };
      response.write(`id: ${event.id}\nevent: task\ndata: ${JSON.stringify(publicEvent)}\n\n`);
    };
    const sendBrowserCommand = (command: TaskBrowserCommand) => {
      if (command.organization_id !== session.user.organizationId) return;
      if (command.assigned_user_id !== session.user.id) return;
      response.write(`event: browser-command\ndata: ${JSON.stringify({
        action: command.action,
        task_id: command.task_id,
        target_user_id: command.assigned_user_id,
        created_at: command.created_at
      })}\n\n`);
    };
    for (const event of await tasks.eventsSince(session.user.organizationId, session.user.id, since)) send(event);
    const heartbeat = setInterval(() => response.write(': heartbeat\n\n'), 20_000);
    const onTask = (event: TaskEvent) => send(event);
    const onBrowserCommand = (command: TaskBrowserCommand) => sendBrowserCommand(command);
    tasks.events.on('task', onTask);
    tasks.events.on('browser-command', onBrowserCommand);
    request.raw.on('close', () => {
      clearInterval(heartbeat);
      tasks.events.off('task', onTask);
      tasks.events.off('browser-command', onBrowserCommand);
    });
  });

  app.setErrorHandler((error, request, reply) => {
    if (error instanceof AuthError || error instanceof TaskError) {
      request.log.warn({
        diagnostic_event: 'http.request.rejected',
        diagnostic_stage: 'http',
        request_id: requestId(request),
        ...(requestTaskId(request) ? { task_id: requestTaskId(request) } : {}),
        status_code: error.statusCode,
        error_code: error.code,
        error_name: error.name,
        ...(error instanceof TaskError ? { error_detail_keys: diagnosticMetadataKeys(error.details) } : {})
      }, 'HTTP request rejected by application guard');
      return reply.code(error.statusCode).send({
        ok: false,
        error_code: error.code,
        message: error.message,
        ...(error instanceof TaskError && Object.keys(error.details).length ? { details: error.details } : {})
      });
    }
    if (error instanceof z.ZodError) {
      request.log.warn({
        diagnostic_event: 'http.request.invalid',
        diagnostic_stage: 'http',
        request_id: requestId(request),
        ...(requestTaskId(request) ? { task_id: requestTaskId(request) } : {}),
        status_code: 400,
        error_code: 'invalid_request',
        validation_paths: error.issues.map((issue) => issue.path.join('.')).slice(0, 100)
      }, 'HTTP request validation failed');
      return reply.code(400).send({ ok: false, error_code: 'invalid_request', message: '请求参数不符合要求。', details: error.issues.map((issue) => issue.path.join('.')) });
    }
    request.log.error({
      diagnostic_event: 'http.request.failed',
      diagnostic_stage: 'http',
      request_id: requestId(request),
      ...(requestTaskId(request) ? { task_id: requestTaskId(request) } : {}),
      status_code: 500,
      ...diagnosticError(error, 'server_error')
    }, 'unhandled request error');
    return reply.code(500).send({ ok: false, error_code: 'server_error', message: '服务暂时不可用。' });
  });

  if (startParserLoop) {
    const interval = setInterval(() => void scheduleParseQueue(), 5_000);
    app.addHook('onClose', async () => clearInterval(interval));
  }

  app.addHook('onClose', async () => {
    app.log.info({
      diagnostic_event: 'service.closing',
      diagnostic_stage: 'shutdown'
    }, 'control plane closing');
    await leaderNotificationService.stopProjector();
    await agentBus?.stop();
    await closePool();
  });
  return { app, auth, tasks, agentBus, channelService, leaderNotificationService };
}

function installProcessDiagnostics(app: Awaited<ReturnType<typeof buildServer>>['app']): void {
  let shuttingDown = false;
  const shutdown = async (reason: string, exitCode: number, error?: unknown): Promise<void> => {
    if (shuttingDown) return;
    shuttingDown = true;
    process.exitCode = exitCode;
    const metadata = {
      diagnostic_event: 'service.shutdown.started',
      diagnostic_stage: 'shutdown',
      shutdown_reason: reason,
      exit_code: exitCode,
      ...(error === undefined ? {} : diagnosticError(error, reason))
    };
    if (exitCode === 0) app.log.info(metadata, 'control plane shutdown started');
    else app.log.error(metadata, 'control plane shutdown started after fatal error');
    try {
      await app.close();
      app.log.info({
        diagnostic_event: 'service.shutdown.completed',
        diagnostic_stage: 'shutdown',
        shutdown_reason: reason,
        exit_code: exitCode
      }, 'control plane shutdown completed');
    } catch (closeError) {
      app.log.error({
        diagnostic_event: 'service.shutdown.failed',
        diagnostic_stage: 'shutdown',
        shutdown_reason: reason,
        exit_code: 1,
        ...diagnosticError(closeError, 'shutdown_failed')
      }, 'control plane shutdown failed');
      process.exitCode = 1;
    }
  };
  process.once('SIGTERM', () => void shutdown('sigterm', 0));
  process.once('SIGINT', () => void shutdown('sigint', 0));
  process.once('uncaughtException', (error) => void shutdown('uncaught_exception', 1, error));
  process.once('unhandledRejection', (error) => void shutdown('unhandled_rejection', 1, error));
}

async function main(): Promise<void> {
  const config = loadConfig();
  await assertDatabaseSchema(config);
  const { app } = await buildServer({ config });
  installProcessDiagnostics(app);
  try {
    await app.listen({ host: config.HOST, port: config.PORT });
    app.log.info({
      diagnostic_event: 'service.listening',
      diagnostic_stage: 'startup',
      host: config.HOST,
      port: config.PORT
    }, 'LianSyn-platform control plane listening');
  } catch (error) {
    app.log.error({
      diagnostic_event: 'service.listen.failed',
      diagnostic_stage: 'startup',
      ...diagnosticError(error, 'listen_failed')
    }, 'control plane failed to listen');
    await app.close().catch(() => undefined);
    throw error;
  }
}

if (process.argv[1] && resolve(process.argv[1]) === resolve(fileURLToPath(import.meta.url))) {
  main().catch(async (error) => {
    writeEmergencyDiagnostic('service.startup_failed', error, { diagnostic_stage: 'startup' });
    await closePool();
    process.exitCode = 1;
  });
}