import { randomUUID } from 'node:crypto';
import { realpathSync } from 'node:fs';
import { fileURLToPath } from 'node:url';
import { performance } from 'node:perf_hooks';
import { Worker } from 'node:worker_threads';
import type { RuntimeDiagnosticSeverity } from '../../../framework/plugins/index.js';
import type {
  BehaviorDecision,
  BehaviorInput,
} from '../../../../sdk/behavior-plugin/index.js';
import {
  behaviorSourceUrl,
  behaviorThreadLoaderConfig,
  pinBehaviorSource,
  resolveBehaviorSource,
} from './behavior-module-loader.js';
import type { ResolvedClawclawBehaviorSource } from './behavior-source.js';
import type { ClawclawBehaviorObservationSnapshot } from '../../behavior-observation-memory.js';

export const PARALLEL_BEHAVIOR_THREAD_POOL_SIZE = 10;
const INITIAL_WORKER_COUNT = 2;
const MAX_POOL_SIZE = 10;
const IDLE_SCALE_DOWN_MS = 30_000;
const START_TIMEOUT_MS = 20_000;
const MAX_WORKER_ERROR_MESSAGE_LENGTH = 512;
const REPLACEMENT_RETRY_BASE_MS = 50;
const REPLACEMENT_RETRY_MAX_MS = 5_000;
const WORKER_BOOTSTRAP = realpathSync(fileURLToPath(
  new URL('../../../framework/modules/runtime-module-worker-bootstrap.mjs', import.meta.url),
));
const WORKER_HOST_URL = new URL('./behavior-analysis-thread-host.ts', import.meta.url).href;

export interface ThreadBehaviorExecutorOptions {
  readonly source: ResolvedClawclawBehaviorSource;
  /** Maximum Worker count and in-flight request window. Tests may reduce it. */
  readonly poolSize?: number;
  /** How long an expanded, idle pool is retained before returning to two Workers. */
  readonly idleWorkerTimeoutMs?: number;
  readonly maxOldGenerationSizeMb?: number;
  readonly diagnostic?: (
    name: string,
    attributes?: Readonly<Record<string, unknown>>,
    severity?: RuntimeDiagnosticSeverity,
  ) => void;
}

export interface ThreadBehaviorExecutorSnapshot {
  readonly state: 'new' | 'starting' | 'running' | 'stopping' | 'stopped';
  readonly workerCount: number;
  readonly idleWorkerCount: number;
  readonly busyWorkerCount: number;
  readonly queuedRequestCount: number;
  readonly inFlightRequestCount: number;
  readonly startedWorkerCount: number;
  readonly retiredWorkerCount: number;
}

interface PendingRequest {
  readonly requestId: string;
  readonly operation: 'decide';
  readonly frame: BehaviorInput;
  readonly observationSnapshot?: Readonly<ClawclawBehaviorObservationSnapshot>;
  readonly signal: AbortSignal;
  readonly enqueuedAt: number;
  readonly resolve: (value: unknown) => void;
  readonly reject: (error: Error) => void;
  readonly onAbort: () => void;
  settled: boolean;
  assignedAt?: number;
  slot?: WorkerSlot;
}

interface WorkerSlot {
  readonly id: number;
  readonly worker: Worker;
  state: 'starting' | 'idle' | 'busy' | 'retiring';
  idleSince?: number;
  current?: PendingRequest;
  readyResolve?: () => void;
  readyReject?: (error: Error) => void;
  retirement?: Promise<void>;
}

interface WorkerMessageBase {
  readonly type: string;
  readonly poolId: string;
  readonly slotId: number;
  readonly requestId?: string;
}

interface WorkerReadyMessage extends WorkerMessageBase {
  readonly type: 'ready';
  readonly behaviorId: string;
  readonly execution: ResolvedClawclawBehaviorSource['execution'];
}

interface WorkerResultMessage extends WorkerMessageBase {
  readonly type: 'result';
  readonly requestId: string;
  readonly result: BehaviorDecision;
  readonly queryCount: number;
  readonly durationMs: number;
}

interface WorkerErrorMessage extends WorkerMessageBase {
  readonly type: 'error';
  readonly code: string;
  readonly message?: string;
}

function asError(value: unknown, fallback: string): Error {
  return value instanceof Error ? value : new Error(typeof value === 'string' ? value : fallback);
}

function abortError(signal: AbortSignal): Error {
  return asError(signal.reason, 'behavior_thread_request_aborted');
}

