Add files using upload-large-folder tool

4e23b01 verified about 1 month ago

19.9 kB

import fsSync from 'node:fs';
import { randomUUID } from 'node:crypto';
import { readFile, rm } from 'node:fs/promises';
import { join } from 'node:path';
import { pathToFileURL } from 'node:url';
import { Worker } from 'node:worker_threads';

import { getTextBuildWorkerRuntimeState } from '@moonshot-ai/minidb/worker-runtime';

import { GlobalSearchError } from '../contract';
import type {
  CoreIndexView,
  CoreLifecycleReport,
  CoreSearchParams,
  CoreSearchResult,
  CoreStatus,
  CoreSyncOutcome,
  SearchCoreLog,
  SyncSessionInput,
} from '../indexCore';
import {
  SEARCH_WORKER_PROTOCOL_VERSION,
  type SearchWorkerData,
  type SearchWorkerEvent,
  type SearchWorkerOpenResult,
  type SearchWorkerCallType,
  type SearchWorkerResultMap,
} from './protocol';
import { getSearchWorkerRuntimeState } from './runtime';

export type SearchWorkerErrorCode =
  | 'runtime-unavailable'
  | 'spawn-failed'
  | 'crashed'
  | 'backoff'
  | 'disposed';

export class SearchWorkerError extends Error {
  constructor(
    readonly code: SearchWorkerErrorCode,
    message: string,
  ) {
    super(message);
    this.name = 'SearchWorkerError';
  }
}

export interface SearchWorkerHostOptions {
  readonly dir: string;
  readonly log: SearchCoreLog;
  readonly readyTimeoutMs?: number;
  readonly closeTimeoutMs?: number;
  readonly requestTimeoutMs?: number;
  readonly syncTimeoutMs?: number;
  readonly maxOldSpaceMb?: number;
  readonly workerFactory?: (entry: { url: URL; data: SearchWorkerData; execArgv: string[] }) => Worker;
}

interface WorkerEntryResolution {
  readonly url: URL;
  readonly execArgv: string[];
}

type PendingRequest = {
  readonly method: SearchWorkerCallType;
  readonly resolve: (value: unknown) => void;
  readonly reject: (error: unknown) => void;
  readonly watchdog?: ReturnType<typeof setTimeout>;
};

const DEFAULT_READY_TIMEOUT_MS = 15_000;
const DEFAULT_CLOSE_TIMEOUT_MS = 30_000;
const DEFAULT_MAX_OLD_SPACE_MB = 1024;
const STABLE_SESSION_MS = 60_000;
const BACKOFF_BASE_MS = 500;
const BACKOFF_CAP_MS = 10_000;
const DEFAULT_REQUEST_TIMEOUT_MS = 60_000;
const DEFAULT_SYNC_TIMEOUT_MS = 30 * 60_000;
const ORPHAN_LOCK_GRACE_MS = 250;

const liveLockTokens = new Set<string>();

export function noteLiveLockToken(token: string): void {
  liveLockTokens.add(token);
}

export function dropLiveLockToken(token: string | undefined): void {
  if (token !== undefined) liveLockTokens.delete(token);
}

export class SearchWorkerHost {
  private worker: Worker | null = null;
  private spawnPromise: Promise<void> | null = null;
  private reapPromise: Promise<void> | null = null;
  private readonly requests = new Map<number, PendingRequest>();
  private nextId = 1;
  private lockToken: string | undefined;
  private failures = 0;
  private nextRetryAfter = 0;
  private lastFailure: string | null = null;
  private readyAt = 0;
  private exiting = false;
  private exitResolve: (() => void) | null = null;
  private orphanCheckScheduled = false;
  private lastCoreLifecycle: CoreLifecycleReport | null = null;

  constructor(private readonly options: SearchWorkerHostOptions) {}

  private get log(): SearchCoreLog {
    return this.options.log;
  }

  private get dir(): string {
    return this.options.dir;
  }

  get reportedLockToken(): string | undefined {
    return this.lockToken;
  }

  lifecycleSnapshot(): CoreLifecycleReport {
    if (this.exiting) {
      return this.worker !== null || this.spawnPromise !== null
        ? { state: 'closing' }
        : { state: 'stopped' };
    }
    if (this.spawnPromise !== null) return { state: 'opening' };
    if (this.worker === null) {
      if (this.lastFailure !== null) {
        const wait = this.nextRetryAfter - Date.now();
        return {
          state: 'degraded',
          detail:
            wait > 0
              ? `search worker restart backing off for ${wait}ms (last failure: ${this.lastFailure})`
              : `search worker restart pending (last failure: ${this.lastFailure})`,
        };
      }
      return { state: 'stopped' };
    }
    return this.lastCoreLifecycle ?? { state: 'opening' };
  }

  async ensureOpen(): Promise<SearchWorkerOpenResult> {
    return this.call('open');
  }

  async search(params: CoreSearchParams): Promise<CoreSearchResult> {
    return this.call('search', params);
  }

  async sync(sessions: readonly SyncSessionInput[]): Promise<CoreSyncOutcome> {
    return this.call('sync', { sessions });
  }

  async refresh(): Promise<SearchWorkerOpenResult> {
    return this.call('refresh');
  }