function errorCode(error: unknown, fallback: string): string {
  const code = error && typeof error === 'object' && typeof (error as any).code === 'string'
    ? (error as any).code.trim()
    : '';
  if (/^[A-Za-z][A-Za-z0-9_.-]{0,119}$/.test(code)) return code;
  const message = error instanceof Error ? error.message : typeof error === 'string' ? error : '';
  return /^([A-Za-z][A-Za-z0-9_.-]{0,119})(?::|$)/.exec(message.trim())?.[1] ?? fallback;
}

function roundedElapsed(startedAt: number): number {
  return Math.round(Math.max(0, performance.now() - startedAt) * 1_000) / 1_000;
}

/** Adaptive Worker Thread pool for cross-tick computations. */
export class ThreadBehaviorExecutor {
  readonly runtimeInstanceId = randomUUID();

  private readonly maxWorkerCount: number;
  private readonly minWorkerCount: number;
  private readonly idleWorkerTimeoutMs: number;
  private readonly maxOldGenerationSizeMb: number;
  private state: ThreadBehaviorExecutorSnapshot['state'] = 'new';
  private readonly slots = new Set<WorkerSlot>();
  private readonly queue: PendingRequest[] = [];
  private readonly inFlight = new Set<PendingRequest>();
  private readonly spawnTasks = new Set<Promise<void>>();
  private startTask?: Promise<void>;
  private stopTask?: Promise<void>;
  private replacementTask?: Promise<void>;
  private replacementRetryTimer?: ReturnType<typeof setTimeout>;
  private idleScaleDownTimer?: ReturnType<typeof setTimeout>;
  private replacementFailureCount = 0;
  private nextSlotId = 1;
  private startedWorkerCount = 0;
  private retiredWorkerCount = 0;
  private desiredWorkerCount: number;

  private readonly options: ThreadBehaviorExecutorOptions;

  constructor(options: ThreadBehaviorExecutorOptions) {
    this.options = Object.freeze({
      ...options,
      source: pinBehaviorSource(options.source),
    });
    if ((options.source.execution ?? 'parallel') !== 'parallel') {
      throw new Error('behavior_thread_execution_invalid');
    }
    const maxWorkerCount = options.poolSize ?? PARALLEL_BEHAVIOR_THREAD_POOL_SIZE;
    if (!Number.isInteger(maxWorkerCount) || maxWorkerCount <= 0 || maxWorkerCount > MAX_POOL_SIZE) {
      throw new TypeError(`poolSize must be between 1 and ${MAX_POOL_SIZE}.`);
    }
    const idleWorkerTimeoutMs = options.idleWorkerTimeoutMs ?? IDLE_SCALE_DOWN_MS;
    if (!Number.isFinite(idleWorkerTimeoutMs) || idleWorkerTimeoutMs < 0) {
      throw new TypeError('idleWorkerTimeoutMs must be a non-negative finite number.');
    }
    this.maxWorkerCount = maxWorkerCount;
    this.minWorkerCount = Math.min(INITIAL_WORKER_COUNT, maxWorkerCount);
    this.desiredWorkerCount = this.minWorkerCount;
    this.idleWorkerTimeoutMs = idleWorkerTimeoutMs;
    this.maxOldGenerationSizeMb = Math.max(16, Math.min(
      256,
      Math.floor(options.maxOldGenerationSizeMb ?? 64),
    ));
  }

  get isAlive(): boolean {
    // A temporarily degraded pool remains the same live executor while it
    // restores a retired slot. Capacity is reported separately by snapshot().
    return this.state === 'running';
  }

  start(signal: AbortSignal): Promise<void> {
    if (this.state === 'running') return Promise.resolve();
    if (this.state === 'starting') return this.awaitStart(signal);
    if (this.state === 'stopping' || this.state === 'stopped') {
      return Promise.reject(new Error('behavior_thread_pool_stopped'));
    }
    if (signal.aborted) return Promise.reject(abortError(signal));
    this.state = 'starting';
    const startedAt = performance.now();
    this.startTask = Promise.all(
      Array.from({ length: this.minWorkerCount }, () => this.spawnSlot('initial')),
    ).then(() => {
      if (this.state !== 'starting') throw new Error('behavior_thread_pool_stopped');
      this.state = 'running';
      this.diagnostic('behavior.thread_pool.started', {
        behavior_id: this.options.source.id,
        execution: this.options.source.execution,
        runtime_instance_id: this.runtimeInstanceId,
        worker_count: this.slots.size,
        min_worker_count: this.minWorkerCount,
        max_worker_count: this.maxWorkerCount,
        started_worker_count: this.startedWorkerCount,
        retired_worker_count: this.retiredWorkerCount,
        duration_ms: roundedElapsed(startedAt),
      }, 'info');
    }).catch(async (error) => {
      if (this.state === 'starting') this.state = 'stopped';
      await Promise.allSettled([...this.slots].map((slot) => this.retireSlot(slot, 'start_failed')));
      throw error;
    });
    return this.awaitStart(signal).catch(async (error) => {
      if (signal.aborted) await this.stop('switched');
      throw error;
    });
  }

  update(
    frame: BehaviorInput,
    signal: AbortSignal,
    observationSnapshot?: Readonly<ClawclawBehaviorObservationSnapshot>,
  ): Promise<BehaviorDecision> {
    return this.submit('decide', frame, signal, observationSnapshot) as Promise<BehaviorDecision>;
  }

  async reset(_signal: AbortSignal): Promise<void> {
    // Parallel Behavior instances are already fresh per input. Only the
    // module/Class is cached by each slot.
  }

  stop(reason: 'switched' | 'scope_ended' | 'shutdown'): Promise<void> {
    if (this.stopTask) return this.stopTask;
    this.stopTask = this.performStop(reason);
    return this.stopTask;
  }

  private async performStop(reason: 'switched' | 'scope_ended' | 'shutdown'): Promise<void> {
    if (this.state === 'stopped') return;
    if (this.state === 'new') {
      this.state = 'stopped';
      return;
    }
    this.state = 'stopping';
    if (this.replacementRetryTimer) {
      clearTimeout(this.replacementRetryTimer);
      this.replacementRetryTimer = undefined;
    }
    this.cancelIdleScaleDown();
    const error = new Error('behavior_thread_pool_stopped');
    for (const request of [...this.inFlight]) this.rejectRequest(request, error);
    await Promise.allSettled([...this.slots].map((slot) => this.retireSlot(slot, reason)));
    await Promise.allSettled([...this.spawnTasks]);
    await this.startTask?.catch(() => {});
    await Promise.allSettled([...this.slots].map((slot) => this.retireSlot(slot, reason)));
    this.state = 'stopped';
    this.diagnostic('behavior.thread_pool.stopped', {
      behavior_id: this.options.source.id,
      runtime_instance_id: this.runtimeInstanceId,
      reason,
      min_worker_count: this.minWorkerCount,
      max_worker_count: this.maxWorkerCount,
      started_worker_count: this.startedWorkerCount,
      retired_worker_count: this.retiredWorkerCount,
    }, 'info');
  }

  snapshot(): ThreadBehaviorExecutorSnapshot {
    const slots = [...this.slots];
    return {
      state: this.state,
      workerCount: slots.length,
      idleWorkerCount: slots.filter(({ state }) => state === 'idle').length,
      busyWorkerCount: slots.filter(({ state }) => state === 'busy').length,
      queuedRequestCount: this.queue.length,
      inFlightRequestCount: this.inFlight.size,
      startedWorkerCount: this.startedWorkerCount,
      retiredWorkerCount: this.retiredWorkerCount,
    };
  }

  private awaitStart(signal: AbortSignal): Promise<void> {
    const task = this.startTask!;
    if (signal.aborted) return Promise.reject(abortError(signal));
    return new Promise<void>((resolve, reject) => {
      const onAbort = (): void => reject(abortError(signal));
      signal.addEventListener('abort', onAbort, { once: true });
      void task.then(resolve, reject).finally(() => signal.removeEventListener('abort', onAbort));
    });
  }