  async reindex(): Promise<SearchWorkerOpenResult> {
    return this.call('reindex');
  }

  async status(): Promise<CoreStatus> {
    return this.call('status');
  }

  async killWorkerForTest(): Promise<void> {
    const worker = this.worker;
    if (worker === null) return;
    await worker.terminate();
    await this.waitForExit();
  }

  private async call<T extends SearchWorkerCallType>(
    type: T,
    params?: unknown,
  ): Promise<SearchWorkerResultMap[T]> {
    await this.ensureWorker();
    const worker = this.worker;
    if (worker === null) {
      throw new SearchWorkerError('crashed', 'search worker is not running');
    }
    return this.sendRequest(worker, type, params) as Promise<SearchWorkerResultMap[T]>;
  }

  private sendRequest(
    worker: Worker,
    type: SearchWorkerCallType,
    params?: unknown,
  ): Promise<unknown> {
    const id = this.nextId++;
    return new Promise<unknown>((resolve, reject) => {
      let watchdog: ReturnType<typeof setTimeout> | undefined;
      if (type !== 'close') {
        const timeoutMs =
          type === 'sync' || type === 'reindex' || type === 'open'
            ? this.options.syncTimeoutMs ?? DEFAULT_SYNC_TIMEOUT_MS
            : this.options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
        watchdog = setTimeout(() => this.onRequestTimeout(id), timeoutMs);
        watchdog.unref?.();
      }
      this.requests.set(id, { method: type, resolve, reject, watchdog });
      try {
        worker.postMessage({ id, v: SEARCH_WORKER_PROTOCOL_VERSION, type, params });
      } catch (error) {
        this.requests.delete(id);
        if (watchdog !== undefined) clearTimeout(watchdog);
        reject(
          new SearchWorkerError(
            'crashed',
            `search worker channel broke: ${error instanceof Error ? error.message : String(error)}`,
          ),
        );
      }
    });
  }

  private onRequestTimeout(id: number): void {
    const pending = this.requests.get(id);
    if (pending === undefined) return;
    this.requests.delete(id);
    const message = `search worker request '${pending.method}' timed out`;
    this.log.warn(`global search: ${message}; terminating the wedged worker`, { dir: this.dir });
    pending.reject(new SearchWorkerError('crashed', message));
    const worker = this.worker;
    if (worker !== null) {
      void worker.terminate().catch(() => {});
    }
  }

  private settleRequest(id: number, error: Error | null, result: unknown): void {
    const pending = this.requests.get(id);
    if (pending === undefined) return;
    this.requests.delete(id);
    if (pending.watchdog !== undefined) clearTimeout(pending.watchdog);
    if (error !== null) {
      pending.reject(error);
      return;
    }
    this.noteResult(result);
    pending.resolve(result);
  }

  private noteResult(result: unknown): void {
    if (result === null || typeof result !== 'object') return;
    const direct = (result as { lockToken?: unknown }).lockToken;
    const nested = (result as { index?: { lockToken?: unknown } }).index?.lockToken;
    const token = typeof direct === 'string' ? direct : typeof nested === 'string' ? nested : undefined;
    if (token !== undefined) this.lockToken = token;
    const readOnly =
      (result as { readOnly?: unknown }).readOnly === true ||
      (result as { index?: { readOnly?: unknown } }).index?.readOnly === true;
    if (readOnly) this.scheduleOrphanCheck();
    const lifecycle = (result as { lifecycle?: CoreLifecycleReport }).lifecycle;
    if (lifecycle !== undefined) {
      this.lastCoreLifecycle = lifecycle;
      return;
    }
    const index = (result as { index?: CoreIndexView }).index;
    if (index !== undefined) {
      this.lastCoreLifecycle = { state: index.state === 'building' ? 'building' : 'ready' };
    }
  }

  private ensureWorker(): Promise<void> {
    if (this.worker !== null) return Promise.resolve();
    if (this.exiting) {
      return Promise.reject(new SearchWorkerError('disposed', 'search worker is disposed'));
    }
    if (this.spawnPromise !== null) return this.spawnPromise;
    const wait = this.nextRetryAfter - Date.now();
    if (wait > 0) {
      return Promise.reject(
        new SearchWorkerError(
          'backoff',
          `search worker restart backing off for ${wait}ms (last failure: ${this.lastFailure ?? 'unknown'})`,
        ),
      );
    }
    this.spawnPromise = this.doSpawn().finally(() => {
      this.spawnPromise = null;
    });
    return this.spawnPromise;
  }

  private resolveEntry(): WorkerEntryResolution {
    const configured = getSearchWorkerRuntimeState();
    if (configured.configured) {
      return { url: pathToFileURL(configured.path), execArgv: [] };
    }
    const source = new URL('./entry.ts', import.meta.url);
    try {
      if (fsSync.statSync(source).isFile()) {
        return {
          url: source,
          execArgv: [
            '--experimental-transform-types',
            '--disable-warning=ExperimentalWarning',
            '--import',
            new URL('./register-dev-hooks.mjs', import.meta.url).href,
          ],
        };
      }
    } catch {
    }
    const bundled = new URL('./search-worker.mjs', import.meta.url);
    try {
      if (fsSync.statSync(bundled).isFile()) {
        return { url: bundled, execArgv: [] };
      }
    } catch {
    }
    throw new SearchWorkerError(
      'runtime-unavailable',
      'search worker entry not found (no configured packaged asset, no sibling source, no bundled sibling)',
    );
  }