  private async submit(
    operation: 'decide',
    frame: BehaviorInput,
    signal: AbortSignal,
    observationSnapshot?: Readonly<ClawclawBehaviorObservationSnapshot>,
  ): Promise<unknown> {
    if (signal.aborted) throw abortError(signal);
    await this.awaitStart(signal);
    if (this.state !== 'running') throw new Error('behavior_thread_pool_not_running');
    if ((this.options.source.execution ?? 'parallel') !== 'parallel') {
      throw new Error('behavior_thread_decide_not_available');
    }
    if (this.inFlight.size >= this.maxWorkerCount) throw new Error('behavior_thread_window_full');
    return new Promise<unknown>((resolve, reject) => {
      const request: PendingRequest = {
        requestId: randomUUID(),
        operation,
        frame: structuredClone(frame),
        ...(observationSnapshot === undefined
          ? {}
          : { observationSnapshot: structuredClone(observationSnapshot) }),
        signal,
        enqueuedAt: performance.now(),
        resolve,
        reject,
        settled: false,
        onAbort: () => this.abortRequest(request),
      };
      signal.addEventListener('abort', request.onAbort, { once: true });
      this.inFlight.add(request);
      this.queue.push(request);
      this.diagnostic(`behavior.${operation}.started`, {
        behavior_id: this.options.source.id,
        executor: 'worker_thread',
        runtime_instance_id: this.runtimeInstanceId,
        request_id: request.requestId,
        epoch_id: frame.epochId,
        ...(frame.gameState.tick === undefined ? {} : { tick: frame.gameState.tick }),
      }, 'debug');
      this.pump();
      this.growForPressure();
    });
  }

  private async spawnSlot(reason: 'initial' | 'queue_pressure' | 'replacement'): Promise<void> {
    const source = resolveBehaviorSource(this.options.source);
    const sourceUrl = behaviorSourceUrl(source);
    const slotId = this.nextSlotId++;
    const loader = behaviorThreadLoaderConfig();
    const worker = new Worker(WORKER_BOOTSTRAP, {
      execArgv: [],
      resourceLimits: { maxOldGenerationSizeMb: this.maxOldGenerationSizeMb },
      stdout: true,
      stderr: true,
      workerData: {
        runtimeModuleBootstrap: {
          hostUrl: WORKER_HOST_URL,
          hooksUrl: loader.hooksUrl,
          sdkAliases: loader.sdkAliases,
          esmModuleRoots: [new URL('.', sourceUrl).href],
        },
        poolId: this.runtimeInstanceId,
        slotId,
        source,
        sourceUrl,
      },
    });
    const slot: WorkerSlot = { id: slotId, worker, state: 'starting' };
    this.slots.add(slot);
    worker.stdout?.on('data', (chunk: Buffer) => this.diagnostic(
      'behavior.thread.stdout',
      { behavior_id: source.id, slot_id: slot.id, bytes: chunk.length },
      'debug',
    ));
    worker.stderr?.on('data', (chunk: Buffer) => this.diagnostic(
      'behavior.thread.stderr',
      { behavior_id: source.id, slot_id: slot.id, bytes: chunk.length },
      'debug',
    ));
    worker.on('message', (message: unknown) => this.handleMessage(slot, message));
    worker.once('error', (error) => this.failSlot(slot, error));
    worker.once('exit', (code) => {
      if (slot.state !== 'retiring') {
        this.failSlot(slot, new Error(`behavior_thread_worker_exited:${code}`));
      }
    });
    let timer: ReturnType<typeof setTimeout> | undefined;
    try {
      await new Promise<void>((resolve, reject) => {
        slot.readyResolve = resolve;
        slot.readyReject = reject;
        timer = setTimeout(() => reject(new Error('behavior_thread_worker_start_timeout')), START_TIMEOUT_MS);
        timer.unref?.();
      });
      if (this.state !== 'starting' && this.state !== 'running') {
        throw new Error('behavior_thread_pool_stopped');
      }
      slot.state = 'idle';
      slot.idleSince = performance.now();
      this.startedWorkerCount += 1;
      this.diagnostic('behavior.thread.started', {
        behavior_id: source.id,
        runtime_instance_id: this.runtimeInstanceId,
        slot_id: slot.id,
        worker_count: this.activeWorkerCount(),
        min_worker_count: this.minWorkerCount,
        max_worker_count: this.maxWorkerCount,
        reason,
      }, 'info');
    } catch (error) {
      await this.retireSlot(slot, 'start_failed');
      throw error;
    } finally {
      if (timer) clearTimeout(timer);
      slot.readyResolve = undefined;
      slot.readyReject = undefined;
    }
  }

  private pump(): void {
    if (this.state !== 'running') return;
    while (this.queue.length > 0) {
      const slot = [...this.slots].find((candidate) => candidate.state === 'idle');
      if (!slot) return;
      const request = this.queue.shift()!;
      if (request.settled || request.signal.aborted) {
        if (!request.settled) this.rejectRequest(request, abortError(request.signal));
        continue;
      }
      slot.state = 'busy';
      slot.idleSince = undefined;
      slot.current = request;
      request.slot = slot;
      request.assignedAt = performance.now();
      try {
        slot.worker.postMessage({
          type: 'run',
          poolId: this.runtimeInstanceId,
          requestId: request.requestId,
          operation: request.operation,
          frame: request.frame,
          ...(request.observationSnapshot === undefined
            ? {}
            : { observationSnapshot: request.observationSnapshot }),
        });
      } catch (error) {
        this.rejectRequest(request, asError(error, 'behavior_thread_submit_failed'));
        void this.retireSlot(slot, 'submit_failed');
      }
    }
  }

  private handleMessage(slot: WorkerSlot, value: unknown): void {
    if (!value || typeof value !== 'object' || Array.isArray(value)) {
      this.failSlot(slot, new Error('behavior_thread_protocol_invalid'));
      return;
    }
    const message = value as WorkerReadyMessage
      | WorkerResultMessage
      | WorkerErrorMessage;
    if (message.poolId !== this.runtimeInstanceId || message.slotId !== slot.id) {
      this.failSlot(slot, new Error('behavior_thread_protocol_identity_mismatch'));
      return;
    }
    if (message.type === 'ready') {
      if (
        slot.state !== 'starting'
        || message.behaviorId !== this.options.source.id
        || message.execution !== this.options.source.execution
      ) {
        this.failSlot(slot, new Error('behavior_thread_ready_invalid'));
        return;
      }
      slot.readyResolve?.();
      return;
    }
    if (message.type === 'error') {
      const code = typeof message.code === 'string'
        ? message.code
        : 'behavior_thread_worker_failed';
      const detail = typeof message.message === 'string'
        ? message.message.trim().slice(0, MAX_WORKER_ERROR_MESSAGE_LENGTH)
        : '';
      const error = new Error(detail ? `${code}: ${detail}` : code);
      if (slot.current && (!message.requestId || message.requestId === slot.current.requestId)) {
        this.rejectRequest(slot.current, error);
      } else {
        slot.readyReject?.(error);
      }
      void this.retireSlot(slot, 'worker_error');
      return;
    }
    const request = slot.current;
    if (!request || message.requestId !== request.requestId) {
      this.failSlot(slot, new Error('behavior_thread_request_mismatch'));
      return;
    }
    if (message.type !== 'result') {
      this.rejectRequest(request, new Error('behavior_thread_response_invalid'));
      void this.retireSlot(slot, 'invalid_response');
      return;
    }
    slot.current = undefined;
    slot.state = 'idle';
    slot.idleSince = performance.now();
    request.slot = undefined;
    this.resolveRequest(request, structuredClone(message.result), message);
    this.pump();
  }

  private resolveRequest(
    request: PendingRequest,
    value: unknown,
    message: WorkerResultMessage,
  ): void {
    if (request.settled) return;
    request.settled = true;
    request.signal.removeEventListener('abort', request.onAbort);
    this.inFlight.delete(request);
    this.diagnostic(`behavior.${request.operation}.completed`, {
      behavior_id: this.options.source.id,
      executor: 'worker_thread',
      runtime_instance_id: this.runtimeInstanceId,
      request_id: request.requestId,
      epoch_id: request.frame.epochId,
      duration_ms: roundedElapsed(request.enqueuedAt),
      queue_duration_ms: Math.round(
        Math.max(0, (request.assignedAt ?? performance.now()) - request.enqueuedAt) * 1_000,
      ) / 1_000,
      worker_duration_ms: message.durationMs,
      read_query_count: message.queryCount,
    }, 'debug');
    request.resolve(value);
    this.scheduleIdleScaleDown();
  }

  private rejectRequest(request: PendingRequest, error: Error): void {
    if (request.settled) return;
    request.settled = true;
    request.signal.removeEventListener('abort', request.onAbort);
    this.inFlight.delete(request);
    const index = this.queue.indexOf(request);
    if (index >= 0) this.queue.splice(index, 1);
    if (request.slot?.current === request) request.slot.current = undefined;
    request.slot = undefined;
    this.diagnostic(
      request.signal.aborted
        ? `behavior.${request.operation}.cancelled`
        : `behavior.${request.operation}.failed`,
      {
        behavior_id: this.options.source.id,
        executor: 'worker_thread',
        runtime_instance_id: this.runtimeInstanceId,
        request_id: request.requestId,
        epoch_id: request.frame.epochId,
        duration_ms: roundedElapsed(request.enqueuedAt),
        aborted: request.signal.aborted,
        error_code: errorCode(error, `BEHAVIOR_${request.operation.toUpperCase()}_FAILED`),
      },
      request.signal.aborted ? 'debug' : 'error',
    );
    request.reject(error);
    this.scheduleIdleScaleDown();
  }