  private async doSpawn(): Promise<void> {
    await this.reapPromise;
    const entry = this.resolveEntry();
    const textBuild = getTextBuildWorkerRuntimeState();
    const data: SearchWorkerData = {
      dir: this.dir,
      bootSalt: randomUUID(),
      textBuildWorkerPath:
        textBuild.configured && textBuild.entry.kind === 'packaged' ? textBuild.entry.path : undefined,
    };
    let worker: Worker;
    try {
      worker =
        this.options.workerFactory?.({
          url: entry.url,
          data,
          execArgv: entry.execArgv,
        }) ??
        new Worker(entry.url, {
          workerData: data,
          execArgv: entry.execArgv,
          resourceLimits: { maxOldGenerationSizeMb: this.options.maxOldSpaceMb ?? DEFAULT_MAX_OLD_SPACE_MB },
          name: 'kimi-search-worker',
        });
    } catch (error) {
      const message = error instanceof Error ? error.message : String(error);
      this.noteFailure(`spawn failed: ${message}`);
      throw new SearchWorkerError('spawn-failed', `search worker failed to start: ${message}`);
    }
    worker.unref();
    this.attach(worker);
    try {
      await this.awaitReady(worker);
    } catch (error) {
      this.detach();
      await worker.terminate().catch(() => {});
      const message = error instanceof Error ? error.message : String(error);
      this.noteFailure(`handshake failed: ${message}`);
      throw new SearchWorkerError('spawn-failed', `search worker handshake failed: ${message}`);
    }
    this.readyAt = Date.now();
    this.worker = worker;
    this.lastFailure = null;
    this.lastCoreLifecycle = null;
    if (this.exiting) {
      try {
        worker.postMessage({ v: SEARCH_WORKER_PROTOCOL_VERSION, type: 'beginClose' });
      } catch {
      }
    }
    this.log.info('global search: worker started', { dir: this.dir });
  }

  private attach(worker: Worker): void {
    worker.on('message', (event: SearchWorkerEvent) => this.onMessage(worker, event));
    worker.on('error', (error: Error) => this.onError(worker, error));
    worker.on('exit', (code: number) => this.onExit(worker, code));
  }

  private detach(): void {
    this.worker = null;
    this.exitResolve?.();
    this.exitResolve = null;
  }

  private onMessage(worker: Worker, event: SearchWorkerEvent): void {
    if (this.worker !== worker) return;
    if (event === null || typeof event !== 'object') return;
    if (event.type === 'log') {
      this.log[event.level](event.message, event.meta);
      return;
    }
    if (event.type === 'lockToken') {
      this.lockToken = event.token;
      liveLockTokens.add(event.token);
      return;
    }
    if (event.type === 'ready') return;
    if (event.type === 'result') {
      this.settleRequest(event.id, null, event.result);
      return;
    }
    if (event.type === 'error') {
      const payload = event.error;
      const error =
        payload.reason !== undefined
          ? new GlobalSearchError(payload.reason, payload.message)
          : new Error(payload.message);
      this.settleRequest(event.id, error, undefined);
    }
  }

  private onError(worker: Worker, error: Error): void {
    if (this.worker !== worker) return;
    this.lastFailure = `worker error: ${error.message}`;
  }

  private onExit(worker: Worker, code: number): void {
    if (this.worker !== worker) return;
    const wasExiting = this.exiting;
    const sessionMs = Date.now() - this.readyAt;
    const deadToken = this.lockToken;
    this.detach();
    this.lockToken = undefined;
    if (deadToken !== undefined) liveLockTokens.delete(deadToken);
    if (this.requests.size > 0) {
      const error = wasExiting
        ? new SearchWorkerError('disposed', 'search worker is disposed')
        : new SearchWorkerError(
            'crashed',
            `search worker exited with in-flight requests (code ${code}${this.lastFailure !== null ? `; ${this.lastFailure}` : ''})`,
          );
      for (const pending of this.requests.values()) {
        if (pending.watchdog !== undefined) clearTimeout(pending.watchdog);
        pending.reject(error);
      }
      this.requests.clear();
    }
    if (wasExiting) return;
    this.lastFailure ??= `worker exited with code ${code}`;
    this.reapPromise = this.reapLockFile(deadToken).finally(() => {
      this.reapPromise = null;
    });
    this.failures = sessionMs > STABLE_SESSION_MS ? Math.max(1, this.failures - 1) : this.failures + 1;
    const backoff = Math.min(BACKOFF_BASE_MS * 2 ** (this.failures - 1), BACKOFF_CAP_MS);
    this.nextRetryAfter = Date.now() + backoff;
    this.log.warn('global search: worker exited unexpectedly; restart backed off', {
      code,
      backoffMs: backoff,
      lastFailure: this.lastFailure ?? undefined,
    });
  }