  private abortRequest(request: PendingRequest): void {
    if (request.settled) return;
    const slot = request.slot;
    this.rejectRequest(request, abortError(request.signal));
    // Synchronous behavior code cannot be interrupted inside one isolate.
    // Replace only its slot; computations in the other Workers remain in flight.
    if (slot) void this.retireSlot(slot, 'request_aborted');
  }

  private failSlot(slot: WorkerSlot, error: Error): void {
    slot.readyReject?.(error);
    if (slot.current) this.rejectRequest(slot.current, error);
    void this.retireSlot(slot, 'failed');
  }

  private retireSlot(slot: WorkerSlot, reason: string): Promise<void> {
    if (slot.retirement) return slot.retirement;
    slot.readyReject?.(new Error(`behavior_thread_worker_retired:${reason}`));
    slot.state = 'retiring';
    if (slot.current) {
      this.rejectRequest(slot.current, new Error(`behavior_thread_worker_retired:${reason}`));
    }
    const retirement = Promise.resolve(slot.worker.terminate()).catch(() => -1).then(() => {
      this.slots.delete(slot);
      this.retiredWorkerCount += 1;
      this.diagnostic('behavior.thread.retired', {
        behavior_id: this.options.source.id,
        runtime_instance_id: this.runtimeInstanceId,
        slot_id: slot.id,
        reason,
        worker_count: this.activeWorkerCount(),
        min_worker_count: this.minWorkerCount,
        max_worker_count: this.maxWorkerCount,
      }, reason === 'request_aborted' ? 'debug' : 'info');
      if (this.state === 'running') this.ensureReplacement('replacement');
    });
    slot.retirement = retirement;
    return retirement;
  }

  private ensureReplacement(reason: 'queue_pressure' | 'replacement'): void {
    if (
      this.state !== 'running'
      || this.activeWorkerCount() >= this.desiredWorkerCount
      || this.replacementTask
      || this.replacementRetryTimer
    ) {
      this.pump();
      return;
    }
    const task = this.spawnSlot(reason).then(() => {
      this.replacementFailureCount = 0;
      this.pump();
      this.scheduleIdleScaleDown();
    }).catch((error) => {
      this.replacementFailureCount += 1;
      const retryDelayMs = Math.min(
        REPLACEMENT_RETRY_MAX_MS,
        REPLACEMENT_RETRY_BASE_MS * (2 ** Math.min(10, this.replacementFailureCount - 1)),
      );
      this.diagnostic('behavior.thread.replacement_failed', {
        behavior_id: this.options.source.id,
        runtime_instance_id: this.runtimeInstanceId,
        error_code: errorCode(error, 'BEHAVIOR_THREAD_REPLACEMENT_FAILED'),
        reason,
        retry_delay_ms: retryDelayMs,
      }, 'error');
      if (this.state === 'running') {
        this.replacementRetryTimer = setTimeout(() => {
          this.replacementRetryTimer = undefined;
          this.ensureReplacement(reason);
        }, retryDelayMs);
        this.replacementRetryTimer.unref?.();
      }
    }).finally(() => {
      this.spawnTasks.delete(task);
      this.replacementTask = undefined;
      if (
        this.state === 'running'
        && this.activeWorkerCount() < this.desiredWorkerCount
        && !this.replacementRetryTimer
      ) this.ensureReplacement(reason);
    });
    this.replacementTask = task;
    this.spawnTasks.add(task);
  }

  private growForPressure(): void {
    const target = Math.min(this.maxWorkerCount, Math.max(this.minWorkerCount, this.inFlight.size));
    if (target <= this.desiredWorkerCount) return;
    const previous = this.desiredWorkerCount;
    this.desiredWorkerCount = target;
    this.diagnostic('behavior.thread_pool.scale_requested', {
      behavior_id: this.options.source.id,
      runtime_instance_id: this.runtimeInstanceId,
      trigger: 'queue_pressure',
      previous_worker_count: previous,
      desired_worker_count: target,
      worker_count: this.activeWorkerCount(),
      min_worker_count: this.minWorkerCount,
      max_worker_count: this.maxWorkerCount,
      queued_request_count: this.queue.length,
      in_flight_request_count: this.inFlight.size,
    }, 'info');
    this.ensureReplacement('queue_pressure');
    this.scheduleIdleScaleDown();
  }

  private scheduleIdleScaleDown(): void {
    if (this.state !== 'running' || this.idleScaleDownTimer) return;
    const delayMs = this.nextIdleScaleDownDelayMs();
    if (delayMs === undefined) return;
    this.idleScaleDownTimer = setTimeout(() => {
      this.idleScaleDownTimer = undefined;
      if (this.state !== 'running') return;
      // inFlight already contains both assigned and queued requests, so it is the current
      // concurrency demand. Never retire a busy Worker or shrink below that demand.
      const demand = Math.min(
        this.maxWorkerCount,
        Math.max(this.minWorkerCount, this.inFlight.size),
      );
      const previous = this.desiredWorkerCount;
      this.desiredWorkerCount = demand;
      const activeWorkerCount = this.activeWorkerCount();
      const retiredWorkerCount = this.trimIdleWorkers(demand);
      this.diagnostic(
        retiredWorkerCount > 0 || previous !== demand
          ? 'behavior.thread_pool.scale_requested'
          : 'behavior.thread_pool.scale_deferred',
        {
          behavior_id: this.options.source.id,
          runtime_instance_id: this.runtimeInstanceId,
          trigger: retiredWorkerCount > 0 ? 'idle' : 'current_demand',
          previous_worker_count: previous,
          desired_worker_count: demand,
          worker_count: activeWorkerCount,
          min_worker_count: this.minWorkerCount,
          max_worker_count: this.maxWorkerCount,
          in_flight_request_count: this.inFlight.size,
          queued_request_count: this.queue.length,
          idle_timeout_ms: this.idleWorkerTimeoutMs,
          retired_worker_count: retiredWorkerCount,
        },
        'info',
      );
      if (this.activeWorkerCount() < this.desiredWorkerCount) {
        this.ensureReplacement('queue_pressure');
      }
      this.scheduleIdleScaleDown();
    }, delayMs);
    this.idleScaleDownTimer.unref?.();
  }

  private nextIdleScaleDownDelayMs(): number | undefined {
    const excess = this.activeWorkerCount() - this.currentWorkerDemand();
    if (excess <= 0) return undefined;
    const oldestIdleSince = [...this.slots].reduce<number | undefined>((oldest, slot) => {
      if (slot.state !== 'idle' || slot.idleSince === undefined) return oldest;
      return oldest === undefined ? slot.idleSince : Math.min(oldest, slot.idleSince);
    }, undefined);
    if (oldestIdleSince === undefined) return undefined;
    return Math.max(
      0,
      Math.ceil(oldestIdleSince + this.idleWorkerTimeoutMs - performance.now()),
    );
  }

  private cancelIdleScaleDown(): void {
    if (!this.idleScaleDownTimer) return;
    clearTimeout(this.idleScaleDownTimer);
    this.idleScaleDownTimer = undefined;
  }

  private trimIdleWorkers(demand = this.currentWorkerDemand()): number {
    let excess = this.activeWorkerCount() - demand;
    if (excess <= 0) return 0;
    const cutoff = performance.now() - this.idleWorkerTimeoutMs;
    const candidates = [...this.slots]
      .filter((slot) => (
        slot.state === 'idle'
        && slot.idleSince !== undefined
        && slot.idleSince <= cutoff
      ))
      .sort((left, right) => left.idleSince! - right.idleSince!);
    let retired = 0;
    for (const slot of candidates) {
      if (excess <= 0) break;
      excess -= 1;
      retired += 1;
      void this.retireSlot(slot, 'idle');
    }
    return retired;
  }

  private currentWorkerDemand(): number {
    return Math.min(
      this.maxWorkerCount,
      Math.max(this.minWorkerCount, this.inFlight.size),
    );
  }

  private activeWorkerCount(): number {
    return [...this.slots].filter((slot) => slot.state !== 'retiring').length;
  }

  private diagnostic(
    name: string,
    attributes?: Readonly<Record<string, unknown>>,
    severity?: RuntimeDiagnosticSeverity,
  ): void {
    try {
      this.options.diagnostic?.(name, attributes, severity);
    } catch {
      // Diagnostics are observability only and must never alter pool lifecycle.
    }
  }
}