  private noteFailure(message: string): void {
    this.failures += 1;
    const backoff = Math.min(BACKOFF_BASE_MS * 2 ** (this.failures - 1), BACKOFF_CAP_MS);
    this.nextRetryAfter = Date.now() + backoff;
    this.lastFailure = message;
  }

  private async reapLockFile(token: string | undefined): Promise<void> {
    if (token === undefined) return;
    const lockPath = join(this.dir, 'db.lock');
    try {
      const raw = await readFile(lockPath, 'utf8');
      const parsed = JSON.parse(raw) as { pid?: unknown; token?: unknown };
      if (parsed.pid === process.pid && parsed.token === token) {
        await rm(lockPath, { force: true });
        this.log.warn('global search: reaped the lock left by the dead worker', { dir: this.dir });
      }
    } catch {
    }
  }

  private awaitReady(worker: Worker): Promise<void> {
    return new Promise<void>((resolve, reject) => {
      const timer = setTimeout(() => {
        cleanup();
        reject(new Error(`ready handshake timed out after ${this.options.readyTimeoutMs ?? DEFAULT_READY_TIMEOUT_MS}ms`));
      }, this.options.readyTimeoutMs ?? DEFAULT_READY_TIMEOUT_MS);
      timer.unref?.();
      const onMessage = (event: SearchWorkerEvent): void => {
        if (event?.type !== 'ready') return;
        cleanup();
        if (event.v !== SEARCH_WORKER_PROTOCOL_VERSION) {
          reject(
            new Error(
              `protocol mismatch: host v${SEARCH_WORKER_PROTOCOL_VERSION}, worker v${event.v ?? 'unknown'}`,
            ),
          );
          return;
        }
        resolve();
      };
      const onError = (error: Error): void => {
        cleanup();
        reject(error);
      };
      const onExit = (code: number): void => {
        cleanup();
        reject(new Error(`worker exited before ready (code ${code})`));
      };
      const cleanup = (): void => {
        clearTimeout(timer);
        worker.off('message', onMessage);
        worker.off('error', onError);
        worker.off('exit', onExit);
      };
      worker.on('message', onMessage);
      worker.on('error', onError);
      worker.on('exit', onExit);
    });
  }

  private waitForExit(): Promise<void> {
    if (this.worker === null) return Promise.resolve();
    return new Promise<void>((resolve) => {
      const previous = this.exitResolve;
      this.exitResolve = () => {
        previous?.();
        resolve();
      };
    });
  }

  beginClose(): void {
    this.exiting = true;
    try {
      this.worker?.postMessage({ v: SEARCH_WORKER_PROTOCOL_VERSION, type: 'beginClose' });
    } catch {
    }
  }

  private scheduleOrphanCheck(): void {
    if (this.orphanCheckScheduled || this.exiting) return;
    this.orphanCheckScheduled = true;
    const timer = setTimeout(() => {
      this.orphanCheckScheduled = false;
      void this.recoverOrphanedLock();
    }, ORPHAN_LOCK_GRACE_MS);
    timer.unref?.();
  }

  private async recoverOrphanedLock(): Promise<void> {
    if (this.exiting || this.worker === null) return;
    const lockPath = join(this.dir, 'db.lock');
    let token: string;
    try {
      const parsed = JSON.parse(await readFile(lockPath, 'utf8')) as {
        pid?: unknown;
        token?: unknown;
      };
      if (parsed.pid !== process.pid || typeof parsed.token !== 'string') return;
      token = parsed.token;
    } catch {
      return;
    }
    if (liveLockTokens.has(token)) return;
    try {
      const again = JSON.parse(await readFile(lockPath, 'utf8')) as {
        pid?: unknown;
        token?: unknown;
      };
      if (again.pid !== process.pid || again.token !== token) return;
      if (liveLockTokens.has(token)) return;
    } catch {
      return;
    }
    this.log.warn('global search: reaping an orphaned same-pid lock and restarting the worker', {
      dir: this.dir,
    });
    await rm(lockPath, { force: true }).catch(() => {});
    const worker = this.worker;
    if (worker === null) return;
    await worker.terminate().catch(() => {});
  }

  dispose(): Promise<void> {
    this.exiting = true;
    this.disposePromise ??= this.doDispose();
    return this.disposePromise;
  }

  private disposePromise: Promise<void> | null = null;

  private async doDispose(): Promise<void> {
    await this.spawnPromise?.catch(() => {});
    const worker = this.worker;
    if (worker === null) return;
    const closeTimeoutMs = this.options.closeTimeoutMs ?? DEFAULT_CLOSE_TIMEOUT_MS;
    let timeout: ReturnType<typeof setTimeout> | null = null;
    try {
      await Promise.race([
        this.sendRequest(worker, 'close'),
        new Promise<never>((_resolve, reject) => {
          timeout = setTimeout(() => {
            reject(new Error(`search worker close timed out after ${closeTimeoutMs}ms`));
          }, closeTimeoutMs);
          timeout.unref?.();
        }),
      ]);
    } catch {
    } finally {
      if (timeout !== null) clearTimeout(timeout);
    }
    if (this.worker !== null) {
      await worker.terminate().catch(() => {});
      await this.waitForExit();
    }
  }
}

	import fsSync from 'node:fs';
	import { randomUUID } from 'node:crypto';
	import { readFile, rm } from 'node:fs/promises';
	import { join } from 'node:path';
	import { pathToFileURL } from 'node:url';
	import { Worker } from 'node:worker_threads';

	import { getTextBuildWorkerRuntimeState } from '@moonshot-ai/minidb/worker-runtime';

	import { GlobalSearchError } from '../contract';
	import type {
	CoreIndexView,
	CoreLifecycleReport,
	CoreSearchParams,
	CoreSearchResult,
	CoreStatus,
	CoreSyncOutcome,
	SearchCoreLog,
	SyncSessionInput,
	} from '../indexCore';
	import {
	SEARCH_WORKER_PROTOCOL_VERSION,
	type SearchWorkerData,
	type SearchWorkerEvent,
	type SearchWorkerOpenResult,
	type SearchWorkerCallType,
	type SearchWorkerResultMap,
	} from './protocol';
	import { getSearchWorkerRuntimeState } from './runtime';

	export type SearchWorkerErrorCode =
	\| 'runtime-unavailable'
	\| 'spawn-failed'
	\| 'crashed'
	\| 'backoff'
	\| 'disposed';

	export class SearchWorkerError extends Error {
	constructor(
	readonly code: SearchWorkerErrorCode,
	message: string,
	) {
	super(message);
	this.name = 'SearchWorkerError';
	}
	}

	export interface SearchWorkerHostOptions {
	readonly dir: string;
	readonly log: SearchCoreLog;
	readonly readyTimeoutMs?: number;
	readonly closeTimeoutMs?: number;
	readonly requestTimeoutMs?: number;
	readonly syncTimeoutMs?: number;
	readonly maxOldSpaceMb?: number;
	readonly workerFactory?: (entry: { url: URL; data: SearchWorkerData; execArgv: string[] }) => Worker;
	}

	interface WorkerEntryResolution {
	readonly url: URL;
	readonly execArgv: string[];
	}

	type PendingRequest = {
	readonly method: SearchWorkerCallType;
	readonly resolve: (value: unknown) => void;
	readonly reject: (error: unknown) => void;
	readonly watchdog?: ReturnType<typeof setTimeout>;
	};

	const DEFAULT_READY_TIMEOUT_MS = 15_000;
	const DEFAULT_CLOSE_TIMEOUT_MS = 30_000;
	const DEFAULT_MAX_OLD_SPACE_MB = 1024;
	const STABLE_SESSION_MS = 60_000;
	const BACKOFF_BASE_MS = 500;
	const BACKOFF_CAP_MS = 10_000;
	const DEFAULT_REQUEST_TIMEOUT_MS = 60_000;
	const DEFAULT_SYNC_TIMEOUT_MS = 30 * 60_000;
	const ORPHAN_LOCK_GRACE_MS = 250;

	const liveLockTokens = new Set<string>();

	export function noteLiveLockToken(token: string): void {
	liveLockTokens.add(token);
	}

	export function dropLiveLockToken(token: string \| undefined): void {
	if (token !== undefined) liveLockTokens.delete(token);
	}

	export class SearchWorkerHost {
	private worker: Worker \| null = null;
	private spawnPromise: Promise<void> \| null = null;
	private reapPromise: Promise<void> \| null = null;
	private readonly requests = new Map<number, PendingRequest>();
	private nextId = 1;
	private lockToken: string \| undefined;
	private failures = 0;
	private nextRetryAfter = 0;
	private lastFailure: string \| null = null;
	private readyAt = 0;
	private exiting = false;
	private exitResolve: (() => void) \| null = null;
	private orphanCheckScheduled = false;
	private lastCoreLifecycle: CoreLifecycleReport \| null = null;

	constructor(private readonly options: SearchWorkerHostOptions) {}

	private get log(): SearchCoreLog {
	return this.options.log;
	}

	private get dir(): string {
	return this.options.dir;
	}

	get reportedLockToken(): string \| undefined {
	return this.lockToken;
	}

	lifecycleSnapshot(): CoreLifecycleReport {
	if (this.exiting) {
	return this.worker !== null \|\| this.spawnPromise !== null
	? { state: 'closing' }
	: { state: 'stopped' };
	}
	if (this.spawnPromise !== null) return { state: 'opening' };
	if (this.worker === null) {
	if (this.lastFailure !== null) {
	const wait = this.nextRetryAfter - Date.now();
	return {
	state: 'degraded',
	detail:
	wait > 0
	? `search worker restart backing off for ${wait}ms (last failure: ${this.lastFailure})`
	: `search worker restart pending (last failure: ${this.lastFailure})`,
	};
	}
	return { state: 'stopped' };
	}
	return this.lastCoreLifecycle ?? { state: 'opening' };
	}

	async ensureOpen(): Promise<SearchWorkerOpenResult> {
	return this.call('open');
	}

	async search(params: CoreSearchParams): Promise<CoreSearchResult> {
	return this.call('search', params);
	}

	async sync(sessions: readonly SyncSessionInput[]): Promise<CoreSyncOutcome> {
	return this.call('sync', { sessions });
	}

	async refresh(): Promise<SearchWorkerOpenResult> {
	return this.call('refresh');
	}

	async reindex(): Promise<SearchWorkerOpenResult> {
	return this.call('reindex');
	}

	async status(): Promise<CoreStatus> {
	return this.call('status');
	}

	async killWorkerForTest(): Promise<void> {
	const worker = this.worker;
	if (worker === null) return;
	await worker.terminate();
	await this.waitForExit();
	}

	private async call<T extends SearchWorkerCallType>(
	type: T,
	params?: unknown,
	): Promise<SearchWorkerResultMap[T]> {
	await this.ensureWorker();
	const worker = this.worker;
	if (worker === null) {
	throw new SearchWorkerError('crashed', 'search worker is not running');
	}
	return this.sendRequest(worker, type, params) as Promise<SearchWorkerResultMap[T]>;
	}

	private sendRequest(
	worker: Worker,
	type: SearchWorkerCallType,
	params?: unknown,
	): Promise<unknown> {
	const id = this.nextId++;
	return new Promise<unknown>((resolve, reject) => {
	let watchdog: ReturnType<typeof setTimeout> \| undefined;
	if (type !== 'close') {
	const timeoutMs =
	type === 'sync' \|\| type === 'reindex' \|\| type === 'open'
	? this.options.syncTimeoutMs ?? DEFAULT_SYNC_TIMEOUT_MS
	: this.options.requestTimeoutMs ?? DEFAULT_REQUEST_TIMEOUT_MS;
	watchdog = setTimeout(() => this.onRequestTimeout(id), timeoutMs);
	watchdog.unref?.();
	}
	this.requests.set(id, { method: type, resolve, reject, watchdog });
	try {
	worker.postMessage({ id, v: SEARCH_WORKER_PROTOCOL_VERSION, type, params });
	} catch (error) {
	this.requests.delete(id);
	if (watchdog !== undefined) clearTimeout(watchdog);
	reject(
	new SearchWorkerError(
	'crashed',
	`search worker channel broke: ${error instanceof Error ? error.message : String(error)}`,
	),
	);
	}
	});
	}

	private onRequestTimeout(id: number): void {
	const pending = this.requests.get(id);
	if (pending === undefined) return;
	this.requests.delete(id);
	const message = `search worker request '${pending.method}' timed out`;
	this.log.warn(`global search: ${message}; terminating the wedged worker`, { dir: this.dir });
	pending.reject(new SearchWorkerError('crashed', message));
	const worker = this.worker;
	if (worker !== null) {
	void worker.terminate().catch(() => {});
	}
	}

	private settleRequest(id: number, error: Error \| null, result: unknown): void {
	const pending = this.requests.get(id);
	if (pending === undefined) return;
	this.requests.delete(id);
	if (pending.watchdog !== undefined) clearTimeout(pending.watchdog);
	if (error !== null) {
	pending.reject(error);
	return;
	}
	this.noteResult(result);
	pending.resolve(result);
	}

	private noteResult(result: unknown): void {
	if (result === null \|\| typeof result !== 'object') return;
	const direct = (result as { lockToken?: unknown }).lockToken;
	const nested = (result as { index?: { lockToken?: unknown } }).index?.lockToken;
	const token = typeof direct === 'string' ? direct : typeof nested === 'string' ? nested : undefined;
	if (token !== undefined) this.lockToken = token;
	const readOnly =
	(result as { readOnly?: unknown }).readOnly === true \|\|
	(result as { index?: { readOnly?: unknown } }).index?.readOnly === true;
	if (readOnly) this.scheduleOrphanCheck();
	const lifecycle = (result as { lifecycle?: CoreLifecycleReport }).lifecycle;
	if (lifecycle !== undefined) {
	this.lastCoreLifecycle = lifecycle;
	return;
	}
	const index = (result as { index?: CoreIndexView }).index;
	if (index !== undefined) {
	this.lastCoreLifecycle = { state: index.state === 'building' ? 'building' : 'ready' };
	}
	}

	private ensureWorker(): Promise<void> {
	if (this.worker !== null) return Promise.resolve();
	if (this.exiting) {
	return Promise.reject(new SearchWorkerError('disposed', 'search worker is disposed'));
	}
	if (this.spawnPromise !== null) return this.spawnPromise;
	const wait = this.nextRetryAfter - Date.now();
	if (wait > 0) {
	return Promise.reject(
	new SearchWorkerError(
	'backoff',
	`search worker restart backing off for ${wait}ms (last failure: ${this.lastFailure ?? 'unknown'})`,
	),
	);
	}
	this.spawnPromise = this.doSpawn().finally(() => {
	this.spawnPromise = null;
	});
	return this.spawnPromise;
	}

	private resolveEntry(): WorkerEntryResolution {
	const configured = getSearchWorkerRuntimeState();
	if (configured.configured) {
	return { url: pathToFileURL(configured.path), execArgv: [] };
	}
	const source = new URL('./entry.ts', import.meta.url);
	try {
	if (fsSync.statSync(source).isFile()) {
	return {
	url: source,
	execArgv: [
	'--experimental-transform-types',
	'--disable-warning=ExperimentalWarning',
	'--import',
	new URL('./register-dev-hooks.mjs', import.meta.url).href,
	],
	};
	}
	} catch {
	}
	const bundled = new URL('./search-worker.mjs', import.meta.url);
	try {
	if (fsSync.statSync(bundled).isFile()) {
	return { url: bundled, execArgv: [] };
	}
	} catch {
	}
	throw new SearchWorkerError(
	'runtime-unavailable',
	'search worker entry not found (no configured packaged asset, no sibling source, no bundled sibling)',
	);
	}

	private async doSpawn(): Promise<void> {
	await this.reapPromise;
	const entry = this.resolveEntry();
	const textBuild = getTextBuildWorkerRuntimeState();
	const data: SearchWorkerData = {
	dir: this.dir,
	bootSalt: randomUUID(),
	textBuildWorkerPath:
	textBuild.configured && textBuild.entry.kind === 'packaged' ? textBuild.entry.path : undefined,
	};
	let worker: Worker;
	try {
	worker =
	this.options.workerFactory?.({
	url: entry.url,
	data,
	execArgv: entry.execArgv,
	}) ??
	new Worker(entry.url, {
	workerData: data,
	execArgv: entry.execArgv,
	resourceLimits: { maxOldGenerationSizeMb: this.options.maxOldSpaceMb ?? DEFAULT_MAX_OLD_SPACE_MB },
	name: 'kimi-search-worker',
	});
	} catch (error) {
	const message = error instanceof Error ? error.message : String(error);
	this.noteFailure(`spawn failed: ${message}`);
	throw new SearchWorkerError('spawn-failed', `search worker failed to start: ${message}`);
	}
	worker.unref();
	this.attach(worker);
	try {
	await this.awaitReady(worker);
	} catch (error) {
	this.detach();
	await worker.terminate().catch(() => {});
	const message = error instanceof Error ? error.message : String(error);
	this.noteFailure(`handshake failed: ${message}`);
	throw new SearchWorkerError('spawn-failed', `search worker handshake failed: ${message}`);
	}
	this.readyAt = Date.now();
	this.worker = worker;
	this.lastFailure = null;
	this.lastCoreLifecycle = null;
	if (this.exiting) {
	try {
	worker.postMessage({ v: SEARCH_WORKER_PROTOCOL_VERSION, type: 'beginClose' });
	} catch {
	}
	}
	this.log.info('global search: worker started', { dir: this.dir });
	}

	private attach(worker: Worker): void {
	worker.on('message', (event: SearchWorkerEvent) => this.onMessage(worker, event));
	worker.on('error', (error: Error) => this.onError(worker, error));
	worker.on('exit', (code: number) => this.onExit(worker, code));
	}

	private detach(): void {
	this.worker = null;
	this.exitResolve?.();
	this.exitResolve = null;
	}

	private onMessage(worker: Worker, event: SearchWorkerEvent): void {
	if (this.worker !== worker) return;
	if (event === null \|\| typeof event !== 'object') return;
	if (event.type === 'log') {
	this.log[event.level](event.message, event.meta);
	return;
	}
	if (event.type === 'lockToken') {
	this.lockToken = event.token;
	liveLockTokens.add(event.token);
	return;
	}
	if (event.type === 'ready') return;
	if (event.type === 'result') {
	this.settleRequest(event.id, null, event.result);
	return;
	}
	if (event.type === 'error') {
	const payload = event.error;
	const error =
	payload.reason !== undefined
	? new GlobalSearchError(payload.reason, payload.message)
	: new Error(payload.message);
	this.settleRequest(event.id, error, undefined);
	}
	}

	private onError(worker: Worker, error: Error): void {
	if (this.worker !== worker) return;
	this.lastFailure = `worker error: ${error.message}`;
	}

	private onExit(worker: Worker, code: number): void {
	if (this.worker !== worker) return;
	const wasExiting = this.exiting;
	const sessionMs = Date.now() - this.readyAt;
	const deadToken = this.lockToken;
	this.detach();
	this.lockToken = undefined;
	if (deadToken !== undefined) liveLockTokens.delete(deadToken);
	if (this.requests.size > 0) {
	const error = wasExiting
	? new SearchWorkerError('disposed', 'search worker is disposed')
	: new SearchWorkerError(
	'crashed',
	`search worker exited with in-flight requests (code ${code}${this.lastFailure !== null ? `; ${this.lastFailure}` : ''})`,
	);
	for (const pending of this.requests.values()) {
	if (pending.watchdog !== undefined) clearTimeout(pending.watchdog);
	pending.reject(error);
	}
	this.requests.clear();
	}
	if (wasExiting) return;
	this.lastFailure ??= `worker exited with code ${code}`;
	this.reapPromise = this.reapLockFile(deadToken).finally(() => {
	this.reapPromise = null;
	});
	this.failures = sessionMs > STABLE_SESSION_MS ? Math.max(1, this.failures - 1) : this.failures + 1;
	const backoff = Math.min(BACKOFF_BASE_MS * 2 ** (this.failures - 1), BACKOFF_CAP_MS);
	this.nextRetryAfter = Date.now() + backoff;
	this.log.warn('global search: worker exited unexpectedly; restart backed off', {
	code,
	backoffMs: backoff,
	lastFailure: this.lastFailure ?? undefined,
	});
	}

	private noteFailure(message: string): void {
	this.failures += 1;
	const backoff = Math.min(BACKOFF_BASE_MS * 2 ** (this.failures - 1), BACKOFF_CAP_MS);
	this.nextRetryAfter = Date.now() + backoff;
	this.lastFailure = message;
	}

	private async reapLockFile(token: string \| undefined): Promise<void> {
	if (token === undefined) return;
	const lockPath = join(this.dir, 'db.lock');
	try {
	const raw = await readFile(lockPath, 'utf8');
	const parsed = JSON.parse(raw) as { pid?: unknown; token?: unknown };
	if (parsed.pid === process.pid && parsed.token === token) {
	await rm(lockPath, { force: true });
	this.log.warn('global search: reaped the lock left by the dead worker', { dir: this.dir });
	}
	} catch {
	}
	}

	private awaitReady(worker: Worker): Promise<void> {
	return new Promise<void>((resolve, reject) => {
	const timer = setTimeout(() => {
	cleanup();
	reject(new Error(`ready handshake timed out after ${this.options.readyTimeoutMs ?? DEFAULT_READY_TIMEOUT_MS}ms`));
	}, this.options.readyTimeoutMs ?? DEFAULT_READY_TIMEOUT_MS);
	timer.unref?.();
	const onMessage = (event: SearchWorkerEvent): void => {
	if (event?.type !== 'ready') return;
	cleanup();
	if (event.v !== SEARCH_WORKER_PROTOCOL_VERSION) {
	reject(
	new Error(
	`protocol mismatch: host v${SEARCH_WORKER_PROTOCOL_VERSION}, worker v${event.v ?? 'unknown'}`,
	),
	);
	return;
	}
	resolve();
	};
	const onError = (error: Error): void => {
	cleanup();
	reject(error);
	};
	const onExit = (code: number): void => {
	cleanup();
	reject(new Error(`worker exited before ready (code ${code})`));
	};
	const cleanup = (): void => {
	clearTimeout(timer);
	worker.off('message', onMessage);
	worker.off('error', onError);
	worker.off('exit', onExit);
	};
	worker.on('message', onMessage);
	worker.on('error', onError);
	worker.on('exit', onExit);
	});
	}

	private waitForExit(): Promise<void> {
	if (this.worker === null) return Promise.resolve();
	return new Promise<void>((resolve) => {
	const previous = this.exitResolve;
	this.exitResolve = () => {
	previous?.();
	resolve();
	};
	});
	}

	beginClose(): void {
	this.exiting = true;
	try {
	this.worker?.postMessage({ v: SEARCH_WORKER_PROTOCOL_VERSION, type: 'beginClose' });
	} catch {
	}
	}

	private scheduleOrphanCheck(): void {
	if (this.orphanCheckScheduled \|\| this.exiting) return;
	this.orphanCheckScheduled = true;
	const timer = setTimeout(() => {
	this.orphanCheckScheduled = false;
	void this.recoverOrphanedLock();
	}, ORPHAN_LOCK_GRACE_MS);
	timer.unref?.();
	}

	private async recoverOrphanedLock(): Promise<void> {
	if (this.exiting \|\| this.worker === null) return;
	const lockPath = join(this.dir, 'db.lock');
	let token: string;
	try {
	const parsed = JSON.parse(await readFile(lockPath, 'utf8')) as {
	pid?: unknown;
	token?: unknown;
	};
	if (parsed.pid !== process.pid \|\| typeof parsed.token !== 'string') return;
	token = parsed.token;
	} catch {
	return;
	}
	if (liveLockTokens.has(token)) return;
	try {
	const again = JSON.parse(await readFile(lockPath, 'utf8')) as {
	pid?: unknown;
	token?: unknown;
	};
	if (again.pid !== process.pid \|\| again.token !== token) return;
	if (liveLockTokens.has(token)) return;
	} catch {
	return;
	}
	this.log.warn('global search: reaping an orphaned same-pid lock and restarting the worker', {
	dir: this.dir,
	});
	await rm(lockPath, { force: true }).catch(() => {});
	const worker = this.worker;
	if (worker === null) return;
	await worker.terminate().catch(() => {});
	}

	dispose(): Promise<void> {
	this.exiting = true;
	this.disposePromise ??= this.doDispose();
	return this.disposePromise;
	}

	private disposePromise: Promise<void> \| null = null;

	private async doDispose(): Promise<void> {
	await this.spawnPromise?.catch(() => {});
	const worker = this.worker;
	if (worker === null) return;
	const closeTimeoutMs = this.options.closeTimeoutMs ?? DEFAULT_CLOSE_TIMEOUT_MS;
	let timeout: ReturnType<typeof setTimeout> \| null = null;
	try {
	await Promise.race([
	this.sendRequest(worker, 'close'),
	new Promise<never>((_resolve, reject) => {
	timeout = setTimeout(() => {
	reject(new Error(`search worker close timed out after ${closeTimeoutMs}ms`));
	}, closeTimeoutMs);
	timeout.unref?.();
	}),
	]);
	} catch {
	} finally {
	if (timeout !== null) clearTimeout(timeout);
	}
	if (this.worker !== null) {
	await worker.terminate().catch(() => {});
	await this.waitForExit();
	}
	}
	}